windy-git/api/tests/test_telemetry.py

"""Field telemetry from the API (step 2): behavioural, no network.

A refusal must become exactly one forge.auth.failed row in the DECLARED shape
(the ledger quarantines anything else); non-refusal errors must not; the
heartbeat must count what happened and never invent a p95 for no traffic.
"""

from __future__ import annotations

import httpx
import pytest
from fastapi import Depends, FastAPI

from api.app import telemetry as tmod
from api.app.errors import RepairPointer

DECLARED_AUTH_KEYS = {"code", "http_status", "caller", "route", "upstream_status", "synthetic"}


def _app(tel: tmod.Telemetry) -> FastAPI:
    """The real middleware + handler, re-registered on a bare app (no DB)."""
    from api.app import main

    app = FastAPI()
    app.state.telemetry = tel
    app.middleware("http")(main._count_requests)
    app.exception_handler(RepairPointer)(main._repair_pointer_handler)

    def refuse(code: str, status: int):
        def dep():
            raise RepairPointer(
                status_code=status,
                code=code,
                speak="no",
                machine_cause="test",
                remediation_tool=None,
            )

        return dep

    @app.get("/api/v1/repos/{repo}/grants", dependencies=[Depends(refuse("passport_revoked", 403))])
    async def grants(repo: str):
        return {}

    @app.get("/api/v1/nope", dependencies=[Depends(refuse("repo_not_found", 404))])
    async def nope():
        return {}

    @app.get("/ok")
    async def ok():
        return {"ok": True}

    return app


async def _get(app, path, headers=None):
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://t") as c:
        return await c.get(path, headers=headers or {})


def _tel():
    return tmod.Telemetry(
        "http://ledger.invalid/v1/events",
        "tok",
        environment="test",
        commit_sha="abc1234",
        version="0.1.0",
    )


@pytest.mark.asyncio
async def test_refusal_emits_one_declared_row_with_route_template_not_path():
    tel = _tel()
    r = await _get(
        _app(tel),
        "/api/v1/repos/grandmas-secret-project/grants",
        {"Authorization": "Bearer eyJhbGciOiJFUzI1NiIsInR5cCI6IkVQVCJ9.e30.x"},
    )
    assert r.status_code == 403
    rows = [e for e in tel.buffer if e["event_type"] == "forge.auth.failed"]
    assert len(rows) == 1
    row = rows[0]
    assert row["actor_type"] == "system" and "actor_id" not in row
    assert set(row["metadata"]) <= DECLARED_AUTH_KEYS
    assert row["metadata"]["code"] == "passport_revoked"
    assert row["metadata"]["http_status"] == 403
    assert row["metadata"]["caller"] == "anonymous_agent"
    assert row["metadata"]["route"] == "/api/v1/repos/{repo}/grants"
    assert "grandmas-secret-project" not in str(row)


@pytest.mark.asyncio
async def test_non_auth_errors_are_counted_but_not_refusal_rows():
    tel = _tel()
    await _get(_app(tel), "/api/v1/nope")
    assert not [e for e in tel.buffer if e["event_type"] == "forge.auth.failed"]
    assert tel.errors_4xx == 1 and tel.refusals_4xx == 0


@pytest.mark.asyncio
async def test_heartbeat_counts_requests_refusals_and_p95():
    tel = _tel()
    app = _app(tel)
    for _ in range(3):
        await _get(app, "/ok")
    await _get(app, "/api/v1/repos/x/grants")
    meta = tel.health_row()
    assert meta["requests"] == 4 and meta["refusals_4xx"] == 1 and meta["errors_5xx"] == 0
    assert isinstance(meta["p95_ms"], int)
    assert {"interval_s", "uptime_s"} <= set(meta)


def test_no_traffic_means_no_p95_not_a_fake_zero():
    assert "p95_ms" not in _tel().health_row()


def test_no_token_sends_and_buffers_nothing():
    tel = tmod.Telemetry("http://ledger.invalid", "")
    tel.boot()
    tel.auth_failed(code="token_invalid", http_status=401, caller="unknown")
    assert tel.buffer == []


def test_unknown_code_is_never_sent_as_a_refusal():
    tel = _tel()
    tel.auth_failed(code="made_up_code", http_status=401, caller="unknown")
    assert tel.buffer == []


def test_boot_omits_an_unknown_commit_rather_than_inventing_one():
    tel = tmod.Telemetry("http://x", "tok", commit_sha=None, version="0.1.0")
    tel.boot()
    assert "commit_sha" not in tel.buffer[0]["metadata"]


def test_caller_classes_are_the_declared_three():
    assert tmod.caller_class({}) == "unknown"
    assert tmod.caller_class({"x-service-token": "s"}) == "unknown"
    assert (
        tmod.caller_class({"authorization": "Bearer eyJhbGciOiJSUzI1NiJ9.e30.x"})
        == "anonymous_human"
    )


@pytest.mark.asyncio
async def test_synthetic_header_marks_the_row_and_absent_means_real():
    async def refusal(headers):
        tel = _tel()
        await _get(_app(tel), "/api/v1/repos/x/grants", headers)
        return [e for e in tel.buffer if e["event_type"] == "forge.auth.failed"][0]["metadata"]["synthetic"]

    assert await refusal({"X-Windy-Synthetic": "1"}) is True
    assert await refusal({}) is False


def test_synthetic_is_forwarded_downstream_only_for_synthetic_requests():
    token = tmod.SYNTHETIC.set(True)
    try:
        assert tmod.synthetic_headers() == {"X-Windy-Synthetic": "1"}
    finally:
        tmod.SYNTHETIC.reset(token)
    assert tmod.synthetic_headers() == {}


# ---- UPDATE 7: the ledger answers 202 even when it quarantines rows ----------


@pytest.mark.asyncio
async def test_quarantined_rows_are_warned_and_counted_on_the_next_heartbeat(monkeypatch, caplog):
    tel = _tel()
    tel.boot()
    monkeypatch.setattr(
        tel, "_post", lambda b: (202, {"accepted": 0, "quarantined": 1, "rejections": ["undeclared key"]})
    )
    with caplog.at_level("WARNING", logger="windy-git.telemetry"):
        await tel.flush()
    assert tel.buffer == []  # sent; the ledger dead-lettered it, retrying won't help
    assert "QUARANTINED" in caplog.text and "undeclared key" in caplog.text
    tel.health()
    assert tel.buffer[-1]["metadata"]["telemetry_quarantined"] == 1
    assert tel.health_row()["telemetry_quarantined"] == 0  # reset per heartbeat window


@pytest.mark.asyncio
async def test_clean_send_reports_zero_and_logs_nothing(monkeypatch, caplog):
    tel = _tel()
    tel.boot()
    monkeypatch.setattr(tel, "_post", lambda b: (202, {"accepted": 1, "quarantined": 0, "rejections": []}))
    with caplog.at_level("WARNING", logger="windy-git.telemetry"):
        await tel.flush()
    assert caplog.text == ""
    row = tel.health_row()
    assert row["telemetry_quarantined"] == 0 and row["telemetry_dropped"] == 0


def test_buffer_overflow_is_counted_as_dropped(monkeypatch):
    monkeypatch.setattr(tmod, "MAX_BUFFER", 3)
    tel = _tel()
    for _ in range(5):
        tel.boot()
    assert len(tel.buffer) == 3
    assert tel.health_row()["telemetry_dropped"] == 2