diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py index 1ab8282ca..68227b68d 100644 --- a/apps/api/src/jobs/asset_scanner_job.py +++ b/apps/api/src/jobs/asset_scanner_job.py @@ -41,6 +41,7 @@ _SCAN_INTERVAL_SEC = 3600 # 每 1 小時 _FIRST_DELAY_SEC = 60 # 啟動後等 60s 再首掃 (其他 service init) _KUBECTL_TIMEOUT_SEC = 30 _LOOP_BACKOFF_SEC = 300 # 異常時 backoff 5 分鐘 +_PROJECT_ID = "awoooi" # 7 個自動化覆蓋維度 (ADR-090 §3.5) _COVERAGE_DIMENSIONS = ( @@ -522,6 +523,7 @@ async def _collect_prometheus_targets() -> tuple[list[dict[str, Any]], list[dict target → host 建 depends_on relationship. """ import httpx + from src.core.config import settings assets: list[dict[str, Any]] = [] @@ -636,9 +638,10 @@ async def _start_discovery_run( """ try: from sqlalchemy import text as _sql + from src.db.base import get_db_context - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: row = await db.execute( _sql(""" INSERT INTO asset_discovery_run ( @@ -677,9 +680,10 @@ async def _finish_discovery_run( ) -> None: try: from sqlalchemy import text as _sql + from src.db.base import get_db_context - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: await db.execute( _sql(""" UPDATE asset_discovery_run @@ -719,13 +723,14 @@ async def _upsert_assets( return 0, 0 from sqlalchemy import text as _sql + from src.db.base import get_db_context new_count = 0 modified_count = 0 try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: for a in assets: # xmax = 0 表示 INSERT (新),否則表示 UPDATE (修改) row = await db.execute( @@ -781,11 +786,12 @@ async def _upsert_relationships(relationships: list[dict[str, str]]) -> int: return 0 from sqlalchemy import text as _sql + from src.db.base import get_db_context written = 0 try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: for rel in relationships: try: # 用 asset_key → asset_id 解析,同時 UPSERT @@ -826,9 +832,10 @@ async def _write_coverage_snapshots(run_id: str) -> None: """ try: from sqlalchemy import text as _sql + from src.db.base import get_db_context - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: # 一次性 INSERT: 取所有 active asset × 7 dimensions await db.execute( _sql(""" @@ -871,6 +878,7 @@ async def _log_aol_asset_discovered( """寫 automation_operation_log(asset_discovered)。失敗只 log 不阻塞。""" try: from sqlalchemy import text as _sql + from src.db.base import get_db_context aol_status = "success" if status == "success" else "failed" @@ -887,7 +895,7 @@ async def _log_aol_asset_discovered( "modified_assets": modified, } - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: await db.execute( _sql(""" INSERT INTO automation_operation_log ( diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py new file mode 100644 index 000000000..6106ca412 --- /dev/null +++ b/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