from __future__ import annotations import json from pathlib import Path import pytest from src.services.portfolio_infrastructure_asset_reconciliation import ( load_portfolio_infrastructure_asset_reconciliation, ) ROOT = Path(__file__).resolve().parents[3] OPERATIONS = ROOT / "docs" / "operations" SNAPSHOT = OPERATIONS / "portfolio-infrastructure-asset-reconciliation.snapshot.json" def test_loader_maps_every_imported_inventory_row_without_claiming_live_truth() -> None: payload = load_portfolio_infrastructure_asset_reconciliation() mapped_rows = [row for asset in payload["assets"] for row in asset["source_rows"]] assert len(mapped_rows) == 131 assert len(mapped_rows) == len(set(mapped_rows)) assert {"L012", "L168", "L206", "L259"}.issubset(mapped_rows) assert payload["rollups"]["runtime_closed_assets"] == 1 assert payload["rollups"]["runtime_closure_status"] == ( "started_partial_1_of_144" ) assert payload["rollups"]["source_reconciled_assets"] == 6 assert payload["rollups"]["conflicts"] == 11 assert payload["rollups"]["conflict_assets"] == 36 assert payload["rollups"]["canonical_assets"] == 144 assert all( asset["live_truth_state"] != "verified_healthy" for asset in payload["assets"] ) assert payload["source_inventory"]["version_history"][0]["source_line_count"] == 241 assert payload["source_inventory"]["source_line_count"] == 284 def test_reconciliation_includes_missing_control_plane_and_provider_assets() -> None: payload = load_portfolio_infrastructure_asset_reconciliation() assets = {asset["canonical_id"]: asset for asset in payload["assets"]} required = { "windows-vmware:host_99", "service:ollama:host110", "host:111", "ai-provider:ollama_gcp_a", "ai-provider:ollama_gcp_b", "ai-provider:claude", "ai-provider:gemini", "control-plane:holmesgpt-investigator", "signal:alertmanager-webhook-chain", "control-plane:km-rag", "control-plane:mcp-gateway", "control-plane:playbook-registry", "control-plane:independent-verifier-registry", } assert required.issubset(assets) imported_omissions = required - {"service:ollama:host110"} assert all( assets[canonical_id]["reconciliation_state"] == "missing_from_imported_inventory" for canonical_id in imported_omissions ) assert assets["service:ollama:host110"]["reconciliation_state"] == ( "retired_tombstone_runtime_absence_verified" ) assert assets["windows-vmware:host_99"]["executor"] == "Agent99" assert assets["host:111"]["executor"] == "host_ansible_executor" assert assets["ai-provider:ollama_gcp_a"]["source_truth_state"] == ( "source_reconciled_runtime_pending" ) assert "WireGuard" in " ".join( assets["ai-provider:ollama_gcp_a"]["findings"] ) or "public HTTP" in " ".join( assets["ai-provider:ollama_gcp_a"]["findings"] ) def test_provider_alert_chain_and_gitea_runtime_gaps_are_explicit() -> None: payload = load_portfolio_infrastructure_asset_reconciliation() assets = {asset["canonical_id"]: asset for asset in payload["assets"]} tombstone = assets["service:ollama:host110"] assert tombstone["runtime_identity"].startswith("retired-tombstone:") assert tombstone["monitoring"]["coverage"] == ( "runtime_absence_verified_20260716" ) assert tombstone["bounded_executor_receipt"] == ( "run-aia-sre-002-20260716-1752" ) host111 = assets["host:111"] assert "macos-launchd" in host111["runtime_identity"] assert {"host120_origin_generation", "host121_origin_generation"}.issubset( host111["monitoring"]["signals"] ) assert host111["monitoring"]["coverage"] == ( "source_candidate_runtime_sensor_missing" ) for provider_id in ("ai-provider:ollama_gcp_a", "ai-provider:ollama_gcp_b"): provider = assets[provider_id] assert provider["transport_policy"] == { "observed": "public_http", "execution_scope": "sanitized_candidate_only", "tool_loop": "fail_closed", "promotion_requires": "secure_mesh_or_tls_plus_runtime_readback", } assert provider["monitoring"]["coverage"] == ( "source_candidate_runtime_target_missing" ) for provider_id in ("ai-provider:claude", "ai-provider:gemini"): provider = assets[provider_id] assert provider["credential_metadata"]["raw_value_recorded"] is False assert provider["credential_metadata"]["raw_value_readback_allowed"] is False assert provider["authorization"] == { "gate": "Gate5", "durable_run_binding_required": True, "status": "pending", } assert provider["canary"]["paired_rollback_required"] is True assert provider["canary"]["cost_caps_verified"] is False alert_chain = assets["signal:alertmanager-webhook-chain"] assert "Host99 Agent99" in " ".join(alert_chain["findings"]) assert "first-hop credential" in " ".join(alert_chain["findings"]) assert alert_chain["monitoring"]["coverage"] == ( "source_primary_and_independent_pull_ready_runtime_pending" ) gitea = assets["service:gitea"] exporter = assets["service:gitea-exporter:host110"] assert gitea["monitoring"]["coverage"] == "runtime_target_missing" assert exporter["monitoring"]["coverage"] == ( "source_candidate_runtime_target_missing" ) def test_source_reconciled_alert_and_gitea_assets_remain_runtime_pending() -> None: payload = load_portfolio_infrastructure_asset_reconciliation() assets = {asset["canonical_id"]: asset for asset in payload["assets"]} assert assets["service:prometheus"]["runtime_identity"] == ( "host110/docker/prometheus" ) assert assets["service:alertmanager"]["runtime_identity"] == ( "host110/docker/alertmanager" ) assert assets["service:alertmanager"]["live_truth_state"] == ( "not_probed_this_reconciliation" ) assert assets["service:gitea-exporter:host110"]["reconciliation_state"] == ( "source_retirement_ready_runtime_pending" ) assert assets["signal:alertmanager-webhook-chain"]["source_truth_state"] == ( "source_reconciled_runtime_pending" ) def test_host112_expansion_and_removed_asset_drift() -> None: payload = load_portfolio_infrastructure_asset_reconciliation() assets = {asset["canonical_id"]: asset for asset in payload["assets"]} required = { "service:wazuh-indexer:host112", "service:wazuh-dashboard:host112", "service:awoooi-guardian:host112", "service:filebeat:host112", "service:wg-easy:host112", "service:node-exporter:host112", "package:metasploit:host112", "package:openvas:host112", "schedule:host112:wazuh-guardian-hourly", "schedule:host112:daily-0200", } assert required.issubset(assets) assert assets["service:wg-easy:host112"]["reconciliation_state"] == ( "github_freeze_and_network_boundary_conflict" ) assert assets["service:fail2ban:host188"]["reconciliation_state"] == ( "missing_from_imported_inventory" ) def test_every_expanded_asset_has_learning_and_typed_execution_contract() -> None: payload = load_portfolio_infrastructure_asset_reconciliation() expected = ["KM", "RAG", "MCP", "PlayBook"] assert all(asset["learning_targets"] == expected for asset in payload["assets"]) assert all( asset["alerting"]["cross_domain_fallback_allowed"] is False for asset in payload["assets"] ) assert all(asset["verifier"] for asset in payload["assets"]) priorities = [item["priority"] for item in payload["reconciliation_work_items"]] assert priorities == ["P0"] * 10 + ["P1"] * 2 def test_loader_rejects_runtime_false_green(tmp_path: Path) -> None: payload = json.loads(SNAPSHOT.read_text(encoding="utf-8")) payload["source_inventory"]["live_truth_accepted"] = True (tmp_path / SNAPSHOT.name).write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(ValueError, match="cannot be live truth"): load_portfolio_infrastructure_asset_reconciliation(tmp_path) def test_loader_rejects_cross_domain_executor_drift(tmp_path: Path) -> None: payload = json.loads(SNAPSHOT.read_text(encoding="utf-8")) payload["asset_groups"][0]["common"]["executor"] = "generic_fallback" (tmp_path / SNAPSHOT.name).write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(ValueError, match="executor/domain mismatch"): load_portfolio_infrastructure_asset_reconciliation(tmp_path) def test_loader_rejects_runtime_closure_without_same_run_receipt( tmp_path: Path, ) -> None: payload = json.loads(SNAPSHOT.read_text(encoding="utf-8")) host110 = next( asset for group in payload["asset_groups"] for asset in group["members"] if asset["canonical_id"] == "service:ollama:host110" ) host110["runtime_closure"].pop("post_verifier") (tmp_path / SNAPSHOT.name).write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(ValueError, match="runtime closure missing"): load_portfolio_infrastructure_asset_reconciliation(tmp_path)