Test hygiene: empty the flow store between modules, unbind demand on shutdown, poll the hook state in test_flow_ids, count only the watcher under test in the revoked-link test, delete runs.spec's runs

backend/fluksio/main.py

@@ -428,8 +428,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         app.state.device_store = devices
         controller.devices = devices
         # What the sockets want and the flows read, mapped onto the devices'
-        # on-demand ports.
+        # on-demand ports. It is process-wide, so it lets go of this
+        # controller when the engine stops.
         demand.bind(controller)
+        started.append(lambda: demand.bind(None))
         # Charts need a deeper series than the default; tell the engine
         # before it starts recording.
         controller.set_history_limits(dashboards.history_requirements())

backend/tests/api/routes/test_flow_ids.py

@@ -20,6 +20,7 @@ from fluksio.flow.dashboards import DashboardDef, WidgetDef
 from fluksio.flow.store import ISSUED_ID
 from fluksio.flow.tombstones import names_of
 from fluksio.models import FlowTombstone, Run
+from tests.api.routes.test_rename import _until
 
 FLOWS = f"{settings.API_V1_STR}/flows"
 DASHBOARDS = f"{settings.API_V1_STR}/dashboards"
@@ -180,8 +181,14 @@ def test_a_flow_runs_under_its_key_and_is_reached_by_its_id(
     hooked = client.post(f"/hooks/{flow_id}/tick", json={"tick": 7.0})
     assert hooked.status_code == 200, hooked.text
     assert client.post(f"/hooks/{key}/tick", json={"tick": 8.0}).status_code == 404
-    state = client.get(f"{FLOWS}/{flow_id}/state", headers=superuser_token_headers)
-    assert state.json()["values"][f"{key}.tick"]["value"] == 7.0
+
+    def values() -> dict[str, Any]:
+        state = client.get(f"{FLOWS}/{flow_id}/state", headers=superuser_token_headers)
+        return dict(state.json()["values"])
+
+    # Once the cascade it queued has run.
+    _until(lambda: f"{key}.tick" in values())
+    assert values()[f"{key}.tick"]["value"] == 7.0
 
     paused = client.post(f"{FLOWS}/{flow_id}/pause", headers=superuser_token_headers)
     assert paused.status_code == 200

backend/tests/conftest.py

@@ -1,3 +1,4 @@
+import shutil
 import uuid
 from collections.abc import Generator
 from datetime import UTC, datetime
@@ -107,6 +108,19 @@ def _clean_engine_rows() -> Generator[None, None, None]:
         session.commit()
 
 
+@pytest.fixture(scope="module", autouse=True)
+def _clean_flow_store() -> Generator[None, None, None]:
+    """Start the next module on an empty flow store.
+
+    The store is session-wide like the database, so the flows, dashboards and
+    devices one module left were loaded by the next module's app at startup.
+    Removed whole, its git with it, after the module's app has stopped.
+    """
+    yield
+    shutil.rmtree(settings.FLOWS_DIR, ignore_errors=True)
+    settings.DEVICES_FILE.unlink(missing_ok=True)
+
+
 @pytest.fixture(scope="module")
 def client() -> Generator[TestClient, None, None]:
     with TestClient(app) as c:

backend/tests/test_cloud.py

@@ -788,7 +788,9 @@ async def test_a_revoked_link_stands_down(enrolled: User) -> None:  # noqa: ARG0
         watcher = asyncio.create_task(connector.watch_enrolment(app))
         await asyncio.sleep(0.05)
         watcher.cancel()
-    assert not started
+    # The module's app runs a watcher of its own on the same patched names and
+    # may dial the enrolment above inside the window, so only this one counts.
+    assert app not in started
 
 
 def test_a_member_removed_at_the_portal_loses_their_user(db: Session) -> None: