iomgaa
b6165ff438
The cache-key test only asserted a hit, so a key degraded to a constant would still pass it. Adding a namespace control group that must miss proves the key still distinguishes inputs; verified by degrading build_cache_key to a constant and watching the case go red. The allow_nan=False branch had no test at all. A ChatRequest built with a nan meta value (bypassing the entry validation, i.e. a future entry point that forgets to validate) must drop the row and not raise; verified red by removing allow_nan=False. Also restore the read-only file permissions in a finally block, so a failing assertion does not get masked by a PermissionError from tmp_path cleanup; rename the warnings fixture to captured_warnings so it stops shadowing the stdlib module; and drop a downstream business term from a fixture value (zero-business-assumption rule).
1141 lines
44 KiB
Python
1141 lines
44 KiB
Python
"""遥测子系统测试: SQLiteRecorder(22 列)+ TelemetryEmitter 单一 helper + TelemetryMW。"""
|
||
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from polygateway.errors import CircuitOpenError, RequestRejectedError
|
||
from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW
|
||
from polygateway.pricing import ModelPrice, PricingTable
|
||
from polygateway.telemetry.sqlite import SQLiteRecorder
|
||
from polygateway.types import ChatRequest, LLMResponse, SourceConfig
|
||
|
||
_REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}], session_id="sess-1")
|
||
|
||
_EXPECTED_COLUMNS = [
|
||
"call_id",
|
||
"parent_call_id",
|
||
"session_id",
|
||
"model",
|
||
"provider",
|
||
"source_name",
|
||
"messages",
|
||
"response",
|
||
"thinking",
|
||
"prompt_tokens",
|
||
"completion_tokens",
|
||
"usage_source",
|
||
"latency_ms",
|
||
"ttft_ms",
|
||
"max_inter_token_ms",
|
||
"cache_hit",
|
||
"error",
|
||
"cost",
|
||
"created_at",
|
||
"cached_prompt_tokens",
|
||
"model_reported",
|
||
"sampling",
|
||
"reasoning_tokens",
|
||
"tenant_id",
|
||
"meta",
|
||
]
|
||
|
||
|
||
def _resp(**overrides):
|
||
base = {
|
||
"content": "ok",
|
||
"thinking": "",
|
||
"model": "m",
|
||
"provider": "p",
|
||
"prompt_tokens": 1,
|
||
"completion_tokens": 2,
|
||
"latency_ms": 30,
|
||
"ttft_ms": None,
|
||
"max_inter_token_ms": None,
|
||
"cache_hit": False,
|
||
"call_id": "cid-1",
|
||
"source_name": "s1",
|
||
"usage_source": "measured",
|
||
}
|
||
base.update(overrides)
|
||
return LLMResponse(**base)
|
||
|
||
|
||
def _source(**overrides):
|
||
base = {
|
||
"name": "s1",
|
||
"provider": "p",
|
||
"base_url": "https://gw.example/v1",
|
||
"api_key": "sk",
|
||
"model": "m",
|
||
"timeout_s": 10.0,
|
||
}
|
||
base.update(overrides)
|
||
return SourceConfig(**base)
|
||
|
||
|
||
# 输出单价 8 元/百万: 改前 `unavailable` 行按兜底的 0/4000 换算恰好是 0.032
|
||
_PRICING = PricingTable({"m": ModelPrice(input_per_1m=1.0, output_per_1m=8.0)})
|
||
|
||
|
||
async def _record_minimal(recorder, call_id="c1", **overrides):
|
||
fields = {
|
||
"call_id": call_id,
|
||
"parent_call_id": None,
|
||
"session_id": "sess-1",
|
||
"model": "m",
|
||
"provider": "p",
|
||
"source_name": "s1",
|
||
"messages": "[]",
|
||
"response": "ok",
|
||
"thinking": "",
|
||
"prompt_tokens": 1,
|
||
"completion_tokens": 2,
|
||
"usage_source": "measured",
|
||
"latency_ms": 10,
|
||
"ttft_ms": None,
|
||
"max_inter_token_ms": None,
|
||
"cache_hit": False,
|
||
"error": None,
|
||
"cost": None,
|
||
"cached_prompt_tokens": None,
|
||
"model_reported": None,
|
||
"sampling": None,
|
||
"reasoning_tokens": None,
|
||
# 到达 recorder 时已由 emitter 归一化: None → '',空 dict → '{}'
|
||
"tenant_id": "",
|
||
"meta": "{}",
|
||
}
|
||
fields.update(overrides)
|
||
await recorder.record_llm_call(**fields)
|
||
|
||
|
||
class TestBackendColumnParity:
|
||
"""两个后端的 `_COLUMNS` 必须逐字同名同序(issue #11)。
|
||
|
||
emitter 只组装一份 `fields`,两个后端各自按自己的 `_COLUMNS` 取值;两份清单
|
||
一旦分叉,同一次调用在 SQLite 上写得进、在 PG 上抛 KeyError 被降级吞掉,
|
||
差异只在换后端时才暴露。列**序**同样断言: INSERT 用位置占位符,顺序错位
|
||
会把值写进错误的列而不报错。
|
||
"""
|
||
|
||
def test_two_backends_agree_on_columns(self):
|
||
from polygateway.telemetry.postgres import _COLUMNS as PG_COLUMNS
|
||
from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS
|
||
|
||
assert SQLITE_COLUMNS == PG_COLUMNS
|
||
|
||
def test_caller_dimensions_are_appended_last(self):
|
||
"""新列只能追加在末尾: 旧表经 ALTER 补列必落末尾,插在中间会让两条路径分叉。"""
|
||
from polygateway.telemetry.postgres import _COLUMNS as PG_COLUMNS
|
||
from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS
|
||
|
||
assert SQLITE_COLUMNS[-2:] == ("tenant_id", "meta")
|
||
assert PG_COLUMNS[-2:] == ("tenant_id", "meta")
|
||
|
||
|
||
class TestSQLiteRecorder:
|
||
async def test_schema_has_frozen_columns(self, tmp_path):
|
||
recorder = SQLiteRecorder(tmp_path / "t.db")
|
||
await _record_minimal(recorder)
|
||
recorder.close()
|
||
cols = [
|
||
r[1] for r in sqlite3.connect(tmp_path / "t.db").execute("PRAGMA table_info(llm_calls)")
|
||
]
|
||
assert cols == _EXPECTED_COLUMNS
|
||
|
||
async def test_call_id_idempotent(self, tmp_path):
|
||
recorder = SQLiteRecorder(tmp_path / "t.db")
|
||
await _record_minimal(recorder, call_id="dup")
|
||
await _record_minimal(recorder, call_id="dup", response="second")
|
||
recorder.close()
|
||
rows = (
|
||
sqlite3.connect(tmp_path / "t.db")
|
||
.execute("SELECT response FROM llm_calls WHERE call_id='dup'")
|
||
.fetchall()
|
||
)
|
||
assert rows == [("ok",)] # INSERT OR IGNORE: 第二次静默忽略
|
||
|
||
async def test_concurrent_writes_all_land(self, tmp_path):
|
||
recorder = SQLiteRecorder(tmp_path / "t.db")
|
||
await asyncio.gather(*(_record_minimal(recorder, call_id=f"c{i}") for i in range(50)))
|
||
recorder.close()
|
||
(count,) = (
|
||
sqlite3.connect(tmp_path / "t.db").execute("SELECT COUNT(*) FROM llm_calls").fetchone()
|
||
)
|
||
assert count == 50
|
||
|
||
async def test_unwritable_path_degrades_silently(self):
|
||
recorder = SQLiteRecorder(Path("/nonexistent-root/deep/t.db"))
|
||
await _record_minimal(recorder) # 不抛
|
||
recorder.close()
|
||
|
||
async def test_observability_columns_round_trip(self, tmp_path):
|
||
recorder = SQLiteRecorder(tmp_path / "t.db")
|
||
await _record_minimal(recorder, call_id="c-hit", cached_prompt_tokens=64)
|
||
await _record_minimal(recorder, call_id="c-zero", cached_prompt_tokens=0)
|
||
await _record_minimal(recorder, call_id="c-none", model_reported="MiniMax-Text-01")
|
||
recorder.close()
|
||
rows = dict(
|
||
sqlite3.connect(tmp_path / "t.db")
|
||
.execute("SELECT call_id, cached_prompt_tokens FROM llm_calls")
|
||
.fetchall()
|
||
)
|
||
assert rows["c-hit"] == 64
|
||
assert rows["c-zero"] == 0 # 真实零命中,读回仍是 0 而非 NULL
|
||
assert rows["c-none"] is None
|
||
|
||
async def test_reasoning_tokens_column_round_trip(self, tmp_path):
|
||
"""issue #6: 7 / 0 / None 三种值各自如实落库,0 与 NULL 不得混同。"""
|
||
recorder = SQLiteRecorder(tmp_path / "t.db")
|
||
await _record_minimal(recorder, call_id="r-some", reasoning_tokens=7)
|
||
await _record_minimal(recorder, call_id="r-zero", reasoning_tokens=0)
|
||
await _record_minimal(recorder, call_id="r-none", reasoning_tokens=None)
|
||
recorder.close()
|
||
rows = dict(
|
||
sqlite3.connect(tmp_path / "t.db")
|
||
.execute("SELECT call_id, reasoning_tokens FROM llm_calls")
|
||
.fetchall()
|
||
)
|
||
assert rows["r-some"] == 7
|
||
assert rows["r-zero"] == 0 # 上报了且确实没推理
|
||
assert rows["r-none"] is None # 本次调用未上报
|
||
|
||
async def test_sampling_column_round_trips(self, tmp_path):
|
||
"""issue #4: 采样参数落库,否则事后无法证明某批数据跑在什么温度下。"""
|
||
recorder = SQLiteRecorder(tmp_path / "t.db")
|
||
await _record_minimal(recorder, call_id="c-s", sampling='{"seed": 42, "temperature": 0}')
|
||
await _record_minimal(recorder, call_id="c-plain")
|
||
recorder.close()
|
||
rows = dict(
|
||
sqlite3.connect(tmp_path / "t.db")
|
||
.execute("SELECT call_id, sampling FROM llm_calls")
|
||
.fetchall()
|
||
)
|
||
assert json.loads(rows["c-s"]) == {"seed": 42, "temperature": 0}
|
||
assert rows["c-plain"] is None # 无采样参数为 NULL,便于 SQL 过滤
|
||
|
||
|
||
class TestSQLiteColumnBackfill:
|
||
"""issue #3: 已存在的 18 列旧表必须自动补列,否则每行写入都被丢弃。"""
|
||
|
||
_LEGACY_DDL = """
|
||
CREATE TABLE llm_calls (
|
||
call_id TEXT PRIMARY KEY,
|
||
parent_call_id TEXT,
|
||
session_id TEXT,
|
||
model TEXT NOT NULL,
|
||
provider TEXT NOT NULL,
|
||
source_name TEXT NOT NULL,
|
||
messages TEXT NOT NULL,
|
||
response TEXT NOT NULL,
|
||
thinking TEXT NOT NULL DEFAULT '',
|
||
prompt_tokens INTEGER NOT NULL,
|
||
completion_tokens INTEGER NOT NULL,
|
||
usage_source TEXT NOT NULL,
|
||
latency_ms INTEGER NOT NULL,
|
||
ttft_ms REAL,
|
||
max_inter_token_ms REAL,
|
||
cache_hit INTEGER NOT NULL DEFAULT 0,
|
||
error TEXT,
|
||
cost REAL,
|
||
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||
);
|
||
"""
|
||
|
||
async def test_legacy_table_is_upgraded_in_place(self, tmp_path):
|
||
db = tmp_path / "legacy.db"
|
||
legacy = sqlite3.connect(db)
|
||
legacy.execute(self._LEGACY_DDL)
|
||
legacy.commit()
|
||
legacy.close()
|
||
|
||
recorder = SQLiteRecorder(db)
|
||
await _record_minimal(recorder, cached_prompt_tokens=7, model_reported="m-real")
|
||
recorder.close()
|
||
|
||
conn = sqlite3.connect(db)
|
||
cols = [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")]
|
||
assert cols == _EXPECTED_COLUMNS # ALTER 追加到末尾,与新建库列序一致
|
||
assert conn.execute(
|
||
"SELECT cached_prompt_tokens, model_reported FROM llm_calls"
|
||
).fetchone() == (7, "m-real")
|
||
|
||
async def test_backfill_failure_keeps_the_recorder_usable(self, tmp_path):
|
||
"""补列失败只能逐行降级,绝不能把 recorder 整体变成 no-op(设计 D1 纪律)。
|
||
|
||
把 llm_calls 建成同名 view: `CREATE TABLE IF NOT EXISTS` 遇 view 静默
|
||
no-op(不抛),随后的 ALTER 才抛 "Cannot add a column to a view"——正是
|
||
补列失败这条分支。`_conn` 必须保持非 None,否则整个 recorder 永久失能。
|
||
"""
|
||
db = tmp_path / "view.db"
|
||
conn = sqlite3.connect(db)
|
||
conn.execute("CREATE TABLE real_rows (call_id TEXT)")
|
||
conn.execute("CREATE VIEW llm_calls AS SELECT call_id FROM real_rows")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
recorder = SQLiteRecorder(db) # 不得抛
|
||
assert recorder._conn is not None # 补列失败 ≠ recorder 失能(D1 纪律)
|
||
await _record_minimal(recorder) # 不得抛
|
||
recorder.close()
|
||
|
||
|
||
# issue #11 之前的表形态: 22 个 recorder 字段 + created_at = 23 个物理列,没有任何租户维度
|
||
_PRE_TENANT_DDL = """
|
||
CREATE TABLE llm_calls (
|
||
call_id TEXT PRIMARY KEY,
|
||
parent_call_id TEXT,
|
||
session_id TEXT,
|
||
model TEXT NOT NULL,
|
||
provider TEXT NOT NULL,
|
||
source_name TEXT NOT NULL,
|
||
messages TEXT NOT NULL,
|
||
response TEXT NOT NULL,
|
||
thinking TEXT NOT NULL DEFAULT '',
|
||
prompt_tokens INTEGER NOT NULL,
|
||
completion_tokens INTEGER NOT NULL,
|
||
usage_source TEXT NOT NULL,
|
||
latency_ms INTEGER NOT NULL,
|
||
ttft_ms REAL,
|
||
max_inter_token_ms REAL,
|
||
cache_hit INTEGER NOT NULL DEFAULT 0,
|
||
error TEXT,
|
||
cost REAL,
|
||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||
cached_prompt_tokens INTEGER,
|
||
model_reported TEXT,
|
||
sampling TEXT,
|
||
reasoning_tokens INTEGER
|
||
);
|
||
"""
|
||
|
||
_PRE_TENANT_INSERT = (
|
||
"INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, "
|
||
"prompt_tokens, completion_tokens, usage_source, latency_ms) "
|
||
"VALUES ('old-row', 'm', 'p', 's1', '[]', 'old body', 1, 2, 'measured', 10)"
|
||
)
|
||
|
||
|
||
def _make_pre_tenant_db(path: Path) -> None:
|
||
"""造一个 issue #11 之前的库: 22 字段旧表 + 一行没有租户归属的历史数据。"""
|
||
conn = sqlite3.connect(path)
|
||
conn.execute(_PRE_TENANT_DDL)
|
||
conn.execute(_PRE_TENANT_INSERT)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
class TestSQLiteCallerDimensionsAcceptance:
|
||
"""issue #11 的机械化验收(SQLite 侧,真实临时文件): 新建库 / 旧表补列 / 补列失败方向。"""
|
||
|
||
async def test_fresh_db_round_trips_the_dimensions(self, tmp_path):
|
||
"""新建库: 列齐全,且维度值原样读回——只验列存在会漏掉写错列位的错。"""
|
||
db = tmp_path / "fresh.db"
|
||
recorder = SQLiteRecorder(db)
|
||
await _record_minimal(
|
||
recorder, call_id="c-dim", tenant_id="tenant-a", meta='{"batch": "b7"}'
|
||
)
|
||
recorder.close()
|
||
|
||
conn = sqlite3.connect(db)
|
||
assert [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")] == _EXPECTED_COLUMNS
|
||
row = conn.execute(
|
||
"SELECT tenant_id, meta FROM llm_calls WHERE call_id = 'c-dim'"
|
||
).fetchone()
|
||
assert row[0] == "tenant-a"
|
||
assert json.loads(row[1]) == {"batch": "b7"}
|
||
|
||
async def test_pre_tenant_table_gains_columns_and_old_rows_stay_auditable(self, tmp_path):
|
||
"""22 字段旧表补列后,新行带维度,而**老行的 tenant_id 是空串而非 NULL**。
|
||
|
||
这条直接验收 issue #11 的核心论点(先启用落库、后加列,补列之前的行没有
|
||
租户归属)。断言方向必须是空串: PG 的 RLS `USING` 表达式对返回 false **或
|
||
NULL** 的行一律隐藏且不报错,故 NULL 的 `tenant_id` 不是"未归属",而是对
|
||
所有人永久不可见的黑洞;哨兵空串则能被一条 `COUNT(*) WHERE tenant_id = ''`
|
||
审计出来,历史欠账是可见、可量化、可补录的。
|
||
"""
|
||
db = tmp_path / "pre_tenant.db"
|
||
_make_pre_tenant_db(db)
|
||
|
||
recorder = SQLiteRecorder(db)
|
||
await _record_minimal(recorder, call_id="new-row", tenant_id="tenant-a", meta='{"k": 1}')
|
||
recorder.close()
|
||
|
||
conn = sqlite3.connect(db)
|
||
cols = [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")]
|
||
assert cols == _EXPECTED_COLUMNS # 22 → 24 个 recorder 字段(+ created_at 共 25 物理列)
|
||
rows = dict(conn.execute("SELECT call_id, tenant_id FROM llm_calls").fetchall())
|
||
assert rows["new-row"] == "tenant-a"
|
||
assert rows["old-row"] == "" # 不是 None: NULL 会被 RLS 静默吞掉
|
||
assert (
|
||
conn.execute("SELECT meta FROM llm_calls WHERE call_id = 'old-row'").fetchone()[0]
|
||
== "{}"
|
||
)
|
||
|
||
async def test_readonly_file_backfill_failure_keeps_the_recorder_alive(self, tmp_path):
|
||
"""补列失败的降级方向(SQLite 等价构造: 文件只读)。
|
||
|
||
SQLite 没有角色权限模型,与 PG「只有 SELECT/INSERT 权限的角色」等价的构造
|
||
是文件本身只读。库文件必须**预先置为 WAL 且干净关闭**,否则 `__init__` 的
|
||
`PRAGMA journal_mode=WAL` 会先撞上只读而让失败点跑到补列之前,测不到本用例
|
||
要测的那条分支(实测: 非 WAL 库 chmod 444 后该 PRAGMA 报 readonly database)。
|
||
只读库连 INSERT 都做不了,故这里**只断言**补列失败不清空 `_conn`、不抛出
|
||
`__init__`(sqlite.py `_backfill_columns` 那条纪律),不断言"写入仍成功"。
|
||
"""
|
||
if os.geteuid() == 0:
|
||
pytest.skip("root 无视文件权限位,只读构造不成立")
|
||
db = tmp_path / "readonly.db"
|
||
conn = sqlite3.connect(db)
|
||
conn.execute("PRAGMA journal_mode=WAL") # 预置 WAL: 让只读连接不必改日志模式
|
||
conn.execute(_PRE_TENANT_DDL)
|
||
conn.execute(_PRE_TENANT_INSERT)
|
||
conn.commit()
|
||
conn.close()
|
||
db.chmod(0o444)
|
||
|
||
# finally 还原权限位: 任一断言先失败时,不还原会让 tmp_path 清理连带报错,
|
||
# 把"某条断言失败"的真因盖成一个无关的 PermissionError
|
||
try:
|
||
recorder = SQLiteRecorder(db) # 不得抛
|
||
assert recorder._conn is not None # 补列失败 ≠ recorder 失能
|
||
await _record_minimal(recorder, call_id="doomed") # 只读库写不进,但不得抛
|
||
recorder.close()
|
||
finally:
|
||
db.chmod(0o644)
|
||
|
||
stale = sqlite3.connect(db)
|
||
assert [r[1] for r in stale.execute("PRAGMA table_info(llm_calls)")] == (
|
||
_EXPECTED_COLUMNS[:-2]
|
||
) # 补列确实没成功,用例不是在只读库上空转
|
||
|
||
|
||
class _FakePgConn:
|
||
"""记录执行过的语句;可让 ALTER/CREATE/探测抛错以模拟权限不足与抖动。
|
||
|
||
`existing` 为空列表即表示**表不存在**(与真实 PG 一致: `to_regclass` 为 NULL
|
||
时列探测必然零行),故 `fetchval` 与 `fetch` 共用同一份事实。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
existing: list[str],
|
||
*,
|
||
fail_alter: bool = False,
|
||
fail_create: bool = False,
|
||
probe_errors: int = 0,
|
||
):
|
||
self.existing = existing
|
||
self.fail_alter = fail_alter
|
||
self.fail_create = fail_create
|
||
self.probe_errors = probe_errors
|
||
self.statements: list[str] = []
|
||
|
||
async def execute(self, sql, *args):
|
||
self.statements.append(sql)
|
||
if sql.startswith("ALTER TABLE") and self.fail_alter:
|
||
raise RuntimeError("must be owner of table llm_calls")
|
||
if sql.lstrip().startswith("CREATE TABLE"):
|
||
if self.fail_create:
|
||
raise RuntimeError("permission denied for schema public")
|
||
self.existing = list(_EXPECTED_COLUMNS)
|
||
|
||
async def fetchval(self, sql, *args):
|
||
self.statements.append(sql)
|
||
if self.probe_errors > 0:
|
||
self.probe_errors -= 1
|
||
raise RuntimeError("connection was closed in the middle of operation")
|
||
return "llm_calls" if self.existing else None
|
||
|
||
async def fetch(self, sql, *args):
|
||
self.statements.append(sql)
|
||
return [{"attname": name} for name in self.existing]
|
||
|
||
|
||
class _FakePgPool:
|
||
def __init__(self, conn):
|
||
self._conn = conn
|
||
|
||
def acquire(self):
|
||
conn = self._conn
|
||
|
||
class _Ctx:
|
||
async def __aenter__(self):
|
||
return conn
|
||
|
||
async def __aexit__(self, *exc):
|
||
return False
|
||
|
||
return _Ctx()
|
||
|
||
|
||
class TestPostgresBackfillDiscipline:
|
||
"""PG 补列必须与 SQLite 侧对称: 失败只逐行降级,且稳态不抢排他锁(issue #3)。"""
|
||
|
||
_LEGACY = ["call_id", "cost", "created_at"]
|
||
_CURRENT = [
|
||
"call_id",
|
||
"cost",
|
||
"created_at",
|
||
"cached_prompt_tokens",
|
||
"model_reported",
|
||
"sampling",
|
||
"reasoning_tokens",
|
||
"tenant_id",
|
||
"meta",
|
||
]
|
||
|
||
def _recorder(self, conn):
|
||
from polygateway.telemetry.postgres import PostgresRecorder
|
||
|
||
return PostgresRecorder("postgresql://u:p@h:5432/polygateway", pool=_FakePgPool(conn))
|
||
|
||
async def test_alter_failure_does_not_disable_the_recorder(self):
|
||
"""ALTER 失败(如账号只有 INSERT 权限)不得置 _failed —— 那会让遥测全灭。"""
|
||
conn = _FakePgConn(self._LEGACY, fail_alter=True)
|
||
recorder = self._recorder(conn)
|
||
await _record_minimal(recorder) # 不得抛
|
||
assert recorder._failed is False
|
||
assert any(s.startswith("INSERT INTO llm_calls") for s in conn.statements)
|
||
|
||
async def test_no_alter_when_columns_already_exist(self):
|
||
"""ADD COLUMN IF NOT EXISTS 即使列已存在也会先抢 ACCESS EXCLUSIVE 锁,
|
||
|
||
而遥测是内联 await——稳态下必须一条 ALTER 都不发,否则每个进程的首次
|
||
写入都会去锁共享审计表。
|
||
"""
|
||
conn = _FakePgConn(self._CURRENT)
|
||
await _record_minimal(self._recorder(conn))
|
||
assert not [s for s in conn.statements if s.startswith("ALTER TABLE")]
|
||
|
||
async def test_missing_columns_are_added_once(self):
|
||
conn = _FakePgConn(self._LEGACY)
|
||
await _record_minimal(self._recorder(conn))
|
||
from polygateway.telemetry.postgres import _BACKFILL
|
||
|
||
altered = [s for s in conn.statements if s.startswith("ALTER TABLE")]
|
||
assert len(altered) == len(_BACKFILL) # 旧表缺全部补列,故一列一条 ALTER
|
||
assert all("IF NOT EXISTS" not in s for s in altered) # 探测已确认缺列,无需再判
|
||
|
||
|
||
class TestPostgresTableProbe:
|
||
"""建表必须先探测,且"判死"只认"确定写不进去"(issue #9)。
|
||
|
||
实测(PostgreSQL 16.14,只有表级 SELECT/INSERT 的角色): `CREATE TABLE IF NOT
|
||
EXISTS` 被拒 permission denied for schema,而同一连接的 `INSERT` 通过——
|
||
PG 对 schema 的 CREATE 权限检查早于 `IF NOT EXISTS` 的存在性判断。无条件发
|
||
DDL 会让这类最小权限部署的整个进程静默失遥测。
|
||
"""
|
||
|
||
_CURRENT = [
|
||
"call_id",
|
||
"cost",
|
||
"created_at",
|
||
"cached_prompt_tokens",
|
||
"model_reported",
|
||
"sampling",
|
||
"reasoning_tokens",
|
||
"tenant_id",
|
||
"meta",
|
||
]
|
||
|
||
def _recorder(self, conn):
|
||
from polygateway.telemetry.postgres import PostgresRecorder
|
||
|
||
return PostgresRecorder("postgresql://u:p@h:5432/polygateway", pool=_FakePgPool(conn))
|
||
|
||
def _created(self, conn):
|
||
return [s for s in conn.statements if s.lstrip().startswith("CREATE TABLE")]
|
||
|
||
async def test_existing_table_is_never_recreated(self):
|
||
"""表已存在就一条 DDL 都不发——这是权限被拒的唯一根治办法。"""
|
||
conn = _FakePgConn(self._CURRENT)
|
||
await _record_minimal(self._recorder(conn))
|
||
assert not self._created(conn)
|
||
|
||
async def test_create_denied_on_existing_table_keeps_recording(self):
|
||
"""就算 DDL 仍被发出并被拒,表存在时也不得判死整个 recorder。"""
|
||
conn = _FakePgConn(self._CURRENT, fail_create=True)
|
||
recorder = self._recorder(conn)
|
||
await _record_minimal(recorder) # 不得抛
|
||
assert recorder._failed is False
|
||
assert any(s.startswith("INSERT INTO llm_calls") for s in conn.statements)
|
||
|
||
async def test_missing_table_is_created_and_not_backfilled(self):
|
||
"""表不存在→建表;新建表列已齐全,不得再发补列 ALTER。"""
|
||
conn = _FakePgConn([])
|
||
recorder = self._recorder(conn)
|
||
await _record_minimal(recorder)
|
||
assert len(self._created(conn)) == 1
|
||
assert not [s for s in conn.statements if s.startswith("ALTER TABLE")]
|
||
assert recorder._failed is False
|
||
assert any(s.startswith("INSERT INTO llm_calls") for s in conn.statements)
|
||
|
||
async def test_create_failure_on_missing_table_degrades_to_noop(self):
|
||
"""表确定不存在且建不出来 = 确定写不进去: 此时才允许永久 no-op。"""
|
||
conn = _FakePgConn([], fail_create=True)
|
||
recorder = self._recorder(conn)
|
||
await _record_minimal(recorder) # 不得抛
|
||
assert recorder._failed is True
|
||
assert not [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]
|
||
|
||
async def test_probe_failure_is_transient_not_terminal(self):
|
||
"""探测失败多为连接抖动: 跳过本次,下次调用必须重试,绝不永久判死。"""
|
||
conn = _FakePgConn(self._CURRENT, probe_errors=1)
|
||
recorder = self._recorder(conn)
|
||
await _record_minimal(recorder, call_id="first") # 不得抛
|
||
assert recorder._failed is False
|
||
assert not [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]
|
||
await _record_minimal(recorder, call_id="second")
|
||
assert [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]
|
||
|
||
|
||
class _MemoryRecorder:
|
||
def __init__(self):
|
||
self.rows = []
|
||
|
||
async def record_llm_call(self, **fields):
|
||
self.rows.append(fields)
|
||
|
||
|
||
class TestEmitterRecorderContract:
|
||
"""emitter 的实参键集合必须与两个后端的 _COLUMNS 完全一致(issue #3)。
|
||
|
||
两个后端的 `row = tuple(fields[col] for col in _COLUMNS)` 都在 try **之外**,
|
||
emitter 漏传一个键就抛 KeyError,被 `_record` 的 except Exception 吞成 warning
|
||
→ 遥测静默全丢。而 8 个 `**fields` 形态的 fake 一个都拦不住,故显式断言。
|
||
"""
|
||
|
||
async def test_emitter_supplies_exactly_the_backend_columns(self):
|
||
from polygateway.telemetry.postgres import _COLUMNS as PG_COLUMNS
|
||
from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS
|
||
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec).emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="cid-1",
|
||
latency_ms=42,
|
||
response=_resp(),
|
||
error=None,
|
||
)
|
||
assert set(rec.rows[0]) == set(SQLITE_COLUMNS) == set(PG_COLUMNS)
|
||
|
||
@pytest.mark.parametrize("emit", ["attempt", "cache_hit", "terminal_failure"])
|
||
async def test_every_entry_point_supplies_the_same_keys(self, emit):
|
||
from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS
|
||
|
||
rec = _MemoryRecorder()
|
||
emitter = TelemetryEmitter(rec)
|
||
if emit == "attempt":
|
||
await emitter.emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="c",
|
||
latency_ms=1,
|
||
response=None,
|
||
error="boom",
|
||
)
|
||
elif emit == "cache_hit":
|
||
await emitter.emit_cache_hit(request=_REQ, response=_resp())
|
||
else:
|
||
await emitter.emit_terminal_failure(
|
||
request=_REQ, call_id="c", latency_ms=1, error="dead"
|
||
)
|
||
assert set(rec.rows[0]) == set(SQLITE_COLUMNS)
|
||
|
||
|
||
class TestEmitterObservabilityFields:
|
||
"""issue #3: 三个入口各自的取值口径(设计 §5 表)。"""
|
||
|
||
async def test_attempt_carries_the_response_values(self):
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec).emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="cid-1",
|
||
latency_ms=42,
|
||
response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7),
|
||
error=None,
|
||
)
|
||
assert rec.rows[0]["cached_prompt_tokens"] == 64
|
||
assert rec.rows[0]["model_reported"] == "m-real"
|
||
assert rec.rows[0]["reasoning_tokens"] == 7
|
||
|
||
async def test_failed_attempt_has_no_provider_facts(self):
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec).emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="cid-2",
|
||
latency_ms=7,
|
||
response=None,
|
||
error="boom",
|
||
)
|
||
assert rec.rows[0]["cached_prompt_tokens"] is None
|
||
assert rec.rows[0]["model_reported"] is None
|
||
assert rec.rows[0]["reasoning_tokens"] is None
|
||
|
||
async def test_cache_hit_replays_the_recorded_values(self):
|
||
"""决策 B1: 命中行原样回放,故命中率统计必须带 WHERE cache_hit = false。"""
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec).emit_cache_hit(
|
||
request=_REQ,
|
||
response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7),
|
||
)
|
||
row = rec.rows[0]
|
||
assert row["cache_hit"] is True
|
||
assert row["cached_prompt_tokens"] == 64 and row["model_reported"] == "m-real"
|
||
assert row["reasoning_tokens"] == 7 # 与 cached 同口径原样回放
|
||
|
||
async def test_terminal_failure_records_none(self):
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec).emit_terminal_failure(
|
||
request=_REQ, call_id="c", latency_ms=1, error="dead"
|
||
)
|
||
assert rec.rows[0]["cached_prompt_tokens"] is None
|
||
assert rec.rows[0]["model_reported"] is None
|
||
assert rec.rows[0]["reasoning_tokens"] is None
|
||
|
||
|
||
class TestEmitterSamplingColumn:
|
||
"""issue #4: sampling 列在三个入口的口径(设计决策 D 表格)。
|
||
|
||
列语义 = 「调用方采样意图 ⊎ 生效源 extra_body」,**不含**结构化注入的
|
||
response_format(列名是采样参数,schema 不是;且数 KB schema 逐行落库会让
|
||
审计表无谓膨胀)。三入口若各读各的层,同一列在不同行含义就不同。
|
||
"""
|
||
|
||
_SAMPLED = ChatRequest(
|
||
messages=[{"role": "user", "content": "hi"}],
|
||
sampling={"seed": 42},
|
||
overlay={"seed": 42, "response_format": {"type": "json_object"}},
|
||
)
|
||
|
||
async def test_attempt_merges_source_extra_body(self):
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec).emit_attempt(
|
||
request=self._SAMPLED,
|
||
source=_source(extra_body={"temperature": 0}),
|
||
call_id="c",
|
||
latency_ms=1,
|
||
response=_resp(),
|
||
error=None,
|
||
)
|
||
assert json.loads(rec.rows[0]["sampling"]) == {"seed": 42, "temperature": 0}
|
||
|
||
async def test_response_format_never_leaks_into_the_column(self):
|
||
"""三行都不得出现 response_format——它不是采样参数。"""
|
||
rec = _MemoryRecorder()
|
||
emitter = TelemetryEmitter(rec)
|
||
await emitter.emit_attempt(
|
||
request=self._SAMPLED,
|
||
source=_source(),
|
||
call_id="c",
|
||
latency_ms=1,
|
||
response=_resp(),
|
||
error=None,
|
||
)
|
||
await emitter.emit_cache_hit(request=self._SAMPLED, response=_resp())
|
||
await emitter.emit_terminal_failure(
|
||
request=self._SAMPLED, call_id="c", latency_ms=1, error="dead"
|
||
)
|
||
assert len(rec.rows) == 3
|
||
for row in rec.rows:
|
||
assert "response_format" not in row["sampling"]
|
||
|
||
@pytest.mark.parametrize("emit", ["cache_hit", "terminal_failure"])
|
||
async def test_sourceless_entries_record_call_level_only(self, emit):
|
||
"""两个最外层入口没有"生效源"可言,与 model/source_name 置空同一先例。"""
|
||
rec = _MemoryRecorder()
|
||
emitter = TelemetryEmitter(rec)
|
||
if emit == "cache_hit":
|
||
await emitter.emit_cache_hit(request=self._SAMPLED, response=_resp())
|
||
else:
|
||
await emitter.emit_terminal_failure(
|
||
request=self._SAMPLED, call_id="c", latency_ms=1, error="dead"
|
||
)
|
||
assert json.loads(rec.rows[0]["sampling"]) == {"seed": 42}
|
||
|
||
async def test_absent_sampling_is_null(self):
|
||
"""无采样参数时为 NULL,而非空字符串或 "{}"——便于 SQL 过滤。"""
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec).emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="c",
|
||
latency_ms=1,
|
||
response=_resp(),
|
||
error=None,
|
||
)
|
||
assert rec.rows[0]["sampling"] is None
|
||
|
||
|
||
class TestEmitterCallerDimensions:
|
||
"""issue #11: 三个 emit 入口统一从 `request` 读维度,`_record` 落库前归一化。
|
||
|
||
维度只有一个读取点(`request`),否则同一列在三种行里口径分叉——那正是
|
||
"遥测调用点收敛为单一 helper"这条铁律要防的形态。
|
||
"""
|
||
|
||
_META = {"z_last": "z", "a_first": 1, "m_mid": True}
|
||
_REQ_A = ChatRequest(
|
||
messages=[{"role": "user", "content": "hi"}],
|
||
session_id="sess-1",
|
||
tenant_id="tenant-a",
|
||
meta=_META,
|
||
)
|
||
|
||
@pytest.mark.parametrize("emit", ["attempt", "cache_hit", "terminal_failure"])
|
||
async def test_every_entry_point_carries_the_dimensions(self, emit):
|
||
"""三条路径写出的行都必须带维度: 漏掉任一条,该租户的账就永远对不上。"""
|
||
rec = _MemoryRecorder()
|
||
emitter = TelemetryEmitter(rec)
|
||
if emit == "attempt":
|
||
await emitter.emit_attempt(
|
||
request=self._REQ_A,
|
||
source=_source(),
|
||
call_id="c",
|
||
latency_ms=1,
|
||
response=_resp(),
|
||
error=None,
|
||
)
|
||
elif emit == "cache_hit":
|
||
await emitter.emit_cache_hit(request=self._REQ_A, response=_resp(cache_hit=True))
|
||
else:
|
||
await emitter.emit_terminal_failure(
|
||
request=self._REQ_A, call_id="c", latency_ms=1, error="dead"
|
||
)
|
||
row = rec.rows[0]
|
||
assert row["tenant_id"] == "tenant-a"
|
||
assert json.loads(row["meta"]) == self._META
|
||
|
||
async def test_cache_hit_records_the_current_caller_not_the_cached_one(self):
|
||
"""缓存命中行的维度是"本次由谁发起",不是历史那次——最容易实现反的一处。
|
||
|
||
历史那次由租户 B 发起并把响应留在了缓存里;本次由租户 A 发起并命中。
|
||
若读了历史那次的归属,租户 A 的调用会记到 B 头上,而 A 的账面凭空少一行
|
||
——两个租户的账同时错,且错得没有任何报错。
|
||
"""
|
||
historical = ChatRequest(
|
||
messages=[{"role": "user", "content": "hi"}],
|
||
tenant_id="tenant-b",
|
||
meta={"batch": "old-batch"},
|
||
)
|
||
rec = _MemoryRecorder()
|
||
mw = TelemetryMW(TelemetryEmitter(rec))
|
||
|
||
async def terminal(request):
|
||
# 缓存层回放的是历史那次的响应对象(其 call_id 属于 historical 那次)
|
||
return _resp(cache_hit=True, latency_ms=0, call_id="cache-cid")
|
||
|
||
assert historical.tenant_id == "tenant-b" # 历史归属确实不同,否则本用例是空转
|
||
await mw(self._REQ_A, terminal)
|
||
|
||
row = rec.rows[0]
|
||
assert row["cache_hit"] is True
|
||
assert row["tenant_id"] == "tenant-a"
|
||
assert "old-batch" not in row["meta"]
|
||
|
||
async def test_absent_dimensions_land_as_sentinels(self):
|
||
"""未传维度落哨兵值: `tenant_id` 空串、`meta` 字面量 `'{}'`,都不是 NULL。
|
||
|
||
NULL 的 `tenant_id` 在 PG 的 RLS policy 下对所有人永久不可见(设计 §4.4),
|
||
空串则可用一条 SQL 审计出还有多少行未归属;`meta` 同理,`'{}'` 可被
|
||
JSON 函数直接查询,NULL 则要每条查询都额外判空。
|
||
"""
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec).emit_attempt(
|
||
request=_REQ, # tenant_id=None, meta={}
|
||
source=_source(),
|
||
call_id="c",
|
||
latency_ms=1,
|
||
response=_resp(),
|
||
error=None,
|
||
)
|
||
row = rec.rows[0]
|
||
assert row["tenant_id"] == ""
|
||
assert row["meta"] == "{}"
|
||
|
||
async def test_meta_is_serialized_with_sorted_keys(self):
|
||
"""键序固定,同一份维度在任意两行里字节一致,可直接做等值比对与去重。"""
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec).emit_terminal_failure(
|
||
request=self._REQ_A, call_id="c", latency_ms=1, error="dead"
|
||
)
|
||
assert list(json.loads(rec.rows[0]["meta"])) == ["a_first", "m_mid", "z_last"]
|
||
|
||
async def test_non_ascii_meta_stays_readable(self):
|
||
"""`ensure_ascii=False`: 中文维度按原文落库,而非 `\\uXXXX` 转义串。"""
|
||
rec = _MemoryRecorder()
|
||
req = ChatRequest(messages=[{"role": "user", "content": "hi"}], meta={"dept": "研发"})
|
||
await TelemetryEmitter(rec).emit_terminal_failure(
|
||
request=req, call_id="c", latency_ms=1, error="dead"
|
||
)
|
||
assert "研发" in rec.rows[0]["meta"]
|
||
|
||
async def test_non_finite_meta_value_drops_the_row_instead_of_poisoning_it(self):
|
||
"""入口失守时 `allow_nan=False` 的真实结果: 整行降级丢弃,且不抛给调用方。
|
||
|
||
直接构造带 `nan` 的 `ChatRequest`(绕过 `validate_caller_dimensions` 这道
|
||
主防线,模拟将来某个新入口忘记校验)。没有 `allow_nan=False` 时,
|
||
`json.dumps` 会写出裸 `NaN` 字面量——PG 的 JSONB 会拒收,但 **SQLite 的
|
||
`meta` 是 TEXT 列不做校验**,那串非法 JSON 会被静默存进去,污染此后一切
|
||
按 JSON 解析 meta 的分析。宁可丢一行遥测,也不要一行毒数据。
|
||
|
||
同时断言不抛: 遥测的降级方向是"静默降级"(铁律),把调用方的一次正常
|
||
业务调用因为一个维度值炸掉,方向反了。
|
||
"""
|
||
rec = _MemoryRecorder()
|
||
req = ChatRequest(messages=[{"role": "user", "content": "hi"}], meta={"k": float("nan")})
|
||
await TelemetryEmitter(rec).emit_terminal_failure(
|
||
request=req, call_id="c", latency_ms=1, error="dead"
|
||
)
|
||
assert rec.rows == []
|
||
|
||
|
||
class TestCostWithCachedTier:
|
||
"""issue #3: 命中部分按缓存单价计费,避免 cost 系统性高估。"""
|
||
|
||
_TABLE = PricingTable(
|
||
{"m": ModelPrice(input_per_1m=10.0, output_per_1m=20.0, cached_input_per_1m=2.0)}
|
||
)
|
||
|
||
async def test_cached_hit_lowers_the_recorded_cost(self):
|
||
rec = _MemoryRecorder()
|
||
emitter = TelemetryEmitter(rec, pricing=self._TABLE)
|
||
full = _resp(prompt_tokens=1_000_000, completion_tokens=0)
|
||
await emitter.emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="c1",
|
||
latency_ms=1,
|
||
response=full,
|
||
error=None,
|
||
)
|
||
await emitter.emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="c2",
|
||
latency_ms=1,
|
||
response=_resp(
|
||
prompt_tokens=1_000_000, completion_tokens=0, cached_prompt_tokens=600_000
|
||
),
|
||
error=None,
|
||
)
|
||
assert rec.rows[0]["cost"] == pytest.approx(10.0)
|
||
assert rec.rows[1]["cost"] == pytest.approx(5.2) # 400k×10 + 600k×2
|
||
|
||
async def test_cache_hit_row_still_costs_zero(self):
|
||
"""缓存命中未产生新调用 → cost 恒 0.0,该短路必须排在任何换算之前。"""
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec, pricing=self._TABLE).emit_cache_hit(
|
||
request=_REQ,
|
||
response=_resp(prompt_tokens=1_000_000, cached_prompt_tokens=600_000),
|
||
)
|
||
assert rec.rows[0]["cost"] == 0.0
|
||
|
||
async def test_unavailable_usage_still_costs_none(self):
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec, pricing=self._TABLE).emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="c",
|
||
latency_ms=1,
|
||
response=_resp(usage_source="unavailable", cached_prompt_tokens=5),
|
||
error=None,
|
||
)
|
||
assert rec.rows[0]["cost"] is None
|
||
|
||
|
||
class TestEmitter:
|
||
async def test_attempt_success_row(self):
|
||
rec = _MemoryRecorder()
|
||
emitter = TelemetryEmitter(rec)
|
||
await emitter.emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="cid-1",
|
||
latency_ms=42,
|
||
response=_resp(),
|
||
error=None,
|
||
)
|
||
row = rec.rows[0]
|
||
assert row["call_id"] == "cid-1" and row["error"] is None
|
||
assert row["session_id"] == "sess-1" and row["source_name"] == "s1"
|
||
assert row["response"] == "ok" and row["cost"] is None
|
||
|
||
async def test_attempt_failure_row(self):
|
||
rec = _MemoryRecorder()
|
||
emitter = TelemetryEmitter(rec)
|
||
await emitter.emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="cid-2",
|
||
latency_ms=7,
|
||
response=None,
|
||
error="TransientError: boom",
|
||
)
|
||
row = rec.rows[0]
|
||
assert row["error"].startswith("TransientError")
|
||
# 失败尝试没有任何用量信息可言 → unavailable(设计 §3.2 #6)
|
||
assert row["response"] == "" and row["usage_source"] == "unavailable"
|
||
assert row["cost"] is None
|
||
|
||
async def test_terminal_failure_row_is_unavailable(self):
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec, pricing=_PRICING).emit_terminal_failure(
|
||
request=_REQ, call_id="cid-t", latency_ms=5, error="cancelled"
|
||
)
|
||
row = rec.rows[0]
|
||
assert row["usage_source"] == "unavailable" and row["cost"] is None
|
||
|
||
@pytest.mark.parametrize(("prompt", "completion"), [(0, 0), (0, 4000)])
|
||
async def test_unavailable_success_row_has_null_cost(self, prompt, completion):
|
||
"""产生了真实调用但用量不可得 → cost 记 NULL(设计 §3.1 不变式)。
|
||
|
||
参数第二组是改前兜底写出的 `0/4000` 形态: 那时换算出 0.032 的假金额。
|
||
"""
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec, pricing=_PRICING).emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="cid-u",
|
||
latency_ms=42,
|
||
response=_resp(
|
||
usage_source="unavailable", prompt_tokens=prompt, completion_tokens=completion
|
||
),
|
||
error=None,
|
||
)
|
||
assert rec.rows[0]["cost"] is None
|
||
|
||
async def test_measured_row_still_priced(self):
|
||
"""对照组: 同一价格表下 measured 行照常换算,证明 None 不是价格表没接上。"""
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec, pricing=_PRICING).emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="cid-m",
|
||
latency_ms=42,
|
||
response=_resp(prompt_tokens=0, completion_tokens=4000),
|
||
error=None,
|
||
)
|
||
assert rec.rows[0]["cost"] == pytest.approx(0.032)
|
||
|
||
async def test_cache_hit_keeps_zero_cost_even_when_unavailable(self):
|
||
"""缓存命中未产生新调用,0.0 是事实而非未知 → 短路必须排在 cache_hit 之后。"""
|
||
rec = _MemoryRecorder()
|
||
await TelemetryEmitter(rec, pricing=_PRICING).emit_cache_hit(
|
||
request=_REQ,
|
||
response=_resp(cache_hit=True, usage_source="unavailable", completion_tokens=4000),
|
||
)
|
||
assert rec.rows[0]["cache_hit"] is True and rec.rows[0]["cost"] == 0.0
|
||
|
||
async def test_multimodal_messages_digested_before_storage(self):
|
||
rec = _MemoryRecorder()
|
||
emitter = TelemetryEmitter(rec)
|
||
big = "data:image/png;base64," + "A" * 100_000
|
||
req = ChatRequest(
|
||
messages=[
|
||
{
|
||
"role": "user",
|
||
"content": [
|
||
{"type": "image_url", "image_url": {"url": big}},
|
||
],
|
||
}
|
||
]
|
||
)
|
||
await emitter.emit_attempt(
|
||
request=req,
|
||
source=_source(),
|
||
call_id="c",
|
||
latency_ms=1,
|
||
response=None,
|
||
error="x",
|
||
)
|
||
assert len(rec.rows[0]["messages"]) < 500 # base64 不整段进库(VT R12)
|
||
|
||
async def test_recorder_failure_swallowed(self):
|
||
class Broken:
|
||
async def record_llm_call(self, **fields):
|
||
raise OSError("disk full")
|
||
|
||
emitter = TelemetryEmitter(Broken())
|
||
await emitter.emit_attempt(
|
||
request=_REQ,
|
||
source=_source(),
|
||
call_id="c",
|
||
latency_ms=1,
|
||
response=_resp(),
|
||
error=None,
|
||
) # 不抛(降级不冒泡)
|
||
|
||
|
||
class TestTelemetryMW:
|
||
async def test_cache_hit_recorded(self):
|
||
rec = _MemoryRecorder()
|
||
mw = TelemetryMW(TelemetryEmitter(rec))
|
||
|
||
async def terminal(request):
|
||
return _resp(cache_hit=True, latency_ms=0, call_id="cache-cid")
|
||
|
||
resp = await mw(_REQ, terminal)
|
||
assert resp.cache_hit
|
||
assert len(rec.rows) == 1
|
||
assert rec.rows[0]["cache_hit"] is True and rec.rows[0]["latency_ms"] == 0
|
||
|
||
async def test_normal_success_not_double_recorded(self):
|
||
"""成功尝试由 RetryMW 逐次记录;最外层不得重复记。"""
|
||
rec = _MemoryRecorder()
|
||
mw = TelemetryMW(TelemetryEmitter(rec))
|
||
|
||
async def terminal(request):
|
||
return _resp(cache_hit=False)
|
||
|
||
await mw(_REQ, terminal)
|
||
assert rec.rows == []
|
||
|
||
async def test_scope_level_failure_recorded(self):
|
||
rec = _MemoryRecorder()
|
||
mw = TelemetryMW(TelemetryEmitter(rec))
|
||
|
||
async def terminal(request):
|
||
raise CircuitOpenError(scope="llm", retry_after_s=30.0)
|
||
|
||
with pytest.raises(CircuitOpenError):
|
||
await mw(_REQ, terminal)
|
||
assert len(rec.rows) == 1 and "circuit_open" in rec.rows[0]["error"]
|
||
|
||
async def test_attempt_level_failure_not_double_recorded(self):
|
||
"""RequestRejected 已被 RetryMW 逐次记录 → 最外层跳过。"""
|
||
rec = _MemoryRecorder()
|
||
mw = TelemetryMW(TelemetryEmitter(rec))
|
||
|
||
async def terminal(request):
|
||
raise RequestRejectedError("400")
|
||
|
||
with pytest.raises(RequestRejectedError):
|
||
await mw(_REQ, terminal)
|
||
assert rec.rows == []
|
||
|
||
|
||
def test_single_emitter_discipline():
|
||
"""铁律执法: record_llm_call 在 src/ 的调用点只允许出现在 telemetry emitter。"""
|
||
out = subprocess.run(
|
||
["grep", "-rln", "record_llm_call(", "src/polygateway"],
|
||
capture_output=True,
|
||
text=True,
|
||
cwd=Path(__file__).resolve().parents[2],
|
||
).stdout.splitlines()
|
||
callers = [
|
||
p
|
||
for p in out
|
||
if not p.endswith(("ports.py", "telemetry/sqlite.py", "telemetry/postgres.py"))
|
||
]
|
||
assert callers == ["src/polygateway/middleware/telemetry.py"]
|