6krrt/tests/test_metrics_endpoint.py

"""Tests for the GET /metrics endpoint on dispatcher.

Seeds a throwaway temp DB and asserts on the actual JSON returned by a real
TestClient GET (never a mock-call assertion), to defeat
``misleading_success_output``.

Also covers the ``/events/decisions`` Server-Sent Events endpoint, which
streams routing decisions to the TUI without polling.
"""

from __future__ import annotations

import json
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest
from starlette.testclient import TestClient

import dispatcher
import events
from config import load_config

ROOT = Path(__file__).resolve().parent.parent
SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()
CFG = load_config(str(ROOT / "config" / "config.yaml"))


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _make_db(tmp_path: Path, extra_sql: str = "") -> sqlite3.Connection:
    conn = sqlite3.connect(str(tmp_path / "test.db"))
    conn.row_factory = sqlite3.Row
    conn.executescript(SCHEMA_SQL + extra_sql)
    return conn


def _seed_models(conn: sqlite3.Connection) -> None:
    for model_id, tier, context, cost, vision in (
        ("cheap", 2, 262128, 0.30, 1),
        ("dear", 2, 262128, 9.00, 0),
        ("tiny", 1, 131072, 0.10, 1),
    ):
        conn.execute(
            """
            INSERT INTO models (
                model_id, provider, base_model_id, tier, context_window,
                effective_context_window, max_output_tokens,
                cost_per_1m_prompt, cost_per_1m_completion,
                supports_vision, supports_json_mode,
                latency_class, reasoning_mode, context_variant,
                access_level, availability, last_updated
            ) VALUES (?, 'neuralwatt', ?, ?, ?, 192500, 16384, ?, ?,
                      ?, 1, 'standard', 'default', 'full', 'public', 'active',
                      '2026-08-22T00:00:00+00:00')
            """,
            (model_id, model_id, tier, context, cost, cost / 3, vision),
        )
    conn.commit()


def _seed_decision(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        INSERT INTO route_decisions (
            observed_at, kind, task_category, task_tier, required_context_tokens,
            confidence, classifier_ms, classification_source, latency_tolerance,
            candidates_considered, selected_model, selected_provider,
            runner_up_models, est_cost_usd, est_proficiency,
            session_key, tools, images, json_mode, streamed,
            flex_preference, flex_swapped, flex_forced
        ) VALUES (?, 'route', 'coding_general', 2, 100, 0.95, 200,
                  'classifier', 'interactive', 5, 'cheap', 'neuralwatt',
                  '[{"model_id":"dear","provider":"neuralwatt"}]',
                  0.001, 0.9, 'abc123', 0, 0, 0, 0,
                  'auto', 0, 1)
        """,
        (_now().isoformat(),),
    )
    conn.commit()


def _seed_energy(conn: sqlite3.Connection) -> None:
    now = _now()
    conn.execute(
        "INSERT INTO energy_observations "
        "(model_id, provider, task_category, completion_tokens, energy_kwh, "
        "cost_usd, carbon_g_co2eq, attribution_ratio, observed_at) "
        "VALUES ('cheap', 'neuralwatt', 'coding_general', 100, 5.0e-05, 0.001, "
        "2.4e-03, 0.25, ?)",
        ((now - timedelta(days=2)).isoformat(),),
    )
    conn.commit()


def _seed_verification(conn: sqlite3.Connection) -> None:
    conn.execute(
        "INSERT INTO verifications (model_id, provider, kind, verdict, observed_at) "
        "VALUES ('cheap', 'neuralwatt', 'structural', 'ok', ?)",
        (_now().isoformat(),),
    )
    conn.commit()


def _seed_proficiency(conn: sqlite3.Connection) -> None:
    conn.execute(
        "INSERT INTO proficiency (model_id, provider, category, blended_score, "
        "source, last_updated) "
        "VALUES ('cheap', 'neuralwatt', 'coding_general', 0.9, "
        "'self_eval_thin', '2026-01-01T00:00:00+00:00')",
    )
    conn.commit()


def _seed_local_energy(conn: sqlite3.Connection) -> None:
    conn.execute(
        "INSERT INTO local_energy_observations "
        "(observed_at, model_id, call_type, avg_power_watts, duration_seconds, "
        "energy_kwh, cost_usd, carbon_g_co2eq, meter) "
        "VALUES (?, 'mistral-nemo:12b', 'classify', 150.0, 1.0, "
        "0.00015, 1.8e-05, 0.06, 'nvidia-smi')",
        (_now().isoformat(),),
    )
    conn.commit()


@pytest.fixture
def seeded_client(tmp_path, monkeypatch):
    """A TestClient wired to a seeded temp DB, at /metrics."""
    conn = _make_db(tmp_path)
    _seed_models(conn)
    for i in range(60):  # exceed the 50 cap
        _seed_decision(conn)
    _seed_energy(conn)
    _seed_verification(conn)
    _seed_proficiency(conn)
    conn.close()

    monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
    monkeypatch.setattr(dispatcher.cfg.local_energy, "enabled", False)
    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")

    with TestClient(dispatcher.app) as client:
        yield client


def test_metrics_endpoint_has_all_top_level_keys(seeded_client):
    """GET /metrics returns 200 with every required top-level key."""
    resp = seeded_client.get("/metrics")
    assert resp.status_code == 200
    data = resp.json()
    for key in (
        "quota",
        "coverage",
        "recent_decisions",
        "per_model",
        "verdict_mix",
        "top_proficiency",
        "local_energy",
        "pinch",
        "cost_calibration",
        "latency",
        "generated_at",
    ):
        assert key in data, f"missing top-level key {key!r}"


def test_metrics_recent_decisions_capped_and_carries_provider(seeded_client):
    """recent_decisions is capped at 50 and each row has selected_provider."""
    resp = seeded_client.get("/metrics")
    data = resp.json()
    decisions = data["recent_decisions"]
    assert isinstance(decisions, list)
    assert len(decisions) <= 50
    assert len(decisions) > 0
    for row in decisions:
        assert "selected_provider" in row
    # DESC order: the first decision has the largest id.
    first_id = decisions[0]["id"]
    for row in decisions[1:]:
        assert row["id"] <= first_id


def test_metrics_aggregations_are_populated(seeded_client):
    """per_model / verdict_mix / top_proficiency reflect seeded data."""
    resp = seeded_client.get("/metrics")
    data = resp.json()
    assert isinstance(data["per_model"], list)
    assert any(r["model_id"] == "cheap" for r in data["per_model"])
    assert data["per_model"][0]["calls"] == 1
    assert data["verdict_mix"]["ok"] == 1
    assert isinstance(data["top_proficiency"], list)
    assert data["top_proficiency"][0]["model_id"] == "cheap"
    assert data["quota"] is not None


def test_metrics_recent_decisions_carry_flex_columns(seeded_client):
    """Each recent_decisions row carries the three flex telemetry columns."""
    resp = seeded_client.get("/metrics")
    data = resp.json()
    decisions = data["recent_decisions"]
    assert decisions
    for row in decisions:
        assert "flex_preference" in row
        assert "flex_swapped" in row
        assert "flex_forced" in row
    assert decisions[0]["flex_preference"] == "auto"
    assert decisions[0]["flex_swapped"] == 0
    assert decisions[0]["flex_forced"] == 1


def test_metrics_exposes_flex_default(seeded_client):
    """The configured flex default is reported (under coverage)."""
    resp = seeded_client.get("/metrics")
    data = resp.json()
    assert "flex_default" in data["coverage"]
    assert data["coverage"]["flex_default"] in ("no-flex", "auto", "prefer-flex", "force-flex")


def test_metrics_contains_no_session_dir(seeded_client):
    """The JSON must never name session_dir or expose conversation text."""
    body = seeded_client.get("/metrics").text
    assert "session_dir" not in body


def test_metrics_quota_carries_accounts_shape(seeded_client):
    """GET /metrics quota carries period/accounts/spend/alarm blocks."""
    resp = seeded_client.get("/metrics")
    assert resp.status_code == 200
    data = resp.json()
    assert data["quota"] is not None
    quota = data["quota"]
    for key in ("period", "accounts", "spend", "alarm"):
        assert key in quota, f"missing quota key {key!r}"

    period = quota["period"]
    for key in ("start", "next_reset", "elapsed_fraction", "source"):
        assert key in period, f"missing period key {key!r}"

    accounts = quota["accounts"]
    assert isinstance(accounts, list)
    for acc in accounts:
        for key in ("provider", "shape", "spend_usd"):
            assert key in acc, f"missing account key {key!r}"
        assert acc["shape"] in ("metered_plan", "prepaid_credit", "self_hosted", "unmetered")

    spend = quota["spend"]
    for key in ("by_provider_usd", "total_usd", "estimated_usd", "estimate_ratio"):
        assert key in spend, f"missing spend key {key!r}"


def test_metrics_quota_survives_missing_provider_balance_table(tmp_path, monkeypatch):
    """A live DB predating provider_balance_observations must not 500 /metrics."""
    conn = _make_db(tmp_path)
    _seed_models(conn)
    _seed_decision(conn)
    conn.execute("DROP TABLE provider_balance_observations")
    conn.commit()
    conn.close()

    monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
    monkeypatch.setattr(dispatcher.cfg.local_energy, "enabled", False)
    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")

    with TestClient(dispatcher.app) as client:
        resp = client.get("/metrics")
    assert resp.status_code == 200
    quota = resp.json()["quota"]
    assert "accounts" in quota
    for acc in quota["accounts"]:
        # Should not crash: missing table means pool is None/burn is None
        assert acc["shape"] in ("metered_plan", "prepaid_credit", "self_hosted", "unmetered")


def test_metrics_local_energy_omitted_when_disabled(seeded_client):
    """With metering off the section is omitted — independent of deployment config."""
    resp = seeded_client.get("/metrics")
    data = resp.json()
    assert "local_energy" in data
    assert data["local_energy"] is None


@pytest.mark.parametrize("local_energy_enabled", [False, True])
def test_metrics_local_energy_follows_enabled_flag(
    tmp_path, monkeypatch, local_energy_enabled
):
    """The suite must not couple to deployment config: both flag values are exercised."""
    conn = _make_db(tmp_path)
    _seed_models(conn)
    if local_energy_enabled:
        _seed_local_energy(conn)
    conn.close()

    monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
    monkeypatch.setattr(
        dispatcher.cfg.local_energy, "enabled", local_energy_enabled
    )
    monkeypatch.setattr(dispatcher.cfg.local_energy, "tariff_usd_per_kwh", 0.12)
    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")

    with TestClient(dispatcher.app) as client:
        resp = client.get("/metrics")
    assert resp.status_code == 200
    data = resp.json()

    if local_energy_enabled:
        assert isinstance(data["local_energy"], dict)
        assert data["local_energy"]["calls_30d"] == 1
        assert "by_type" in data["local_energy"]
    else:
        assert data["local_energy"] is None


def test_metrics_local_energy_enabled_reflects_seeded_row(
    tmp_path, monkeypatch
):
    """When enabled and a row is seeded, local_energy carries expected totals."""
    conn = _make_db(tmp_path)
    _seed_models(conn)
    _seed_local_energy(conn)
    conn.close()

    monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
    monkeypatch.setattr(dispatcher.cfg.local_energy, "enabled", True)
    monkeypatch.setattr(dispatcher.cfg.local_energy, "tariff_usd_per_kwh", 0.12)
    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")

    with TestClient(dispatcher.app) as client:
        resp = client.get("/metrics")
    assert resp.status_code == 200
    data = resp.json()
    le = data["local_energy"]
    assert isinstance(le, dict)
    assert "metered_kwh_30d" in le
    assert "metered_cost_usd_30d" in le
    assert "calls_30d" in le
    assert "by_type" in le
    assert "reset_date" in le
    assert le["metered_kwh_30d"] == round(0.00015, 5)
    assert le["metered_cost_usd_30d"] == round(1.8e-05, 5)
    assert le["calls_30d"] == 1
    assert "classify" in le["by_type"]
    assert le["by_type"]["classify"]["calls"] == 1
    assert le["by_type"]["classify"]["kwh"] == round(0.00015, 5)
    assert le["by_type"]["classify"]["cost_usd"] == round(1.8e-05, 5)


def test_metrics_empty_db_returns_200(monkeypatch, tmp_path):
    """Fresh empty temp DB: 200 with empty arrays, no exception."""
    conn = _make_db(tmp_path)
    conn.close()
    monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
    monkeypatch.setattr(dispatcher.cfg.local_energy, "enabled", False)
    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")

    with TestClient(dispatcher.app) as client:
        resp = client.get("/metrics")
    assert resp.status_code == 200
    data = resp.json()
    assert data["recent_decisions"] == []
    assert data["per_model"] == []
    assert data["verdict_mix"] == {}
    assert data["top_proficiency"] == []
    assert "local_energy" in data
    assert data["local_energy"] is None
    assert "generated_at" in data


# admin_model_overrides is NOT in schema.sql — dispatcher creates it at
# startup via ensure_route_decisions-style migration, so endpoint tests that
# need it build it with extra_sql (pattern from tests/test_metrics.py).
_ADMIN_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS admin_model_overrides (
    model_id         TEXT NOT NULL,
    provider         TEXT NOT NULL,
    availability     TEXT NOT NULL,
    reason           TEXT,
    updated_at       TEXT NOT NULL,
    PRIMARY KEY (model_id, provider)
);
CREATE INDEX IF NOT EXISTS idx_admin_model_overrides_availability
    ON admin_model_overrides (availability);
"""


def test_metrics_coverage_warnings_include_rejection_and_capability(
    tmp_path, monkeypatch
):
    """GET /metrics surfaces the new rejection and capability warnings.

    Seeds the 2026-09-04 incident state — kimi-k3 and kimi-k3-fast (the only
    vision-capable tier-1 models) deprecated via admin overrides, plus two
    image rejections — into the endpoint's temp DB, and asserts the warnings
    reach the same coverage.warnings list the admin bell and TUI render.
    The rejection warning is the deterministic assertion; on this exact seed
    the capability warning fires too.
    """
    now = _now().isoformat()
    conn = _make_db(tmp_path, extra_sql=_ADMIN_TABLE_SQL)
    for model_id, vision, eff_ctx in (
        ("kimi-k3", 1, 782324),
        ("kimi-k3-fast", 1, 782324),
        ("deepseek-v4-flash", 0, 262128),
    ):
        conn.execute(
            """
            INSERT INTO models (
                model_id, provider, base_model_id, tier, context_window,
                effective_context_window, max_output_tokens,
                cost_per_1m_prompt, cost_per_1m_completion,
                supports_vision, supports_json_mode,
                latency_class, reasoning_mode, context_variant,
                access_level, availability, last_updated
            ) VALUES (?, 'neuralwatt', ?, 1, ?, ?, 16384, 0.30, 0.10,
                      ?, 1, 'standard', 'default', 'full', 'public', 'active',
                      ?)
            """,
            (model_id, model_id, eff_ctx, eff_ctx, vision, now),
        )
    for model_id in ("kimi-k3", "kimi-k3-fast"):
        conn.execute(
            "INSERT INTO admin_model_overrides "
            "(model_id, provider, availability, reason, updated_at) "
            "VALUES (?, 'neuralwatt', 'deprecated', 'cost', ?)",
            (model_id, now),
        )
    for tokens in (242486, 195000):
        conn.execute(
            """
            INSERT INTO route_decisions (
                observed_at, kind, task_category, task_tier,
                required_context_tokens, latency_tolerance,
                selected_model, selected_provider, rejected_reason,
                tools, images, json_mode, streamed
            ) VALUES (?, 'chat', 'coding_general', 1, 200000,
                      'interactive', NULL, NULL, ?, 0, 1, 0, 0)
            """,
            (
                now,
                f"context >= {tokens} tokens; "
                f"vision-capable model (request carries image(s))",
            ),
        )
    conn.commit()
    conn.close()

    monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
    monkeypatch.setattr(dispatcher.cfg.local_energy, "enabled", False)
    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")

    with TestClient(dispatcher.app) as client:
        resp = client.get("/metrics")
    assert resp.status_code == 200
    data = resp.json()
    warnings = data["coverage"]["warnings"]
    assert len(warnings) > 0
    assert any("rejection" in w for w in warnings)
    # The capability warning is secondary per the plan (it depends on the
    # seeded models) but this seed reproduces the incident, so it fires.
    assert any("vision" in w for w in warnings)


def _sse_frame(line: str | bytes) -> dict:
    """Parse one ``data: <json>`` SSE line and return the JSON payload."""
    if isinstance(line, bytes):
        line = line.decode()
    assert line.startswith("data: "), f"unexpected SSE frame: {line!r}"
    return json.loads(line[len("data: ") :])


def test_events_decisions_returns_sse_headers(seeded_client, monkeypatch):
    """The endpoint announces text/event-stream and no-cache headers.

    The real stream is unbounded (it blocks for heartbeats), so the generator
    is stubbed to a bounded one to let TestClient read the whole body without
    hanging. The live replay/stream behaviour is covered by the deterministic
    generator test below.
    """
    events.clear()
    try:

        def _bounded():
            yield "retry: 3000\n\n"
            yield "data: {\"id\": 9}\n\n"

        monkeypatch.setattr(dispatcher, "_decision_event_stream", _bounded)
        resp = seeded_client.get("/events/decisions")
        assert resp.status_code == 200
        assert "text/event-stream" in resp.headers["content-type"]
        assert "no-cache" in resp.headers.get("cache-control", "")
        assert "data: {\"id\": 9}" in resp.text
    finally:
        events.clear()


@pytest.mark.anyio
async def test_decision_event_stream_replays_then_streams_live():
    """Drive the generator directly: replay first, then a live publish arrives
    as the next frame. No HTTP client or threads, so it is deterministic."""
    events.clear()
    try:
        events.publish_decision(
            {"id": 1, "selected_model": "cheap", "task_category": "coding"}
        )
        stream = dispatcher._decision_event_stream()
        retry = await anext(stream)
        assert retry.startswith("retry:")
        replayed = await anext(stream)
        assert _sse_frame(replayed)["id"] == 1

        events.publish_decision(
            {"id": 2, "selected_model": "tiny", "task_category": "debugging"}
        )
        live = await anext(stream)
        assert _sse_frame(live)["id"] == 2
    finally:
        events.clear()


@pytest.mark.anyio
async def test_persist_route_decision_publishes_profile_in_sse_payload(
    tmp_path, monkeypatch,
):
    import asyncio

    events.clear()
    db_path = tmp_path / "profile-sse.db"
    conn = sqlite3.connect(db_path)
    conn.executescript(SCHEMA_SQL)
    conn.close()

    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
    monkeypatch.setattr(dispatcher.cfg.logging, "log_route_decisions", True)

    sse_queue: asyncio.Queue[dict] = asyncio.Queue()
    try:
        events.subscribe_sse(sse_queue, replay=True)

        dispatcher.persist_route_decision(
            "route",
            classification=dispatcher.RouteResponse(
                classification=dispatcher.Classification(
                    task_category="coding_general", task_tier=2,
                    required_context_tokens=100, confidence=0.9,
                ),
                latency_tolerance="interactive",
                profile="batch",
                selected=dispatcher.Candidate(
                    model_id="cheap", provider="neuralwatt",
                    tier=2, latency_class="standard", reasoning_mode="default",
                    context_variant="full", effective_context_window=128000,
                    composite=0.5, cost_score=0.5, proficiency_score=0.9,
                ),
                candidates_considered=1,
            ),
            latency_tolerance="interactive",
            selected_model="cheap",
            selected_provider="neuralwatt",
        )

        decision = await asyncio.wait_for(sse_queue.get(), timeout=2.0)
        assert decision["profile"] == "batch"
    finally:
        events.unsubscribe_sse(sse_queue)
        events.clear()


# ---------------------------------------------------------------------------
# content_fault_warnings unit tests
# ---------------------------------------------------------------------------

from types import SimpleNamespace as _SN  # noqa: E402


def _cfw_cfg():
    return _SN(
        objective=_SN(
            rejection_warning_window_hours=1,
            rejection_warning_baseline_hours=24,
            rejection_warning_min_count=6,
        )
    )


def _cfw_conn(tmp_path):
    conn = sqlite3.connect(str(tmp_path / "cfw_test.db"))
    conn.row_factory = sqlite3.Row
    conn.execute("""
        CREATE TABLE verifications (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            model_id TEXT NOT NULL,
            provider TEXT NOT NULL,
            request_id TEXT,
            task_category TEXT,
            kind TEXT NOT NULL,
            verdict TEXT NOT NULL,
            detail TEXT,
            completion_tokens INTEGER,
            observed_at TEXT NOT NULL,
            model_attributable INTEGER DEFAULT 1
        )
    """)
    conn.commit()
    return conn


def _cfw_insert(conn, model, provider, observed_at, attributable=1):
    conn.execute(
        "INSERT INTO verifications "
        "(model_id, provider, task_category, request_id, kind, verdict, "
        " detail, completion_tokens, model_attributable, observed_at) "
        "VALUES (?, ?, 'coding_general', ?, 'structural', 'malformed', "
        " 'empty response', 0, ?, ?)",
        (model, provider, f"req-{model}", attributable, observed_at),
    )


def test_content_fault_no_malformed_returns_empty(tmp_path):
    from metrics import content_fault_warnings
    conn = _cfw_conn(tmp_path)
    try:
        assert content_fault_warnings(conn, _cfw_cfg()) == []
    finally:
        conn.close()


def test_content_fault_high_rate_triggers_known_warning(tmp_path):
    from metrics import content_fault_warnings
    conn = _cfw_conn(tmp_path)
    try:
        now = datetime.now(timezone.utc)
        for i in range(7):
            _cfw_insert(conn, "bad-model", "openrouter",
                        (now - timedelta(minutes=i)).isoformat())
        conn.commit()
        warnings = content_fault_warnings(conn, _cfw_cfg())
        assert len(warnings) >= 1
        assert any("bad-model" in w for w in warnings)
        assert any("content fault:" in w for w in warnings)
    finally:
        conn.close()


def test_content_fault_novel_pattern_triggers_new_warning(tmp_path):
    from metrics import content_fault_warnings
    conn = _cfw_conn(tmp_path)
    try:
        now = datetime.now(timezone.utc)
        for i in range(3):
            _cfw_insert(conn, "novel-model", "openrouter",
                        (now - timedelta(minutes=i)).isoformat())
        conn.commit()
        warnings = content_fault_warnings(conn, _cfw_cfg())
        assert len(warnings) >= 1
        assert any("new content fault:" in w for w in warnings)
    finally:
        conn.close()


def test_content_fault_low_count_does_not_trigger(tmp_path):
    from metrics import content_fault_warnings
    conn = _cfw_conn(tmp_path)
    try:
        now = datetime.now(timezone.utc)
        _cfw_insert(conn, "low-model", "openrouter",
                    (now - timedelta(hours=20)).isoformat())
        for i in range(3):
            _cfw_insert(conn, "low-model", "openrouter",
                        (now - timedelta(minutes=i)).isoformat())
        conn.commit()
        assert content_fault_warnings(conn, _cfw_cfg()) == []
    finally:
        conn.close()


def test_content_fault_non_attributable_excluded(tmp_path):
    from metrics import content_fault_warnings
    conn = _cfw_conn(tmp_path)
    try:
        now = datetime.now(timezone.utc)
        for i in range(10):
            _cfw_insert(conn, "tool-model", "openrouter",
                        (now - timedelta(minutes=i)).isoformat(),
                        attributable=0)
        conn.commit()
        assert content_fault_warnings(conn, _cfw_cfg()) == []
    finally:
        conn.close()