from __future__ import annotations # ruff: noqa: E402, I001 import asyncio import json import os from datetime import UTC, datetime, timedelta from types import SimpleNamespace os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") from fastapi import FastAPI from fastapi.testclient import TestClient from src.api.v1.iwooos import router from src.services.iwooos_security_asset_control_plane import ( IwoooSSecurityAssetControlPlaneService, build_iwooos_security_asset_control_plane, build_unavailable_iwooos_security_asset_control_plane, ) def _row(**values): return SimpleNamespace(**values) def _ready_payload() -> dict: now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC) discovery = _row( latest_status="success", latest_scan_depth="full", latest_started_at=now - timedelta(minutes=31), latest_ended_at=now - timedelta(minutes=30), latest_success_ended_at=now - timedelta(minutes=30), latest_total_assets=24, latest_new_assets=2, latest_modified_assets=1, latest_disappeared_assets=0, ) inventory = [ _row(asset_type="host", cnt=3, unowned_count=1, stale_count=0), _row(asset_type="k8s_workload", cnt=5, unowned_count=0, stale_count=1), _row(asset_type="website", cnt=2, unowned_count=0, stale_count=0), _row(asset_type="api_endpoint", cnt=4, unowned_count=0, stale_count=0), _row(asset_type="database", cnt=2, unowned_count=0, stale_count=0), _row(asset_type="log_stream", cnt=4, unowned_count=0, stale_count=0), _row(asset_type="monitoring_target", cnt=3, unowned_count=0, stale_count=0), _row(asset_type="ai_agent", cnt=1, unowned_count=0, stale_count=0), ] coverage = [ _row(dimension=dimension, status="green", cnt=20) for dimension in ( "auto_monitoring", "auto_alerting", "auto_rule_creation", "auto_rule_matching", "auto_playbook", "auto_remediation", "auto_km_creation", ) ] compliance = [ _row(dimension=dimension, status="compliant", cnt=20) for dimension in ( "ssl_cert_valid", "cve_scan", "secret_rotated", "backup_tested", "audit_log_enabled", "access_reviewed", "encryption_at_rest", ) ] automation = [ _row(operation_type="asset_discovered", status="success", cnt=1), _row(operation_type="rule_matched", status="success", cnt=3), _row(operation_type="rule_created", status="success", cnt=2), _row(operation_type="ansible_check_mode_executed", status="dry_run", cnt=2), _row(operation_type="ansible_apply_executed", status="success", cnt=1), _row(operation_type="remediation_verified", status="success", cnt=1), _row(operation_type="km_created", status="success", cnt=1), _row(operation_type="ansible_apply_executed", status="failed", cnt=1), ] return build_iwooos_security_asset_control_plane( discovery_row=discovery, inventory_rows=inventory, coverage_rows=coverage, relationship_row=_row(relationship_count=21, orphan_asset_count=3), compliance_rows=compliance, automation_rows=automation, ai_trace_row=_row(trace_count=4, accepted_trace_count=3), siem_row=_row( rule_total=12, approved_rule_count=10, fired_rule_count=8, noisy_rule_count=1, ai_generated_rule_count=3, incident_total_24h=5, open_incident_count_24h=2, resolved_incident_count_24h=3, decision_chain_count_24h=5, normalized_event_count_24h=18, mttr_minutes_24h=14.25, ), audit_row=_row( k8s_audit_count_24h=2, k8s_audit_failure_count_24h=0, mcp_audit_count_24h=11, mcp_audit_failure_count_24h=1, asset_change_count_24h=4, asset_change_ai_analysis_count_24h=3, ), generated_at=now, ) def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure() -> ( None ): payload = _ready_payload() assert payload["schema_version"] == "iwooos_security_asset_control_plane_v1" assert payload["status"] == "ready" assert payload["source_status"] == "live_database_aggregate" assert payload["summary"]["managed_asset_count"] == 24 assert payload["summary"]["present_asset_type_count"] == 8 assert payload["summary"]["siem_stage_coverage_percent"] == 100 assert payload["summary"]["ai_loop_stage_coverage_percent"] == 100 assert payload["summary"]["verified_remediation_receipt_count_24h"] == 1 assert payload["discovery"]["fresh"] is True assert payload["ai_automation"]["same_run_closed_loop_proven"] is False assert payload["ai_automation"]["same_run_closed_loop_count"] == 0 assert len(payload["nist_csf_functions"]) == 6 assert len(payload["siem"]["pipeline"]) == 8 assert payload["security_audit"]["mcp_tool_audit_count_24h"] == 11 assert any( item["work_item_id"] == "AIA-P0-006-02" for item in payload["work_items"] ) assert any( item["work_item_id"] == "AIA-P1-001-01" for item in payload["work_items"] ) public_text = json.dumps(payload, ensure_ascii=False) assert "192.168." not in public_text assert "postgresql://" not in public_text assert "target_resource" not in public_text assert '"event_payload":' not in public_text assert 'secret_value_collection_allowed": true' not in public_text.lower() def test_unknown_coverage_and_compliance_create_p0_work_items() -> None: now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC) payload = build_iwooos_security_asset_control_plane( discovery_row=_row( latest_status="success", latest_scan_depth="shallow", latest_started_at=now - timedelta(minutes=10), latest_ended_at=now - timedelta(minutes=9), latest_success_ended_at=now - timedelta(minutes=9), latest_total_assets=1, ), inventory_rows=[ _row(asset_type="host", cnt=1, unowned_count=1, stale_count=0), ], coverage_rows=[ _row(dimension="auto_monitoring", status="unknown", cnt=1), ], relationship_row=_row(relationship_count=0, orphan_asset_count=1), compliance_rows=[ _row(dimension="cve_scan", status="unknown", cnt=1), ], automation_rows=[], ai_trace_row=_row(trace_count=0, accepted_trace_count=0), siem_row=_row(), audit_row=_row(), generated_at=now, ) assert payload["summary"]["automation_coverage_green_percent"] == 0 assert payload["summary"]["automation_coverage_unknown_count"] == 1 assert payload["summary"]["compliance_unknown_count"] == 1 assert payload["summary"]["siem_observed_stage_count"] == 0 assert payload["summary"]["ai_loop_observed_stage_count"] == 0 work_item_ids = {item["work_item_id"] for item in payload["work_items"]} assert { "AIA-P0-006-02", "AIA-P0-006-03", "AIA-P0-006-04", "AIA-P0-006-05", "AIA-P0-006-06", "AIA-P0-007-01", "AIA-P0-008-01", } <= work_item_ids def test_unavailable_payload_is_fixed_degraded_contract_without_raw_error() -> None: payload = build_unavailable_iwooos_security_asset_control_plane( "database_query_failed" ) assert payload["status"] == "degraded" assert payload["source_status"] == "live_database_unavailable" assert payload["reason_code"] == "database_query_failed" assert payload["summary"]["managed_asset_count"] == 0 assert payload["discovery"]["fresh"] is False assert payload["boundaries"]["raw_asset_identity_returned"] is False assert payload["boundaries"]["raw_event_payload_returned"] is False def test_service_redacts_database_exception_details(monkeypatch) -> None: service = IwoooSSecurityAssetControlPlaneService() async def fail_load(): raise RuntimeError("postgresql://operator:raw-secret@private-host/runtime") monkeypatch.setattr(service, "_load_snapshot", fail_load) payload = asyncio.run(service.get_snapshot()) text = json.dumps(payload) assert payload["reason_code"] == "database_query_failed" assert "raw-secret" not in text assert "private-host" not in text def test_public_api_returns_live_aggregate(monkeypatch) -> None: payload = _ready_payload() class FakeService: async def get_snapshot(self): return payload monkeypatch.setattr( "src.api.v1.iwooos.get_iwooos_security_asset_control_plane_service", lambda: FakeService(), ) app = FastAPI() app.include_router(router) response = TestClient(app).get("/api/v1/iwooos/security-asset-control-plane") assert response.status_code == 200 data = response.json() assert data["schema_version"] == "iwooos_security_asset_control_plane_v1" assert data["summary"]["managed_asset_count"] == 24 assert data["summary"]["siem_stage_coverage_percent"] == 100 assert data["ai_automation"]["same_run_closed_loop_proven"] is False assert "192.168." not in response.text assert "raw-secret" not in response.text