test: stop fixtures leaking config overrides and stop retrying a failed Prefect setup

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()

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:

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