From 09be4153d10051560f45685af820e2a3334e4fd7 Mon Sep 17 00:00:00 2001 From: Infrahub Date: Wed, 19 Aug 2026 14:24:41 +0000 Subject: [PATCH 1/2] test: stop the cache and message-bus fixtures from leaking their config overrides MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `memory_cache` and `bus_simulator` each set a `config.OVERRIDE` field and never put it back. The `dependency_provider.scope(...)` unwinds on teardown, but `build_cache()` and `build_message_bus()` consult `config.OVERRIDE` *first*, so the override outlives the class that installed it and every later resolution in that xdist worker gets the previous class's throwaway adapter. For the cache that surfaces as ResourceNotFoundError: Diff summary for pipeline was not found in the cache in `TestProposedChange::test_run_generators_validate_requested_jobs`. The test writes the diff summary through a cache built from `config.SETTINGS.cache.driver` (Redis) and `run_generators` reads it back via `get_cache()`. Once the override leaks, the write goes to Redis and the read goes to the leftover MemoryCache. It reads as flaky but it is scheduling: it fails exactly when xdist puts `test_artifact_regen_e2e.py`, which uses `memory_cache`, on the same worker earlier in the session. Runs 32238651760 and 32153198842 had both files on gw3 and failed; run 32251034448 had them on gw3 and gw1 and passed. For the message bus nothing fails today — a stale BusSimulator swallows messages instead of raising — so it is fixed here before it costs a debugging session. Save and restore in a `finally`, matching the neighbouring `workflow_local` fixture and every other override site in the suite. Co-Authored-By: Claude Opus 5 (1M context) --- backend/tests/helpers/test_app.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/backend/tests/helpers/test_app.py b/backend/tests/helpers/test_app.py index c550b891304..b51f7e46635 100644 --- a/backend/tests/helpers/test_app.py +++ b/backend/tests/helpers/test_app.py @@ -96,20 +96,28 @@ async def bus_simulator( # Creating another service object to get service correctly initialized is a hack. # We should either reuse `service` fixture (leading to circular fixture dependencies issue atm), # or ideally properly patch production code responsible for Bus instantiation instead + original = config.OVERRIDE.message_bus bus = BusSimulator() _ = await InfrahubServices.new(database=db, workflow=WorkflowLocalExecution(), message_bus=bus) config.OVERRIDE.message_bus = bus - with dependency_provider.scope(build_message_bus, lambda: bus): - yield bus + try: + with dependency_provider.scope(build_message_bus, lambda: bus): + yield bus + finally: + config.OVERRIDE.message_bus = original @pytest.fixture(scope="class") async def memory_cache( self, db: InfrahubDatabase, dependency_provider: Provider ) -> AsyncGenerator[MemoryCache, None]: + original = config.OVERRIDE.cache cache = MemoryCache() config.OVERRIDE.cache = cache - with dependency_provider.scope(build_cache, lambda: cache): - yield cache + try: + with dependency_provider.scope(build_cache, lambda: cache): + yield cache + finally: + config.OVERRIDE.cache = original @pytest.fixture(scope="class") async def register_internal_schema(self, db: InfrahubDatabase, default_branch: Branch) -> SchemaBranch: From d28c010ecbb4746b67a54944f5af277f967dee1d Mon Sep 17 00:00:00 2001 From: Infrahub Date: Wed, 19 Aug 2026 14:28:14 +0000 Subject: [PATCH 2/2] test: do not retry a failed Prefect task manager setup once per test class MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `setup_task_manager_once` recorded only success, so a Prefect test server that came up and then stopped responding was retried by every later test class in that xdist worker. The retry is not cheap. The setup does not fail fast against an unreachable server — it blocks on the API until the pytest timeout fires — so each retry cost the full 300s. In run 32238651760 that turned one broken worker into 46 `Failed: Timeout >300.0s` errors across five test files and pushed the session into its 1800s limit, with the original httpx.ReadTimeout buried under 45 identical copies. Remember the failure alongside the success and re-raise it, chained, on every later call. The worker still fails, but once, in seconds, with the cause attached to the first error rather than the forty-sixth. `except BaseException` is deliberate: the pytest timeout raises `Failed`, which does not derive from `Exception`, and that is exactly the failure worth remembering. The once-per-process state moves onto a `TaskManagerSetup` object that takes the setup callable as a constructor argument, so the tests drive it with recording and failing doubles instead of patching the module — the adapter pattern the testing guidelines ask for. `setup_task_manager_once()` keeps its signature and callers. Co-Authored-By: Claude Opus 5 (1M context) --- backend/tests/helpers/task_manager.py | 39 +++++++++-- .../tests/unit/helpers/test_task_manager.py | 70 +++++++++++++++++++ 2 files changed, 104 insertions(+), 5 deletions(-) create mode 100644 backend/tests/unit/helpers/test_task_manager.py diff --git a/backend/tests/helpers/task_manager.py b/backend/tests/helpers/task_manager.py index 60ae083975a..13416d0a570 100644 --- a/backend/tests/helpers/task_manager.py +++ b/backend/tests/helpers/task_manager.py @@ -5,17 +5,46 @@ the calls are slow (several seconds of API round-trips), so fixtures should reuse a single setup per process instead of repeating it for every test or test class. +A failure is remembered the same way a success is. An unreachable Prefect test +server does not fail fast — the setup blocks until the pytest timeout fires — so +retrying it for every later test class costs that timeout each time and buries the +original cause under a wall of identical errors. + Tests that intentionally corrupt the shared task manager state must restore it themselves before yielding back, otherwise later tests will observe the corruption. """ +from collections.abc import Awaitable, Callable + from infrahub.workflows.initialization import setup_task_manager -_state = {"initialized": False} + +class TaskManagerSetup: + def __init__(self, setup: Callable[[], Awaitable[None]] = setup_task_manager) -> None: + self._setup = setup + self._initialized = False + self._failure: BaseException | None = None + + async def run_once(self) -> None: + if self._failure is not None: + raise RuntimeError("Prefect task manager setup already failed in this process") from self._failure + + if self._initialized: + return + + try: + await self._setup() + # The pytest timeout raises Failed, which derives from BaseException, and that is + # the failure worth remembering most. + except BaseException as exc: + self._failure = exc + raise + + self._initialized = True + + +_setup = TaskManagerSetup() async def setup_task_manager_once() -> None: - if _state["initialized"]: - return - await setup_task_manager() - _state["initialized"] = True + await _setup.run_once() diff --git a/backend/tests/unit/helpers/test_task_manager.py b/backend/tests/unit/helpers/test_task_manager.py new file mode 100644 index 00000000000..2ad537a9331 --- /dev/null +++ b/backend/tests/unit/helpers/test_task_manager.py @@ -0,0 +1,70 @@ +import pytest + +from tests.helpers.task_manager import TaskManagerSetup + + +class TimeoutFailure(BaseException): + """Stands in for pytest's Failed, which derives from BaseException rather than Exception.""" + + +class RecordingSetup: + """Counts how many times the task manager setup was actually run.""" + + def __init__(self) -> None: + self.calls = 0 + + async def __call__(self) -> None: + self.calls += 1 + + +class FailingSetup(RecordingSetup): + """Stands in for a Prefect test server that accepts connections but never answers.""" + + def __init__(self, error: BaseException) -> None: + super().__init__() + self.error = error + + async def __call__(self) -> None: + await super().__call__() + raise self.error + + +async def test_setup_runs_once_across_repeated_calls() -> None: + setup = RecordingSetup() + once = TaskManagerSetup(setup=setup) + + await once.run_once() + await once.run_once() + await once.run_once() + + assert setup.calls == 1 + + +async def test_failed_setup_is_reported_without_being_rerun() -> None: + setup = FailingSetup(TimeoutError("prefect server is unreachable")) + once = TaskManagerSetup(setup=setup) + + with pytest.raises(TimeoutError, match=r"^prefect server is unreachable$"): + await once.run_once() + + for _ in range(3): + with pytest.raises( + RuntimeError, match=r"^Prefect task manager setup already failed in this process$" + ) as exc_info: + await once.run_once() + assert isinstance(exc_info.value.__cause__, TimeoutError) + + assert setup.calls == 1 + + +async def test_failure_that_bypasses_exception_is_remembered() -> None: + setup = FailingSetup(TimeoutFailure("Timeout >300.0s")) + once = TaskManagerSetup(setup=setup) + + with pytest.raises(TimeoutFailure, match=r"^Timeout >300\.0s$"): + await once.run_once() + + with pytest.raises(RuntimeError, match=r"^Prefect task manager setup already failed in this process$"): + await once.run_once() + + assert setup.calls == 1