fix for "Aborting test can fail when running distributed with MQ" (#333)

* reformat behave exception error messages

* rewrite of async-messaged worker

still not quite done with forcing the process to stop when receiving sigint.

* fix for issue #323

start `router` in seperate process, which in turn runs a `ThreadPoolExecutor`, where each worker is a seperate thread.

install the signal handler and share a `threading.Event` with all processes/threads, that if set would abort their respective I/O loops. the signal handler will set the event in case of SIGTERM.

in the main loop, wait for the event to be set, then terminate the process with the `ThreadPoolExecutor`.

the router/frontend will check which threads has stopped by their own, and the others it will send a "abort" message to their clients.

the abort message will result in `AsyncMessageAbort` exception, which is handled by users and client tasks in the form of a `StopUser`.

* py3.12 compatible rmtree wrapper

* fixed unit tests

* bump static python version from 3.11 to 3.12 in workflows

* fix fixed unit tests

* refactoring so that each scenario has their own jinja2 Environment

this is done in preparation of allow same variable names in all scenarios, but them still beeing separated (except for Atomic* variables).

first step though will be the possibility to reuse request templates between scenarios, which will be needed when scenarios includes steps from other scenarios `{% scenario ... %}`.

.devcontainer/devcontainer.json

@@ -24,6 +24,7 @@
                     "*.j2.json": "jinja-json",
                     "*.j2.xml": "jinja-xml"
                 },
+                "mypy-type-checker.ignorePatterns": ["tests/project/**/*.py"],
                 "mypy.targets": [
                     "grizzly/",
                     "grizzly_extras/",

.github/workflows/code-quality.yaml

@@ -21,7 +21,7 @@ jobs:
       id: setup-python
       uses: actions/setup-python@v5
       with:
-        python-version: '3.11'
+        python-version: '3.12'
         cache: 'pip'
 
     - name: install python dependencies
@@ -52,7 +52,7 @@ jobs:
         python-version: ['3.9', '3.10', '3.11', '3.12']
         runs-on: ['ubuntu-latest']
         include:
-          - python-version: '3.11'
+          - python-version: '3.12'
             runs-on: windows-latest
 
     env:
@@ -118,7 +118,7 @@ jobs:
       id: setup-python
       uses: actions/setup-python@v5
       with:
-        python-version: '3.11'
+        python-version: '3.12'
         cache: 'pip'
 
     - name: setup environment
@@ -195,7 +195,7 @@ jobs:
       id: setup-python
       uses: actions/setup-python@v5
       with:
-        python-version: '3.11'
+        python-version: '3.12'
         cache: 'pip'
 
     - name: install python dependencies
@@ -225,7 +225,7 @@ jobs:
       id: setup-python
       uses: actions/setup-python@v5
       with:
-        python-version: '3.11'
+        python-version: '3.12'
         cache: 'pip'
 
     - name: setup node

.github/workflows/release.yaml

@@ -42,7 +42,7 @@ jobs:
       id: setup-python
       uses: actions/setup-python@v5
       with:
-        python-version: '3.11'
+        python-version: '3.12'
         cache: 'pip'
 
     - name: setup node

.gitignore

@@ -30,3 +30,4 @@ docs/_site
 docs/_build*
 grizzly/__version__.py
 .pytest_tmp/
+tests/project

grizzly/auth/__init__.py

@@ -197,7 +197,7 @@ def refresh_token(client: GrizzlyHttpAuthClient, *args: P.args, **kwargs: P.kwar
 
 def render(client: GrizzlyHttpAuthClient) -> None:
     variables = cast(dict[str, Any], client._context.get('variables', {}))
-    host = client.grizzly.state.jinja2.from_string(client.host).render(**variables)
+    host = client._scenario.jinja2.from_string(client.host).render(**variables)
     parsed = urlparse(host)
 
     client.host = f'{parsed.scheme}://{parsed.netloc}'

grizzly/behave.py

@@ -154,7 +154,7 @@ def after_feature(context: Context, feature: Feature, *_args: Any, **_kwargs: An
                 buffer.append(f'Scenario: {scenario_name}')
             else:
                 buffer.append('')
-            buffer.extend([indent(str(exception), '    ') for exception in exceptions])
+            buffer.extend([str(exception).replace('\n', '\n    ') for exception in exceptions])
             buffer.append('')
 
         failure_summary = indent('\n'.join(buffer), '    ')

grizzly/context.py

@@ -8,7 +8,7 @@
 from typing import TYPE_CHECKING, Any, Callable, Optional, Union, cast
 
 import yaml
-from jinja2 import Environment
+from jinja2 import Environment, FileSystemLoader
 from jinja2.filters import FILTERS
 
 from grizzly.types import MessageCallback, MessageDirection
@@ -86,9 +86,9 @@ def destroy(cls) -> None:
 
     def __init__(self) -> None:
         if not self._initialized:
-            self._state = GrizzlyContextState()
             self._setup = GrizzlyContextSetup()
             self._scenarios = GrizzlyContextScenarios()
+            self._state = GrizzlyContextState(self._scenarios)
             self._initialized = True
 
     @property
@@ -114,28 +114,26 @@ def scenarios(self) -> GrizzlyContextScenarios:
 
 
 def jinja2_environment_factory() -> Environment:
-    """Create a Jinja2 environment, so same instance is used throughout grizzly, with custom filters."""
-    return Environment(autoescape=False)
+    """Create a Jinja2 environment, so same instance is used throughout each grizzly scenario, with custom filters."""
+    return Environment(autoescape=False, loader=FileSystemLoader(Path(environ['GRIZZLY_CONTEXT_ROOT']) / 'requests'))
 
 
 @dataclass
 class GrizzlyContextState:
+    _scenarios: GrizzlyContextScenarios = field(init=True)
     spawning_complete: bool = field(default=False)
     background_section_done: bool = field(default=False)
-    variables: GrizzlyVariables = field(init=False, default_factory=GrizzlyVariables)
+    variables: GrizzlyVariables = field(init=False)
     configuration: dict[str, Any] = field(init=False, default_factory=load_configuration_file)
     alias: dict[str, str] = field(init=False, default_factory=dict)
     verbose: bool = field(default=False)
     locust: Union[MasterRunner, WorkerRunner, LocalRunner] = field(init=False, repr=False)
     persistent: dict[str, str] = field(init=False, repr=False, default_factory=dict)
-    _jinja2: Environment = field(init=False, repr=False, default_factory=jinja2_environment_factory)
 
-    @property
-    def jinja2(self) -> Environment:
-        # something might have changed in the filters department
-        self._jinja2.filters = FILTERS
-
-        return self._jinja2
+    def __post_init__(self) -> None:
+        """Workaround until grizzly.state.variables has been deprecated."""
+        self.variables = GrizzlyVariables(scenarios=self._scenarios)
+        del self._scenarios
 
 
 @dataclass
@@ -276,6 +274,14 @@ class GrizzlyContextScenario:
     validation: GrizzlyContextScenarioValidation = field(init=False, hash=False, compare=False, default_factory=GrizzlyContextScenarioValidation)
     failure_exception: Optional[type[Exception]] = field(init=False, default=None)
     orphan_templates: list[str] = field(init=False, repr=False, hash=False, compare=False, default_factory=list)
+    _jinja2: Environment = field(init=False, repr=False, default_factory=jinja2_environment_factory)
+
+    @property
+    def jinja2(self) -> Environment:
+        # something might have changed in the filters department
+        self._jinja2.filters = FILTERS
+
+        return self._jinja2
 
     def __post_init__(self) -> None:
         self.name = self.behave.name
@@ -372,4 +378,8 @@ def create(self, behave_scenario: Scenario) -> None:
         """Create a new scenario based on the behave Scenario, and add it to the current list of scenarios in this context."""
         grizzly_scenario = GrizzlyContextScenario(len(self) + 1, behave=behave_scenario)
 
+        # inherit jinja2.globals from previous scenario
+        if len(self) > 0:
+            grizzly_scenario.jinja2.globals.update({**self[-1].jinja2.globals})
+
         self.append(grizzly_scenario)

grizzly/events/request_logger.py

@@ -9,8 +9,6 @@
 from typing import TYPE_CHECKING, Any, Optional
 from urllib.parse import urlparse, urlunparse
 
-from jinja2 import Template
-
 from grizzly.events import GrizzlyEventHandler
 from grizzly.utils import normalize
 from grizzly_extras.transformer import JsonBytesEncoder
@@ -182,7 +180,7 @@ def __call__(
         name = normalize(name)
 
         log_name = f'{name}.{log_date.strftime("%Y%m%dT%H%M%S%f")}.log'
-        contents = Template(LOG_FILE_TEMPLATE).render(**variables)
+        contents = self.user._scenario.jinja2.from_string(LOG_FILE_TEMPLATE).render(**variables)
 
         log_file = self.log_dir / log_name
         log_file.write_text(contents)

grizzly/events/response_handler.py

@@ -54,7 +54,7 @@ def get_match(
 
         """
         input_content_type, input_payload = input_context
-        j2env = self.grizzly.state.jinja2
+        j2env = user._scenario.jinja2
         rendered_expression = j2env.from_string(self.expression).render(user.context_variables)
         rendered_match_with = j2env.from_string(self.match_with).render(user.context_variables)
         rendered_expected_matches = int(j2env.from_string(self.expected_matches).render(user.context_variables))

grizzly/exceptions.py

@@ -13,6 +13,7 @@
 
 __all__ = [
     'StopUser',
+    'AsyncMessageAbort',
 ]
 
 
@@ -27,7 +28,7 @@ class RestartScenario(Exception):  # noqa: N818
     pass
 
 
-class StopScenario(AsyncMessageAbort):
+class StopScenario(Exception):  # noqa: N818
     pass
 
 class AssertionErrors(Exception):  # noqa: N818
@@ -64,7 +65,7 @@ def __init__(self, error: AssertionError | str, step: Step) -> None:
         self.error = error
 
     def __str__(self) -> str:
-        return '\n'.join([f'{self.step.keyword} {self.step.name} # {self.step.location}', f'    ! {self.error!s}'])
+        return '\n'.join([f'    {self.step.keyword} {self.step.name} # {self.step.location}', f'    ! {self.error!s}'])
 
 
 class ScenarioError(AssertionError):
@@ -73,12 +74,12 @@ def __init__(self, error: AssertionError, scenario: Scenario) -> None:
         self.error = error
 
     def __str__(self) -> str:
-        return f'    ! {self.error!s}'
+        return f'! {self.error!s}'
 
 
 class FeatureError(Exception):
     def __init__(self, error: Exception) -> None:
         self.error = error
 
     def __str__(self) -> str:
-        return f'    {self.error!s}'
+        return f'{self.error!s}'

grizzly/locust.py

@@ -871,13 +871,9 @@ def shutdown_external_processes(processes: dict[str, subprocess.Popen], greenlet
         else:
             process.terminate()
 
-        if not abort_test:
-            process.wait()
-        else:
-            process.kill()
-            process.returncode = 1
+        process.wait()
 
-        logger.debug('process.returncode=%d', process.returncode)
+        logger.debug('%s: process.returncode=%d', dependency, process.returncode)
 
     processes.clear()
 
@@ -1017,10 +1013,14 @@ def start_watching_external_processes(processes: dict[str, subprocess.Popen]) ->
                 def watch_running_external_processes() -> None:
                     while runner.user_count > 0:
                         _processes = processes.copy()
+                        if len(_processes) < 1:
+                            logger.error('no running processes')
+                            break
+
                         for dependency, process in _processes.items():
                             if process.poll() is not None:
-                                logger.error('%s is not running, restarting', dependency)
-                                raise SystemExit(1)
+                                logger.error('%s is not running, stop', dependency)
+                                del processes[dependency]
 
                         gevent.sleep(10.0)
 
@@ -1199,7 +1199,12 @@ def wrapper() -> None:
                     return
 
                 logger.info('handling signal %s (%d)', signame, signum)
+
+                logger.debug('shutdown external processes')
+
                 abort_test = True
+                shutdown_external_processes(external_processes, watch_running_external_processes_greenlet)
+
                 if isinstance(runner, WorkerRunner):
                     runner._send_stats()
                     runner.client.send(Message('client_aborted', None, runner.client_id))
@@ -1228,7 +1233,8 @@ def wrapper() -> None:
 
         return code
     finally:
-        shutdown_external_processes(external_processes, watch_running_external_processes_greenlet)
+        if not abort_test:
+            shutdown_external_processes(external_processes, watch_running_external_processes_greenlet)
 
 
 def _grizzly_sort_stats(stats: lstats.RequestStats) -> list[tuple[str, str, int]]:

grizzly/scenarios/__init__.py

@@ -70,7 +70,7 @@ def render(self, template: str, variables: Optional[dict[str, Any]] = None, *, e
         else:
             render_variables = {**self.user._context['variables'], **variables}
 
-        return self.grizzly.state.jinja2.from_string(template).render(**render_variables)
+        return self.user._scenario.jinja2.from_string(template).render(**render_variables)
 
     def prefetch(self) -> None:
         """Do not prefetch anything by default."""

grizzly/steps/_helpers.py

@@ -36,29 +36,16 @@ def create_request_task(
     substitutes: Optional[dict[str, str]] = None,
     content_type: Optional[TransformerContentType] = None,
 ) -> RequestTask:
-    return _create_request_task(context.config.base_dir, method, source, endpoint, name, substitutes=substitutes, content_type=content_type)
-
-
-def _create_request_task(
-    base_dir: str,
-    method: RequestMethod,
-    source: Optional[str],
-    endpoint: str,
-    name: Optional[str] = None,
-    substitutes: Optional[dict[str, str]] = None,
-    content_type: Optional[TransformerContentType] = None,
-) -> RequestTask:
-    path = Path(base_dir) / 'requests'
-    j2env = j2.Environment(
-        autoescape=False,
-        loader=j2.FileSystemLoader(path),
-    )
+    grizzly = cast(GrizzlyContext, context.grizzly)
+    path = Path(context.config.base_dir) / 'requests'
 
     template: Optional[j2.Template] = None
 
     if source is not None:
         original_source = source
 
+        print(f'{source=}')
+
         try:
             possible_file = path / source
             if possible_file.is_file():
@@ -71,7 +58,8 @@ def _create_request_task(
                         fd.seek(0)
                         source = fd.read()
 
-                template = j2env.get_template(source)
+                if name is None:
+                    name, _ = possible_file.name.split('.', 1)
         except (j2.exceptions.TemplateNotFound, OSError) as e:
             # `TemplateNotFound` inherits `OSError`...
             if not isinstance(e, j2.exceptions.TemplateNotFound) and e.errno != ENAMETOOLONG:
@@ -83,6 +71,8 @@ def _create_request_task(
         for key, value in (substitutes or {}).items():
             source = source.replace(f'{{{{ {key} }}}}', value)
 
+        template = grizzly.scenario.jinja2.from_string(source)
+
     if name is None:
         name = '<unknown>'