guke
52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
"""接口指标可观测(observe)单测:配置门槛 / 队列 / 中间件 / worker。
|
|
|
|
沿用仓库约定:TestClient + monkeypatch,绝不打真网络。observe 默认关(conftest 未设
|
|
OBSERVE_*),需要开启的用例用 monkeypatch 改 settings 单例属性。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from app.core.config import settings
|
|
from app.core import observe
|
|
|
|
|
|
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} # 保留的是先到的
|