feat(governance): 新增 Status Cleanup Dashboard read model
This commit is contained in:
Your Name
@@ -73,6 +73,9 @@ from src.services.ai_agent_critic_reviewer_result_capture import (
|
||||
from src.services.ai_agent_deployment_layout import (
|
||||
load_latest_ai_agent_deployment_layout,
|
||||
)
|
||||
from src.services.awoooi_status_cleanup_dashboard import (
|
||||
load_latest_awoooi_status_cleanup_dashboard,
|
||||
)
|
||||
from src.services.ai_agent_failure_receipt_no_send_replay import (
|
||||
load_latest_ai_agent_failure_receipt_no_send_replay,
|
||||
)
|
||||
@@ -763,6 +766,36 @@ async def get_agent_deployment_layout() -> dict[str, Any]:
|
||||
) from exc
|
||||
|
||||
|
||||
@router.get(
|
||||
"/awoooi-status-cleanup-dashboard",
|
||||
response_model=dict[str, Any],
|
||||
summary="取得 AWOOOI 狀態清理儀表板",
|
||||
description=(
|
||||
"讀取最新已提交的 AWOOOI Status Cleanup Dashboard 只讀快照；"
|
||||
"此端點只呈現狀態清理完成度、owner gate、apply gate、artifact sync 與 Wazuh handoff 邊界。"
|
||||
"它不更新 project_current_status / memory、不同步 raw Codex App DB / auth / conversations / sessions、"
|
||||
"不呼叫 Wazuh live API、不執行 active response、不改主機、不修改 workflow / repo refs、"
|
||||
"不執行 backup / restore / migration。"
|
||||
),
|
||||
)
|
||||
async def get_awoooi_status_cleanup_dashboard() -> dict[str, Any]:
|
||||
"""回傳最新 AWOOOI 狀態清理儀表板只讀快照。"""
|
||||
try:
|
||||
payload = await asyncio.to_thread(load_latest_awoooi_status_cleanup_dashboard)
|
||||
return redact_public_lan_topology(payload)
|
||||
except FileNotFoundError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=str(exc),
|
||||
) from exc
|
||||
except (json.JSONDecodeError, ValueError) as exc:
|
||||
logger.error("awoooi_status_cleanup_dashboard_invalid", error=str(exc))
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail="AWOOOI 狀態清理儀表板快照無效",
|
||||
) from exc
|
||||
|
||||
|
||||
@router.get(
|
||||
"/agent-12-agent-war-room",
|
||||
response_model=dict[str, Any],
|
||||
|
||||
276
apps/api/src/services/awoooi_status_cleanup_dashboard.py
Normal file
276
apps/api/src/services/awoooi_status_cleanup_dashboard.py
Normal file
@@ -0,0 +1,276 @@
|
||||
"""
|
||||
AWOOOI Status Cleanup dashboard snapshot.
|
||||
|
||||
Loads the read-only dashboard that combines status cleanup preflight, owner
|
||||
review, owner response preflight, dry-run execution plan, apply gate, and
|
||||
artifact sync readback. This module never writes memory, syncs raw Codex app
|
||||
data, fetches repos, modifies workflows, queries Wazuh live APIs, or authorizes
|
||||
runtime writes.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from src.services.snapshot_paths import default_operations_dir
|
||||
|
||||
_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
|
||||
_SNAPSHOT_PATTERN = "awoooi-status-cleanup-dashboard.snapshot.json"
|
||||
_SCHEMA_VERSION = "awoooi_status_cleanup_dashboard_v1"
|
||||
|
||||
|
||||
def load_latest_awoooi_status_cleanup_dashboard(
|
||||
operations_dir: Path | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Load the committed AWOOOI Status Cleanup dashboard snapshot."""
|
||||
directory = operations_dir or _DEFAULT_OPERATIONS_DIR
|
||||
candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
|
||||
if not candidates:
|
||||
raise FileNotFoundError(f"no AWOOOI status cleanup dashboard found in {directory}")
|
||||
|
||||
latest = candidates[-1]
|
||||
with latest.open(encoding="utf-8") as handle:
|
||||
payload = json.load(handle)
|
||||
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError(f"{latest}: expected JSON object")
|
||||
_require_schema(payload, _SCHEMA_VERSION, str(latest))
|
||||
_require_read_only_boundaries(payload, str(latest))
|
||||
_require_summary(payload, str(latest))
|
||||
_require_cards(payload, str(latest))
|
||||
_require_risk_controls(payload, str(latest))
|
||||
_require_wazuh_boundary(payload, str(latest))
|
||||
_require_hard_gates(payload, str(latest))
|
||||
return payload
|
||||
|
||||
|
||||
def _require_schema(payload: dict[str, Any], expected: str, label: str) -> None:
|
||||
actual = payload.get("schema_version")
|
||||
if actual != expected:
|
||||
raise ValueError(f"{label}: expected schema_version={expected}, got {actual!r}")
|
||||
|
||||
|
||||
def _require_read_only_boundaries(payload: dict[str, Any], label: str) -> None:
|
||||
blocked_flags = {
|
||||
"secret_values_collected",
|
||||
"remote_write_performed",
|
||||
"local_product_write_performed",
|
||||
"execution_authorized",
|
||||
"memory_write_authorized",
|
||||
"wazuh_api_live_query_authorized",
|
||||
"runtime_execution_authorized",
|
||||
"ui_implementation_authorized",
|
||||
}
|
||||
invalid = sorted(flag for flag in blocked_flags if payload.get(flag) is not False)
|
||||
if invalid:
|
||||
raise ValueError(f"{label}: dashboard boundaries must remain false: {invalid}")
|
||||
|
||||
operation_boundaries = payload.get("operation_boundaries") or {}
|
||||
invalid_ops = sorted(key for key, value in operation_boundaries.items() if value is not False)
|
||||
if invalid_ops:
|
||||
raise ValueError(f"{label}: operation boundaries must remain false: {invalid_ops}")
|
||||
|
||||
|
||||
def _require_summary(payload: dict[str, Any], label: str) -> None:
|
||||
summary = payload.get("summary") or {}
|
||||
gates = payload.get("gate_cards") or []
|
||||
blockers = payload.get("blocking_reasons") or []
|
||||
actions = payload.get("next_actions") or []
|
||||
if payload.get("target_route") != "/workspace/status-cleanup":
|
||||
raise ValueError(f"{label}: target_route must remain /workspace/status-cleanup")
|
||||
if summary.get("dashboard_status") != "blocked_status_cleanup_apply_not_authorized":
|
||||
raise ValueError(f"{label}: dashboard_status must remain blocked")
|
||||
if summary.get("gate_count") != len(gates):
|
||||
raise ValueError(f"{label}: gate_count mismatch")
|
||||
if summary.get("blocked_gate_count") != sum(1 for item in gates if item.get("blocked") is True):
|
||||
raise ValueError(f"{label}: blocked_gate_count mismatch")
|
||||
if summary.get("blocking_reason_count") != len(blockers):
|
||||
raise ValueError(f"{label}: blocking_reason_count mismatch")
|
||||
if summary.get("next_action_count") != len(actions):
|
||||
raise ValueError(f"{label}: next_action_count mismatch")
|
||||
for flag in (
|
||||
"apply_allowed",
|
||||
"memory_write_authorized",
|
||||
"wazuh_api_live_query_authorized",
|
||||
"runtime_execution_authorized",
|
||||
"ui_implementation_allowed",
|
||||
):
|
||||
if summary.get(flag) is not False:
|
||||
raise ValueError(f"{label}: summary.{flag} must remain false")
|
||||
if summary.get("gate_count") != 5:
|
||||
raise ValueError(f"{label}: dashboard must contain five status cleanup gates")
|
||||
if summary.get("wazuh_handoff_status") != "blocked_not_released":
|
||||
raise ValueError(f"{label}: Wazuh handoff status must remain blocked")
|
||||
if summary.get("wazuh_handoff_base_commit") != "b540fc0c":
|
||||
raise ValueError(f"{label}: Wazuh handoff base must remain b540fc0c")
|
||||
if summary.get("wazuh_handoff_commit_count") != 7:
|
||||
raise ValueError(f"{label}: Wazuh handoff commit count must be 7")
|
||||
if summary.get("wazuh_handoff_patch_count") != 7:
|
||||
raise ValueError(f"{label}: Wazuh handoff patch count must be 7")
|
||||
zero_wazuh_summary = (
|
||||
"wazuh_release_owner_request_sent_count",
|
||||
"wazuh_release_owner_response_accepted_count",
|
||||
"wazuh_acknowledged_release_owner_ack_count",
|
||||
"wazuh_accepted_evidence_count",
|
||||
"wazuh_live_metadata_owner_count",
|
||||
"wazuh_secret_metadata_count",
|
||||
"wazuh_live_agent_registry_readback",
|
||||
"iwooos_wazuh_runtime_gate",
|
||||
"wazuh_active_response_count",
|
||||
"wazuh_agent_visibility_runtime_gate_count",
|
||||
"wazuh_push_gate_count",
|
||||
"wazuh_deploy_gate_count",
|
||||
"wazuh_readback_gate_count",
|
||||
"wazuh_runtime_gate_count",
|
||||
)
|
||||
non_zero_wazuh = sorted(key for key in zero_wazuh_summary if summary.get(key) != 0)
|
||||
if non_zero_wazuh:
|
||||
raise ValueError(f"{label}: Wazuh release summary gates must remain zero: {non_zero_wazuh}")
|
||||
|
||||
|
||||
def _require_cards(payload: dict[str, Any], label: str) -> None:
|
||||
gate_ids = {item.get("gate_id") for item in payload.get("gate_cards") or []}
|
||||
required_gates = {
|
||||
"status_cleanup_preflight",
|
||||
"owner_review_package",
|
||||
"owner_response_preflight",
|
||||
"execution_plan",
|
||||
"apply_gate",
|
||||
}
|
||||
missing_gates = sorted(required_gates - gate_ids)
|
||||
if missing_gates:
|
||||
raise ValueError(f"{label}: gate_cards missing required gates: {missing_gates}")
|
||||
|
||||
metric_ids = {item.get("metric_id") for item in payload.get("metric_cards") or []}
|
||||
required_metrics = {
|
||||
"overall_completion",
|
||||
"project_status_age",
|
||||
"owner_flags",
|
||||
"approved_sections",
|
||||
"execution_preview",
|
||||
"apply_confirmation",
|
||||
"artifact_sync",
|
||||
}
|
||||
missing_metrics = sorted(required_metrics - metric_ids)
|
||||
if missing_metrics:
|
||||
raise ValueError(f"{label}: metric_cards missing required metrics: {missing_metrics}")
|
||||
|
||||
unsafe = [
|
||||
item.get("gate_id")
|
||||
for item in payload.get("gate_cards") or []
|
||||
if item.get("memory_write_authorized") is not False
|
||||
or item.get("execution_authorized") is not False
|
||||
]
|
||||
if unsafe:
|
||||
raise ValueError(f"{label}: gate cards must remain read-only: {sorted(unsafe)}")
|
||||
|
||||
|
||||
def _require_risk_controls(payload: dict[str, Any], label: str) -> None:
|
||||
control_ids = {item.get("control_id") for item in payload.get("risk_controls") or []}
|
||||
required = {
|
||||
"memory_write",
|
||||
"raw_codex_history",
|
||||
"wazuh_runtime",
|
||||
"workflow",
|
||||
"repo_refs",
|
||||
"host",
|
||||
"backup_restore",
|
||||
"ui",
|
||||
}
|
||||
missing = sorted(required - control_ids)
|
||||
if missing:
|
||||
raise ValueError(f"{label}: risk_controls missing controls: {missing}")
|
||||
unsafe = sorted(
|
||||
item.get("control_id")
|
||||
for item in payload.get("risk_controls") or []
|
||||
if item.get("authorized") is not False
|
||||
)
|
||||
if unsafe:
|
||||
raise ValueError(f"{label}: risk controls must not authorize actions: {unsafe}")
|
||||
|
||||
|
||||
def _require_wazuh_boundary(payload: dict[str, Any], label: str) -> None:
|
||||
handoff = payload.get("wazuh_handoff") or {}
|
||||
if handoff.get("status") != "blocked_not_released":
|
||||
raise ValueError(f"{label}: Wazuh handoff status must remain blocked_not_released")
|
||||
if handoff.get("base_commit") != "b540fc0c":
|
||||
raise ValueError(f"{label}: Wazuh handoff base must remain b540fc0c")
|
||||
if handoff.get("commit_count") != 7:
|
||||
raise ValueError(f"{label}: Wazuh handoff commit_count must be 7")
|
||||
if handoff.get("patch_count") != 7:
|
||||
raise ValueError(f"{label}: Wazuh handoff patch_count must be 7")
|
||||
zero_handoff_keys = (
|
||||
"release_owner_request_sent_count",
|
||||
"release_owner_response_accepted_count",
|
||||
"acknowledged_release_owner_ack_count",
|
||||
"accepted_evidence_count",
|
||||
"live_metadata_owner_count",
|
||||
"secret_metadata_count",
|
||||
"wazuh_live_agent_registry_readback",
|
||||
"iwooos_wazuh_runtime_gate",
|
||||
"active_response",
|
||||
"push_gate_count",
|
||||
"deploy_gate_count",
|
||||
"readback_gate_count",
|
||||
"runtime_gate_count",
|
||||
)
|
||||
non_zero_handoff = sorted(key for key in zero_handoff_keys if handoff.get(key) != 0)
|
||||
if non_zero_handoff:
|
||||
raise ValueError(f"{label}: Wazuh handoff gates must remain zero: {non_zero_handoff}")
|
||||
if handoff.get("runtime_execution_authorized") is not False:
|
||||
raise ValueError(f"{label}: Wazuh runtime execution must remain false")
|
||||
if handoff.get("wazuh_api_live_query_authorized") is not False:
|
||||
raise ValueError(f"{label}: Wazuh live query must remain false")
|
||||
if handoff.get("stored_api_secret_metadata_changed") is not False:
|
||||
raise ValueError(f"{label}: Wazuh stored API secret metadata must remain unchanged")
|
||||
if handoff.get("agent_visibility_status") != "blocked_waiting_manager_agent_registry_readback":
|
||||
raise ValueError(f"{label}: Wazuh agent visibility must remain blocked")
|
||||
visibility_false = (
|
||||
"manager_agent_registry_readback_passed",
|
||||
"iwooos_live_route_readback_passed",
|
||||
"dashboard_agent_list_recovered",
|
||||
)
|
||||
invalid_visibility = sorted(key for key in visibility_false if handoff.get(key) is not False)
|
||||
if invalid_visibility:
|
||||
raise ValueError(f"{label}: Wazuh visibility flags must remain false: {invalid_visibility}")
|
||||
if handoff.get("agent_visibility_runtime_gate_count") != 0:
|
||||
raise ValueError(f"{label}: Wazuh visibility runtime gate count must remain zero")
|
||||
boundary = str(handoff.get("boundary", ""))
|
||||
for token in (
|
||||
"base=b540fc0c",
|
||||
"38dc3c2f",
|
||||
"9a53d3e1",
|
||||
"e9972d47",
|
||||
"758d419e",
|
||||
"04db4b8a",
|
||||
"8eec298e",
|
||||
"325f262a",
|
||||
"release_lane_preflight=ready0_acks0of6_evidence0of6_push0_deploy0_readback0_runtime0",
|
||||
"owner_gate=request_sent0_response_accepted0_acks0of6_evidence0of6_push0_deploy0_readback0_runtime0",
|
||||
"live_metadata_env_gate=owner0_secret_metadata0_push0_deploy0_readback0_runtime0",
|
||||
"wazuh_live_agent_registry_readback=0",
|
||||
"iwooos_wazuh_runtime_gate=0",
|
||||
"active_response=0",
|
||||
"manager_agent_registry_readback_passed=false",
|
||||
"iwooos_live_route_readback_passed=false",
|
||||
"dashboard_agent_list_recovered=false",
|
||||
"push_blocked",
|
||||
):
|
||||
if token not in boundary:
|
||||
raise ValueError(f"{label}: Wazuh handoff boundary missing {token}")
|
||||
|
||||
|
||||
def _require_hard_gates(payload: dict[str, Any], label: str) -> None:
|
||||
gate_text = "\n".join(payload.get("hard_gates") or [])
|
||||
for token in (
|
||||
"project_current_status",
|
||||
"memory_write_authorized=false",
|
||||
"raw Codex App",
|
||||
"Wazuh",
|
||||
".gitea/workflows",
|
||||
"apps/web",
|
||||
):
|
||||
if token not in gate_text:
|
||||
raise ValueError(f"{label}: hard_gates missing {token}")
|
||||
@@ -30,3 +30,8 @@ def resolve_repo_root(anchor: Path) -> Path:
|
||||
def default_evaluations_dir(anchor: Path) -> Path:
|
||||
"""Resolve the default committed evaluations snapshot directory."""
|
||||
return resolve_repo_root(anchor) / "docs" / "evaluations"
|
||||
|
||||
|
||||
def default_operations_dir(anchor: Path) -> Path:
|
||||
"""Resolve the default committed operations snapshot directory."""
|
||||
return resolve_repo_root(anchor) / "docs" / "operations"
|
||||
|
||||
112
apps/api/tests/test_awoooi_status_cleanup_dashboard.py
Normal file
112
apps/api/tests/test_awoooi_status_cleanup_dashboard.py
Normal file
@@ -0,0 +1,112 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from src.services.awoooi_status_cleanup_dashboard import (
|
||||
load_latest_awoooi_status_cleanup_dashboard,
|
||||
)
|
||||
|
||||
|
||||
def test_load_latest_awoooi_status_cleanup_dashboard_reads_committed_snapshot():
|
||||
data = load_latest_awoooi_status_cleanup_dashboard()
|
||||
|
||||
assert data["schema_version"] == "awoooi_status_cleanup_dashboard_v1"
|
||||
assert data["summary"]["dashboard_status"] == "blocked_status_cleanup_apply_not_authorized"
|
||||
assert data["summary"]["gate_count"] == 5
|
||||
assert data["summary"]["blocked_gate_count"] == 5
|
||||
assert data["summary"]["accepted_owner_flag_count"] == 0
|
||||
assert data["summary"]["required_owner_flag_count"] == 6
|
||||
assert data["summary"]["apply_allowed"] is False
|
||||
assert data["summary"]["memory_write_authorized"] is False
|
||||
assert data["summary"]["wazuh_api_live_query_authorized"] is False
|
||||
assert data["summary"]["runtime_execution_authorized"] is False
|
||||
assert data["summary"]["wazuh_handoff_status"] == "blocked_not_released"
|
||||
assert data["summary"]["wazuh_handoff_base_commit"] == "b540fc0c"
|
||||
assert data["summary"]["wazuh_handoff_commit_count"] == 7
|
||||
assert data["summary"]["wazuh_handoff_patch_count"] == 7
|
||||
assert data["summary"]["wazuh_live_metadata_owner_count"] == 0
|
||||
assert data["summary"]["wazuh_secret_metadata_count"] == 0
|
||||
assert data["summary"]["wazuh_live_agent_registry_readback"] == 0
|
||||
assert data["summary"]["iwooos_wazuh_runtime_gate"] == 0
|
||||
assert data["summary"]["wazuh_active_response_count"] == 0
|
||||
assert data["summary"]["wazuh_agent_visibility_status"] == "blocked_waiting_manager_agent_registry_readback"
|
||||
assert data["summary"]["wazuh_manager_agent_registry_readback_passed"] is False
|
||||
assert data["summary"]["wazuh_iwooos_live_route_readback_passed"] is False
|
||||
assert data["summary"]["wazuh_dashboard_agent_list_recovered"] is False
|
||||
assert data["summary"]["wazuh_agent_visibility_runtime_gate_count"] == 0
|
||||
assert data["summary"]["wazuh_runtime_gate_count"] == 0
|
||||
assert data["memory_write_authorized"] is False
|
||||
assert data["runtime_execution_authorized"] is False
|
||||
assert data["wazuh_handoff"]["wazuh_api_live_query_authorized"] is False
|
||||
assert data["wazuh_handoff"]["base_commit"] == "b540fc0c"
|
||||
assert data["wazuh_handoff"]["patch_count"] == 7
|
||||
assert data["wazuh_handoff"]["live_metadata_owner_count"] == 0
|
||||
assert data["wazuh_handoff"]["wazuh_live_agent_registry_readback"] == 0
|
||||
assert data["wazuh_handoff"]["active_response"] == 0
|
||||
assert data["wazuh_handoff"]["agent_visibility_status"] == "blocked_waiting_manager_agent_registry_readback"
|
||||
assert data["wazuh_handoff"]["manager_agent_registry_readback_passed"] is False
|
||||
assert "base=b540fc0c" in data["wazuh_handoff"]["boundary"]
|
||||
assert "release_lane_preflight=ready0_acks0of6_evidence0of6_push0_deploy0_readback0_runtime0" in data["wazuh_handoff"]["boundary"]
|
||||
assert "owner_gate=request_sent0_response_accepted0_acks0of6_evidence0of6_push0_deploy0_readback0_runtime0" in data["wazuh_handoff"]["boundary"]
|
||||
assert "live_metadata_env_gate=owner0_secret_metadata0_push0_deploy0_readback0_runtime0" in data["wazuh_handoff"]["boundary"]
|
||||
assert "wazuh_live_agent_registry_readback=0" in data["wazuh_handoff"]["boundary"]
|
||||
assert "manager_agent_registry_readback_passed=false" in data["wazuh_handoff"]["boundary"]
|
||||
assert {item["gate_id"] for item in data["gate_cards"]} >= {
|
||||
"status_cleanup_preflight",
|
||||
"owner_review_package",
|
||||
"owner_response_preflight",
|
||||
"execution_plan",
|
||||
"apply_gate",
|
||||
}
|
||||
|
||||
|
||||
def test_awoooi_status_cleanup_dashboard_rejects_memory_write_authorization(tmp_path):
|
||||
snapshot = _snapshot()
|
||||
snapshot["memory_write_authorized"] = True
|
||||
_write_snapshot(tmp_path, snapshot)
|
||||
|
||||
with pytest.raises(ValueError, match="dashboard boundaries"):
|
||||
load_latest_awoooi_status_cleanup_dashboard(tmp_path)
|
||||
|
||||
|
||||
def test_awoooi_status_cleanup_dashboard_rejects_wazuh_live_query(tmp_path):
|
||||
snapshot = _snapshot()
|
||||
snapshot["wazuh_handoff"]["wazuh_api_live_query_authorized"] = True
|
||||
_write_snapshot(tmp_path, snapshot)
|
||||
|
||||
with pytest.raises(ValueError, match="Wazuh live query"):
|
||||
load_latest_awoooi_status_cleanup_dashboard(tmp_path)
|
||||
|
||||
|
||||
def test_awoooi_status_cleanup_dashboard_rejects_live_metadata_gate(tmp_path):
|
||||
snapshot = _snapshot()
|
||||
snapshot["summary"]["wazuh_live_metadata_owner_count"] = 1
|
||||
_write_snapshot(tmp_path, snapshot)
|
||||
|
||||
with pytest.raises(ValueError, match="Wazuh release summary gates"):
|
||||
load_latest_awoooi_status_cleanup_dashboard(tmp_path)
|
||||
|
||||
|
||||
def test_awoooi_status_cleanup_dashboard_rejects_missing_risk_control(tmp_path):
|
||||
snapshot = _snapshot()
|
||||
snapshot["risk_controls"] = [
|
||||
item for item in snapshot["risk_controls"] if item["control_id"] != "host"
|
||||
]
|
||||
_write_snapshot(tmp_path, snapshot)
|
||||
|
||||
with pytest.raises(ValueError, match="risk_controls"):
|
||||
load_latest_awoooi_status_cleanup_dashboard(tmp_path)
|
||||
|
||||
|
||||
def _snapshot() -> dict:
|
||||
return copy.deepcopy(load_latest_awoooi_status_cleanup_dashboard())
|
||||
|
||||
|
||||
def _write_snapshot(tmp_path, snapshot: dict) -> None:
|
||||
(tmp_path / "awoooi-status-cleanup-dashboard.snapshot.json").write_text(
|
||||
json.dumps(snapshot),
|
||||
encoding="utf-8",
|
||||
)
|
||||
25
apps/api/tests/test_awoooi_status_cleanup_dashboard_api.py
Normal file
25
apps/api/tests/test_awoooi_status_cleanup_dashboard_api.py
Normal file
@@ -0,0 +1,25 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from src.api.v1.agents import router
|
||||
|
||||
|
||||
def test_awoooi_status_cleanup_dashboard_endpoint_returns_snapshot():
|
||||
app = FastAPI()
|
||||
app.include_router(router, prefix="/api/v1")
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/api/v1/agents/awoooi-status-cleanup-dashboard")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["schema_version"] == "awoooi_status_cleanup_dashboard_v1"
|
||||
assert data["summary"]["dashboard_status"] == "blocked_status_cleanup_apply_not_authorized"
|
||||
assert data["summary"]["gate_count"] == 5
|
||||
assert data["summary"]["apply_allowed"] is False
|
||||
assert data["summary"]["memory_write_authorized"] is False
|
||||
assert data["summary"]["wazuh_api_live_query_authorized"] is False
|
||||
assert data["memory_write_authorized"] is False
|
||||
assert data["runtime_execution_authorized"] is False
|
||||
Reference in New Issue
Block a user