新增 AI 自動化 Smoke Dashboard

tests/test_ai_automation_smoke_service.py

@@ -0,0 +1,38 @@
+def test_event_router_smoke_reports_queued_deliveries(tmp_path, monkeypatch):
+    from services import ai_automation_metrics as metrics
+    from services import ai_automation_smoke_service as smoke
+    from services import event_router
+
+    queue_path = tmp_path / "failed_deliveries.jsonl"
+    queue_path.write_text('{"event_key":"a"}\n{"event_key":"b"}\n', encoding="utf-8")
+    monkeypatch.setattr(event_router, "_QUEUE_PATH", str(queue_path))
+    metrics.reset_for_tests()
+    metrics.record_event_router_dispatch(
+        tier="L1",
+        event_type="crawler_timeout",
+        delivered=False,
+        queued=True,
+        latency_ms=12,
+    )
+
+    result = smoke._event_router_check()
+
+    assert result["status"] == "warning"
+    assert result["details"]["queued_deliveries"] == 2
+    assert result["details"]["dispatch_metric_total"] == 1
+    assert result["details"]["dispatch_sync"] is True
+
+
+def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch):
+    from services import ai_automation_smoke_service as smoke
+
+    monkeypatch.setattr(smoke, "_event_router_check", lambda: smoke._check("event", "ok", "ok"))
+    monkeypatch.setattr(smoke, "_autoheal_check", lambda: smoke._check("autoheal", "warning", "warn"))
+    monkeypatch.setattr(smoke, "_nemotron_check", lambda: smoke._check("nemotron", "ok", "ok"))
+    monkeypatch.setattr(smoke, "_embedding_queue_check", lambda: smoke._check("embedding", "critical", "boom"))
+    monkeypatch.setattr(smoke, "_elephant_hitl_check", lambda: smoke._check("elephant", "ok", "ok"))
+
+    result = smoke.collect_ai_automation_smoke()
+
+    assert result["status"] == "critical"
+    assert result["summary"] == {"ok": 3, "warning": 1, "critical": 1, "total": 5}