Add a final statistics report

Author: xverges

examples/test_load_example.py

@@ -7,67 +7,46 @@
 1. Weighted test selection (tests run with different frequencies)
 2. Shared state tracking across workers using shared_json_fixture_factory
 3. Conditional stopping based on shared state
+4. Automatic execution report at session end
 
 TEST_CODE:
 ```python
 result = pytester.runpytest('--load-test', '-n', '2', '-v')
 result.stdout.fnmatch_lines([
-    '*Interrupted: System health critical*',
+    '*Interrupted: Test session completed*',
 ])
 assert result.ret == pytest.ExitCode.INTERRUPTED
 ```
 """
 
-import logging
-import sys
-
 import pytest
 
 from pytest_load_testing import stop_load_testing, weight
 
-logger = logging.getLogger(__name__)
-
 
 @pytest.fixture(scope="session")
-def progress_tracker(request, shared_json_fixture_factory):
+def progress_tracker(shared_json_fixture_factory):
     """Track errors across all workers."""
-
-    def on_last_worker(shared):
-        """Called once by the last worker to finish."""
-        # Read the data that was collected during test execution
-        data = shared.read()
-
-        # Write to a file to prove callback executed
-        with open("/tmp/load_test_report.txt", "w") as f:
-            f.write("=" * 60 + "\n")
-            f.write("LOAD TEST REPORT\n")
-            f.write("=" * 60 + "\n")
-            f.write(f"Total test executions: {data.get('count', 0)}\n")
-            f.write("=" * 60 + "\n")
-
-        # Also write to stderr
-        report = []
-        report.append("\n" + "=" * 60)
-        report.append("LOAD TEST REPORT")
-        report.append("=" * 60)
-        report.append(f"\nTotal test executions: {data.get('count', 0)}")
-        report.append("=" * 60 + "\n")
-        sys.stderr.write("\n".join(report))
-        sys.stderr.flush()
-
-    return shared_json_fixture_factory(name="counter", on_first_worker={"count": 0}, on_last_worker=on_last_worker)
+    return shared_json_fixture_factory(name="progress", on_first_worker={"count": 0, "results_by_node": {}})
 
 
 @pytest.fixture
 def iteration_counter(request, progress_tracker):
     """Common fixture that tracks iterations and stops after threshold."""
+    node_id = request.node.nodeid
+
     with progress_tracker.locked_dict() as data:
         data["count"] = data.get("count", 0) + 1
         current_count = data["count"]
 
-    # Stop after 20 total test executions across all workers
+        # Track results by node id
+        results_by_node = data.get("results_by_node", {})
+        results_by_node[node_id] = results_by_node.get(node_id, 0) + 1
+        data["results_by_node"] = results_by_node
+
+    # Stop after 100 total test executions across all workers
     if current_count >= 100:
-        stop_load_testing(request, "System health critical")
+        stop_load_testing(request, "Test session completed")
 
 
 @weight(70)

src/pytest_load_testing/plugin.py

@@ -1,4 +1,5 @@
 """pytest-xdist-load-testing plugin implementation."""
+
 import logging
 
 import pytest
@@ -73,7 +74,6 @@ def pytest_runtest_setup(self, item: pytest.Item):
 
         # If item has been run before, we need to clear fixtures for re-execution
         if has_been_run and request_obj is not None:  # type: ignore[attr-defined]
-
             self._cleanup_previous_run(item)
 
         # Get ready for next time
@@ -92,7 +92,6 @@ def _save_weight_for_controller(self, item: pytest.Item) -> None:
             item.user_properties.append((LoadTestPlugin.WEIGHT_PROPERTY_KEY, weight_value))
 
     def _cleanup_previous_run(self, item: pytest.Item) -> None:
-
         setupstate = item.session._setupstate
 
         # If item is in stack, run its teardown functions first
@@ -164,7 +163,7 @@ def pytest_runtest_logreport(self, report: pytest.TestReport):
             return
 
         # Check for stop signal and weight in user_properties
-        if hasattr(report, 'user_properties'):
+        if hasattr(report, "user_properties"):
             for key, value in report.user_properties:
                 if key == LOAD_TEST_STOP_SIGNAL and isinstance(value, str):
                     scheduler.stop(value)
@@ -194,6 +193,13 @@ def pytest_sessionstart(self, session: pytest.Session):
 
     def pytest_sessionfinish(self, session: pytest.Session, exitstatus: int):
         """Hook called after whole test run finished, right before returning exit status."""
+        # Print load test statistics if enabled
+        if self.is_enabled(session.config):
+            scheduler = session.config.stash.get(stash_key_scheduler, None)
+            if scheduler:
+                # Use stderr=False (logger) by default, can be changed to True for stderr output
+                scheduler.print_final_statistics(use_stderr=True)
+
         if stash_key_session in session.config.stash:
             del session.config.stash[stash_key_session]
 
@@ -206,8 +212,7 @@ def pytest_configure(config: pytest.Config):
     """
     # Register custom markers
     config.addinivalue_line(
-        "markers",
-        "weight(value): Set the weight of a test for load testing (higher = more likely to be selected)"
+        "markers", "weight(value): Set the weight of a test for load testing (higher = more likely to be selected)"
     )
 
     # Only register plugin instance if load testing is enabled
@@ -218,11 +223,11 @@ def pytest_configure(config: pytest.Config):
 
 def pytest_addoption(parser):
     """Add plugin-specific command line options."""
-    group = parser.getgroup('xdist-load-testing')
+    group = parser.getgroup("xdist-load-testing")
     group.addoption(
-        '--load-test',
-        action='store_true',
+        "--load-test",
+        action="store_true",
         dest=LoadTestPlugin.PYTEST_OPTION_NAME,
         default=False,
-        help='Enable load testing mode with continuous test execution.'
+        help="Enable load testing mode with continuous test execution.",
     )

src/pytest_load_testing/scheduler.py

@@ -1,7 +1,10 @@
 """Load testing scheduler implementation."""
+
 import logging
 import random
+import sys
 import time
+from dataclasses import dataclass
 from typing import Any, Optional
 
 import pytest
@@ -13,6 +16,21 @@
 logger = logging.getLogger(__name__)
 
 
+@dataclass
+class TestStatistic:
+    """Statistics for a single test's execution."""
+
+    nodeid: str
+    total: int
+    passes: int
+    failures: int
+
+    @property
+    def failure_rate(self) -> float:
+        """Calculate the failure rate for this test."""
+        return self.failures / self.total if self.total > 0 else 0.0
+
+
 class LoadTestScheduler(LoadScheduling):
     """
     Custom scheduler for pytest-xdist that continuously supplies tests to workers.
@@ -31,6 +49,7 @@ class LogProducer:
         Compatible with xdist's Producer interface but routes messages through
         Python's logging system.
         """
+
         def __init__(self, name: str, *, enabled: bool = True) -> None:
             self.name = name
             self.enabled = enabled
@@ -168,11 +187,7 @@ def _send_weighted_tests(self, node: WorkerController, num: int) -> None:
         # Select 'num' tests randomly with weights
         # Tests with weight 0 will never be selected
         try:
-            selected_indices = random.choices(
-                range(len(self.collection)),
-                weights=self.weights,
-                k=num
-            )
+            selected_indices = random.choices(range(len(self.collection)), weights=self.weights, k=num)
         except ValueError:
             # This shouldn't happen if we checked for all-zero weights above
             self.log("Error selecting tests with weights, stopping scheduler")
@@ -215,9 +230,7 @@ def schedule(self) -> None:
         for node in self.nodes:
             self.check_schedule(node)
 
-    def mark_test_complete(
-        self, node: WorkerController, item_index: int, duration: float = 0
-    ) -> None:
+    def mark_test_complete(self, node: WorkerController, item_index: int, duration: float = 0) -> None:
         """Called when a test completes - schedule more work."""
         # Call parent, which will call check_schedule
         super().mark_test_complete(node, item_index, duration)
@@ -286,6 +299,48 @@ def remove_node(self, node: WorkerController) -> Optional[str]:
         if pending:
             # Return the first item that was running
             return self.collection[pending[0]] if self.collection else None
+
+    def _calculate_test_statistics(self) -> list[TestStatistic]:
+        """Calculate execution statistics for all tests."""
+        all_tests = set(self.test_passes.keys()) | set(self.test_failures.keys())
+        stats = []
+
+        for nodeid in all_tests:
+            passes = self.test_passes.get(nodeid, 0)
+            failures = self.test_failures.get(nodeid, 0)
+            stats.append(TestStatistic(nodeid, passes + failures, passes, failures))
+
+        return sorted(stats, key=lambda x: x.total, reverse=True)
+
+    def _format_statistics_report(self, stats: list[TestStatistic]) -> list[str]:
+        """Format statistics into report lines."""
+        SEPARATOR_WIDTH = 70
+        total_executions = sum(s.total for s in stats)
+
+        lines = [
+            "=" * SEPARATOR_WIDTH,
+            "Load Test Execution Report",
+            "=" * SEPARATOR_WIDTH,
+        ]
+
+        for stat in stats:
+            status = f"✓ {stat.passes} ✗ {stat.failures}"
+            lines.append(f"{stat.nodeid}: {stat.total} executions ({status})")
+
+        lines.append(f"Total executions: {total_executions}")
+        lines.append("=" * SEPARATOR_WIDTH)
+
+        return lines
+
+    def _output_report(self, lines: list[str], use_stderr: bool = False) -> None:
+        """Output report lines to stderr or logger."""
+        if use_stderr:
+            sys.stderr.write("\n" + "\n".join(lines) + "\n")
+            sys.stderr.flush()
+        else:
+            for line in lines:
+                logger.info(line)
+
         return None
 
     def stop(self, reason: str = "Interrupted"):
@@ -302,3 +357,16 @@ def stop(self, reason: str = "Interrupted"):
         for node in self.nodes:
             if not node.shutting_down:
                 node.shutdown()
+
+    def print_final_statistics(self, use_stderr: bool = False) -> None:
+        """Print test execution statistics at the end of the session.
+
+        Args:
+            use_stderr: If True, write to stderr. If False, use logger.info (default).
+        """
+        if not self.test_passes and not self.test_failures:
+            return
+
+        stats = self._calculate_test_statistics()
+        lines = self._format_statistics_report(stats)
+        self._output_report(lines, use_stderr)