iomgaa
82f065e4ef
asyncio.to_thread 桥接 SQLite，字段与 TelemetryRecorder Protocol 一一对应。 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
"""adapters/telemetry.py 单元测试 — SQLiteTelemetryRecorder。"""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from adapters.telemetry import SQLiteTelemetryRecorder
|
|
from core.protocols import TelemetryRecorder
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture()
|
|
def db_path(tmp_path):
|
|
"""返回临时数据库路径。"""
|
|
return tmp_path / "telemetry_test.db"
|
|
|
|
|
|
@pytest.fixture()
|
|
def recorder(db_path):
|
|
"""构造 SQLiteTelemetryRecorder 实例。"""
|
|
return SQLiteTelemetryRecorder(db_path=db_path)
|
|
|
|
|
|
def _make_call_kwargs(*, cache_hit: bool = False, error: str | None = None):
|
|
"""构造 record_llm_call 的标准参数字典。"""
|
|
return dict(
|
|
call_id=str(uuid.uuid4()),
|
|
parent_call_id=None,
|
|
session_id="sess-001",
|
|
model_name="gpt-4o",
|
|
provider="openai",
|
|
messages='[{"role":"user","content":"hi"}]',
|
|
response="hello",
|
|
thinking="",
|
|
prompt_tokens=10,
|
|
completion_tokens=5,
|
|
latency_ms=120,
|
|
ttft_ms=45.2,
|
|
max_inter_token_ms=12.3,
|
|
cache_hit=cache_hit,
|
|
error=error,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_satisfies_protocol(recorder):
|
|
"""SQLiteTelemetryRecorder 满足 TelemetryRecorder Protocol。"""
|
|
assert isinstance(recorder, TelemetryRecorder)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_creates_table_and_inserts(recorder, db_path):
|
|
"""首次写入应懒创建表并成功插入一条记录。"""
|
|
kwargs = _make_call_kwargs()
|
|
await recorder.record_llm_call(**kwargs)
|
|
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.row_factory = sqlite3.Row
|
|
rows = conn.execute("SELECT * FROM llm_calls").fetchall()
|
|
conn.close()
|
|
|
|
assert len(rows) == 1
|
|
row = rows[0]
|
|
assert row["call_id"] == kwargs["call_id"]
|
|
assert row["model_name"] == "gpt-4o"
|
|
assert row["prompt_tokens"] == 10
|
|
assert row["completion_tokens"] == 5
|
|
assert row["cache_hit"] == 0 # False → INTEGER 0
|
|
assert row["error"] is None
|
|
assert row["created_at"] is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_with_error(recorder, db_path):
|
|
"""error 字段非 None 时应正确存储。"""
|
|
kwargs = _make_call_kwargs(error="RateLimitError: 429")
|
|
await recorder.record_llm_call(**kwargs)
|
|
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.row_factory = sqlite3.Row
|
|
row = conn.execute("SELECT error FROM llm_calls WHERE call_id = ?", (kwargs["call_id"],)).fetchone()
|
|
conn.close()
|
|
|
|
assert row["error"] == "RateLimitError: 429"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_cache_hit(recorder, db_path):
|
|
"""cache_hit=True 时应存储为 INTEGER 1。"""
|
|
kwargs = _make_call_kwargs(cache_hit=True)
|
|
await recorder.record_llm_call(**kwargs)
|
|
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.row_factory = sqlite3.Row
|
|
row = conn.execute("SELECT cache_hit FROM llm_calls WHERE call_id = ?", (kwargs["call_id"],)).fetchone()
|
|
conn.close()
|
|
|
|
assert row["cache_hit"] == 1
|