86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
|
|
from src.jobs import coverage_evaluator_job
|
|
|
|
|
|
class _Result:
|
|
rowcount = 1
|
|
|
|
def scalar(self):
|
|
return "00000000-0000-0000-0000-000000000001"
|
|
|
|
def fetchall(self):
|
|
return []
|
|
|
|
def fetchone(self):
|
|
return None
|
|
|
|
|
|
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_coverage_evaluator_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()
|
|
|
|
async def fake_prometheus_targets() -> list[str]:
|
|
return ["192.0.2.10"]
|
|
|
|
monkeypatch.setattr("src.db.base.get_db_context", fake_db_context)
|
|
monkeypatch.setattr(
|
|
coverage_evaluator_job,
|
|
"_fetch_prometheus_target_ips",
|
|
fake_prometheus_targets,
|
|
)
|
|
|
|
async def exercise_evaluator_operations() -> None:
|
|
run_id = await coverage_evaluator_job._get_latest_run_id()
|
|
assert run_id == "00000000-0000-0000-0000-000000000001"
|
|
|
|
await coverage_evaluator_job._fetch_red_summary()
|
|
await coverage_evaluator_job._evaluate_monitoring(run_id)
|
|
await coverage_evaluator_job._evaluate_alerting(run_id)
|
|
await coverage_evaluator_job._evaluate_km_coverage(run_id)
|
|
await coverage_evaluator_job._evaluate_playbook_coverage(run_id)
|
|
await coverage_evaluator_job._evaluate_remediation_coverage(run_id)
|
|
await coverage_evaluator_job._evaluate_rule_matching_coverage(run_id)
|
|
await coverage_evaluator_job._evaluate_rule_creation_coverage(run_id)
|
|
await coverage_evaluator_job._log_aol(
|
|
stats={"monitoring_updated": 1},
|
|
duration_ms=10,
|
|
error=None,
|
|
)
|
|
assert (
|
|
await coverage_evaluator_job._auto_create_rules_for_uncovered_assets(
|
|
run_id
|
|
)
|
|
== 0
|
|
)
|
|
|
|
asyncio.run(exercise_evaluator_operations())
|
|
|
|
assert requested_projects == ["awoooi"] * 11
|
|
|
|
|
|
def test_coverage_evaluator_has_no_unscoped_database_context() -> None:
|
|
source = inspect.getsource(coverage_evaluator_job)
|
|
|
|
assert "get_db_context()" not in source
|
|
assert source.count("get_db_context(_PROJECT_ID)") == 12
|