From 384630563464779bd108d2606e3d813cdda9c95f Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 22 Jul 2026 10:32:15 -0400 Subject: [PATCH] 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. --- tests/integration/test_postgres_telemetry.py | 60 ++++++++++++++------ 1 file changed, 44 insertions(+), 16 deletions(-) diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index aa0bb82..9371bf4 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -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()