awoooi/apps/api/tests/test_iwooos_security_tool_closure_readback.py

from __future__ import annotations

from fastapi import FastAPI
from fastapi.testclient import TestClient

from src.api.v1.iwooos import router
from src.services.iwooos_security_tool_closure_readback import (
    load_latest_iwooos_security_tool_closure_readback,
)


def _client() -> TestClient:
    app = FastAPI()
    app.include_router(router)
    return TestClient(app)


def test_iwooos_security_tool_closure_readback_rolls_up_tool_tracks() -> None:
    payload = load_latest_iwooos_security_tool_closure_readback()

    assert payload["schema_version"] == "iwooos_security_tool_closure_readback_v1"
    assert payload["status"] == "tool_closure_readback_ready_runtime_actions_closed"
    assert payload["summary"]["source_readback_count"] == 7
    assert payload["summary"]["tool_track_count"] == 8
    assert payload["summary"]["p0_tool_track_count"] == 3
    assert payload["summary"]["wazuh_expected_host_scope_count"] == 6
    assert payload["summary"]["wazuh_manager_registry_accepted_count"] == 6
    assert payload["summary"]["wazuh_registry_complete_count"] == 1
    assert payload["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 0
    assert payload["summary"]["supply_chain_scan_closed_count"] == 0
    assert payload["summary"]["runtime_detection_closed_count"] == 0
    assert payload["summary"]["web_api_dast_closed_count"] == 0
    assert payload["summary"]["runtime_gate_count"] == 0

    track_ids = {track["track_id"] for track in payload["tool_tracks"]}
    assert track_ids == {
        "wazuh_detection_response",
        "supply_chain_sbom_scan",
        "source_control_scorecard",
        "k8s_policy_attestation",
        "runtime_threat_detection",
        "web_api_dast",
        "normalized_detection_schema",
        "ai_agent_controlled_apply",
    }


def test_iwooos_security_tool_closure_readback_keeps_dangerous_boundaries_closed() -> None:
    payload = load_latest_iwooos_security_tool_closure_readback()

    assert payload["boundaries"]["runtime_execution_authorized"] is False
    assert payload["boundaries"]["runtime_gate_open"] is False
    assert payload["boundaries"]["host_write_authorized"] is False
    assert payload["boundaries"]["wazuh_active_response_authorized"] is False
    assert payload["boundaries"]["secret_value_collection_allowed"] is False
    assert payload["boundaries"]["not_authorization"] is True
    assert all(track["runtime_gate_open"] is False for track in payload["tool_tracks"])
    assert all(track["active_response_authorized"] is False for track in payload["tool_tracks"])
    assert all(track["host_write_authorized"] is False for track in payload["tool_tracks"])
    assert all(
        track["secret_value_collection_allowed"] is False
        for track in payload["tool_tracks"]
    )


def test_iwooos_security_tool_closure_readback_marks_partial_vs_missing_tracks() -> None:
    payload = load_latest_iwooos_security_tool_closure_readback()
    tracks = {track["track_id"]: track for track in payload["tool_tracks"]}

    assert tracks["wazuh_detection_response"]["closure_state"] == (
        "partial_readback_missing_verifier"
    )
    assert tracks["wazuh_detection_response"]["ready_signal_count"] == 1
    assert tracks["wazuh_detection_response"]["missing_signal_count"] == 5
    assert tracks["supply_chain_sbom_scan"]["closure_state"] == "required_not_closed"
    assert tracks["runtime_threat_detection"]["tool_ids"] == ["falco"]
    assert tracks["web_api_dast"]["tool_ids"] == ["owasp_zap_or_equivalent_dast"]
    assert tracks["ai_agent_controlled_apply"]["ready_signal_count"] > 0
    assert tracks["ai_agent_controlled_apply"]["missing_signal_count"] >= 0


def test_iwooos_security_tool_closure_readback_accepts_alert_receipt_signal() -> None:
    payload = load_latest_iwooos_security_tool_closure_readback(
        alert_receipt_readback={
            "events": [{"provider_event_id": "alertmanager:received:tool-closure"}],
            "total": 1,
            "limit": 20,
            "source_status": "ready",
        }
    )
    tracks = {track["track_id"]: track for track in payload["tool_tracks"]}

    assert payload["summary"]["alert_receipt_accepted_count"] == 1
    assert payload["summary"]["alert_receipt_observed_count"] == 1
    assert tracks["wazuh_detection_response"]["ready_signal_count"] == 2
    assert tracks["normalized_detection_schema"]["ready_signal_count"] == 1
    assert payload["summary"]["runtime_gate_count"] == 0


def test_iwooos_security_tool_closure_readback_api_is_public_safe(monkeypatch) -> None:
    async def fake_recent_events(**_: object) -> dict[str, object]:
        return {
            "events": [{"provider_event_id": "alertmanager:received:tool-closure"}],
            "total": 1,
            "limit": 20,
            "source_status": "ready",
        }

    monkeypatch.setattr(
        "src.api.v1.iwooos.list_recent_channel_events",
        fake_recent_events,
    )

    response = _client().get("/api/v1/iwooos/security-tool-closure-readback")

    assert response.status_code == 200
    data = response.json()
    assert data["schema_version"] == "iwooos_security_tool_closure_readback_v1"
    assert data["summary"]["tool_track_count"] == 8
    assert data["summary"]["alert_receipt_accepted_count"] == 1
    assert data["summary"]["runtime_gate_count"] == 0
    assert any(
        item["track_id"] == "supply_chain_sbom_scan"
        for item in data["tool_tracks"]
    )
    assert "192.168.0." not in response.text
    assert "工作視窗" not in response.text
    assert "批准！繼續" not in response.text
    assert "WAZUH_API_PASSWORD" not in response.text