From 171b4795d1ab1fbfe964afc6508a572e3534d3d6 Mon Sep 17 00:00:00 2001 From: danceratopz Date: Tue, 1 Sep 2026 10:07:47 +0200 Subject: [PATCH] fix(consume): detect missing enginex hive results Track executed, reported, and pre-Hive skipped test IDs per worker, then verify the aggregate after the session-wide EngineX suite finishes. Skip the invariant for distributed early stops and fail both undercounts and overcounts. Leave non-idempotent Hive requests single-shot so ambiguous response failures cannot duplicate them; ethereum-hive owns safe connection-establishment retries. Add focused accounting and real xdist early-stop coverage, and make the pytest_hive directories regular packages to avoid test import collisions. --- .../consume/simulators/enginex/conftest.py | 36 +- .../consume/tests/test_enginex_args.py | 21 + .../plugins/pytest_hive/__init__.py | 1 + .../plugins/pytest_hive/pytest_hive.py | 45 ++- .../plugins/pytest_hive/reporting.py | 203 ++++++++++ .../plugins/pytest_hive/tests/__init__.py | 1 + .../pytest_hive/tests/test_reporting.py | 376 ++++++++++++++++++ 7 files changed, 664 insertions(+), 19 deletions(-) create mode 100644 packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/__init__.py create mode 100644 packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/reporting.py create mode 100644 packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/__init__.py create mode 100644 packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/test_reporting.py diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/enginex/conftest.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/enginex/conftest.py index f0275269e97..6e28301c21f 100644 --- a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/enginex/conftest.py +++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/enginex/conftest.py @@ -11,16 +11,18 @@ import json import logging import time +from pathlib import Path from typing import TYPE_CHECKING, Generator, cast import pytest from hive.client import Client, ClientType -from hive.testing import HiveTest +from hive.testing import HiveTest, HiveTestSuite from execution_testing.fixtures import BlockchainEngineXFixture from execution_testing.fixtures.blockchain import FixtureHeader from execution_testing.fixtures.pre_alloc_groups import PreAllocGroup +from ....pytest_hive.reporting import write_reported_test_count from ..helpers.test_tracker import ( PreAllocGroupTestTracker, enginex_group_counts_key, @@ -48,6 +50,13 @@ def pytest_configure(config: pytest.Config) -> None: """Set the supported fixture formats for the enginex simulator.""" config.supported_fixture_formats = [BlockchainEngineXFixture] # type: ignore[attr-defined] + # Detect silent test loss: after the run, assert that every collected + # test reported its result to hive (see `_verify_reported_test_count` + # in the pytest_hive plugin). + config.assert_reported_test_count = True # type: ignore[attr-defined] + # The count check covers the whole pytest session, so keep one shared hive + # suite alive until every module and xdist worker has finished. + config.test_suite_scope = "session" # type: ignore[attr-defined] @pytest.hookimpl(trylast=True) @@ -155,13 +164,13 @@ def _configure_client_manager( multi_test_client_manager.set_test_tracker(pre_alloc_group_test_tracker) -@pytest.fixture(scope="module") +@pytest.fixture(scope="session") def test_suite_name() -> str: """The name of the hive test suite used in this simulator.""" return "eels/consume-enginex" -@pytest.fixture(scope="module") +@pytest.fixture(scope="session") def test_suite_description() -> str: """The description of the hive test suite used in this simulator.""" return ( @@ -190,6 +199,27 @@ def _per_test_reporting( hive_test.register_multi_test_client(client) +@pytest.fixture(scope="session", autouse=True) +def _reported_test_count_flush( + test_suite: HiveTestSuite, + session_temp_folder: Path, + request: pytest.FixtureRequest, +) -> Generator[None, None, None]: + """ + Persist this worker's reported-test counts before suite teardown. + + Depending on `test_suite` guarantees this fixture's teardown runs + before the suite teardown, in which the last-finishing xdist worker + aggregates all workers' counts and asserts that no test results + were silently lost (never started or never reported to hive). + """ + del test_suite + yield + write_reported_test_count( + request.config, request.session, session_temp_folder + ) + + @pytest.fixture(scope="function") def client( multi_test_hive_test: HiveTest, diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/tests/test_enginex_args.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/tests/test_enginex_args.py index 9090e4ffaa3..800a777ff20 100644 --- a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/tests/test_enginex_args.py +++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/tests/test_enginex_args.py @@ -1,12 +1,33 @@ """Tests for EngineX consume argument processing.""" +from types import SimpleNamespace + import pytest +from execution_testing.cli.pytest_commands.plugins.consume.simulators.enginex.conftest import ( # noqa: E501 + pytest_configure as configure_enginex, +) +from execution_testing.cli.pytest_commands.plugins.consume.simulators.enginex.conftest import ( # noqa: E501 + test_suite_description, + test_suite_name, +) from execution_testing.cli.pytest_commands.processors import ( HiveEnvironmentProcessor, ) +def test_enginex_uses_session_scoped_hive_suite() -> None: + """The session-wide result check must outlive every test module.""" + config = SimpleNamespace() + + configure_enginex(config) # type: ignore[arg-type] + + assert config.assert_reported_test_count is True + assert config.test_suite_scope == "session" + assert test_suite_name._fixture_function_marker.scope == "session" + assert test_suite_description._fixture_function_marker.scope == "session" + + @pytest.mark.parametrize( "parallelism_args", [ diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/__init__.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/__init__.py new file mode 100644 index 00000000000..2b1293ccbf9 --- /dev/null +++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/__init__.py @@ -0,0 +1 @@ +"""Hive pytest plugin providing common functionality for simulators.""" diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/pytest_hive.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/pytest_hive.py index fbe2ece3766..2bba0201ddd 100644 --- a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/pytest_hive.py +++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/pytest_hive.py @@ -51,6 +51,14 @@ from ..shared.helpers import is_help_or_collectonly_mode from .hive_info import ClientFile, HiveInfo +from .reporting import ( + HiveReportedTestCountError, + count_skipped_test, + end_test_and_count, + mark_hive_test_started, + record_test_execution, + verify_reported_test_count, +) logger = get_logger(__name__) @@ -174,6 +182,10 @@ def pytest_runtest_makereport( outcome = yield report = outcome.get_result() setattr(item, f"result_{report.when}", report) + if report.when == "setup": + record_test_execution(item) + if report.skipped: + count_skipped_test(item) @pytest.fixture(scope="session") @@ -211,6 +223,7 @@ def get_test_suite_scope(fixture_name: str, config: pytest.Config) -> str: @pytest.fixture(scope=get_test_suite_scope) # type: ignore[arg-type] def test_suite( + request: pytest.FixtureRequest, simulator: Simulation, session_temp_folder: Path, test_suite_name: str, @@ -247,6 +260,7 @@ def test_suite( yield suite + count_check_error: str | None = None with FileLock(users_lock_file): with open(users_file, "r") as f: users = json.load(f) @@ -254,9 +268,15 @@ def test_suite( with open(users_file, "w") as f: json.dump(users, f) if users == 0: + if getattr(request.config, "assert_reported_test_count", False): + count_check_error = verify_reported_test_count( + request, suite, session_temp_folder + ) suite.end() suite_file.unlink() users_file.unlink() + if count_check_error is not None: + raise HiveReportedTestCountError(count_check_error) @pytest.fixture(scope="module") @@ -336,11 +356,11 @@ def hive_test( "by the simulator or pytest plugin using this plugin!" ) - test_parameter_string = request.node.name test: HiveTest = test_suite.start_test( - name=test_parameter_string, + name=request.node.name, description=test_case_description, ) + mark_hive_test_started(request.node) yield test try: @@ -415,15 +435,6 @@ def hive_test( "unknown).\n\n" + captured_output ) - test.end( - result=HiveTestResult( - test_pass=test_passed, details=test_result_details - ) - ) - logger.verbose( - f"Finished processing logs for test: {request.node.nodeid}" - ) - except Exception as e: logger.verbose( f"Error processing logs for test {request.node.nodeid}: {str(e)}" @@ -432,8 +443,10 @@ def hive_test( test_result_details = ( f"Exception whilst processing test result: {str(e)}" ) - test.end( - result=HiveTestResult( - test_pass=test_passed, details=test_result_details - ) - ) + end_test_and_count( + request.config, + request.node.nodeid, + test, + HiveTestResult(test_pass=test_passed, details=test_result_details), + ) + logger.verbose(f"Finished processing logs for test: {request.node.nodeid}") diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/reporting.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/reporting.py new file mode 100644 index 00000000000..5fa92ae0022 --- /dev/null +++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/reporting.py @@ -0,0 +1,203 @@ +""" +Helpers detecting silent hive test-result loss. + +At very high test throughput (e.g. `consume-enginex`), the simulator +can transiently exhaust its ephemeral port range toward the hive API +endpoint: `connect()` then fails with `EADDRNOTAVAIL` for a few seconds +on all xdist workers at once. Tests in flight during such a burst +either error at setup (`start_test` failed: the test is then silently +missing from the hive results), or run and pass but never report their +verdict (`end_test` failed: the dangling hive test case is later +force-closed as "Test was terminated by host"). + +The root fix is connection pooling and connection-establishment retries +in the `ethereum-hive` package. The helpers here provide complementary +simulator-side detection by comparing the number of test results +successfully reported to hive against the tests pytest actually ran. +""" + +import json +import os +from pathlib import Path +from typing import Dict, TypedDict + +import pytest +from hive.testing import HiveTest, HiveTestResult, HiveTestSuite + +from execution_testing.logging import get_logger + +logger = get_logger(__name__) + +reported_test_ids_key = pytest.StashKey[set[str]]() +"""Node IDs whose results this process successfully reported to hive.""" + +skipped_test_ids_key = pytest.StashKey[set[str]]() +"""Node IDs skipped before their hive test was started.""" + +executed_test_ids_key = pytest.StashKey[set[str]]() +"""Node IDs for which this process produced a setup report.""" + +hive_test_started_key = pytest.StashKey[bool]() +"""Whether a pytest item has successfully started its hive test.""" + + +class WorkerTestCounts(TypedDict): + """Persisted test-accounting state for one pytest worker.""" + + reported: int + skipped: int + executed: int + interrupted: bool + + +class HiveReportedTestCountError(Exception): + """The number of hive test results does not match the pytest run.""" + + +def record_test_execution(item: pytest.Item) -> None: + """Record that pytest produced a setup report for an item.""" + item.config.stash.setdefault(executed_test_ids_key, set()).add(item.nodeid) + + +def mark_hive_test_started(item: pytest.Item) -> None: + """Record that an item's hive test was successfully started.""" + item.stash[hive_test_started_key] = True + + +def end_test_and_count( + config: pytest.Config, + nodeid: str, + test: HiveTest, + result: HiveTestResult, +) -> None: + """End a hive test once and record its successfully reported result.""" + test.end(result=result) + config.stash.setdefault(reported_test_ids_key, set()).add(nodeid) + + +def count_skipped_test(item: pytest.Item) -> None: + """ + Record a test skipped before its hive test was started. + + A later fixture can skip an item after `hive_test` has started. That + item must be accounted for by the hive result instead of being counted + as both skipped and reported. + """ + if item.stash.get(hive_test_started_key, False): + return + item.config.stash.setdefault(skipped_test_ids_key, set()).add(item.nodeid) + + +def write_reported_test_count( + config: pytest.Config, + session: pytest.Session, + session_temp_folder: Path, +) -> None: + """ + Persist this process's test-accounting state to the shared session + folder so the last-finishing xdist worker can verify the full run. + """ + worker_id = os.environ.get("PYTEST_XDIST_WORKER", "master") + counts: WorkerTestCounts = { + "reported": len(config.stash.get(reported_test_ids_key, set())), + "skipped": len(config.stash.get(skipped_test_ids_key, set())), + "executed": len(config.stash.get(executed_test_ids_key, set())), + "interrupted": bool(session.shouldstop or session.shouldfail), + } + count_file = ( + session_temp_folder / f"hive_reported_test_count_{worker_id}.json" + ) + with open(count_file, "w") as f: + json.dump(counts, f) + + +def verify_reported_test_count( + request: pytest.FixtureRequest, + suite: HiveTestSuite, + session_temp_folder: Path, +) -> str | None: + """ + Verify that every collected test reported a result to hive. + + Return an error message if results were lost, `None` otherwise. On + loss, additionally report a failing meta-test case to hive so that + the loss is visible in the hive results, where the affected tests + are otherwise silently missing. + """ + session = request.session + collected = session.testscollected + reported = 0 + skipped = 0 + executed = 0 + interrupted_workers: list[str] = [] + worker_counts: Dict[str, WorkerTestCounts] = {} + for count_file in sorted( + session_temp_folder.glob("hive_reported_test_count_*.json") + ): + with open(count_file, "r") as f: + counts: WorkerTestCounts = json.load(f) + worker_id = count_file.stem.removeprefix("hive_reported_test_count_") + worker_counts[worker_id] = counts + reported += counts["reported"] + skipped += counts["skipped"] + executed += counts["executed"] + if counts["interrupted"]: + interrupted_workers.append(worker_id) + count_file.unlink() + + if ( + session.shouldstop + or session.shouldfail + or interrupted_workers + or executed < collected + ): + logger.warning( + "Test run was interrupted; skipping the collected-vs-reported " + f"hive test count check ({executed}/{collected} tests ran; " + f"interrupted workers: {interrupted_workers})." + ) + return None + + accounted = reported + skipped + if executed == collected and accounted == collected: + logger.info( + f"All {collected} collected tests are accounted for " + f"({reported} reported to hive, {skipped} skipped)." + ) + return None + + if executed > collected: + message = ( + f"{executed - collected} more test execution(s) than collected " + f"tests were recorded: {executed} ran, but only {collected} were " + f"collected (per-worker counts: {worker_counts})." + ) + elif accounted < collected: + message = ( + f"{collected - accounted} test result(s) were not reported to " + f"hive: collected and ran {collected} tests, but only {reported} " + f"result(s) were reported and {skipped} test(s) were skipped " + f"(per-worker counts: {worker_counts}). Results are typically " + f"lost when hive API calls fail, e.g. due to client startup " + f"failures or ephemeral port exhaustion (EADDRNOTAVAIL) at high " + f"test throughput." + ) + else: + message = ( + f"{accounted - collected} more hive result(s) than collected " + f"tests were recorded: {reported} reported + {skipped} skipped > " + f"{collected} collected (per-worker counts: {worker_counts})." + ) + logger.error(message) + try: + check_test = suite.start_test( + name="reported-test-count-check", + description=( + "Meta-test verifying that every collected test reported " + "its result to hive; fails if test results were lost." + ), + ) + check_test.end(result=HiveTestResult(test_pass=False, details=message)) + except Exception as e: + logger.error(f"Failed to report the test count mismatch: {e}") + return message diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/__init__.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/__init__.py new file mode 100644 index 00000000000..a74e7674cdb --- /dev/null +++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for the pytest_hive plugin.""" diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/test_reporting.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/test_reporting.py new file mode 100644 index 00000000000..2f4c62a5c0b --- /dev/null +++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/test_reporting.py @@ -0,0 +1,376 @@ +"""Unit tests for the hive reporting guard helpers.""" + +import json +from pathlib import Path +from typing import Any, List + +import pytest +from hive.testing import HiveTestResult + +from execution_testing.cli.pytest_commands.plugins.pytest_hive.reporting import ( # noqa: E501 + count_skipped_test, + end_test_and_count, + executed_test_ids_key, + hive_test_started_key, + mark_hive_test_started, + record_test_execution, + reported_test_ids_key, + skipped_test_ids_key, + verify_reported_test_count, + write_reported_test_count, +) + + +class FakeSession: + """Session double exposing what `verify_reported_test_count` reads.""" + + def __init__(self, testscollected: int): # noqa: D107 + self.testscollected = testscollected + self.shouldstop: bool | str = False + self.shouldfail: bool | str = False + + +class FakeRequest: + """FixtureRequest double carrying only a session.""" + + def __init__(self, session: FakeSession): # noqa: D107 + self.session = session + + +class FakeItem: + """Item double exposing the state used by reporting hooks.""" + + def __init__(self, config: pytest.Config, nodeid: str): # noqa: D107 + self.config = config + self.nodeid = nodeid + self.stash = pytest.Stash() + + +class FakeHiveTest: + """HiveTest double recording its end result.""" + + def __init__(self, test_id: int): # noqa: D107 + self.id = test_id + self.result: HiveTestResult | None = None + self.end_calls = 0 + + def end(self, *, result: HiveTestResult) -> None: # noqa: D102 + self.end_calls += 1 + self.result = result + + +class FakeHiveTestSuite: + """HiveTestSuite double recording started meta-tests.""" + + def __init__(self) -> None: # noqa: D107 + self.started: List[FakeHiveTest] = [] + + def start_test(self, name: str, description: str) -> FakeHiveTest: # noqa: D102 + del name, description + test = FakeHiveTest(test_id=len(self.started)) + self.started.append(test) + return test + + +def write_count_file( + folder: Path, + worker_id: str, + *, + reported: int, + skipped: int, + executed: int, + interrupted: bool = False, +) -> None: + """Write a per-worker reported-test count file.""" + file = folder / f"hive_reported_test_count_{worker_id}.json" + with open(file, "w") as f: + json.dump( + { + "reported": reported, + "skipped": skipped, + "executed": executed, + "interrupted": interrupted, + }, + f, + ) + + +def test_end_test_records_reported_nodeid( + pytestconfig: pytest.Config, monkeypatch: Any +) -> None: + """A successfully ended hive test is recorded exactly once.""" + monkeypatch.setitem(pytestconfig.stash, reported_test_ids_key, set()) + test = FakeHiveTest(test_id=1) + result = HiveTestResult(test_pass=True, details="passed") + + end_test_and_count( + pytestconfig, + "test_module.py::test_case", + test, # type: ignore[arg-type] + result, + ) + + assert test.end_calls == 1 + assert test.result == result + assert pytestconfig.stash[reported_test_ids_key] == { + "test_module.py::test_case" + } + + +def test_end_test_failure_is_not_retried_or_recorded( + pytestconfig: pytest.Config, monkeypatch: Any +) -> None: + """An ambiguous end failure propagates without a duplicate request.""" + monkeypatch.setitem(pytestconfig.stash, reported_test_ids_key, set()) + + class FailingHiveTest(FakeHiveTest): + def end(self, *, result: HiveTestResult) -> None: # noqa: D102 + del result + self.end_calls += 1 + raise RuntimeError("response disconnected") + + test = FailingHiveTest(test_id=1) + result = HiveTestResult(test_pass=True, details="passed") + + with pytest.raises(RuntimeError, match="response disconnected"): + end_test_and_count( + pytestconfig, + "test_module.py::test_case", + test, # type: ignore[arg-type] + result, + ) + + assert test.end_calls == 1 + assert pytestconfig.stash[reported_test_ids_key] == set() + + +def test_write_reported_test_count( + pytestconfig: pytest.Config, tmp_path: Path, monkeypatch: Any +) -> None: + """Counts from the config stash are written to a per-worker file.""" + monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw7") + monkeypatch.setitem( + pytestconfig.stash, + reported_test_ids_key, + {"test_a", "test_b", "test_c"}, + ) + monkeypatch.setitem( + pytestconfig.stash, + skipped_test_ids_key, + {"test_d"}, + ) + monkeypatch.setitem( + pytestconfig.stash, + executed_test_ids_key, + {"test_a", "test_b", "test_c", "test_d"}, + ) + write_reported_test_count( + pytestconfig, + FakeSession(testscollected=4), # type: ignore[arg-type] + tmp_path, + ) + file = tmp_path / "hive_reported_test_count_gw7.json" + with open(file, "r") as f: + assert json.load(f) == { + "reported": 3, + "skipped": 1, + "executed": 4, + "interrupted": False, + } + + +def test_count_skipped_test( + pytestconfig: pytest.Config, monkeypatch: Any +) -> None: + """A test skipped before hive startup is accounted for once.""" + monkeypatch.setitem( + pytestconfig.stash, + skipped_test_ids_key, + set(), + ) + item = FakeItem(pytestconfig, "test_module.py::test_skipped") + count_skipped_test(item) # type: ignore[arg-type] + count_skipped_test(item) # type: ignore[arg-type] + assert pytestconfig.stash[skipped_test_ids_key] == { + "test_module.py::test_skipped" + } + + +def test_skip_after_hive_test_start_is_not_counted( + pytestconfig: pytest.Config, monkeypatch: Any +) -> None: + """A late setup skip is accounted for only by its hive result.""" + monkeypatch.setitem(pytestconfig.stash, skipped_test_ids_key, set()) + item = FakeItem(pytestconfig, "test_module.py::test_late_skip") + mark_hive_test_started(item) # type: ignore[arg-type] + + count_skipped_test(item) # type: ignore[arg-type] + + assert item.stash[hive_test_started_key] is True + assert pytestconfig.stash[skipped_test_ids_key] == set() + + +def test_record_test_execution_is_unique( + pytestconfig: pytest.Config, monkeypatch: Any +) -> None: + """Repeated reports for one item do not inflate execution counts.""" + monkeypatch.setitem(pytestconfig.stash, executed_test_ids_key, set()) + item = FakeItem(pytestconfig, "test_module.py::test_case") + + record_test_execution(item) # type: ignore[arg-type] + record_test_execution(item) # type: ignore[arg-type] + + assert pytestconfig.stash[executed_test_ids_key] == { + "test_module.py::test_case" + } + + +def test_verify_passes_when_all_tests_reported(tmp_path: Path) -> None: + """No error and no meta-test when all collected tests are reported.""" + write_count_file(tmp_path, "gw0", reported=6, skipped=0, executed=6) + write_count_file(tmp_path, "gw1", reported=3, skipped=1, executed=4) + suite = FakeHiveTestSuite() + request = FakeRequest(FakeSession(testscollected=10)) + error = verify_reported_test_count(request, suite, tmp_path) # type: ignore[arg-type] + assert error is None + assert suite.started == [] + # Count files are consumed by the check. + assert list(tmp_path.glob("hive_reported_test_count_*.json")) == [] + + +def test_verify_detects_lost_test_results(tmp_path: Path) -> None: + """A shortfall returns an error and reports a failing hive meta-test.""" + write_count_file(tmp_path, "gw0", reported=5, skipped=0, executed=6) + write_count_file(tmp_path, "gw1", reported=3, skipped=0, executed=4) + suite = FakeHiveTestSuite() + request = FakeRequest(FakeSession(testscollected=10)) + error = verify_reported_test_count(request, suite, tmp_path) # type: ignore[arg-type] + assert error is not None + assert "2 test result(s) were not reported to hive" in error + assert len(suite.started) == 1 + meta_test = suite.started[0] + assert meta_test.result is not None + assert meta_test.result.test_pass is False + assert error in meta_test.result.details + + +def test_verify_skipped_when_peer_worker_was_interrupted( + tmp_path: Path, +) -> None: + """An interrupted peer prevents a false failure on the last worker.""" + write_count_file( + tmp_path, + "gw0", + reported=1, + skipped=0, + executed=1, + interrupted=True, + ) + write_count_file(tmp_path, "gw1", reported=1, skipped=0, executed=1) + suite = FakeHiveTestSuite() + session = FakeSession(testscollected=10) + request = FakeRequest(session) + error = verify_reported_test_count(request, suite, tmp_path) # type: ignore[arg-type] + assert error is None + assert suite.started == [] + + +def test_verify_skipped_when_not_all_collected_tests_ran( + tmp_path: Path, +) -> None: + """A partial run is detected even without a local interruption flag.""" + write_count_file(tmp_path, "gw0", reported=2, skipped=0, executed=2) + suite = FakeHiveTestSuite() + request = FakeRequest(FakeSession(testscollected=10)) + + error = verify_reported_test_count(request, suite, tmp_path) # type: ignore[arg-type] + + assert error is None + assert suite.started == [] + + +def test_verify_detects_overcount(tmp_path: Path) -> None: + """An overcount fails instead of masking another missing result.""" + write_count_file(tmp_path, "gw0", reported=10, skipped=1, executed=10) + suite = FakeHiveTestSuite() + request = FakeRequest(FakeSession(testscollected=10)) + + error = verify_reported_test_count(request, suite, tmp_path) # type: ignore[arg-type] + + assert error is not None + assert "1 more hive result(s) than collected tests" in error + assert len(suite.started) == 1 + + +def test_xdist_early_stop_persists_interruption_across_workers( + pytester: pytest.Pytester, +) -> None: + """The aggregate check recognizes `-n 2 -x` as a partial run.""" + count_folder = pytester.mkdir("worker-counts") + pytester.makeconftest( + f""" + import importlib + import os + from pathlib import Path + + import pytest + + reporting = importlib.import_module( + "execution_testing.cli.pytest_commands.plugins." + "pytest_hive.reporting" + ) + + count_folder = Path({str(count_folder)!r}) + + @pytest.hookimpl(hookwrapper=True) + def pytest_runtest_makereport(item, call): + outcome = yield + report = outcome.get_result() + if report.when == "setup": + reporting.record_test_execution(item) + + @pytest.fixture(scope="session", autouse=True) + def persist_worker_counts(request): + yield + reporting.write_reported_test_count( + request.config, request.session, count_folder + ) + worker = os.environ["PYTEST_XDIST_WORKER"] + (count_folder / f"collected_{{worker}}").write_text( + str(request.session.testscollected) + ) + """ + ) + pytester.makepyfile( + """ + import time + + import pytest + + @pytest.mark.parametrize("case", range(40)) + def test_cases(case): + if case == 0: + pytest.fail("stop the distributed run") + time.sleep(0.02) + """ + ) + + result = pytester.runpytest("-q", "-n", "2", "-x") + + assert result.ret == pytest.ExitCode.INTERRUPTED + count_files = list(count_folder.glob("hive_reported_test_count_*.json")) + counts = [json.loads(file.read_text()) for file in count_files] + assert any(count["interrupted"] for count in counts) + assert sum(count["executed"] for count in counts) < 40 + collected_files = count_folder.glob("collected_*") + assert {int(file.read_text()) for file in collected_files} == {40} + + suite = FakeHiveTestSuite() + request = FakeRequest(FakeSession(testscollected=40)) + error = verify_reported_test_count( + request, # type: ignore[arg-type] + suite, # type: ignore[arg-type] + count_folder, + ) + assert error is None + assert suite.started == []