stack/tests/bcda/test_ui_metrics.py

"""bcda.ui.metrics — instruments, store gauges and the /metrics server."""

from __future__ import annotations

import re
import socket
import urllib.request

import pytest

from bcda.store import Store
from bcda.ui import metrics
from bcda.ui.runner import Runner


def _sample(body: str, name: str, **labels: str) -> float | None:
    """Value of the first *name* series whose labels include *labels*."""
    for line in body.splitlines():
        if not line.startswith((name + "{", name + " ")):
            continue
        braces = line[len(name) : line.index("}")] if "{" in line else ""
        got = dict(re.findall(r'(\w+)="([^"]*)"', braces))
        if all(got.get(k) == v for k, v in labels.items()):
            return float(line.rsplit(" ", 1)[1])
    return None


@pytest.fixture(autouse=True)
def _fresh_instruments(monkeypatch):
    """Rebuild the instrument set per test; keep any server from leaking state."""
    monkeypatch.setattr(metrics, "_instruments", None)
    yield
    monkeypatch.setattr(metrics, "_instruments", None)


def test_mock_meter_path_is_free(monkeypatch) -> None:
    monkeypatch.setenv("STACK_TELEMETRY", "false")
    metrics.run_started("export")
    metrics.run_finished("export", "done", 1.5)
    metrics.run_finished("bogus kind", "failed", -1)  # clamps and folds to "other"
    metrics.page_view("connect")
    metrics.observe_store(None)
    assert metrics.serve(port=0) is None
    assert metrics._kind("load") == "load" and metrics._kind("nope") == "other"


def test_store_observations(tmp_path, monkeypatch) -> None:
    assert (
        metrics.store_job_counts(tmp_path) == {} and metrics.store_bytes(tmp_path) == 0
    )
    store = Store(path=str(tmp_path))
    store._append_job(key="A", status="complete")
    store._append_job(key="B", status="failed")
    store._append_job(key="C", status="complete")
    d = tmp_path / "exports" / "A"
    d.mkdir(parents=True)
    (d / "Patient.ndjson").write_bytes(b"x" * 10)
    assert metrics.store_job_counts(tmp_path) == {"complete": 2, "failed": 1}
    assert metrics.store_bytes(tmp_path) == 10
    # errors never propagate to a scrape
    from bcda.ui import service as svc

    def boom(*a, **k):  # noqa: ANN002, ANN003, ANN202
        raise OSError("disk gone")

    monkeypatch.setattr(svc, "make_store", boom)
    monkeypatch.setattr(svc, "export_dirs", boom)
    assert (
        metrics.store_job_counts(tmp_path) == {} and metrics.store_bytes(tmp_path) == 0
    )


def test_runner_records_runs(monkeypatch) -> None:
    seen: list = []
    monkeypatch.setattr(
        metrics, "run_started", lambda kind: seen.append(("start", kind))
    )
    monkeypatch.setattr(
        metrics,
        "run_finished",
        lambda kind, state, secs: seen.append(("end", kind, state, secs >= 0)),
    )
    r = Runner()
    r.wait(r.submit("ok", lambda: 1, kind="load").id)

    def boom() -> None:
        raise RuntimeError("x")

    r.wait(r.submit("bad", boom).id)
    assert seen == [
        ("start", "load"),
        ("end", "load", "done", True),
        ("start", "other"),
        ("end", "other", "failed", True),
    ]
    assert r.all()[0].kind == "other" and r.all()[1].kind == "load"


def _free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def test_serve_publishes_registry(monkeypatch, tmp_path) -> None:
    """With telemetry on, /metrics carries stack_bcda_* series incl. the store gauges."""
    import perf
    from perf import _meter

    monkeypatch.setenv("STACK_TELEMETRY", "true")
    monkeypatch.setattr(perf, "_initialised", False)
    monkeypatch.setattr(_meter, "_provider_ready", False)
    monkeypatch.setattr(metrics, "_server_port", None)
    store = Store(path=str(tmp_path))
    store._append_job(key="A", status="complete")

    port = metrics.serve(port=_free_port())
    assert port and metrics.serve() == port  # idempotent
    metrics.observe_store(tmp_path)
    metrics.run_started("export")
    metrics.run_finished("export", "done", 2.0)
    metrics.page_view("exports")

    body = (
        urllib.request.urlopen(f"http://127.0.0.1:{port}/metrics", timeout=5)
        .read()
        .decode()
    )
    # The OTel Prometheus exporter (>= 0.66) adds otel_scope_* labels to
    # every series, so match the labels we own, not the whole label set.
    assert _sample(body, "stack_bcda_runs_total", kind="export", state="done") == 1.0
    assert _sample(body, "stack_bcda_page_views_total", page="exports") == 1.0
    assert _sample(body, "stack_bcda_store_jobs", status="complete") == 1.0
    assert _sample(body, "stack_bcda_store_bytes") == 0.0
    assert "stack_bcda_run_seconds" in body

    # the default port comes from the environment
    monkeypatch.setattr(metrics, "_server_port", None)
    env_port = _free_port()
    monkeypatch.setenv("BCDA_METRICS_PORT", str(env_port))
    assert metrics.serve() == env_port
    perf.shutdown()