294 lines
11 KiB
Python
294 lines
11 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
|
||
import pytest
|
||
|
||
from src.services.observability_contract_matrix import load_latest_observability_contract_matrix
|
||
|
||
|
||
def test_load_latest_observability_contract_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 / "observability_contract_matrix_2026-06-04.json").write_text(
|
||
json.dumps(older),
|
||
encoding="utf-8",
|
||
)
|
||
(tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
|
||
json.dumps(newer),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
loaded = load_latest_observability_contract_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_surfaces"] == 2
|
||
assert loaded["operation_boundaries"]["alertmanager_to_openclaw_allowed"] is False
|
||
|
||
|
||
def test_observability_contract_matrix_requires_read_only_mode(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["program_status"]["read_only_mode"] = False
|
||
(tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="read_only_mode"):
|
||
load_latest_observability_contract_matrix(tmp_path)
|
||
|
||
|
||
def test_observability_contract_matrix_blocks_route_and_rule_mutations(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["operation_boundaries"]["prometheus_rule_write_allowed"] = True
|
||
snapshot["operation_boundaries"]["alertmanager_to_openclaw_allowed"] = True
|
||
(tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="operation boundaries"):
|
||
load_latest_observability_contract_matrix(tmp_path)
|
||
|
||
|
||
def test_observability_contract_matrix_requires_rollup_consistency(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["rollups"]["surface_ids_requiring_action"] = []
|
||
(tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="surface_ids_requiring_action"):
|
||
load_latest_observability_contract_matrix(tmp_path)
|
||
|
||
|
||
def test_observability_contract_matrix_requires_noise_candidates_to_be_proposal_only(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["noise_reduction_opportunities"][0]["proposal_only"] = False
|
||
(tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="proposal_only"):
|
||
load_latest_observability_contract_matrix(tmp_path)
|
||
|
||
|
||
def test_observability_contract_matrix_requires_openclaw_receiver_denial(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["operator_contract"]["must_not_interpret_as"].remove(
|
||
"Alertmanager 指向 OpenClaw receiver 批准"
|
||
)
|
||
(tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="operator_contract"):
|
||
load_latest_observability_contract_matrix(tmp_path)
|
||
|
||
|
||
def test_observability_contract_matrix_rejects_secret_payload_keys(tmp_path):
|
||
snapshot = _snapshot()
|
||
snapshot["latest_observations"][0]["webhook_secret"] = "redacted"
|
||
(tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
|
||
json.dumps(snapshot),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="forbidden secret payload key"):
|
||
load_latest_observability_contract_matrix(tmp_path)
|
||
|
||
|
||
def test_observability_contract_matrix_fails_when_missing(tmp_path):
|
||
with pytest.raises(FileNotFoundError):
|
||
load_latest_observability_contract_matrix(tmp_path)
|
||
|
||
|
||
def _snapshot(
|
||
*,
|
||
generated_at: str = "2026-06-05T00:00:00+08:00",
|
||
completion: int = 100,
|
||
) -> dict:
|
||
surfaces = [
|
||
_surface(
|
||
"prometheus_alert_rule_catalog",
|
||
"Prometheus 告警規則合約",
|
||
"prometheus_rules",
|
||
"action_required",
|
||
"proposal_only",
|
||
),
|
||
_surface(
|
||
"alertmanager_awoooi_route",
|
||
"Alertmanager → AWOOOI API 路由",
|
||
"alertmanager_route",
|
||
"verified",
|
||
"proposal_only",
|
||
),
|
||
]
|
||
opportunities = [
|
||
_opportunity("prometheus_noise_rule_tuning", "approval_required"),
|
||
_opportunity("alertmanager_grouping_inhibit_tuning", "approval_required"),
|
||
_opportunity("success_notification_quiet_policy", "preserved"),
|
||
]
|
||
gaps = [
|
||
{
|
||
"gap_id": "prometheus_alert_rule_catalog_seed",
|
||
"display_name": "Alert rule catalog seed 未正式產品化",
|
||
"status": "action_required",
|
||
"severity": "high",
|
||
"summary": "只讀矩陣已建立，尚未產生 catalog seed。",
|
||
"evidence_refs": ["docs/adr/ADR-090-monitoring-blindspot-governance.md"],
|
||
"next_action": "先產 proposal，不改 rule。",
|
||
}
|
||
]
|
||
return {
|
||
"schema_version": "observability_contract_matrix_v1",
|
||
"generated_at": generated_at,
|
||
"program_status": {
|
||
"overall_completion_percent": completion,
|
||
"current_priority": "P1",
|
||
"current_task_id": "P1-003",
|
||
"next_task_id": "P1-004",
|
||
"read_only_mode": True,
|
||
},
|
||
"source_refs": ["docs/schemas/observability_contract_matrix_v1.schema.json"],
|
||
"rollups": {
|
||
"total_surfaces": len(surfaces),
|
||
"by_kind": _count_by(surfaces, "kind"),
|
||
"by_status": _count_by(surfaces, "status"),
|
||
"by_evidence_status": _count_by(surfaces, "evidence_status"),
|
||
"by_noise_policy_status": _count_by(surfaces, "noise_policy_status"),
|
||
"surface_ids_requiring_action": ["prometheus_alert_rule_catalog"],
|
||
"surface_ids_with_proposal_only_noise_policy": [
|
||
"alertmanager_awoooi_route",
|
||
"prometheus_alert_rule_catalog",
|
||
],
|
||
"noise_reduction_opportunities_total": len(opportunities),
|
||
"approval_required_opportunity_ids": [
|
||
"alertmanager_grouping_inhibit_tuning",
|
||
"prometheus_noise_rule_tuning",
|
||
],
|
||
"classification_gap_ids": ["prometheus_alert_rule_catalog_seed"],
|
||
"read_only_denials_total": 12,
|
||
},
|
||
"observability_surfaces": surfaces,
|
||
"noise_reduction_opportunities": opportunities,
|
||
"classification_gaps": gaps,
|
||
"latest_observations": [
|
||
{
|
||
"observation_id": "alertmanager_receiver_guard",
|
||
"status": "verified",
|
||
"summary": "Alertmanager 不得指向 OpenClaw。",
|
||
"evidence_refs": ["docs/HARD_RULES.md#alertmanager-routing"],
|
||
}
|
||
],
|
||
"operator_contract": {
|
||
"display_mode": "read_only_observability_contract_matrix",
|
||
"must_not_interpret_as": [
|
||
"Prometheus alert rule 修改批准",
|
||
"Alertmanager receiver / route 修改批准",
|
||
"Alertmanager 指向 OpenClaw receiver 批准",
|
||
"Silence 建立或維護窗口批准",
|
||
"Grafana dashboard 寫入批准",
|
||
"SigNoz / Sentry webhook 設定修改批准",
|
||
"Secret 已讀取或可輸出",
|
||
"Telegram 測試通知批准",
|
||
"deploy / reload / workflow 觸發批准",
|
||
"runtime execution 授權",
|
||
],
|
||
"secret_display_policy": "只顯示 redacted metadata。",
|
||
"alertmanager_route_policy": "OpenClaw 不接收 Alertmanager webhook；receiver 維持 AWOOOI API。",
|
||
"noise_reduction_policy": "只產生 proposal。",
|
||
"notification_policy": "成功不洗版。",
|
||
},
|
||
"operation_boundaries": {
|
||
"read_only_api_allowed": True,
|
||
"prometheus_rule_write_allowed": False,
|
||
"prometheus_reload_allowed": False,
|
||
"alertmanager_route_write_allowed": False,
|
||
"alertmanager_receiver_change_allowed": False,
|
||
"alertmanager_to_openclaw_allowed": False,
|
||
"silence_create_allowed": False,
|
||
"grafana_dashboard_write_allowed": False,
|
||
"grafana_api_write_allowed": False,
|
||
"signoz_query_mutation_allowed": False,
|
||
"signoz_webhook_change_allowed": False,
|
||
"sentry_webhook_change_allowed": False,
|
||
"otel_collector_deploy_allowed": False,
|
||
"event_exporter_restart_allowed": False,
|
||
"secret_read_allowed": False,
|
||
"secret_plaintext_allowed": False,
|
||
"notification_send_allowed": False,
|
||
"external_api_call_allowed": False,
|
||
"live_prometheus_query_allowed": False,
|
||
"workflow_trigger_allowed": False,
|
||
"deploy_trigger_allowed": False,
|
||
"reload_trigger_allowed": False,
|
||
"runtime_execution_allowed": False,
|
||
},
|
||
"approval_boundaries": {
|
||
"prometheus_rule_change_authorized": False,
|
||
"prometheus_reload_authorized": False,
|
||
"alertmanager_route_change_authorized": False,
|
||
"alertmanager_receiver_change_authorized": False,
|
||
"alertmanager_to_openclaw_authorized": False,
|
||
"silence_authorized": False,
|
||
"grafana_write_authorized": False,
|
||
"signoz_write_authorized": False,
|
||
"sentry_write_authorized": False,
|
||
"otel_deploy_authorized": False,
|
||
"event_exporter_restart_authorized": False,
|
||
"notification_send_authorized": False,
|
||
"external_call_authorized": False,
|
||
"secret_plaintext_allowed": False,
|
||
"workflow_trigger_authorized": False,
|
||
"deploy_reload_authorized": False,
|
||
"runtime_execution_authorized": False,
|
||
},
|
||
}
|
||
|
||
|
||
def _surface(
|
||
surface_id: str,
|
||
display_name: str,
|
||
kind: str,
|
||
status: str,
|
||
noise_policy_status: str,
|
||
) -> dict:
|
||
return {
|
||
"surface_id": surface_id,
|
||
"display_name": display_name,
|
||
"kind": kind,
|
||
"status": status,
|
||
"risk_level": "critical",
|
||
"evidence_status": "committed_manifest",
|
||
"noise_policy_status": noise_policy_status,
|
||
"coverage_contract": "只讀 committed evidence。",
|
||
"current_contract": "不得改 live 設定。",
|
||
"evidence_refs": ["docs/HARD_RULES.md"],
|
||
"next_action": "只產 proposal。",
|
||
}
|
||
|
||
|
||
def _opportunity(opportunity_id: str, status: str) -> dict:
|
||
return {
|
||
"opportunity_id": opportunity_id,
|
||
"display_name": opportunity_id,
|
||
"status": status,
|
||
"proposal_only": True,
|
||
"impact": "降噪提案。",
|
||
"evidence_refs": ["docs/HARD_RULES.md"],
|
||
"next_action": "人工批准前不執行。",
|
||
}
|
||
|
||
|
||
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
|