fix: scope postgres telemetry test to run-prefixed rows

The fixture used to DROP the shared llm_calls table on every run, wiping
concurrent migration-batch telemetry (and its own count assertion was
polluted in return). Assertions now filter by a per-run call_id prefix
and teardown deletes only its own rows.
This commit is contained in:
2026-07-22 10:32:15 -04:00
parent 547141cf0a
commit 3846305634
+44 -16
View File
@@ -2,12 +2,17 @@
DSN 走 .env `PGW_TELEMETRY_PG_DSN`,缺则 skip。该实例上有 app/chs_prod 等
在用库——本测试只允许连 polygateway 专用库(fixture 里守卫)。
隔离纪律(M4 事故教训): `llm_calls` 是与真实批跑/迁移项目共享的表,
**严禁 DROP/TRUNCATE**——本测试以 run 级 call_id 前缀隔离,断言只看
自己写入的行,teardown 只删自己的行。
"""
from __future__ import annotations
import asyncio
import os
from uuid import uuid4
import pytest
from dotenv import dotenv_values
@@ -36,6 +41,13 @@ _EXPECTED_COLUMNS = [
"created_at",
]
# run 级前缀: 同库并存的其他运行(迁移批跑/另一开发机)互不可见
_RUN_PREFIX = f"pgwtest-{uuid4().hex[:8]}"
def _cid(suffix: str) -> str:
return f"{_RUN_PREFIX}-{suffix}"
def _dsn() -> str | None:
merged = {**dotenv_values(".env"), **os.environ}
@@ -54,19 +66,23 @@ async def dsn():
# 隔离守卫: 该实例有 app/chs_prod/mimiciv 等在用库,只许打 polygateway 专用库
if not value.rstrip("/").endswith("/polygateway"):
pytest.fail(f"遥测测试只允许连 polygateway 专用库,当前 DSN 库名不符: {value!r}")
yield value
# teardown: 只删本 run 写入的行;表可能尚不存在(全新库)则忽略
import asyncpg
conn = await asyncpg.connect(value, timeout=10)
try:
await conn.execute("DROP TABLE IF EXISTS llm_calls")
if await conn.fetchval("SELECT to_regclass('llm_calls')") is not None:
await conn.execute(
"DELETE FROM llm_calls WHERE call_id LIKE $1", f"{_RUN_PREFIX}-%"
)
finally:
await conn.close()
return value
async def _record_minimal(recorder: PostgresRecorder, call_id: str = "c1", **overrides) -> None:
async def _record_minimal(recorder: PostgresRecorder, call_id: str | None = None, **overrides) -> None:
fields = {
"call_id": call_id,
"call_id": call_id if call_id is not None else _cid("c1"),
"parent_call_id": None,
"session_id": "sess-1",
"model": "m",
@@ -89,12 +105,12 @@ async def _record_minimal(recorder: PostgresRecorder, call_id: str = "c1", **ove
await recorder.record_llm_call(**fields)
async def _fetch(dsn: str, sql: str):
async def _fetch(dsn: str, sql: str, *args):
import asyncpg
conn = await asyncpg.connect(dsn, timeout=10)
try:
return await conn.fetch(sql)
return await conn.fetch(sql, *args)
finally:
await conn.close()
@@ -116,9 +132,11 @@ class TestSchema:
async def test_call_id_idempotent(self, dsn):
recorder = PostgresRecorder(dsn)
try:
await _record_minimal(recorder, call_id="dup")
await _record_minimal(recorder, call_id="dup", response="second")
rows = await _fetch(dsn, "SELECT response FROM llm_calls WHERE call_id='dup'")
await _record_minimal(recorder, call_id=_cid("dup"))
await _record_minimal(recorder, call_id=_cid("dup"), response="second")
rows = await _fetch(
dsn, "SELECT response FROM llm_calls WHERE call_id = $1", _cid("dup")
)
assert [r["response"] for r in rows] == ["ok"] # ON CONFLICT DO NOTHING
finally:
await recorder.aclose()
@@ -126,8 +144,14 @@ class TestSchema:
async def test_concurrent_writes_all_land(self, dsn):
recorder = PostgresRecorder(dsn)
try:
await asyncio.gather(*(_record_minimal(recorder, call_id=f"c{i}") for i in range(50)))
rows = await _fetch(dsn, "SELECT count(*) AS n FROM llm_calls")
await asyncio.gather(
*(_record_minimal(recorder, call_id=_cid(f"c{i}")) for i in range(50))
)
rows = await _fetch(
dsn,
"SELECT count(*) AS n FROM llm_calls WHERE call_id LIKE $1",
f"{_RUN_PREFIX}-c%",
)
assert rows[0]["n"] == 50
finally:
await recorder.aclose()
@@ -138,17 +162,21 @@ class TestDegradation:
"""结构性失败(建池不通)→ warning 一次后永久降级,业务零感知。"""
recorder = PostgresRecorder("postgresql://u:p@127.0.0.1:1/x")
await _record_minimal(recorder) # 不抛
await _record_minimal(recorder, call_id="c2") # 已降级短路,同样不抛
await _record_minimal(recorder, call_id=_cid("c2")) # 已降级短路,同样不抛
await recorder.aclose()
async def test_row_failure_does_not_poison_later_rows(self, dsn):
"""运行时单条写失败(NUL 字节文本被 PG 拒)→ 丢该行,后续行照常落库。"""
recorder = PostgresRecorder(dsn)
try:
await _record_minimal(recorder, call_id="bad", response="nul\x00byte")
await _record_minimal(recorder, call_id="good")
rows = await _fetch(dsn, "SELECT call_id FROM llm_calls ORDER BY call_id")
assert [r["call_id"] for r in rows] == ["good"]
await _record_minimal(recorder, call_id=_cid("bad"), response="nul\x00byte")
await _record_minimal(recorder, call_id=_cid("good"))
rows = await _fetch(
dsn,
"SELECT call_id FROM llm_calls WHERE call_id = ANY($1::text[]) ORDER BY call_id",
[_cid("bad"), _cid("good")],
)
assert [r["call_id"] for r in rows] == [_cid("good")]
finally:
await recorder.aclose()