from __future__ import annotations import os 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_control_coverage import ( load_latest_iwooos_security_control_coverage, ) def _client() -> TestClient: app = FastAPI() app.include_router(router) return TestClient(app) def test_iwooos_security_control_coverage_rolls_up_core_scopes() -> None: payload = load_latest_iwooos_security_control_coverage() assert payload["schema_version"] == "iwooos_security_control_coverage_v1" assert payload["status"] == "committed_scope_rollup_ready_with_controlled_apply_exception" assert payload["summary"]["source_snapshot_count"] == 8 assert payload["summary"]["control_domain_count"] == 8 assert payload["summary"]["visible_scope_unit_count"] == 160 assert payload["summary"]["asset_group_count"] == 16 assert payload["summary"]["host_service_surface_count"] == 9 assert payload["summary"]["monitoring_surface_count"] == 60 assert payload["summary"]["ssh_network_surface_count"] == 16 assert payload["summary"]["runtime_surface_count"] == 22 assert payload["summary"]["wazuh_expected_host_scope_count"] == 6 assert payload["summary"]["agent_bounty_product_surface_count"] == 7 assert payload["summary"]["ai_agent_asset_count"] == 24 assert payload["summary"]["professional_review_finding_count"] == 6 assert payload["summary"]["professional_review_p0_work_item_count"] == 12 assert payload["summary"]["external_security_tool_track_count"] == 8 assert payload["summary"]["ai_security_automation_stage_count"] == 7 assert payload["summary"]["full_scope_contract_count"] == 8 assert payload["summary"]["ui_ux_control_room_requirement_count"] == 6 domain_ids = {domain["domain_id"] for domain in payload["domains"]} assert domain_ids == { "high_value_asset_control", "host_service_runtime", "monitoring_alerting_observability", "ssh_firewall_network_access", "awoooi_runtime_surfaces", "wazuh_managed_host_coverage", "agent_bounty_protocol", "ai_agent_automation", } def test_iwooos_security_control_coverage_exposes_professional_review() -> None: payload = load_latest_iwooos_security_control_coverage() review = payload["professional_review"] assert review["schema_version"] == "iwooos_security_governance_professional_review_v1" assert review["completion_diagnosis"]["diagnosis"] == ( "control_plane_exists_runtime_security_closure_not_complete" ) assert review["completion_diagnosis"]["actual_runtime_acceptance_percent"] == 0 assert review["root_cause_summary"][0]["finding_id"] == "F01" assert review["root_cause_summary"][0]["severity"] == "P0" assert {scope["scope_id"] for scope in review["scope_contract"]} == { "hosts", "services", "products", "websites", "tools", "packages", "ai_agents", "runtime_actions", } assert [item["priority"] for item in review["priority_work_items"][:3]] == [ "P0-01", "P0-02", "P0-03", ] assert {tool["tool_id"] for tool in review["security_tool_integration_matrix"]} >= { "wazuh", "trivy", "syft_grype", "openssf_scorecard", "kyverno_cosign", "falco", } assert review["ai_automation_closure_loop"][0]["stage_id"] == "collect" assert all(stage["runtime_write"] is False for stage in review["ai_automation_closure_loop"]) def test_iwooos_security_control_coverage_keeps_runtime_gates_closed() -> None: payload = load_latest_iwooos_security_control_coverage() summary = payload["summary"] assert summary["actual_runtime_acceptance_percent"] == 0 assert summary["runtime_gate_count"] == 0 assert summary["owner_response_received_count"] == 0 assert summary["owner_response_accepted_count"] == 0 assert summary["live_evidence_accepted_count"] == 0 assert summary["wazuh_manager_registry_accepted_count"] == 6 assert summary["active_scan_authorized_count"] == 0 assert summary["active_response_authorized_count"] == 0 assert summary["telegram_send_authorized_count"] == 0 assert summary["host_write_authorized_count"] == 0 assert summary["secret_value_collected_count"] == 0 assert summary["agent_bounty_runtime_gate_open_count"] == 0 assert summary["ai_agent_runtime_write_gate_open_count"] == 0 assert summary["all_scope_runtime_controlled"] is False assert summary["allowlisted_controlled_apply_bypasses_iwooos_ledger"] is False assert summary["allowlisted_controlled_apply_requires_iwooos_ledger_receipt"] is True assert summary["controlled_apply_ledger_contract"] == ( "same_run_iwooos_aisoc_trace_required" ) assert ( summary["controlled_apply_policy"] == "low_medium_high_allowed_after_allowlist_check_mode_rollback_verifier_km" ) assert summary["critical_break_glass_required"] is True assert all( domain["accepted_count"] == (6 if domain["domain_id"] == "wazuh_managed_host_coverage" else 0) for domain in payload["domains"] ) def test_iwooos_security_control_coverage_api_is_public_safe() -> None: response = _client().get("/api/v1/iwooos/security-control-coverage") assert response.status_code == 200 data = response.json() assert data["schema_version"] == "iwooos_security_control_coverage_v1" assert data["summary"]["runtime_gate_count"] == 0 assert data["summary"]["visible_scope_unit_count"] == 160 assert data["professional_review"]["schema_version"] == ( "iwooos_security_governance_professional_review_v1" ) assert any(action["priority"] == "P0-01" for action in data["p0_next_actions"]) assert "192.168.0." not in response.text assert "runtime_control_blocked" not in response.text assert "工作視窗" not in response.text assert "批准！繼續" not in response.text def test_iwooos_security_control_coverage_p0_actions_are_executable_contracts() -> None: payload = load_latest_iwooos_security_control_coverage() actions = payload["p0_next_actions"] assert len(actions) == 12 assert all(action["work_item_id"].startswith("IWOOOS-SEC-P0-") for action in actions) assert all(action["requires_iwooos_ledger_receipt"] is True for action in actions) assert all(action["next_executable_action"] for action in actions) assert all(action["asset_scope_ids"] for action in actions) assert all( action["runtime_credit_policy"] == "no_credit_until_same_run_iwooos_aisoc_receipts_exist" for action in actions ) assert all( "bounded_execution_receipt" in action["required_controlled_apply_stages"] for action in actions )