fix(test-consume): harden enginex hive reporting against connection errors

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,

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",
     [

packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/__init__.py

@@ -0,0 +1 @@
+"""Hive pytest plugin providing common functionality for simulators."""

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,16 +260,23 @@ 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)
         users -= 1
         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}")