fix(security): restore scoped asset scanner writes

apps/api/tests/test_asset_scanner_job.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import asyncio
+
+from src.jobs import asset_scanner_job
+
+
+class _Result:
+    def scalar(self):
+        return "00000000-0000-0000-0000-000000000001"
+
+    def one(self):
+        return 1, True
+
+
+class _Database:
+    async def execute(self, statement, parameters=None):
+        return _Result()
+
+
+class _Context:
+    async def __aenter__(self):
+        return _Database()
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        return False
+
+
+def test_background_scanner_scopes_every_database_operation(monkeypatch) -> None:
+    requested_projects: list[str | None] = []
+
+    def fake_db_context(project_id: str | None = None):
+        requested_projects.append(project_id)
+        return _Context()
+
+    monkeypatch.setattr("src.db.base.get_db_context", fake_db_context)
+
+    async def exercise_scanner_writes() -> None:
+        run_id = await asset_scanner_job._start_discovery_run(
+            "cron", ["k8s", "prometheus"], "shallow"
+        )
+        assert run_id == "00000000-0000-0000-0000-000000000001"
+
+        await asset_scanner_job._finish_discovery_run(
+            run_id=run_id,
+            status="success",
+            total_assets=1,
+            new_assets=1,
+            modified_assets=0,
+            duration_ms=10,
+            error=None,
+        )
+        inserted, modified = await asset_scanner_job._upsert_assets(
+            [
+                {
+                    "asset_key": "host/test",
+                    "asset_type": "host",
+                    "host": "test",
+                    "namespace": None,
+                    "name": "test",
+                    "metadata": {},
+                    "tags": ["source:test"],
+                }
+            ],
+            run_id,
+        )
+        assert (inserted, modified) == (1, 0)
+
+        written = await asset_scanner_job._upsert_relationships(
+            [
+                {
+                    "from_key": "host/test",
+                    "to_key": "monitoring/test",
+                    "relationship_type": "monitors",
+                }
+            ]
+        )
+        assert written == 1
+
+        await asset_scanner_job._write_coverage_snapshots(run_id)
+        await asset_scanner_job._log_aol_asset_discovered(
+            run_id=run_id,
+            triggered_by="cron",
+            total=1,
+            new=1,
+            modified=0,
+            duration_ms=10,
+            status="success",
+            error=None,
+        )
+
+    asyncio.run(exercise_scanner_writes())
+
+    assert requested_projects == ["awoooi"] * 6