[release] 1.0.1rc2: support xdist properly

src/buildkite_test_collector/collector/constants.py

@@ -3,4 +3,4 @@
 """This module defines collector-level constants."""
 
 COLLECTOR_NAME='buildkite-test-collector'
-VERSION='1.0.1rc1'
+VERSION='1.0.1rc2'

src/buildkite_test_collector/collector/payload.py

@@ -5,6 +5,8 @@
 from datetime import timedelta
 from uuid import UUID
 
+from ..pytest_plugin.logger import logger
+
 from .instant import Instant
 from .run_env import RuntimeEnvironment
 
@@ -215,6 +217,7 @@ class Payload:
     @classmethod
     def init(cls, run_env: RuntimeEnvironment) -> 'Payload':
         """Create a new instance of payload with the provided runtime environment"""
+
         return cls(
             run_env=run_env,
             data=(),
@@ -224,10 +227,13 @@ def init(cls, run_env: RuntimeEnvironment) -> 'Payload':
 
     def as_json(self) -> JsonDict:
         """Convert into a Dict suitable for eventual serialisation to JSON"""
-        finished_data = filter(
+        finished_data = list(filter(
             lambda td: td.is_finished(),
             self.data
-        )
+        ))
+
+        if len(finished_data) < len(self.data):
+            logger.warning('Unexpected unfinished test data, skipping unfinished test records...')
 
         return {
             "format": "json",

src/buildkite_test_collector/collector/run_env.py

@@ -77,7 +77,7 @@ def __circle_ci_env() -> Optional['RuntimeEnvironment']:
     )
 
 
-def __generic_env() -> Optional['RuntimeEnvironment']:
+def __generic_env() -> 'RuntimeEnvironment':
     return RuntimeEnvironment(
         ci="generic",
         key=str(uuid4()),
@@ -120,7 +120,7 @@ def as_json(self) -> Dict[str, str]:
         return {k: v for k, v in attrs.items() if v is not None}
 
 
-def detect_env() -> Optional['RuntimeEnvironment']:
+def detect_env() -> RuntimeEnvironment:
     """Attempt to detect the CI system we're running in"""
     return __buildkite_env() or \
         __github_actions_env() or \

src/buildkite_test_collector/pytest_plugin/__init__.py

@@ -37,15 +37,29 @@ def pytest_unconfigure(config):
     plugin = getattr(config, '_buildkite', None)
 
     if plugin:
-        # Only submit if this is not an xdist worker,
-        # this prevents duplicate payload submissions
-        # see https://github.com/pytest-dev/pytest-xdist/blob/fabdbe3fd2dbaf0e2764697ba4c79938d565dc44/src/xdist/plugin.py#L305
-        if not hasattr(config, "workerinput"):
+        xdist_enabled = (
+            config.pluginmanager.getplugin("xdist") is not None
+            and config.getoption("numprocesses") is not None
+        )
+        is_xdist_worker = hasattr(config, 'workerinput')
+
+        is_controller = not xdist_enabled or (xdist_enabled and not is_xdist_worker)
+
+        # When xdist is not installed, or when it's installed and not enabled
+        if not xdist_enabled:
+            submit(plugin.payload)
+
+        # When xdist is activated, we want to submit from worker thread only, because they have access to tag data
+        if xdist_enabled and is_xdist_worker:
             submit(plugin.payload)
 
-        jsonpath = config.option.jsonpath
-        if jsonpath:
-            plugin.save_payload_as_json(jsonpath)
+        # We only want a single thread to write to the json file.
+        # When xdist is enabled, that will be the controller thread.
+        if is_controller:
+            # Note that when xdist is used, this JSON output file will NOT contain tags.
+            jsonpath = config.option.jsonpath
+            if jsonpath:
+                plugin.save_payload_as_json(jsonpath)
 
         del config._buildkite
         config.pluginmanager.unregister(plugin)

src/buildkite_test_collector/pytest_plugin/buildkite_plugin.py

@@ -60,14 +60,19 @@ def pytest_runtest_logreport(self, report):
             # so we can get the correct result when we process it during the teardown hook.
             self.in_flight[nodeid] = test_data
 
-    def pytest_runtest_teardown(self, item):
+    # This hook only runs in xdist worker thread, not controller thread.
+    # We used to rely on pytest_runtest_teardown, but somehow xdist will ignore it
+    # in both controller and worker thread.
+    def pytest_runtest_makereport(self, item, call):
         """pytest_runtest_hook hook callback to mark test as finished and add it to the payload"""
-        logger.debug('Enter pytest_runtest_teardown for %s', item.nodeid)
+
+        if call.when != 'teardown':
+            return
+
+        logger.debug('Enter pytest_runtest_makereport for %s', item.nodeid)
         test_data = self.in_flight.get(item.nodeid)
 
         if test_data:
-            test_data = test_data.finish()
-
             tags = item.iter_markers("execution_tag")
             for tag in tags:
                 test_data = test_data.tag_execution(tag.args[0], tag.args[1])
@@ -76,27 +81,29 @@ def pytest_runtest_teardown(self, item):
 
             self.finalize_test(item.nodeid)
         else:
-            logger.warning('Unexpected missing test_data during pytest_runtest_teardown')
+            logger.warning('Unexpected missing test_data during pytest_runtest_makereport')
 
-    # Strictly speaking, we do not need this hook.
-    # But in pytest it's hard to predict how plugins interfere each other.
-    # So let's be defensive here.
+    # If pytest_runtest_makereport runs properly then this hook is unnecessary.
+    # But as we commented above, in some cases, in pytest-xdist controller thread,
+    # pytest_runtest_makereport will be skipped.
+    # In that case, it's necessary for this hook to work as a fallback mechanism.
     def pytest_runtest_logfinish(self, nodeid, location):  # pylint: disable=unused-argument
         """pytest_runtest_logfinish hook always runs in the very end"""
         logger.debug('Enter pytest_runtest_logfinish for %s', nodeid)
         if self.finalize_test(nodeid):
-            logger.warning(
-                'Detected possible interference in pytest_runtest_teardown hook. '
-                'Falling back to pytest_runtest_logfinish, but note that test tags '
-                'will not be uploaded.'
+            # This is expected to happen in xdist controller thread.
+            # Where it would skip many pytest_runtest_xxx hooks
+            logger.debug(
+                'Detected possible interference in pytest_runtest_makereport hook (xdist?). '
+                'Falling back to pytest_runtest_logfinish'
             )
 
-
     def finalize_test(self, nodeid):
         """ Attempting to move test data for a nodeid to payload area for upload """
         test_data = self.in_flight.get(nodeid)
         if test_data:
             del self.in_flight[nodeid]
+            test_data = test_data.finish()
             self.payload = self.payload.push_test_data(test_data)
             return True
         return False

src/buildkite_test_collector/pytest_plugin/logger.py

@@ -1,14 +1,11 @@
-"""A plugin internal logger, use DEBUG=1 env var to turn on all debug logs"""
+"""A plugin internal logger"""
 import os
 import logging
 
 def setup_logger(name=__name__):
     """
     Configure and return a logger with the specified name.
 
-    The logger's level is set based on the DEBUG environment variable.
-    If DEBUG=1, the level is set to DEBUG, otherwise it's set to INFO.
-
     Args:
         name (str): The name for the logger. Defaults to the current module name.
 
@@ -18,7 +15,8 @@ def setup_logger(name=__name__):
     l = logging.getLogger(name)
 
     # Set level based on DEBUG env var
-    l.setLevel(logging.DEBUG if os.getenv("DEBUG") == "1" else logging.INFO)
+    debug_enabled = os.getenv("BUILDKITE_ANALYTICS_DEBUG_ENABLED") == "1"
+    l.setLevel(logging.DEBUG if debug_enabled else logging.INFO)
 
     # Add handler only if none exists (prevents duplicate logs)
     if not l.handlers: