140 lines
4.7 KiB
Python
140 lines
4.7 KiB
Python
def test_ai_automation_metrics_snapshot_records_event_router_and_autoheal():
|
|
from services import ai_automation_metrics as metrics
|
|
|
|
metrics.reset_for_tests()
|
|
|
|
metrics.record_event_router_dispatch(
|
|
tier="L2",
|
|
event_type="scheduler_task_failure",
|
|
delivered=False,
|
|
queued=True,
|
|
latency_ms=123,
|
|
)
|
|
metrics.record_event_router_safe_action("retry_task", "ok")
|
|
metrics.record_event_router_replay(attempted=2, sent=1, failed=1)
|
|
metrics.record_autoheal_action(
|
|
action="ALERT_ONLY",
|
|
error_type="DB_UNREACHABLE",
|
|
success=True,
|
|
duration_ms=45,
|
|
)
|
|
|
|
snap = metrics.snapshot()
|
|
counters = {
|
|
(metric, dict(labels).get("outcome") or dict(labels).get("status") or dict(labels).get("result")): value
|
|
for (metric, labels), value in snap["counters"].items()
|
|
}
|
|
|
|
assert counters[("event_router_dispatch_total", "queued")] == 1
|
|
assert counters[("event_router_safe_action_total", "ok")] == 1
|
|
assert counters[("event_router_replay_total", "attempted")] == 2
|
|
assert counters[("event_router_replay_total", "sent")] == 1
|
|
assert counters[("autoheal_action_total", "success")] == 1
|
|
|
|
latency = next(
|
|
values for (metric, _labels), values in snap["latency"].items()
|
|
if metric == "event_router_latency_ms"
|
|
)
|
|
assert latency["count"] == 1
|
|
assert latency["sum"] == 123
|
|
assert latency["max"] == 123
|
|
|
|
|
|
def test_system_metrics_exports_ai_automation_metrics():
|
|
from prometheus_client import CollectorRegistry, Gauge, generate_latest
|
|
from routes.system_public_routes import _register_ai_automation_metrics
|
|
from services import ai_automation_metrics as metrics
|
|
|
|
metrics.reset_for_tests()
|
|
metrics.record_event_router_dispatch(
|
|
tier="L1",
|
|
event_type="crawler_timeout",
|
|
delivered=True,
|
|
latency_ms=7,
|
|
)
|
|
|
|
registry = CollectorRegistry()
|
|
_register_ai_automation_metrics(registry, Gauge, metrics.snapshot())
|
|
output = generate_latest(registry).decode("utf-8")
|
|
|
|
assert "momo_ai_event_router_dispatch_total" in output
|
|
assert 'event_type="crawler_timeout"' in output
|
|
assert 'outcome="delivered"' in output
|
|
assert "momo_ai_event_router_latency_ms_count" in output
|
|
|
|
|
|
def test_system_metrics_exports_ai_automation_zero_baseline():
|
|
from prometheus_client import CollectorRegistry, Gauge, generate_latest
|
|
from routes.system_public_routes import _register_ai_automation_metrics
|
|
from services import ai_automation_metrics as metrics
|
|
|
|
metrics.reset_for_tests()
|
|
registry = CollectorRegistry()
|
|
|
|
_register_ai_automation_metrics(registry, Gauge, metrics.snapshot())
|
|
|
|
output = generate_latest(registry).decode("utf-8")
|
|
assert (
|
|
'momo_ai_event_router_dispatch_total{event_type="baseline",outcome="none",tier="baseline"} 0.0'
|
|
in output
|
|
)
|
|
assert (
|
|
'momo_ai_event_router_safe_action_total{action="baseline",status="none"} 0.0'
|
|
in output
|
|
)
|
|
assert 'momo_ai_event_router_replay_total{status="none"} 0.0' in output
|
|
assert (
|
|
'momo_ai_autoheal_action_total{action="baseline",error_type="none",result="none"} 0.0'
|
|
in output
|
|
)
|
|
assert (
|
|
'momo_ai_event_router_latency_ms_count{event_type="baseline",tier="baseline"} 0.0'
|
|
in output
|
|
)
|
|
assert (
|
|
'momo_ai_autoheal_duration_ms_count{action="baseline",error_type="none"} 0.0'
|
|
in output
|
|
)
|
|
|
|
|
|
def test_system_metrics_counts_sales_records_with_raw_count_query():
|
|
from prometheus_client import CollectorRegistry, Gauge, generate_latest
|
|
from routes.system_public_routes import _set_database_record_counts
|
|
|
|
class FakeQuery:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def count(self):
|
|
return self.value
|
|
|
|
class FakeResult:
|
|
def scalar(self):
|
|
return 789
|
|
|
|
class FakeSession:
|
|
def __init__(self):
|
|
self.executed_sql = []
|
|
|
|
def query(self, model):
|
|
if model.__name__ == "Product":
|
|
return FakeQuery(123)
|
|
if model.__name__ == "PriceRecord":
|
|
return FakeQuery(456)
|
|
raise AssertionError(f"不應透過 ORM 查詢 {model.__name__}")
|
|
|
|
def execute(self, statement):
|
|
self.executed_sql.append(str(statement))
|
|
return FakeResult()
|
|
|
|
session = FakeSession()
|
|
registry = CollectorRegistry()
|
|
|
|
_set_database_record_counts(registry, Gauge, session)
|
|
|
|
output = generate_latest(registry).decode("utf-8")
|
|
assert "momo_products_total 123.0" in output
|
|
assert "momo_price_records_total 456.0" in output
|
|
assert "momo_sales_records_total 789.0" in output
|
|
assert session.executed_sql == ["SELECT COUNT(*) FROM realtime_sales_monthly"]
|