shaguabijia-app-server/tests/test_observe.py

"""接口指标可观测(observe)单测:配置门槛 / 队列 / 中间件 / worker。

沿用仓库约定:TestClient + monkeypatch,绝不打真网络。observe 默认关(conftest 未设
OBSERVE_*),需要开启的用例用 monkeypatch 改 settings 单例属性。
"""
from __future__ import annotations

import asyncio

import httpx
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.core import observe, observe_worker
from app.core.config import settings


def test_observe_configured_requires_switch_and_creds(monkeypatch):
    # 开关开 + endpoint(默认 localhost)+ user + password 齐全 → True
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", True)
    monkeypatch.setattr(settings, "OBSERVE_USER", "u")
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p")
    assert settings.observe_configured is True

    # 缺密码 → False
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "")
    assert settings.observe_configured is False

    # 缺用户名 → False
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p")
    monkeypatch.setattr(settings, "OBSERVE_USER", "")
    assert settings.observe_configured is False

    # 开关关 → False(即便凭证齐全)
    monkeypatch.setattr(settings, "OBSERVE_USER", "u")
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
    assert settings.observe_configured is False


def test_record_event_enqueues(monkeypatch):
    q = asyncio.Queue(maxsize=10)
    monkeypatch.setattr(observe, "_queue", q)
    observe.record_event({"route": "/x"})
    assert q.get_nowait() == {"route": "/x"}


def test_record_event_drops_when_full(monkeypatch):
    q = asyncio.Queue(maxsize=1)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(observe, "_dropped", 0)
    observe.record_event({"n": 1})   # 占满
    observe.record_event({"n": 2})   # 满 → 丢弃当前,不抛异常
    assert observe.take_dropped() == 1
    assert observe.take_dropped() == 0   # 取出后清零
    assert q.get_nowait() == {"n": 1}    # 保留的是先到的


def _make_probe_app() -> FastAPI:
    """独立最小 app:只挂中间件 + 两个无鉴权路由,不碰真业务 DB/auth。"""
    app = FastAPI()
    app.add_middleware(observe.RequestMetricsMiddleware)

    @app.get("/things/{tid}")
    def get_thing(tid: str):
        return {"tid": tid}

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

    return app


@pytest.fixture
def observe_on(monkeypatch):
    """开启观测 + 换一个干净小队列,返回该队列供断言。"""
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", True)
    monkeypatch.setattr(settings, "OBSERVE_USER", "u")
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p")
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    return q


def test_middleware_records_route_template(observe_on):
    client = TestClient(_make_probe_app())
    r = client.get("/things/42")
    assert r.status_code == 200
    evt = observe_on.get_nowait()
    assert evt["route"] == "/things/{tid}"   # 模板,不是 /things/42
    assert evt["method"] == "GET"
    assert evt["status"] == 200
    assert evt["duration_ms"] >= 0
    assert evt["service"] and "env" in evt and isinstance(evt["_timestamp"], int)


def test_middleware_skips_health(observe_on):
    client = TestClient(_make_probe_app())
    client.get("/health")
    assert observe_on.empty()


def test_middleware_unmatched_route_is_normalized(observe_on):
    client = TestClient(_make_probe_app())
    r = client.get("/definitely-not-a-route")
    assert r.status_code == 404
    evt = observe_on.get_nowait()
    assert evt["route"] == "__unmatched__"
    assert evt["status"] == 404


def test_middleware_noop_when_disabled(monkeypatch):
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    client = TestClient(_make_probe_app())
    client.get("/things/1")
    assert q.empty()   # 未配置观测 → 零入队


async def test_collect_batch_drains_up_to_batch_max(monkeypatch):
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(settings, "OBSERVE_FLUSH_INTERVAL_SEC", 0.1)
    monkeypatch.setattr(settings, "OBSERVE_BATCH_MAX", 200)
    for i in range(3):
        q.put_nowait({"n": i})
    batch = await observe_worker._collect_batch()
    assert [e["n"] for e in batch] == [0, 1, 2]


async def test_collect_batch_timeout_returns_empty(monkeypatch):
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(settings, "OBSERVE_FLUSH_INTERVAL_SEC", 0.05)
    batch = await observe_worker._collect_batch()
    assert batch == []


async def test_post_batch_hits_json_ingest_url(monkeypatch):
    monkeypatch.setattr(settings, "OBSERVE_ORG", "default")
    monkeypatch.setattr(settings, "OBSERVE_STREAM", "app_requests")
    captured = {}

    def handler(request: httpx.Request) -> httpx.Response:
        captured["url"] = str(request.url)
        captured["json"] = request.content
        return httpx.Response(200, json={"code": 200})

    client = httpx.AsyncClient(
        base_url="http://oo", transport=httpx.MockTransport(handler)
    )
    await observe_worker._post_batch(client, [{"route": "/x", "status": 200}])
    await client.aclose()
    assert captured["url"] == "http://oo/api/default/app_requests/_json"
    assert b"/x" in captured["json"]


def test_start_observe_worker_noop_when_not_configured(monkeypatch):
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
    assert observe_worker.start_observe_worker() is None