"""PostgresRecorder 集成测试(M2 设计 §5;真实实验室 Postgres,polygateway 专用库)。 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 json import os import re from datetime import UTC, datetime, timedelta from uuid import uuid4 import pytest from dotenv import dotenv_values from polygateway.telemetry.postgres import PostgresRecorder _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", ] # 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} raw = merged.get("PGW_TELEMETRY_PG_DSN") if not raw: return None scheme, sep, rest = raw.partition("://") return f"{scheme.partition('+')[0]}{sep}{rest}" @pytest.fixture async def dsn(): value = _dsn() if value is None: pytest.skip("PGW_TELEMETRY_PG_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: 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() async def _record_minimal( recorder: PostgresRecorder, call_id: str | None = None, **overrides ) -> None: fields = { "call_id": call_id if call_id is not None else _cid("c1"), "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) async def _fetch(dsn: str, sql: str, *args): import asyncpg conn = await asyncpg.connect(dsn, timeout=10) try: return await conn.fetch(sql, *args) finally: await conn.close() _LEGACY_DDL = """ CREATE TABLE {schema}.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 DOUBLE PRECISION, max_inter_token_ms DOUBLE PRECISION, cache_hit BOOLEAN NOT NULL DEFAULT FALSE, error TEXT, cost DOUBLE PRECISION, created_at TIMESTAMPTZ NOT NULL DEFAULT now() ) """ @pytest.fixture async def legacy_schema(dsn): """在**自建的临时 schema** 里造一张 18 列旧表,验证补列(issue #3)。 绝不碰共享的 public.llm_calls: 用 search_path 把 recorder 指向临时 schema, teardown 只 DROP 自己建的 schema。 """ import asyncpg name = f"pgwtest_{uuid4().hex[:8]}" conn = await asyncpg.connect(dsn, timeout=10) try: await conn.execute(f"CREATE SCHEMA {name}") await conn.execute(_LEGACY_DDL.format(schema=name)) finally: await conn.close() sep = "&" if "?" in dsn else "?" yield f"{dsn}{sep}options=-csearch_path%3D{name}", name conn = await asyncpg.connect(dsn, timeout=10) try: await conn.execute(f"DROP SCHEMA {name} CASCADE") finally: await conn.close() class TestObservabilityColumns: """issue #3: 两列写入可回读,且已存在的 18 列旧表会被自动补列。""" async def test_values_round_trip(self, dsn): recorder = PostgresRecorder(dsn) try: await _record_minimal(recorder, call_id=_cid("hit"), cached_prompt_tokens=64) await _record_minimal(recorder, call_id=_cid("zero"), cached_prompt_tokens=0) await _record_minimal(recorder, call_id=_cid("model"), model_reported="MiniMax-01") await _record_minimal( recorder, call_id=_cid("samp"), sampling='{"seed": 42, "temperature": 0}' ) rows = await _fetch( dsn, "SELECT call_id, cached_prompt_tokens, model_reported, sampling FROM llm_calls " "WHERE call_id LIKE $1", f"{_RUN_PREFIX}-%", ) by_id = {r["call_id"]: r for r in rows} assert by_id[_cid("hit")]["cached_prompt_tokens"] == 64 assert by_id[_cid("zero")]["cached_prompt_tokens"] == 0 # 真实零命中 ≠ NULL assert by_id[_cid("model")]["cached_prompt_tokens"] is None assert by_id[_cid("model")]["model_reported"] == "MiniMax-01" # issue #4: PG 侧也须验非空 sampling 能读回原值(不只是列存在) assert json.loads(by_id[_cid("samp")]["sampling"]) == {"seed": 42, "temperature": 0} assert by_id[_cid("hit")]["sampling"] is None finally: await recorder.aclose() async def test_legacy_table_is_upgraded_in_place(self, legacy_schema): """18 列旧表不补列的话,每行写入都会被逐行 warning 丢弃(遥测静默全失)。""" schema_dsn, schema = legacy_schema recorder = PostgresRecorder(schema_dsn) try: await _record_minimal( recorder, call_id=_cid("legacy"), cached_prompt_tokens=7, model_reported="m-real" ) cols = await _fetch( schema_dsn, "SELECT column_name FROM information_schema.columns " "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position", schema, ) # ALTER 只能追加到末尾: 与新建库的列序一致才不会分叉 assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS rows = await _fetch( schema_dsn, "SELECT cached_prompt_tokens, model_reported FROM llm_calls WHERE call_id = $1", _cid("legacy"), ) assert (rows[0]["cached_prompt_tokens"], rows[0]["model_reported"]) == (7, "m-real") finally: await recorder.aclose() class TestSchema: async def test_schema_has_frozen_columns_in_order(self, dsn): recorder = PostgresRecorder(dsn) try: await _record_minimal(recorder) rows = await _fetch( dsn, "SELECT column_name FROM information_schema.columns " "WHERE table_name='llm_calls' ORDER BY ordinal_position", ) assert [r["column_name"] for r in rows] == _EXPECTED_COLUMNS finally: await recorder.aclose() async def test_call_id_idempotent(self, dsn): recorder = PostgresRecorder(dsn) try: 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() async def test_concurrent_writes_all_land(self, dsn): recorder = PostgresRecorder(dsn) try: 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() class TestDegradation: async def test_unreachable_server_degrades_silently(self): """结构性失败(建池不通)→ warning 一次后永久降级,业务零感知。""" recorder = PostgresRecorder("postgresql://u:p@127.0.0.1:1/x") await _record_minimal(recorder) # 不抛 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=_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() async def test_aclose_idempotent(self, dsn): recorder = PostgresRecorder(dsn) await _record_minimal(recorder) await recorder.aclose() await recorder.aclose() _PROBE_PASSWORD = "pgw_issue9_probe" # 临时角色,teardown 删除;非任何真实凭据 @pytest.fixture async def least_privilege_dsn(dsn): """临时 schema + 临时角色: 只授表级 SELECT/INSERT,**不授 schema CREATE**。 这是 issue #9 的现场——最小权限部署的标准形态。fixture 建的一切 (schema、表、角色)都在 teardown 里删净,共享的 public.llm_calls 不受影响; 连不上或无权建角色(非超级用户)时 skip,不让 CI 假绿。 """ import asyncpg from polygateway.telemetry.schema import PG_DDL name = f"pgwtest_lp_{uuid4().hex[:8]}" admin = await asyncpg.connect(dsn, timeout=10) try: if not await admin.fetchval( "SELECT rolcreaterole OR rolsuper FROM pg_roles WHERE rolname = current_user" ): pytest.skip("当前账号无权建临时角色,跳过最小权限用例") await admin.execute(f"CREATE ROLE {name} LOGIN PASSWORD '{_PROBE_PASSWORD}'") await admin.execute(f"CREATE SCHEMA {name}") await admin.execute(f"SET search_path = {name}") await admin.execute(PG_DDL) # 表由**别的账号**建好,与现场一致 await admin.execute(f"GRANT USAGE ON SCHEMA {name} TO {name}") await admin.execute(f"GRANT SELECT, INSERT ON {name}.llm_calls TO {name}") # 关键: 绝不 GRANT CREATE ON SCHEMA —— 缺的正是这一项 finally: await admin.close() low = re.sub(r"//[^@/]+@", f"//{name}:{_PROBE_PASSWORD}@", dsn, count=1) sep = "&" if "?" in low else "?" yield f"{low}{sep}options=-csearch_path%3D{name}", name admin = await asyncpg.connect(dsn, timeout=10) try: await admin.execute(f"DROP SCHEMA IF EXISTS {name} CASCADE") await admin.execute(f"DROP OWNED BY {name}") await admin.execute(f"DROP ROLE IF EXISTS {name}") finally: await admin.close() class TestLeastPrivilegeDeployment: """issue #9: 只有表级写权限的账号,遥测必须照常落库而不是整体判死。""" async def test_create_table_if_not_exists_is_denied_for_this_role(self, least_privilege_dsn): """库外事实先钉死: 表存在、写得进去,DDL 仍被拒——PG 的权限检查早于 IF NOT EXISTS。 修复依赖的是这条 PG 语义;若某天它变了,这里先红,而不是让下面那条 用例悄悄变成"永远通过"的空断言。 """ import asyncpg low_dsn, _ = least_privilege_dsn conn = await asyncpg.connect(low_dsn, timeout=10) try: assert await conn.fetchval("SELECT to_regclass('llm_calls')") is not None with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): await conn.execute("CREATE TABLE IF NOT EXISTS llm_calls (call_id TEXT)") finally: await conn.close() async def test_records_land_without_schema_create_privilege(self, least_privilege_dsn): """修复前: 建表被拒 → _failed → 整个进程一条不落(下游 150 次调用全丢)。""" low_dsn, schema = least_privilege_dsn recorder = PostgresRecorder(low_dsn) try: await _record_minimal(recorder, call_id=_cid("lp1")) await _record_minimal(recorder, call_id=_cid("lp2"), cost=1.5) assert recorder._failed is False # 判死开关不得被建表权限触发 rows = await _fetch( low_dsn, "SELECT call_id, cost FROM llm_calls WHERE call_id LIKE $1 ORDER BY call_id", f"{_RUN_PREFIX}-lp%", ) assert [(r["call_id"], r["cost"]) for r in rows] == [ (_cid("lp1"), None), (_cid("lp2"), 1.5), ] assert schema # teardown 会连表带角色删净 finally: await recorder.aclose() # issue #11 之前的表形态: 22 个 recorder 字段 + created_at = 23 个物理列,没有任何租户维度 _PRE_TENANT_DDL = """ CREATE TABLE {schema}.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 DOUBLE PRECISION, max_inter_token_ms DOUBLE PRECISION, cache_hit BOOLEAN NOT NULL DEFAULT FALSE, error TEXT, cost DOUBLE PRECISION, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), cached_prompt_tokens INTEGER, model_reported TEXT, sampling TEXT, reasoning_tokens INTEGER ) """ _PRE_TENANT_INSERT = ( "INSERT INTO {schema}.llm_calls (call_id, model, provider, source_name, messages, response, " "prompt_tokens, completion_tokens, usage_source, latency_ms) " "VALUES ($1, 'm', 'p', 's1', '[]', 'old body', 1, 2, 'measured', 10)" ) def _search_path_dsn(dsn: str, schema: str) -> str: sep = "&" if "?" in dsn else "?" return f"{dsn}{sep}options=-csearch_path%3D{schema}" @pytest.fixture async def captured_warnings(): """捕获库发出的 WARNING;loguru 不经标准 logging,pytest 的 caplog 抓不到。 名字避开裸 `warnings`: 那会遮蔽标准库模块名,本文件将来任何一次 `import warnings` 都会与它静默互相顶掉,而报错点离真因很远。 """ from loguru import logger messages: list[str] = [] sink_id = logger.add(messages.append, level="WARNING") yield messages logger.remove(sink_id) @pytest.fixture async def pre_tenant_schema(dsn): """自建临时 schema 里造一张 **22 字段的 issue #11 之前的表**,并留一行历史数据。 绝不碰共享的 public.llm_calls——本机那张表早已被 `_BACKFILL` 真实补过列, 指望它还是旧形态的测试第二次跑就会空转。schema 名带 uuid,可重复运行。 """ import asyncpg name = f"pgwtest_pre_{uuid4().hex[:8]}" conn = await asyncpg.connect(dsn, timeout=10) try: await conn.execute(f"CREATE SCHEMA {name}") await conn.execute(_PRE_TENANT_DDL.format(schema=name)) await conn.execute(_PRE_TENANT_INSERT.format(schema=name), _cid("old")) finally: await conn.close() yield _search_path_dsn(dsn, name), name conn = await asyncpg.connect(dsn, timeout=10) try: await conn.execute(f"DROP SCHEMA {name} CASCADE") finally: await conn.close() @pytest.fixture async def fresh_schema(dsn): """空 schema: recorder 自己建表,验"新建库"这条路径而不依赖共享表的历史状态。""" import asyncpg name = f"pgwtest_new_{uuid4().hex[:8]}" conn = await asyncpg.connect(dsn, timeout=10) try: await conn.execute(f"CREATE SCHEMA {name}") finally: await conn.close() yield _search_path_dsn(dsn, name), name conn = await asyncpg.connect(dsn, timeout=10) try: await conn.execute(f"DROP SCHEMA {name} CASCADE") finally: await conn.close() @pytest.fixture async def least_privilege_pre_tenant_dsn(dsn): """22 字段旧表 + 只有 `SELECT, INSERT` 权限的角色: 补列必然失败的现场。 与 `least_privilege_dsn` 分开而非复用: 那个 fixture 建的是列已齐全的当前表 (测的是 CREATE 被拒),这里必须是缺列的旧表,才能让 `ALTER TABLE` 真的发出去 并撞上 ownership 检查(该检查早于 `IF NOT EXISTS` 的存在性判断)。 """ import asyncpg name = f"pgwtest_lppre_{uuid4().hex[:8]}" admin = await asyncpg.connect(dsn, timeout=10) try: if not await admin.fetchval( "SELECT rolcreaterole OR rolsuper FROM pg_roles WHERE rolname = current_user" ): pytest.skip("当前账号无权建临时角色,跳过最小权限用例") await admin.execute(f"CREATE ROLE {name} LOGIN PASSWORD '{_PROBE_PASSWORD}'") await admin.execute(f"CREATE SCHEMA {name}") await admin.execute(_PRE_TENANT_DDL.format(schema=name)) # 表属主是 admin,不是应用账号 await admin.execute(f"GRANT USAGE ON SCHEMA {name} TO {name}") await admin.execute(f"GRANT SELECT, INSERT ON {name}.llm_calls TO {name}") finally: await admin.close() low = re.sub(r"//[^@/]+@", f"//{name}:{_PROBE_PASSWORD}@", dsn, count=1) yield _search_path_dsn(low, name) admin = await asyncpg.connect(dsn, timeout=10) try: await admin.execute(f"DROP SCHEMA IF EXISTS {name} CASCADE") await admin.execute(f"DROP OWNED BY {name}") await admin.execute(f"DROP ROLE IF EXISTS {name}") finally: await admin.close() class TestCallerDimensionsAcceptance: """issue #11 的机械化验收(PG 侧,真实实例): 新建库 / 旧表补列 / 补列失败方向。""" async def test_fresh_schema_round_trips_the_dimensions(self, fresh_schema): """新建库: 列齐全,且维度值原样读回——只验列存在会漏掉写错列位的错。""" fresh_dsn, schema = fresh_schema recorder = PostgresRecorder(fresh_dsn) try: await _record_minimal( recorder, call_id=_cid("dim"), tenant_id="tenant-a", meta='{"batch": "b7"}' ) cols = await _fetch( fresh_dsn, "SELECT column_name FROM information_schema.columns " "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position", schema, ) assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS rows = await _fetch( fresh_dsn, "SELECT tenant_id, meta FROM llm_calls WHERE call_id = $1", _cid("dim"), ) assert rows[0]["tenant_id"] == "tenant-a" assert json.loads(rows[0]["meta"]) == {"batch": "b7"} finally: await recorder.aclose() async def test_pre_tenant_table_gains_columns_and_old_rows_stay_auditable( self, pre_tenant_schema ): """22 字段旧表补列后,新行带维度,而**老行的 tenant_id 是空串而非 NULL**。 这条直接验收 issue #11 的核心论点(先启用落库、后加列,补列之前的行没有 租户归属)。断言方向必须是空串: PG 的 RLS `USING` 表达式对返回 false **或 NULL** 的行一律隐藏且不报错,故 NULL 的 `tenant_id` 不是"未归属",而是对 所有人永久不可见的黑洞;哨兵空串则能被一条 `COUNT(*) WHERE tenant_id = ''` 审计出来,历史欠账是可见、可量化、可补录的。 """ schema_dsn, schema = pre_tenant_schema recorder = PostgresRecorder(schema_dsn) try: await _record_minimal( recorder, call_id=_cid("new"), tenant_id="tenant-a", meta='{"k": 1}' ) cols = await _fetch( schema_dsn, "SELECT column_name FROM information_schema.columns " "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position", schema, ) # 22 → 24 个 recorder 字段(加 created_at 共 25 个物理列),且新列追加在末尾 assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS rows = await _fetch( schema_dsn, "SELECT call_id, tenant_id, meta FROM llm_calls " "WHERE call_id = ANY($1::text[]) ORDER BY call_id", [_cid("new"), _cid("old")], ) by_id = {r["call_id"]: r for r in rows} assert by_id[_cid("new")]["tenant_id"] == "tenant-a" assert json.loads(by_id[_cid("new")]["meta"]) == {"k": 1} assert by_id[_cid("old")]["tenant_id"] == "" # 不是 None: NULL 会被 RLS 静默吞掉 assert json.loads(by_id[_cid("old")]["meta"]) == {} finally: await recorder.aclose() async def test_alter_is_denied_for_a_role_that_can_still_insert( self, least_privilege_pre_tenant_dsn ): """库外事实先钉死: 表存在、写得进去,补列的 ALTER 仍被拒(ownership 检查早于存在性判断)。 没有这条,下面那个降级用例可能因为 ALTER 其实成功了而变成"永远通过"的空断言。 """ import asyncpg conn = await asyncpg.connect(least_privilege_pre_tenant_dsn, timeout=10) try: assert await conn.fetchval("SELECT to_regclass('llm_calls')") is not None with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): await conn.execute("ALTER TABLE llm_calls ADD COLUMN IF NOT EXISTS tenant_id TEXT") finally: await conn.close() async def test_backfill_failure_degrades_per_row_not_wholesale( self, least_privilege_pre_tenant_dsn, captured_warnings ): """补列失败的降级方向: 记 warning、不置 `_failed`、后续 INSERT 仍照发。 置 `_failed` 会让整个进程从此一条遥测都不写(比逐行丢弃严重得多), 且一旦 DBA 补上列也不会自愈——必须等重启。 """ recorder = PostgresRecorder(least_privilege_pre_tenant_dsn) try: await _record_minimal(recorder, call_id=_cid("lpp1")) # 不得抛 assert recorder._failed is False assert any("补列失败" in m for m in captured_warnings) # 缺列的表上 INSERT 必然失败;逐行 warning 正是"INSERT 照发了"的证据 assert any("写入失败" in m for m in captured_warnings) finally: await recorder.aclose() # issue #12 的目标表形态: 按 created_at 做 RANGE 分区(过期清理 DROP PARTITION 而非 DELETE)。 # PG 强制分区表的唯一约束必须包含分区键,故主键只能是 (call_id, created_at) —— # 这正是带目标的 `ON CONFLICT (call_id)` 再也匹配不到约束的现场。 _PARTITIONED_DDL = """ CREATE TABLE {schema}.llm_calls ( call_id TEXT NOT NULL, 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 DOUBLE PRECISION, max_inter_token_ms DOUBLE PRECISION, cache_hit BOOLEAN NOT NULL DEFAULT FALSE, error TEXT, cost DOUBLE PRECISION, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), cached_prompt_tokens INTEGER, model_reported TEXT, sampling TEXT, reasoning_tokens INTEGER, tenant_id TEXT NOT NULL DEFAULT '', meta JSONB NOT NULL DEFAULT '{{}}'::jsonb, PRIMARY KEY (call_id, created_at) ) PARTITION BY RANGE (created_at) """ _PARTITION_DDL = ( "CREATE TABLE {schema}.llm_calls_current PARTITION OF {schema}.llm_calls " "FOR VALUES FROM ('{start}') TO ('{end}')" ) def _current_month_bounds() -> tuple[str, str]: """当前月的 [月初, 下月初) 边界字面量;分区键落在区间外会因找不到分区而写失败。""" now = datetime.now(UTC) start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0) end = (start + timedelta(days=32)).replace(day=1) fmt = "%Y-%m-%d %H:%M:%S%z" return start.strftime(fmt), end.strftime(fmt) @pytest.fixture async def partitioned_schema(dsn): """自建临时 schema 里造一张按 created_at RANGE 分区的表 + 覆盖当前月的分区。 与 legacy_schema 同款隔离: 绝不碰共享的 public.llm_calls,teardown 只 DROP 自己建的 schema(CASCADE 连分区一并删)。 """ import asyncpg name = f"pgwtest_part_{uuid4().hex[:8]}" start, end = _current_month_bounds() conn = await asyncpg.connect(dsn, timeout=10) try: await conn.execute(f"CREATE SCHEMA {name}") await conn.execute(_PARTITIONED_DDL.format(schema=name)) await conn.execute(_PARTITION_DDL.format(schema=name, start=start, end=end)) finally: await conn.close() yield _search_path_dsn(dsn, name), name conn = await asyncpg.connect(dsn, timeout=10) try: await conn.execute(f"DROP SCHEMA {name} CASCADE") finally: await conn.close() class TestConflictTargetFreeInsert: """issue #13: INSERT 不绑定冲突目标,普通表与分区表两种形态都写得进去。""" async def test_plain_table_still_dedupes_by_call_id(self, fresh_schema, captured_warnings): """普通表上语义不变: 重复 call_id 仍只落一行,且不是被拒后丢弃。 表上只有主键这一个唯一约束,故无目标的 DO NOTHING 与 `(call_id)` 逐字等价; 断言"无写入失败 warning"是为了区分"冲突被忽略"与"整条被 PG 拒收"。 """ fresh_dsn, _ = fresh_schema recorder = PostgresRecorder(fresh_dsn) try: await _record_minimal(recorder, call_id=_cid("nodup")) await _record_minimal(recorder, call_id=_cid("nodup"), response="second") assert [m for m in captured_warnings if "写入失败" in m] == [] rows = await _fetch( fresh_dsn, "SELECT response FROM llm_calls WHERE call_id = $1", _cid("nodup") ) assert [r["response"] for r in rows] == ["ok"] # 首行胜出,写入幂等 finally: await recorder.aclose() async def test_partitioned_table_accepts_writes(self, partitioned_schema, captured_warnings): """分区表上写入成功且能读回——改动前这里必红。 带目标的 `ON CONFLICT (call_id)` 在主键为 `(call_id, created_at)` 的表上 匹配不到任何约束,PG 报 "there is no unique or exclusion constraint matching the ON CONFLICT specification";该错误被逐行降级吞成 warning,于是分区部署下 遥测全线写不进去却一声不吭,只能靠"读不回来"暴露。 """ part_dsn, _ = partitioned_schema recorder = PostgresRecorder(part_dsn) try: await _record_minimal(recorder, call_id=_cid("part"), tenant_id="tenant-p") assert [m for m in captured_warnings if "写入失败" in m] == [] rows = await _fetch( part_dsn, "SELECT call_id, tenant_id FROM llm_calls WHERE call_id = $1", _cid("part"), ) assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(_cid("part"), "tenant-p")] finally: await recorder.aclose()