d734362768
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 4m3s
CD Pipeline / build-and-deploy (push) Failing after 10m5s
CD Pipeline / post-deploy-checks (push) Has been skipped
798 lines
24 KiB
Python
798 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
from src.jobs import asset_scanner_job
|
|
|
|
|
|
class _Result:
|
|
rowcount = 1
|
|
|
|
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"
|
|
|
|
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)
|
|
|
|
disappeared = await asset_scanner_job._deprecate_missing_assets(
|
|
[
|
|
{
|
|
"asset_key": "host/test",
|
|
}
|
|
],
|
|
("host/",),
|
|
run_id,
|
|
)
|
|
assert disappeared == 1
|
|
|
|
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._finalize_discovery_run(
|
|
run_id=run_id,
|
|
triggered_by="cron",
|
|
status="success",
|
|
total_assets=1,
|
|
new_assets=1,
|
|
modified_assets=0,
|
|
disappeared_assets=1,
|
|
duration_ms=10,
|
|
error=None,
|
|
scope=["k8s", "prometheus", "database_catalog"],
|
|
scan_depth="shallow",
|
|
)
|
|
|
|
asyncio.run(exercise_scanner_writes())
|
|
|
|
assert requested_projects == ["awoooi"] * 6
|
|
|
|
|
|
def test_asset_upsert_propagates_database_failure(monkeypatch) -> None:
|
|
class FailingDatabase:
|
|
async def execute(self, statement, parameters=None):
|
|
raise RuntimeError("constraint rejected")
|
|
|
|
class FailingContext:
|
|
async def __aenter__(self):
|
|
return FailingDatabase()
|
|
|
|
async def __aexit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
monkeypatch.setattr(
|
|
"src.db.base.get_db_context",
|
|
lambda project_id=None: FailingContext(),
|
|
)
|
|
|
|
async def exercise() -> None:
|
|
try:
|
|
await asset_scanner_job._upsert_assets(
|
|
[
|
|
{
|
|
"asset_key": "postgres/database/awoooi",
|
|
"asset_type": "database",
|
|
"host": None,
|
|
"namespace": None,
|
|
"name": "awoooi",
|
|
"metadata": {},
|
|
"tags": ["source:test"],
|
|
}
|
|
],
|
|
"00000000-0000-0000-0000-000000000001",
|
|
)
|
|
except RuntimeError as exc:
|
|
assert str(exc) == "constraint rejected"
|
|
else:
|
|
raise AssertionError("asset upsert failure must propagate")
|
|
|
|
asyncio.run(exercise())
|
|
|
|
|
|
def test_asset_key_integrity_preflight_accepts_unique_inventory(monkeypatch) -> None:
|
|
requested_projects: list[str | None] = []
|
|
statements: list[str] = []
|
|
|
|
class IntegrityResult:
|
|
def one(self):
|
|
return 0, True
|
|
|
|
class IntegrityDatabase:
|
|
async def execute(self, statement, parameters=None):
|
|
statements.append(str(statement))
|
|
return IntegrityResult()
|
|
|
|
class IntegrityContext:
|
|
async def __aenter__(self):
|
|
return IntegrityDatabase()
|
|
|
|
async def __aexit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
def fake_db_context(project_id=None):
|
|
requested_projects.append(project_id)
|
|
return IntegrityContext()
|
|
|
|
monkeypatch.setattr("src.db.base.get_db_context", fake_db_context)
|
|
|
|
asyncio.run(asset_scanner_job._assert_asset_key_integrity())
|
|
|
|
assert requested_projects == ["awoooi"]
|
|
assert any("enable_indexscan" in statement for statement in statements)
|
|
assert any("enable_indexonlyscan" in statement for statement in statements)
|
|
assert any("enable_bitmapscan" in statement for statement in statements)
|
|
assert any("HAVING COUNT(*) > 1" in statement for statement in statements)
|
|
assert any("asset_inventory_asset_key_key" in statement for statement in statements)
|
|
|
|
|
|
def test_asset_key_integrity_preflight_fails_closed_on_duplicate_key(
|
|
monkeypatch,
|
|
) -> None:
|
|
class IntegrityResult:
|
|
def one(self):
|
|
return 1, True
|
|
|
|
class IntegrityDatabase:
|
|
async def execute(self, statement, parameters=None):
|
|
return IntegrityResult()
|
|
|
|
class IntegrityContext:
|
|
async def __aenter__(self):
|
|
return IntegrityDatabase()
|
|
|
|
async def __aexit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
monkeypatch.setattr(
|
|
"src.db.base.get_db_context",
|
|
lambda project_id=None: IntegrityContext(),
|
|
)
|
|
|
|
async def exercise() -> None:
|
|
try:
|
|
await asset_scanner_job._assert_asset_key_integrity()
|
|
except RuntimeError as exc:
|
|
assert str(exc) == (
|
|
"asset_inventory_key_integrity_failed "
|
|
"duplicate_groups=1 unique_index_healthy=true"
|
|
)
|
|
else:
|
|
raise AssertionError("duplicate canonical keys must fail closed")
|
|
|
|
asyncio.run(exercise())
|
|
|
|
|
|
def test_deprecate_missing_assets_is_bounded_to_successful_prefixes(
|
|
monkeypatch,
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
class CapturingDatabase:
|
|
async def execute(self, statement, parameters=None):
|
|
captured["sql"] = str(statement)
|
|
captured["parameters"] = parameters
|
|
return _Result()
|
|
|
|
class CapturingContext:
|
|
async def __aenter__(self):
|
|
return CapturingDatabase()
|
|
|
|
async def __aexit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
monkeypatch.setattr(
|
|
"src.db.base.get_db_context",
|
|
lambda project_id=None: CapturingContext(),
|
|
)
|
|
|
|
disappeared = asyncio.run(
|
|
asset_scanner_job._deprecate_missing_assets(
|
|
[
|
|
{"asset_key": "k8s/pod/prod/current"},
|
|
{"asset_key": "prometheus_target/node/192.168.0.110:9100"},
|
|
],
|
|
("k8s/pod/",),
|
|
"00000000-0000-0000-0000-000000000001",
|
|
)
|
|
)
|
|
|
|
assert disappeared == 1
|
|
assert "last_seen_at <" in str(captured["sql"])
|
|
assert captured["parameters"] == {
|
|
"patterns": ["k8s/pod/%"],
|
|
"seen_keys": ["k8s/pod/prod/current"],
|
|
"rid": "00000000-0000-0000-0000-000000000001",
|
|
}
|
|
|
|
|
|
def test_runtime_collection_reports_failed_sources_without_reconciling_them(
|
|
monkeypatch,
|
|
) -> None:
|
|
async def fake_kubectl(resource: str, all_namespaces: bool = True):
|
|
if resource == "nodes":
|
|
raise RuntimeError("nodes unavailable")
|
|
return {"items": []}
|
|
|
|
async def fake_prometheus():
|
|
return [], []
|
|
|
|
async def fake_database_catalog():
|
|
return [], []
|
|
|
|
async def fake_ai_catalog():
|
|
return [], []
|
|
|
|
monkeypatch.setattr(asset_scanner_job, "_fetch_kubectl_json", fake_kubectl)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_collect_prometheus_targets",
|
|
fake_prometheus,
|
|
)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_collect_database_catalog_assets",
|
|
fake_database_catalog,
|
|
)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_collect_ai_catalog_assets",
|
|
fake_ai_catalog,
|
|
)
|
|
|
|
assets, relationships, prefixes, errors = asyncio.run(
|
|
asset_scanner_job._collect_runtime_assets()
|
|
)
|
|
|
|
assert assets == []
|
|
assert relationships == []
|
|
assert "k8s/node/" not in prefixes
|
|
assert "k8s/pod/" in prefixes
|
|
assert "prometheus_target/" in prefixes
|
|
assert errors == ("k8s_nodes",)
|
|
|
|
|
|
def test_ingress_collector_builds_website_api_and_certificate_without_key_data() -> (
|
|
None
|
|
):
|
|
assets, relationships = asset_scanner_job._collect_ingress_assets(
|
|
{
|
|
"items": [
|
|
{
|
|
"metadata": {"namespace": "awoooi-prod", "name": "awoooi"},
|
|
"spec": {
|
|
"rules": [
|
|
{
|
|
"host": "awoooi.example.test",
|
|
"http": {
|
|
"paths": [
|
|
{
|
|
"path": "/api",
|
|
"pathType": "Prefix",
|
|
"backend": {
|
|
"service": {"name": "awoooi-api"}
|
|
},
|
|
}
|
|
]
|
|
},
|
|
}
|
|
],
|
|
"tls": [
|
|
{
|
|
"hosts": ["awoooi.example.test"],
|
|
"secretName": "awoooi-tls",
|
|
}
|
|
],
|
|
},
|
|
}
|
|
]
|
|
}
|
|
)
|
|
|
|
by_type = {asset["asset_type"]: asset for asset in assets}
|
|
assert set(by_type) == {"website", "api_endpoint", "certificate"}
|
|
assert by_type["api_endpoint"]["metadata"]["request_or_response_body_read"] is False
|
|
assert by_type["certificate"]["metadata"]["certificate_material_read"] is False
|
|
assert by_type["certificate"]["metadata"]["secret_object_queried"] is False
|
|
route_key = by_type["api_endpoint"]["asset_key"]
|
|
assert relationships == [
|
|
{
|
|
"from_key": "k8s/website/awoooi.example.test",
|
|
"to_key": route_key,
|
|
"relationship_type": "routes_to",
|
|
},
|
|
{
|
|
"from_key": route_key,
|
|
"to_key": "k8s/service/awoooi-prod/awoooi-api",
|
|
"relationship_type": "routes_to",
|
|
},
|
|
{
|
|
"from_key": "k8s/website/awoooi.example.test",
|
|
"to_key": "k8s/certificate-ref/awoooi-prod/awoooi-tls",
|
|
"relationship_type": "authenticates_via",
|
|
},
|
|
]
|
|
|
|
|
|
def test_ingress_route_identity_is_stable_when_manifest_order_changes() -> None:
|
|
def payload(paths: list[dict[str, object]]) -> dict[str, object]:
|
|
return {
|
|
"items": [
|
|
{
|
|
"metadata": {"namespace": "prod", "name": "public"},
|
|
"spec": {
|
|
"rules": [
|
|
{
|
|
"host": "service.example.test",
|
|
"http": {"paths": paths},
|
|
}
|
|
]
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
api_path = {
|
|
"path": "/api",
|
|
"backend": {"service": {"name": "api"}},
|
|
}
|
|
health_path = {
|
|
"path": "/health",
|
|
"backend": {"service": {"name": "api"}},
|
|
}
|
|
forward_assets, _ = asset_scanner_job._collect_ingress_assets(
|
|
payload([api_path, health_path])
|
|
)
|
|
reversed_assets, _ = asset_scanner_job._collect_ingress_assets(
|
|
payload([health_path, api_path])
|
|
)
|
|
|
|
forward_keys = {
|
|
asset["name"]: asset["asset_key"]
|
|
for asset in forward_assets
|
|
if asset["asset_type"] == "api_endpoint"
|
|
}
|
|
reversed_keys = {
|
|
asset["name"]: asset["asset_key"]
|
|
for asset in reversed_assets
|
|
if asset["asset_type"] == "api_endpoint"
|
|
}
|
|
assert forward_keys == reversed_keys
|
|
assert len(set(forward_keys.values())) == 2
|
|
|
|
|
|
def test_k8s_metadata_assets_never_read_volume_secret_or_command_data() -> None:
|
|
pvc = asset_scanner_job._build_pvc_asset(
|
|
{
|
|
"metadata": {"namespace": "prod", "name": "data"},
|
|
"spec": {
|
|
"storageClassName": "local-path",
|
|
"accessModes": ["ReadWriteOnce"],
|
|
"resources": {"requests": {"storage": "10Gi"}},
|
|
},
|
|
"status": {"phase": "Bound"},
|
|
}
|
|
)
|
|
cronjob = asset_scanner_job._build_cronjob_asset(
|
|
{
|
|
"metadata": {"namespace": "prod", "name": "scan"},
|
|
"spec": {"schedule": "0 * * * *", "suspend": False},
|
|
"status": {"active": []},
|
|
}
|
|
)
|
|
secret = asset_scanner_job._build_secret_reference_asset(
|
|
"prod",
|
|
"database-ref",
|
|
reference_kind="PodVolume",
|
|
)
|
|
|
|
assert pvc["asset_type"] == "volume"
|
|
assert pvc["metadata"]["data_read"] is False
|
|
assert cronjob["asset_type"] == "scheduled_job"
|
|
assert cronjob["metadata"]["command_or_env_read"] is False
|
|
assert secret["asset_type"] == "secret"
|
|
assert secret["metadata"]["secret_object_queried"] is False
|
|
assert secret["metadata"]["secret_value_read"] is False
|
|
|
|
|
|
def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefixes(
|
|
monkeypatch,
|
|
) -> None:
|
|
async def fake_kubectl(resource: str, all_namespaces: bool = True):
|
|
if resource == "pods":
|
|
return {
|
|
"items": [
|
|
{
|
|
"metadata": {"namespace": "prod", "name": "api"},
|
|
"spec": {
|
|
"volumes": [
|
|
{
|
|
"name": "credential",
|
|
"secret": {"secretName": "database-ref"},
|
|
}
|
|
]
|
|
},
|
|
}
|
|
]
|
|
}
|
|
if resource == "persistentvolumeclaims":
|
|
return {
|
|
"items": [
|
|
{
|
|
"metadata": {"namespace": "prod", "name": "data"},
|
|
"spec": {},
|
|
"status": {},
|
|
}
|
|
]
|
|
}
|
|
if resource == "ingresses":
|
|
return {
|
|
"items": [
|
|
{
|
|
"metadata": {"namespace": "prod", "name": "public"},
|
|
"spec": {
|
|
"rules": [
|
|
{
|
|
"host": "service.example.test",
|
|
"http": {
|
|
"paths": [
|
|
{
|
|
"path": "/api",
|
|
"backend": {"service": {"name": "api"}},
|
|
}
|
|
]
|
|
},
|
|
}
|
|
],
|
|
"tls": [
|
|
{
|
|
"hosts": ["service.example.test"],
|
|
"secretName": "public-tls",
|
|
}
|
|
],
|
|
},
|
|
}
|
|
]
|
|
}
|
|
if resource == "cronjobs":
|
|
return {
|
|
"items": [
|
|
{
|
|
"metadata": {"namespace": "prod", "name": "scan"},
|
|
"spec": {"schedule": "0 * * * *"},
|
|
"status": {},
|
|
}
|
|
]
|
|
}
|
|
return {"items": []}
|
|
|
|
async def empty_collector():
|
|
return [], []
|
|
|
|
monkeypatch.setattr(asset_scanner_job, "_fetch_kubectl_json", fake_kubectl)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_collect_prometheus_targets",
|
|
empty_collector,
|
|
)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_collect_database_catalog_assets",
|
|
empty_collector,
|
|
)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_collect_ai_catalog_assets",
|
|
empty_collector,
|
|
)
|
|
|
|
assets, relationships, prefixes, errors = asyncio.run(
|
|
asset_scanner_job._collect_runtime_assets()
|
|
)
|
|
|
|
assert errors == ()
|
|
assert {
|
|
"container",
|
|
"secret",
|
|
"volume",
|
|
"website",
|
|
"api_endpoint",
|
|
"certificate",
|
|
"scheduled_job",
|
|
} <= {asset["asset_type"] for asset in assets}
|
|
assert {
|
|
"k8s/secret-ref/",
|
|
"k8s/pvc/",
|
|
"k8s/website/",
|
|
"k8s/ingress-route/",
|
|
"k8s/certificate-ref/",
|
|
"k8s/cronjob/",
|
|
} <= set(prefixes)
|
|
assert any(
|
|
relationship["to_key"] == "k8s/secret-ref/prod/database-ref"
|
|
for relationship in relationships
|
|
)
|
|
|
|
|
|
def test_ai_catalog_collector_reads_aggregates_and_model_names_only(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
class CatalogResult:
|
|
def __init__(self, rows):
|
|
self._rows = rows
|
|
|
|
def fetchall(self):
|
|
return self._rows
|
|
|
|
class CatalogDatabase:
|
|
async def execute(self, statement, parameters=None):
|
|
sql = str(statement)
|
|
if "FROM agent_sessions" in sql:
|
|
return CatalogResult(
|
|
[
|
|
type(
|
|
"AgentRow",
|
|
(),
|
|
{
|
|
"_mapping": {
|
|
"agent_role": "security_reviewer",
|
|
"session_count": 4,
|
|
"latest_seen_at": "2026-07-14T00:00:00+00:00",
|
|
}
|
|
},
|
|
)()
|
|
]
|
|
)
|
|
assert parameters == {"project_id": "awoooi"}
|
|
return CatalogResult(
|
|
[
|
|
type(
|
|
"KnowledgeRow",
|
|
(),
|
|
{
|
|
"_mapping": {
|
|
"entry_type": "runbook",
|
|
"entry_count": 9,
|
|
"latest_updated_at": "2026-07-14T00:00:00+00:00",
|
|
}
|
|
},
|
|
)()
|
|
]
|
|
)
|
|
|
|
class CatalogContext:
|
|
async def __aenter__(self):
|
|
return CatalogDatabase()
|
|
|
|
async def __aexit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
monkeypatch.setattr(
|
|
"src.db.base.get_db_context",
|
|
lambda project_id=None: CatalogContext(),
|
|
)
|
|
registry = tmp_path / "models.json"
|
|
registry.write_text(
|
|
json.dumps(
|
|
{
|
|
"providers": {
|
|
"ollama": {
|
|
"models": {
|
|
"default": "model-a",
|
|
"summary": "model-a",
|
|
}
|
|
}
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assets, relationships = asyncio.run(
|
|
asset_scanner_job._collect_ai_catalog_assets(registry)
|
|
)
|
|
by_type: dict[str, list[dict]] = {}
|
|
for asset in assets:
|
|
by_type.setdefault(asset["asset_type"], []).append(asset)
|
|
|
|
assert set(by_type) == {
|
|
"third_party_service",
|
|
"llm_model",
|
|
"ai_agent",
|
|
"km_entry",
|
|
}
|
|
assert len(by_type["llm_model"]) == 1
|
|
assert by_type["llm_model"][0]["metadata"]["purposes"] == [
|
|
"default",
|
|
"summary",
|
|
]
|
|
assert by_type["ai_agent"][0]["metadata"]["session_input_or_output_read"] is False
|
|
assert by_type["km_entry"][0]["metadata"]["title_or_content_read"] is False
|
|
assert relationships == [
|
|
{
|
|
"from_key": "model_registry/model/ollama/model-a",
|
|
"to_key": "model_registry/provider/ollama",
|
|
"relationship_type": "depends_on",
|
|
}
|
|
]
|
|
|
|
|
|
def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) -> None:
|
|
terminal: dict[str, object] = {}
|
|
|
|
async def fake_start(*args, **kwargs):
|
|
return "00000000-0000-0000-0000-000000000001"
|
|
|
|
async def fake_collect():
|
|
return [], [], (), ("prometheus_targets",)
|
|
|
|
async def fake_integrity():
|
|
return None
|
|
|
|
async def fake_upsert(*args, **kwargs):
|
|
return 0, 0
|
|
|
|
async def fake_deprecate(*args, **kwargs):
|
|
return 0
|
|
|
|
async def fake_relationships(*args, **kwargs):
|
|
return 0
|
|
|
|
async def fake_coverage(*args, **kwargs):
|
|
return None
|
|
|
|
async def fake_finalize(**kwargs):
|
|
terminal.update(kwargs)
|
|
|
|
monkeypatch.setattr(asset_scanner_job, "_start_discovery_run", fake_start)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_assert_asset_key_integrity",
|
|
fake_integrity,
|
|
)
|
|
monkeypatch.setattr(asset_scanner_job, "_collect_runtime_assets", fake_collect)
|
|
monkeypatch.setattr(asset_scanner_job, "_upsert_assets", fake_upsert)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_deprecate_missing_assets",
|
|
fake_deprecate,
|
|
)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_upsert_relationships",
|
|
fake_relationships,
|
|
)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_write_coverage_snapshots",
|
|
fake_coverage,
|
|
)
|
|
monkeypatch.setattr(
|
|
asset_scanner_job,
|
|
"_finalize_discovery_run",
|
|
fake_finalize,
|
|
)
|
|
|
|
run_id = asyncio.run(asset_scanner_job.scan_once())
|
|
|
|
assert run_id == "00000000-0000-0000-0000-000000000001"
|
|
assert terminal["status"] == "failed"
|
|
assert terminal["error"] == (
|
|
"RuntimeError: asset_collector_failures=prometheus_targets"
|
|
)
|
|
|
|
|
|
def test_database_catalog_collector_reads_metadata_only(monkeypatch) -> None:
|
|
requested_projects: list[str | None] = []
|
|
|
|
class CatalogResult:
|
|
def fetchall(self):
|
|
return [
|
|
type(
|
|
"CatalogRow",
|
|
(),
|
|
{
|
|
"_mapping": {
|
|
"database_name": "awoooi",
|
|
"schema_name": "public",
|
|
"relation_name": "incidents",
|
|
"relation_kind": "table",
|
|
"row_security_enabled": True,
|
|
}
|
|
},
|
|
)()
|
|
]
|
|
|
|
class CatalogDatabase:
|
|
async def execute(self, statement, parameters=None):
|
|
return CatalogResult()
|
|
|
|
class CatalogContext:
|
|
async def __aenter__(self):
|
|
return CatalogDatabase()
|
|
|
|
async def __aexit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
def fake_db_context(project_id: str | None = None):
|
|
requested_projects.append(project_id)
|
|
return CatalogContext()
|
|
|
|
monkeypatch.setattr("src.db.base.get_db_context", fake_db_context)
|
|
|
|
assets, relationships = asyncio.run(
|
|
asset_scanner_job._collect_database_catalog_assets()
|
|
)
|
|
|
|
assert requested_projects == ["awoooi"]
|
|
assert [asset["asset_type"] for asset in assets] == ["database", "table"]
|
|
assert assets[0]["metadata"]["row_data_read"] is False
|
|
assert assets[1]["metadata"] == {
|
|
"database": "awoooi",
|
|
"schema": "public",
|
|
"relation_kind": "table",
|
|
"row_security_enabled": True,
|
|
"row_data_read": False,
|
|
}
|
|
assert relationships == [
|
|
{
|
|
"from_key": "postgres/relation/awoooi/public/incidents",
|
|
"to_key": "postgres/database/awoooi",
|
|
"relationship_type": "depends_on",
|
|
}
|
|
]
|