302 lines
12 KiB
Python
302 lines
12 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
|
||
import pytest
|
||
|
||
from src.services.service_health_gap_matrix import load_latest_service_health_gap_matrix
|
||
|
||
|
||
def test_load_latest_service_health_gap_matrix_reads_newest_file(tmp_path):
|
||
older = _snapshot(generated_at="2026-06-04T00:00:00+08:00", completion=40)
|
||
newer = _snapshot(generated_at="2026-06-05T00:00:00+08:00", completion=100)
|
||
(tmp_path / "service_health_gap_matrix_2026-06-04.json").write_text(
|
||
json.dumps(older),
|
||
encoding="utf-8",
|
||
)
|
||
(tmp_path / "service_health_gap_matrix_2026-06-05.json").write_text(
|
||
json.dumps(newer),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
loaded = load_latest_service_health_gap_matrix(tmp_path)
|
||
|
||
assert loaded["generated_at"] == "2026-06-05T00:00:00+08:00"
|
||
assert loaded["program_status"]["overall_completion_percent"] == 100
|
||
assert loaded["rollups"]["total_targets"] == 7
|
||
assert loaded["operation_boundaries"]["active_probe_allowed"] is False
|
||
|
||
|
||
def test_service_health_gap_matrix_requires_read_only_mode(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["program_status"]["read_only_mode"] = False
|
||
(tmp_path / "service_health_gap_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="read_only_mode"):
|
||
load_latest_service_health_gap_matrix(tmp_path)
|
||
|
||
|
||
def test_service_health_gap_matrix_blocks_probe_restart_endpoint_and_notifications(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["operation_boundaries"]["active_probe_allowed"] = True
|
||
snapshot["operation_boundaries"]["service_restart_allowed"] = True
|
||
snapshot["operation_boundaries"]["endpoint_change_allowed"] = True
|
||
snapshot["operation_boundaries"]["notification_send_allowed"] = True
|
||
(tmp_path / "service_health_gap_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="operation boundaries"):
|
||
load_latest_service_health_gap_matrix(tmp_path)
|
||
|
||
|
||
def test_service_health_gap_matrix_requires_rollup_consistency(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["rollups"]["target_ids_requiring_action"] = []
|
||
(tmp_path / "service_health_gap_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="target_ids_requiring_action"):
|
||
load_latest_service_health_gap_matrix(tmp_path)
|
||
|
||
|
||
def test_service_health_gap_matrix_requires_required_gaps_and_stale_endpoints(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["health_gaps"] = snapshot["health_gaps"][:-1]
|
||
snapshot["stale_endpoints"] = snapshot["stale_endpoints"][:-1]
|
||
_refresh_rollups(snapshot)
|
||
(tmp_path / "service_health_gap_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="missing required health gaps"):
|
||
load_latest_service_health_gap_matrix(tmp_path)
|
||
|
||
|
||
def test_service_health_gap_matrix_requires_operator_denials(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["operator_contract"]["must_not_interpret_as"].remove("active probe 批准")
|
||
(tmp_path / "service_health_gap_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="operator_contract"):
|
||
load_latest_service_health_gap_matrix(tmp_path)
|
||
|
||
|
||
def test_service_health_gap_matrix_requires_endpoint_and_notification_policies(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["operator_contract"]["endpoint_policy"] = "可修改端點"
|
||
(tmp_path / "service_health_gap_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="endpoint_policy"):
|
||
load_latest_service_health_gap_matrix(tmp_path)
|
||
|
||
|
||
def test_service_health_gap_matrix_rejects_secret_payload_keys(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["latest_observations"][0]["authorization_header"] = "redacted"
|
||
(tmp_path / "service_health_gap_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="forbidden secret payload key"):
|
||
load_latest_service_health_gap_matrix(tmp_path)
|
||
|
||
|
||
def test_service_health_gap_matrix_fails_when_missing(tmp_path):
|
||
with pytest.raises(FileNotFoundError):
|
||
load_latest_service_health_gap_matrix(tmp_path)
|
||
|
||
|
||
def _snapshot(
|
||
*,
|
||
generated_at: str = "2026-06-05T00:00:00+08:00",
|
||
completion: int = 100,
|
||
) -> dict:
|
||
targets = [
|
||
_target("production_api_health_public", "api_health", "verified", "critical", "fresh_readback"),
|
||
_target("awoooi_web_health_manifest", "web_health", "verified", "high", "manifest_mapped"),
|
||
_target("ollama_three_layer_health_contract", "ai_provider_health", "action_required", "critical", "source_mismatch"),
|
||
_target("openclaw_health_endpoint_contract", "ai_provider_health", "action_required", "critical", "source_mismatch"),
|
||
_target("prometheus_alertmanager_endpoint_reference", "observability_health", "action_required", "high", "source_mismatch"),
|
||
_target("gitea_workflow_runner_health_contract", "devops_health", "action_required", "critical", "stale_evidence"),
|
||
_target("kali_scanner_health_reference", "security_edge_health", "action_required", "medium", "manifest_only"),
|
||
]
|
||
gaps = [
|
||
_gap("endpoint_reference_stale_hosts", ["ollama_three_layer_health_contract"]),
|
||
_gap("gitea_runner_attestation_health_gap", ["gitea_workflow_runner_health_contract"]),
|
||
_gap("health_check_script_not_authoritative", ["production_api_health_public"]),
|
||
_gap("openclaw_nemo_rca_health_review", ["openclaw_health_endpoint_contract"]),
|
||
_gap("security_scanner_health_evidence_gap", ["kali_scanner_health_reference"]),
|
||
]
|
||
endpoints = [
|
||
_endpoint("legacy_188_ollama_provider_endpoint"),
|
||
_endpoint("prometheus_alertmanager_110_188_split"),
|
||
_endpoint("openclaw_8088_comment_vs_8089_contract"),
|
||
]
|
||
payload = {
|
||
"schema_version": "service_health_gap_matrix_v1",
|
||
"generated_at": generated_at,
|
||
"program_status": {
|
||
"overall_completion_percent": completion,
|
||
"current_priority": "P1",
|
||
"current_task_id": "P1-005",
|
||
"next_task_id": "P1-006",
|
||
"read_only_mode": True,
|
||
},
|
||
"source_refs": ["docs/schemas/service_health_gap_matrix_v1.schema.json"],
|
||
"rollups": {},
|
||
"service_health_targets": targets,
|
||
"health_gaps": gaps,
|
||
"stale_endpoints": endpoints,
|
||
"latest_observations": [
|
||
{
|
||
"observation_id": "service_health_gap_matrix_seed",
|
||
"status": "verified",
|
||
"summary": "只讀 health gap seed。",
|
||
"evidence_refs": ["docs/reference/SERVICE-ENDPOINTS.md"],
|
||
}
|
||
],
|
||
"operator_contract": {
|
||
"display_mode": "read_only_service_health_gap_matrix",
|
||
"must_not_interpret_as": [
|
||
"服務重啟批准",
|
||
"endpoint 修改批准",
|
||
"active probe 批准",
|
||
"live health check 已執行",
|
||
"Secret payload 已讀取或可輸出",
|
||
"Telegram 成功通知批准",
|
||
"workflow / deploy / reload 觸發批准",
|
||
"provider 切換批准",
|
||
"runtime execution 授權",
|
||
"OpenClaw 取代或降級批准",
|
||
],
|
||
"secret_display_policy": "只顯示來源檔案與 env var 名稱。",
|
||
"restart_policy": "P1-005 只能列缺口；任何 restart 都需另行批准。",
|
||
"endpoint_policy": "P1-005 不修改端點，只提出 source truth cleanup。",
|
||
"notification_policy": "成功 smoke 不通知；失敗或 stale 才進 P1-007 通知合約。",
|
||
},
|
||
"operation_boundaries": {
|
||
"read_only_api_allowed": True,
|
||
"service_restart_allowed": False,
|
||
"pod_restart_allowed": False,
|
||
"host_restart_allowed": False,
|
||
"rollout_restart_allowed": False,
|
||
"endpoint_change_allowed": False,
|
||
"configmap_patch_allowed": False,
|
||
"active_probe_allowed": False,
|
||
"external_health_probe_allowed": False,
|
||
"live_benchmark_allowed": False,
|
||
"provider_switch_allowed": False,
|
||
"paid_api_call_allowed": False,
|
||
"secret_read_allowed": False,
|
||
"secret_plaintext_allowed": False,
|
||
"notification_send_allowed": False,
|
||
"workflow_trigger_allowed": False,
|
||
"deploy_trigger_allowed": False,
|
||
"reload_trigger_allowed": False,
|
||
"runtime_execution_allowed": False,
|
||
},
|
||
"approval_boundaries": {
|
||
"service_restart_approved": False,
|
||
"endpoint_change_approved": False,
|
||
"configmap_patch_approved": False,
|
||
"active_probe_approved": False,
|
||
"secret_read_approved": False,
|
||
"notification_send_approved": False,
|
||
"workflow_trigger_approved": False,
|
||
"provider_switch_approved": False,
|
||
"cost_change_approved": False,
|
||
},
|
||
}
|
||
_refresh_rollups(payload)
|
||
return payload
|
||
|
||
|
||
def _target(target_id: str, kind: str, status: str, risk_level: str, freshness_status: str) -> dict:
|
||
return {
|
||
"target_id": target_id,
|
||
"display_name": target_id,
|
||
"kind": kind,
|
||
"status": status,
|
||
"risk_level": risk_level,
|
||
"freshness_status": freshness_status,
|
||
"health_contract": "只讀健康合約。",
|
||
"endpoint_contract": "不修改 endpoint。",
|
||
"evidence_refs": ["docs/reference/SERVICE-ENDPOINTS.md"],
|
||
"next_action": "只補證據，不執行。",
|
||
}
|
||
|
||
|
||
def _gap(gap_id: str, target_ids: list[str]) -> dict:
|
||
return {
|
||
"gap_id": gap_id,
|
||
"display_name": gap_id,
|
||
"status": "action_required",
|
||
"severity": "high",
|
||
"summary": "只讀缺口。",
|
||
"target_ids": target_ids,
|
||
"evidence_refs": ["docs/reference/SERVICE-ENDPOINTS.md"],
|
||
"next_action": "只整理 owner attestation。",
|
||
}
|
||
|
||
|
||
def _endpoint(endpoint_id: str) -> dict:
|
||
return {
|
||
"endpoint_id": endpoint_id,
|
||
"display_name": endpoint_id,
|
||
"status": "action_required",
|
||
"severity": "medium",
|
||
"stale_ref": "舊端點參考。",
|
||
"current_truth": "只標記 source drift，不判定 live down。",
|
||
"evidence_refs": ["docs/reference/SERVICE-ENDPOINTS.md"],
|
||
"next_action": "只整理 source cleanup。",
|
||
}
|
||
|
||
|
||
def _refresh_rollups(payload: dict) -> None:
|
||
targets = payload["service_health_targets"]
|
||
gaps = payload["health_gaps"]
|
||
endpoints = payload["stale_endpoints"]
|
||
payload["rollups"] = {
|
||
"total_targets": len(targets),
|
||
"by_kind": _count_by(targets, "kind"),
|
||
"by_status": _count_by(targets, "status"),
|
||
"by_freshness_status": _count_by(targets, "freshness_status"),
|
||
"target_ids_requiring_action": sorted(
|
||
target["target_id"] for target in targets if target["status"] == "action_required"
|
||
),
|
||
"health_gap_ids": sorted(gap["gap_id"] for gap in gaps),
|
||
"stale_endpoint_ids": sorted(endpoint["endpoint_id"] for endpoint in endpoints),
|
||
"critical_target_ids": sorted(
|
||
target["target_id"] for target in targets if target["risk_level"] == "critical"
|
||
),
|
||
"read_only_denials_total": 10,
|
||
"service_restart_allowed_count": 0,
|
||
"endpoint_change_allowed_count": 0,
|
||
"active_probe_allowed_count": 0,
|
||
"notification_send_allowed_count": 0,
|
||
"runtime_execution_allowed_count": 0,
|
||
}
|
||
|
||
|
||
def _count_by(items: list[dict], key: str) -> dict[str, int]:
|
||
counts: dict[str, int] = {}
|
||
for item in items:
|
||
value = item[key]
|
||
counts[value] = counts.get(value, 0) + 1
|
||
return counts
|