fix(metrics): read latest sitewide visual qa artifact
Some checks failed
CD Pipeline / deploy (push) Has been cancelled

This commit is contained in:
ogt
2026-07-03 07:45:59 +08:00
parent 023ba0631f
commit 4ccd3c413a
2 changed files with 110 additions and 0 deletions

View File

@@ -214,6 +214,9 @@ def build_scheduled_automation_health_summary(
sitewide_details = sitewide_readback.get("details") or {}
visual_qa_readback = _find_check(source_result, "Sitewide visual QA readback")
visual_qa_details = visual_qa_readback.get("details") or {}
if not visual_qa_readback or not visual_qa_details:
visual_qa_readback = _latest_sitewide_visual_qa_readback_check()
visual_qa_details = visual_qa_readback.get("details") or {}
smoke_status = source_result.get("status") or ("warning" if latest_history else "warning")
freshness_family = _history_freshness_family(
latest_history,
@@ -1019,6 +1022,60 @@ def _sitewide_visual_qa_check() -> Dict[str, Any]:
)
def _latest_sitewide_visual_qa_readback_check() -> Dict[str, Any]:
"""Read the latest visual QA artifact for scheduled summaries without running smoke."""
try:
from services.ai_surface_html_readback_service import build_sitewide_visual_qa_readback
readback = build_sitewide_visual_qa_readback()
summary = readback.get("summary") or {}
status = readback.get("status") or "warning"
route_count = int(summary.get("route_count") or 0)
viewport_count = int(summary.get("viewport_count") or 0)
failed_count = int(summary.get("failed_count") or 0)
return _check(
"Sitewide visual QA readback",
status,
(
f"Sitewide visual QA ok{route_count} routes x {viewport_count} viewports 無 overflow 退化"
if status == "ok"
else (
f"Sitewide visual QA 需更新或修復failed={failed_count}"
f"routes={route_count}、stale={bool(summary.get('stale'))}"
)
),
{
"policy": readback.get("policy"),
"result_count": int(summary.get("result_count") or 0),
"route_count": route_count,
"viewport_count": viewport_count,
"pass_count": int(summary.get("pass_count") or 0),
"failed_count": failed_count,
"overflow_issue_count": int(summary.get("overflow_issue_count") or 0),
"visual_offender_count": int(summary.get("visual_offender_count") or 0),
"artifact_path": readback.get("artifact_path"),
"artifact_exists": bool(readback.get("artifact_exists")),
"artifact_generated_at": readback.get("artifact_generated_at"),
"age_hours": summary.get("age_hours"),
"stale": bool(summary.get("stale")),
"next_machine_action": readback.get("next_machine_action"),
"writes_database": False,
"writes_database_count": 0,
"primary_human_gate_count": 0,
},
)
except Exception as exc:
return _check(
"Sitewide visual QA readback",
"warning",
f"Sitewide visual QA artifact fallback 無法讀取:{exc}",
{
"writes_database": False,
"primary_human_gate_count": 0,
},
)
def collect_ai_automation_smoke(*, record_history: bool = True, history_limit: int = 20) -> Dict[str, Any]:
checks: List[Dict[str, Any]] = [
_event_router_check(),

View File

@@ -335,6 +335,59 @@ def test_scheduled_automation_health_summary_can_use_current_smoke_without_recor
assert not history_path.exists()
def test_scheduled_automation_health_summary_falls_back_to_visual_qa_artifact(tmp_path, monkeypatch):
from datetime import datetime
from services import ai_automation_smoke_service as smoke
from services import ai_surface_html_readback_service as surface_service
history_path = tmp_path / "smoke_history.jsonl"
history_path.write_text(
json.dumps({
"generated_at": datetime.now().isoformat(timespec="seconds"),
"status": "ok",
"summary": {"ok": 9, "warning": 0, "critical": 0, "total": 9},
"checks": [],
}, ensure_ascii=False) + "\n",
encoding="utf-8",
)
monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path))
monkeypatch.setattr(
surface_service,
"build_sitewide_visual_qa_readback",
lambda **_kwargs: {
"policy": "read_only_sitewide_visual_qa_readback_v1",
"status": "ok",
"artifact_path": "data/ai_automation/sitewide_visual_qa_latest.json",
"artifact_exists": True,
"artifact_generated_at": "2026-07-03T07:40:00",
"next_machine_action": "keep_sitewide_visual_qa_monitoring",
"summary": {
"result_count": 147,
"route_count": 49,
"viewport_count": 3,
"pass_count": 147,
"failed_count": 0,
"overflow_issue_count": 0,
"visual_offender_count": 0,
"stale": False,
},
},
)
summary = smoke.build_scheduled_automation_health_summary(
history_limit=5,
freshness_max_age_hours=9999,
)
visual_family = next(
item for item in summary["families"]
if item["key"] == "sitewide_visual_qa"
)
assert visual_family["status"] == "ok"
assert visual_family["details"]["route_count"] == 49
assert visual_family["details"]["failed_count"] == 0
def test_scheduled_automation_health_summary_route_returns_compact_payload(tmp_path, monkeypatch):
from datetime import datetime
from flask import Flask