fix: step tree incorrect when spawn thread nestly #872

allure-pytest/src/listener.py

@@ -30,7 +30,6 @@
 
 
 class AllureListener:
-
     SUITE_LABELS = {
         LabelType.PARENT_SUITE,
         LabelType.SUITE,
@@ -57,6 +56,10 @@ def stop_step(self, uuid, exc_type, exc_val, exc_tb):
                                      status=get_status(exc_val),
                                      statusDetails=get_status_details(exc_type, exc_val, exc_tb))
 
+    @allure_commons.hookimpl
+    def init_thread(self, source_thread, parent_uuid):
+        self.allure_logger.init_thread(source_thread, parent_uuid)
+
     @allure_commons.hookimpl
     def start_fixture(self, parent_uuid, uuid, name):
         after_fixture = TestAfterResult(name=name, start=now())

allure-python-commons/src/allure_commons/_allure.py

@@ -1,3 +1,4 @@
+import threading
 from functools import wraps
 from typing import Any, Callable, TypeVar, Union, overload
 
@@ -15,6 +16,7 @@ def safely(result):
     else:
         def dummy(function):
             return function
+
         return dummy
 
 
@@ -187,11 +189,22 @@ def __init__(self, title, params):
 
     def __enter__(self):
         plugin_manager.hook.start_step(uuid=self.uuid, title=self.title, params=self.params)
+        return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         plugin_manager.hook.stop_step(uuid=self.uuid, title=self.title, exc_type=exc_type, exc_val=exc_val,
                                       exc_tb=exc_tb)
 
+    def get_thread_initializer(self):
+        # save states of current thread, which will be used in spawned threads.
+        parent_uuid = self.uuid
+        current_thread = threading.current_thread()
+
+        def initializer():
+            plugin_manager.hook.init_thread(source_thread=current_thread, parent_uuid=parent_uuid)
+
+        return initializer
+
     def __call__(self, func: _TFunc) -> _TFunc:
         @wraps(func)
         def impl(*a, **kw):

allure-python-commons/src/allure_commons/_hooks.py

@@ -58,6 +58,10 @@ def start_step(self, uuid, title, params):
     def stop_step(self, uuid, exc_type, exc_val, exc_tb):
         """ step """
 
+    @hookspec
+    def init_thread(self, source_thread, parent_uuid):
+        """ init for step which executed in new spawned thread """
+
     @hookspec
     def attach_data(self, body, name, attachment_type, extension):
         """ attach data """

allure-python-commons/src/allure_commons/reporter.py

@@ -10,7 +10,6 @@
 
 
 class ThreadContextItems:
-
     _thread_context = defaultdict(OrderedDict)
     _init_thread: threading.Thread
 
@@ -52,6 +51,9 @@ def cleanup(self):
         for thread in stopped_threads:
             del self._thread_context[thread]
 
+    def init_thread(self, source_thread, parent_uuid):
+        self.thread_context[parent_uuid] = self._thread_context[source_thread][parent_uuid]
+
 
 class AllureReporter:
     def __init__(self):
@@ -139,6 +141,9 @@ def stop_step(self, uuid, **kwargs):
             self._update_item(uuid, **kwargs)
             self._items.pop(uuid)
 
+    def init_thread(self, source_thread, parent_uuid):
+        self._items.init_thread(source_thread, parent_uuid)
+
     def _attach(self, uuid, name=None, attachment_type=None, extension=None, parent_uuid=None):
         mime_type = attachment_type
         extension = extension if extension else 'attach'

tests/allure_pytest/acceptance/step/test_step_with_several_step_inside_thread.py

@@ -2,7 +2,7 @@
 from tests.allure_pytest.pytest_runner import AllurePytestRunner
 
 from allure_commons_test.report import has_test_case
-from allure_commons_test.result import has_step
+from allure_commons_test.result import has_step, not_
 
 
 def test_step_with_thread(allure_pytest_runner: AllurePytestRunner):
@@ -84,3 +84,57 @@ def test_step_with_reused_threads(allure_pytest_runner: AllurePytestRunner):
             )
         )
     )
+
+
+def test_step_within_thread_nested(allure_pytest_runner: AllurePytestRunner):
+    """
+    >>> from concurrent.futures import ThreadPoolExecutor
+    ... import allure
+    ... import time
+    ...
+    ...
+    ... def session_on_node(node_id, session):
+    ...     with allure.step(f"Session#{session} on Node#{node_id}"):
+    ...         pass
+    ...
+    ...
+    ... def parallel_task_for_specific_node(node_id):
+    ...     with allure.step(f"Node#{node_id}") as step:
+    ...         with ThreadPoolExecutor(initializer=step.get_thread_initializer()) as executor:
+    ...             for session in range(100):
+    ...                 executor.submit(session_on_node, node_id, session)
+    ...
+    ...             executor.shutdown(wait=True)
+    ...
+    ...
+    ... def test_multithreaded():
+    ...     with allure.step("Root") as step:
+    ...         with ThreadPoolExecutor(initializer=step.get_thread_initializer()) as executor:
+    ...             for node_id in range(2):
+    ...                 executor.submit(parallel_task_for_specific_node, node_id)
+    ...
+    ...             executor.shutdown(wait=True)
+    ...
+    """
+
+    allure_results = allure_pytest_runner.run_docstring()
+
+    assert_that(
+        allure_results,
+        has_test_case(
+            "test_multithreaded",
+            has_step(
+                "Root",
+                has_step(
+                    "Node#0",
+                    *[has_step(f"Session#{s} on Node#0") for s in range(100)],
+                    *[not_(has_step(f"Session#{s} on Node#1")) for s in range(100)],
+                ),
+                has_step(
+                    "Node#1",
+                    *[has_step(f"Session#{s} on Node#1") for s in range(100)],
+                    *[not_(has_step(f"Session#{s} on Node#0")) for s in range(100)],
+                ),
+            ),
+        )
+    )