964beb2564
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m36s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled
73 lines
1.8 KiB
Python
73 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from src.jobs import compliance_scanner_job
|
|
|
|
|
|
class _AssetRow:
|
|
asset_id = 1
|
|
asset_key = "host/test"
|
|
asset_type = "host"
|
|
metadata = {}
|
|
|
|
|
|
class _Result:
|
|
def fetchall(self):
|
|
return [_AssetRow()]
|
|
|
|
|
|
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_compliance_scanner_scopes_database_operations(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)
|
|
monkeypatch.setattr(
|
|
compliance_scanner_job,
|
|
"_evaluate_all_dimensions",
|
|
lambda asset: {"audit_log_enabled": ("compliant", {"source": "test"})},
|
|
)
|
|
|
|
async def exercise_compliance_operations() -> None:
|
|
assets = await compliance_scanner_job._fetch_active_assets()
|
|
assert assets == [
|
|
{
|
|
"asset_id": 1,
|
|
"asset_key": "host/test",
|
|
"asset_type": "host",
|
|
"metadata": {},
|
|
}
|
|
]
|
|
|
|
written, violations, warnings = (
|
|
await compliance_scanner_job._write_compliance_for_asset(assets[0])
|
|
)
|
|
assert (written, violations, warnings) == (1, 0, 0)
|
|
|
|
await compliance_scanner_job._log_aol(
|
|
stats={"assets_scanned": 1, "snapshots_written": 1},
|
|
duration_ms=10,
|
|
triggered_by="cron",
|
|
error=None,
|
|
)
|
|
|
|
asyncio.run(exercise_compliance_operations())
|
|
|
|
assert requested_projects == ["awoooi"] * 3
|