diff --git a/Makefile b/Makefile index 5f3bf3b..8557dfb 100644 --- a/Makefile +++ b/Makefile @@ -1,21 +1,34 @@ -.PHONY: install test lint format check ci wiki wiki-check +.PHONY: install test lint format check ci wiki wiki-check shared-table-gate ENV := PolyGateway +# 集成测试触碰共享表 llm_calls 的字面量门(issue #18)。 +# 这道门是**烟雾报警器,不是隔离证明**: 它拦不住 f"{schema}.{table}" 拼接、 +# 参数化查询,或不带限定名的 DELETE 配上 admin 的默认 search_path。真正的隔离 +# 来自两处——沙箱工厂不把管理连接交给用例,以及清理脚本以无权角色运行。 +# 留着它是因为字面量回归最常见、也最便宜拦。 +shared-table-gate: + @if grep -rn --include='*.py' 'public\.llm_calls' tests/; then \ + echo ""; \ + echo "错误: 集成测试不得触碰共享表(见上面的命中行)。"; \ + echo "改用 tests/integration/conftest.py 的 pg_sandbox 工厂;注释里提到它请写「共享表 llm_calls」。"; \ + exit 1; \ + fi + install: conda run -n $(ENV) pip install -e ".[redis,postgres,structured,dev]" test: conda run -n $(ENV) pytest tests/ --cov=src/polygateway --cov-report=term-missing -lint: +lint: shared-table-gate conda run -n $(ENV) ruff check src/ tests/ --fix conda run -n $(ENV) lint-imports format: conda run -n $(ENV) ruff format src/ tests/ -check: +check: shared-table-gate conda run -n $(ENV) ruff format --check src/ tests/ conda run -n $(ENV) ruff check src/ tests/ conda run -n $(ENV) lint-imports diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index 77a01ee..0159873 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -1,11 +1,15 @@ """PostgresRecorder 集成测试(M2 设计 §5;真实实验室 Postgres,polygateway 专用库)。 DSN 走 .env `PGW_TELEMETRY_PG_DSN`,缺则 skip。该实例上有 app/chs_prod 等 -在用库——本测试只允许连 polygateway 专用库(fixture 里守卫)。 +在用库——本测试只允许连 polygateway 专用库(`conftest.py` 的工厂里守卫)。 -隔离纪律(M4 事故教训): `llm_calls` 是与真实批跑/迁移项目共享的表, -**严禁 DROP/TRUNCATE**——本测试以 run 级 call_id 前缀隔离,断言只看 -自己写入的行,teardown 只删自己的行。 +隔离纪律(issue #18): 本文件对共享表 `llm_calls` **零触碰**——每条用例都在 +`pg_sandbox` 建的一次性 schema 里跑,建/删都只发生在自己的 schema 内。 +此前那套 run 级 call_id 前缀隔离已随之删除: schema 隔离完全取代了它, +两套并存只会让"这一行归谁"重新变成需要论证的事。 + +**唯一的例外是连接**: 连接是实例级共享资源,schema 隔离对它无效,故 +`TestPoolFootprint` 仍靠一个就地生成的唯一 `application_name` 认领本池连接。 """ from __future__ import annotations @@ -55,15 +59,9 @@ _EXPECTED_COLUMNS = [ "thinking_observation", ] -# run 级前缀: 同库并存的其他运行(迁移批跑/另一开发机)互不可见 -_RUN_PREFIX = f"pgwtest-{uuid4().hex[:8]}" - - -def _cid(suffix: str) -> str: - return f"{_RUN_PREFIX}-{suffix}" - def _dsn() -> str | None: + """读 `.env` 的 DSN 并剥掉 SQLAlchemy 风格的 `+driver` 后缀;未配置返回 None。""" merged = {**dotenv_values(".env"), **os.environ} raw = merged.get("PGW_TELEMETRY_PG_DSN") if not raw: @@ -73,23 +71,24 @@ def _dsn() -> str | None: @pytest.fixture -async def dsn(): +async def template_admin_dsn() -> str: + """管理连接串,**只服务 `production_template` 一个 fixture**。 + + 它没有随其余六个 fixture 一起收敛到 `pg_sandbox`,是因为 `production_template` + 要自建三个角色、跑 README 解析出的整套模板 SQL、按月建分区,权限语义与失败期 + 清理都是它自己的(设计 §7.1 末段),工厂强行接管会把这些语义压扁。 + + 名字不叫 `dsn`: 叫 `dsn` 等于把一个能动共享表的连接摆在每条用例的参数位上, + 而设计 §7.1 约束 3 要的正是"用例拿不到管理连接"。此处的窄命名是那条约束在 + 本文件能做到的最接近的形态。 + """ 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() + return value async def _record_minimal( @@ -101,7 +100,7 @@ async def _record_minimal( "库写错列位"的形态误报,而漏抄的列则悄悄不被验证。 """ fields: dict[str, object] = { - "call_id": call_id if call_id is not None else _cid("c1"), + "call_id": call_id if call_id is not None else "c1", "parent_call_id": None, "session_id": "sess-1", "model": "m", @@ -176,8 +175,9 @@ async def _execute_script(dsn: str, sql: str) -> None: await conn.close() +# 裸表名: 由 `pg_sandbox` 在沙箱 schema 的 search_path 下执行(工厂的 DDL 执行契约) _LEGACY_DDL = """ -CREATE TABLE {schema}.llm_calls ( +CREATE TABLE llm_calls ( call_id TEXT PRIMARY KEY, parent_call_id TEXT, session_id TEXT, @@ -202,56 +202,43 @@ CREATE TABLE {schema}.llm_calls ( @pytest.fixture -async def legacy_schema(dsn): - """在**自建的临时 schema** 里造一张 18 列旧表,验证补列(issue #3)。 +async def legacy_schema(pg_sandbox) -> tuple[str, str]: + """在一次性沙箱 schema 里造一张 18 列旧表,验证补列(issue #3)。 - 绝不碰共享的 public.llm_calls: 用 search_path 把 recorder 指向临时 schema, - teardown 只 DROP 自己建的 schema。 + 共享表 `llm_calls` 一个字节都不碰: recorder 由 search_path 指向沙箱 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() + sandbox = await pg_sandbox(ddl=_LEGACY_DDL) + return sandbox.dsn, sandbox.schema class TestObservabilityColumns: """issue #3: 两列写入可回读,且已存在的 18 列旧表会被自动补列。""" - async def test_values_round_trip(self, dsn): - recorder = _recorder(dsn, auto_migrate=True) + async def test_values_round_trip(self, pg_sandbox): + sandbox = await pg_sandbox() + recorder = _recorder(sandbox.dsn, auto_migrate=True) 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="hit", cached_prompt_tokens=64) + await _record_minimal(recorder, call_id="zero", cached_prompt_tokens=0) + await _record_minimal(recorder, call_id="model", model_reported="MiniMax-01") await _record_minimal( - recorder, call_id=_cid("samp"), sampling='{"seed": 42, "temperature": 0}' + recorder, call_id="samp", sampling='{"seed": 42, "temperature": 0}' ) rows = await _fetch( - dsn, + sandbox.dsn, "SELECT call_id, cached_prompt_tokens, model_reported, sampling FROM llm_calls " - "WHERE call_id LIKE $1", - f"{_RUN_PREFIX}-%", + "WHERE call_id = ANY($1::text[])", + ["hit", "zero", "model", "samp"], ) 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" + assert by_id["hit"]["cached_prompt_tokens"] == 64 + assert by_id["zero"]["cached_prompt_tokens"] == 0 # 真实零命中 ≠ NULL + assert by_id["model"]["cached_prompt_tokens"] is None + assert by_id["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 + assert json.loads(by_id["samp"]["sampling"]) == {"seed": 42, "temperature": 0} + assert by_id["hit"]["sampling"] is None finally: await recorder.aclose() @@ -261,7 +248,7 @@ class TestObservabilityColumns: recorder = _recorder(schema_dsn, auto_migrate=True) try: await _record_minimal( - recorder, call_id=_cid("legacy"), cached_prompt_tokens=7, model_reported="m-real" + recorder, call_id="legacy", cached_prompt_tokens=7, model_reported="m-real" ) cols = await _fetch( schema_dsn, @@ -274,7 +261,7 @@ class TestObservabilityColumns: rows = await _fetch( schema_dsn, "SELECT cached_prompt_tokens, model_reported FROM llm_calls WHERE call_id = $1", - _cid("legacy"), + "legacy", ) assert (rows[0]["cached_prompt_tokens"], rows[0]["model_reported"]) == (7, "m-real") finally: @@ -282,42 +269,44 @@ class TestObservabilityColumns: class TestSchema: - async def test_schema_has_frozen_columns_in_order(self, dsn): - recorder = _recorder(dsn, auto_migrate=True) + async def test_schema_has_frozen_columns_in_order(self, pg_sandbox): + sandbox = await pg_sandbox() + recorder = _recorder(sandbox.dsn, auto_migrate=True) try: await _record_minimal(recorder) rows = await _fetch( - dsn, + sandbox.dsn, + # `table_schema = $1` 不可省: 不带它,库里任何一个残留 schema 下的同名表 + # 都会把自己的列拼进结果,这条断言于是以"列数不符"的形态被别人的残留误伤 "SELECT column_name FROM information_schema.columns " - "WHERE table_name='llm_calls' ORDER BY ordinal_position", + "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position", + sandbox.schema, ) assert [r["column_name"] for r in rows] == _EXPECTED_COLUMNS finally: await recorder.aclose() - async def test_call_id_idempotent(self, dsn): - recorder = _recorder(dsn, auto_migrate=True) + async def test_call_id_idempotent(self, pg_sandbox): + sandbox = await pg_sandbox() + recorder = _recorder(sandbox.dsn, auto_migrate=True) try: - await _record_minimal(recorder, call_id=_cid("dup")) - await _record_minimal(recorder, call_id=_cid("dup"), response="second") + 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 = $1", _cid("dup") + sandbox.dsn, "SELECT response FROM llm_calls WHERE call_id = $1", "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 = _recorder(dsn, auto_migrate=True) + async def test_concurrent_writes_all_land(self, pg_sandbox): + sandbox = await pg_sandbox() + recorder = _recorder(sandbox.dsn, auto_migrate=True) 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%", - ) + await asyncio.gather(*(_record_minimal(recorder, call_id=f"c{i}") for i in range(50))) + # 沙箱 schema 里只有这一批行,故全表 COUNT 就是本用例写入的行数—— + # 前缀过滤在这里已无事可做(它当年存在只是为了从共享表里认领自己的行) + rows = await _fetch(sandbox.dsn, "SELECT count(*) AS n FROM llm_calls") assert rows[0]["n"] == 50 finally: await recorder.aclose() @@ -341,7 +330,7 @@ class TestDegradation: """服务端连不上 → warning 一次后降级,业务零感知(不抛、不拖)。""" recorder = _recorder("postgresql://u:p@127.0.0.1:1/x", auto_migrate=True) await _record_minimal(recorder) # 不抛 - await _record_minimal(recorder, call_id=_cid("c2")) # 已降级短路,同样不抛 + await _record_minimal(recorder, call_id="c2") # 已降级短路,同样不抛 await recorder.aclose() async def test_refused_connection_cools_down_and_retries_after_cooldown(self): @@ -364,7 +353,7 @@ class TestDegradation: now=clock, ) try: - await _record_minimal(recorder, call_id=_cid("deg1")) + await _record_minimal(recorder, call_id="deg1") first = recorder.telemetry_status # 非 fatal 正是 issue #15 的核心: 连接被拒过去在建池那一步被一刀判死, # 整进程从此一行遥测都不落、只有重启能恢复 @@ -375,14 +364,14 @@ class TestDegradation: assert "建表探测失败" in (first.reason or "") clock.advance(30.0) - await _record_minimal(recorder, call_id=_cid("deg2")) + await _record_minimal(recorder, call_id="deg2") mid = recorder.telemetry_status # 冷却窗口没被刷新 = 这次调用压根没去连库(降级期间零成本短路) assert mid.retry_after_s == pytest.approx(30.0) assert mid.dropped_rows == 2 clock.advance(30.1) - await _record_minimal(recorder, call_id=_cid("deg3")) + await _record_minimal(recorder, call_id="deg3") after = recorder.telemetry_status # 冷却窗口被重新拉满 = 真的重连了一次(照旧被拒,故仍降级但仍可自愈) assert after.retry_after_s == pytest.approx(60.0) @@ -391,23 +380,25 @@ class TestDegradation: finally: await recorder.aclose() - async def test_row_failure_does_not_poison_later_rows(self, dsn): + async def test_row_failure_does_not_poison_later_rows(self, pg_sandbox): """运行时单条写失败(NUL 字节文本被 PG 拒)→ 丢该行,后续行照常落库。""" - recorder = _recorder(dsn, auto_migrate=True) + sandbox = await pg_sandbox() + recorder = _recorder(sandbox.dsn, auto_migrate=True) try: - await _record_minimal(recorder, call_id=_cid("bad"), response="nul\x00byte") - await _record_minimal(recorder, call_id=_cid("good")) + await _record_minimal(recorder, call_id="bad", response="nul\x00byte") + await _record_minimal(recorder, call_id="good") rows = await _fetch( - dsn, + sandbox.dsn, "SELECT call_id FROM llm_calls WHERE call_id = ANY($1::text[]) ORDER BY call_id", - [_cid("bad"), _cid("good")], + ["bad", "good"], ) - assert [r["call_id"] for r in rows] == [_cid("good")] + assert [r["call_id"] for r in rows] == ["good"] finally: await recorder.aclose() - async def test_aclose_idempotent(self, dsn): - recorder = _recorder(dsn, auto_migrate=True) + async def test_aclose_idempotent(self, pg_sandbox): + sandbox = await pg_sandbox() + recorder = _recorder(sandbox.dsn, auto_migrate=True) await _record_minimal(recorder) await recorder.aclose() await recorder.aclose() @@ -427,7 +418,7 @@ def _tagged(dsn: str, app_name: str) -> str: async def _pool_backend_count(dsn: str, app_name: str) -> int: """数**本池**在服务端的连接数(只读查询,不改实例任何状态)。 - 只按 run 级唯一的 `application_name` 过滤: 这台实例被多项目共用,按库名或 + 只按用例级唯一的 `application_name` 过滤: 这台实例被多项目共用,按库名或 用户名计数会把别人的连接算进来,做出的是设计上就会间歇红的用例 (CLAUDE.md §4.6)。本查询自己那条连接走未打 tag 的 DSN,故不会数到自己。 """ @@ -458,73 +449,53 @@ class TestPoolFootprint: 那么多连接"——两件事,只有真实 PG 能证后者。 """ - async def test_pool_does_not_preconnect_and_stays_within_pool_max(self, dsn): - app_name = f"{_RUN_PREFIX}-pool" # run 级唯一,与并跑的其他运行互不可见 - recorder = _recorder(_tagged(dsn, app_name), auto_migrate=True) + async def test_pool_does_not_preconnect_and_stays_within_pool_max(self, pg_sandbox): + sandbox = await pg_sandbox() + # `application_name` 的唯一性必须**就地**造,不能跟着行隔离前缀一起删掉: + # 连接是实例级资源,schema 隔离对 `pg_stat_activity` 完全无效,换成固定名字 + # 会把并跑进程的连接数进来,等于把偶发红从表层搬到连接层(设计 §6.1)。 + app_name = f"pgwtest-pool-{uuid4().hex[:12]}" + recorder = _recorder(_tagged(sandbox.dsn, app_name), auto_migrate=True) try: # 构造只记参数、不触库: 这一条与下一条合起来才是钉子——修复前 # `create_pool` 继承 asyncpg 的 min_size=10,首次写入后下面会是 10 - assert await _pool_backend_count(dsn, app_name) == 0 + assert await _pool_backend_count(sandbox.dsn, app_name) == 0 - await _record_minimal(recorder, call_id=_cid("fp1")) + await _record_minimal(recorder, call_id="fp1") # **时序前提**: 写入已 await 到返回,连接必然已建立(没建立就写不成功), # 归还只是还进池而不断开,asyncpg 空闲回收是 300s 不会在用例内触发。 # 故这是个确定值,不是"某一刻恰好的采样" - assert await _pool_backend_count(dsn, app_name) == 1 + assert await _pool_backend_count(sandbox.dsn, app_name) == 1 await asyncio.gather( - *(_record_minimal(recorder, call_id=_cid(f"fp{i}")) for i in range(2, 22)) + *(_record_minimal(recorder, call_id=f"fp{i}") for i in range(2, 22)) ) - steady = await _pool_backend_count(dsn, app_name) + steady = await _pool_backend_count(sandbox.dsn, app_name) # 上界由 max_size 保证;下界 ≥1 不是凑数——它确保过滤条件真的命中了本池, # 否则 tag 一旦拼错,上面那条 ==0 会以"永远绿"的形态通过 assert 1 <= steady <= _POOL_MAX finally: await recorder.aclose() - assert await _settled_backend_count(dsn, app_name) == 0 # 关闭即归还全部连接 + # 关闭即归还全部连接 + assert await _settled_backend_count(sandbox.dsn, app_name) == 0 _PROBE_PASSWORD = "pgw_issue9_probe" # 临时角色,teardown 删除;非任何真实凭据 @pytest.fixture -async def least_privilege_dsn(dsn): - """临时 schema + 临时角色: 只授表级 SELECT/INSERT,**不授 schema CREATE**。 +async def least_privilege_dsn(pg_sandbox) -> tuple[str, str]: + """一次性 schema + 独占角色: 只授表级 SELECT/INSERT,**不授 schema CREATE**。 - 这是 issue #9 的现场——最小权限部署的标准形态。fixture 建的一切 - (schema、表、角色)都在 teardown 里删净,共享的 public.llm_calls 不受影响; - 连不上或无权建角色(非超级用户)时 skip,不让 CI 假绿。 + 这是 issue #9 的现场——最小权限部署的标准形态。`role="grantee"` 的语义恰是它: + 表由 admin 建好(属主不是应用账号),角色只拿到 `USAGE` 加表级 grants, + 唯独没有 `CREATE ON SCHEMA`——缺的正是这一项。 + 无权建角色(非超级用户)时工厂自己 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() + sandbox = await pg_sandbox(ddl=PG_DDL, role="grantee") + return sandbox.dsn, sandbox.schema class TestLeastPrivilegeDeployment: @@ -552,26 +523,28 @@ class TestLeastPrivilegeDeployment: low_dsn, schema = least_privilege_dsn recorder = _recorder(low_dsn, auto_migrate=True) try: - await _record_minimal(recorder, call_id=_cid("lp1")) - await _record_minimal(recorder, call_id=_cid("lp2"), cost=1.5) + await _record_minimal(recorder, call_id="lp1") + await _record_minimal(recorder, call_id="lp2", cost=1.5) assert recorder.telemetry_status.degraded 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%", + "SELECT call_id, cost FROM llm_calls " + "WHERE call_id = ANY($1::text[]) ORDER BY call_id", + ["lp1", "lp2"], ) assert [(r["call_id"], r["cost"]) for r in rows] == [ - (_cid("lp1"), None), - (_cid("lp2"), 1.5), + ("lp1", None), + ("lp2", 1.5), ] assert schema # teardown 会连表带角色删净 finally: await recorder.aclose() -# issue #11 之前的表形态: 22 个 recorder 字段 + created_at = 23 个物理列,没有任何租户维度 +# issue #11 之前的表形态: 22 个 recorder 字段 + created_at = 23 个物理列,没有任何租户维度。 +# 裸表名: 由 `pg_sandbox` 在沙箱 schema 的 search_path 下执行(工厂的 DDL 执行契约) _PRE_TENANT_DDL = """ -CREATE TABLE {schema}.llm_calls ( +CREATE TABLE llm_calls ( call_id TEXT PRIMARY KEY, parent_call_id TEXT, session_id TEXT, @@ -598,10 +571,12 @@ CREATE TABLE {schema}.llm_calls ( ) """ +# 工厂的 `extra` 逐条裸执行、不接受查询参数,故这行历史数据的 call_id 直接内联成 +# 字面量('old' 是本文件固定的测试常量,不是外部输入)。 _PRE_TENANT_INSERT = ( - "INSERT INTO {schema}.llm_calls (call_id, model, provider, source_name, messages, response, " + "INSERT INTO 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)" + "VALUES ('old', 'm', 'p', 's1', '[]', 'old body', 1, 2, 'measured', 10)" ) @@ -637,82 +612,36 @@ async def captured_warnings(): @pytest.fixture -async def pre_tenant_schema(dsn): - """自建临时 schema 里造一张 **22 字段的 issue #11 之前的表**,并留一行历史数据。 +async def pre_tenant_schema(pg_sandbox) -> tuple[str, str]: + """一次性沙箱里造一张 **22 字段的 issue #11 之前的表**,并留一行历史数据。 - 绝不碰共享的 public.llm_calls——本机那张表早已被 `_BACKFILL` 真实补过列, - 指望它还是旧形态的测试第二次跑就会空转。schema 名带 uuid,可重复运行。 + 共享表 `llm_calls` 一个字节都不碰——本机那张表早已被 `_BACKFILL` 真实补过列, + 指望它还是旧形态的测试第二次跑就会空转。 """ - 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() + sandbox = await pg_sandbox(ddl=_PRE_TENANT_DDL, extra=(_PRE_TENANT_INSERT,)) + return sandbox.dsn, sandbox.schema @pytest.fixture -async def fresh_schema(dsn): +async def fresh_schema(pg_sandbox) -> tuple[str, str]: """空 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() + sandbox = await pg_sandbox() + return sandbox.dsn, sandbox.schema @pytest.fixture -async def least_privilege_pre_tenant_dsn(dsn): +async def least_privilege_pre_tenant_dsn(pg_sandbox) -> str: """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() + `role="grantee"` 正是这个现场: 表由 admin 建好(属主不是应用账号),角色只拿到 + 表级 SELECT/INSERT。 + """ + sandbox = await pg_sandbox(ddl=_PRE_TENANT_DDL, role="grantee") + return sandbox.dsn class TestCallerDimensionsAcceptance: @@ -724,7 +653,7 @@ class TestCallerDimensionsAcceptance: recorder = _recorder(fresh_dsn, auto_migrate=True) try: await _record_minimal( - recorder, call_id=_cid("dim"), tenant_id="tenant-a", meta='{"batch": "b7"}' + recorder, call_id="dim", tenant_id="tenant-a", meta='{"batch": "b7"}' ) cols = await _fetch( fresh_dsn, @@ -736,7 +665,7 @@ class TestCallerDimensionsAcceptance: rows = await _fetch( fresh_dsn, "SELECT tenant_id, meta FROM llm_calls WHERE call_id = $1", - _cid("dim"), + "dim", ) assert rows[0]["tenant_id"] == "tenant-a" assert json.loads(rows[0]["meta"]) == {"batch": "b7"} @@ -757,9 +686,7 @@ class TestCallerDimensionsAcceptance: schema_dsn, schema = pre_tenant_schema recorder = _recorder(schema_dsn, auto_migrate=True) try: - await _record_minimal( - recorder, call_id=_cid("new"), tenant_id="tenant-a", meta='{"k": 1}' - ) + await _record_minimal(recorder, call_id="new", tenant_id="tenant-a", meta='{"k": 1}') cols = await _fetch( schema_dsn, "SELECT column_name FROM information_schema.columns " @@ -772,13 +699,13 @@ class TestCallerDimensionsAcceptance: 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")], + ["new", "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"]) == {} + assert by_id["new"]["tenant_id"] == "tenant-a" + assert json.loads(by_id["new"]["meta"]) == {"k": 1} + assert by_id["old"]["tenant_id"] == "" # 不是 None: NULL 会被 RLS 静默吞掉 + assert json.loads(by_id["old"]["meta"]) == {} finally: await recorder.aclose() @@ -809,7 +736,7 @@ class TestCallerDimensionsAcceptance: """ recorder = _recorder(least_privilege_pre_tenant_dsn, auto_migrate=True) try: - await _record_minimal(recorder, call_id=_cid("lpp1")) # 不得抛 + await _record_minimal(recorder, call_id="lpp1") # 不得抛 assert recorder.telemetry_status.degraded is False assert any("补列失败" in m for m in captured_warnings) # 缺列的表上 INSERT 必然失败;逐行 warning 正是"INSERT 照发了"的证据 @@ -821,8 +748,9 @@ class TestCallerDimensionsAcceptance: # issue #12 的目标表形态: 按 created_at 做 RANGE 分区(过期清理 DROP PARTITION 而非 DELETE)。 # PG 强制分区表的唯一约束必须包含分区键,故主键只能是 (call_id, created_at) —— # 这正是带目标的 `ON CONFLICT (call_id)` 再也匹配不到约束的现场。 +# 裸表名: 由 `pg_sandbox` 在沙箱 schema 的 search_path 下执行(工厂的 DDL 执行契约) _PARTITIONED_DDL = """ -CREATE TABLE {schema}.llm_calls ( +CREATE TABLE llm_calls ( call_id TEXT NOT NULL, parent_call_id TEXT, session_id TEXT, @@ -847,14 +775,14 @@ CREATE TABLE {schema}.llm_calls ( sampling TEXT, reasoning_tokens INTEGER, tenant_id TEXT NOT NULL DEFAULT '', - meta JSONB NOT NULL DEFAULT '{{}}'::jsonb, + meta JSONB NOT NULL DEFAULT '{}'::jsonb, PRIMARY KEY (call_id, created_at) ) PARTITION BY RANGE (created_at) """ +# 仍带 `.format`,但只为月份边界——表名两处都已是裸名,由 search_path 定位 _PARTITION_DDL = ( - "CREATE TABLE {schema}.llm_calls_current PARTITION OF {schema}.llm_calls " - "FOR VALUES FROM ('{start}') TO ('{end}')" + "CREATE TABLE llm_calls_current PARTITION OF llm_calls FOR VALUES FROM ('{start}') TO ('{end}')" ) @@ -868,29 +796,18 @@ def _current_month_bounds() -> tuple[str, str]: @pytest.fixture -async def partitioned_schema(dsn): - """自建临时 schema 里造一张按 created_at RANGE 分区的表 + 覆盖当前月的分区。 +async def partitioned_schema(pg_sandbox) -> tuple[str, str]: + """一次性沙箱里造一张按 created_at RANGE 分区的表 + 覆盖当前月的分区。 - 与 legacy_schema 同款隔离: 绝不碰共享的 public.llm_calls,teardown 只 DROP - 自己建的 schema(CASCADE 连分区一并删)。 + 与 `legacy_schema` 同款隔离: 共享表 `llm_calls` 一个字节都不碰,工厂的 + `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() + sandbox = await pg_sandbox( + ddl=_PARTITIONED_DDL, + extra=(_PARTITION_DDL.format(start=start, end=end),), + ) + return sandbox.dsn, sandbox.schema class TestConflictTargetFreeInsert: @@ -905,11 +822,11 @@ class TestConflictTargetFreeInsert: fresh_dsn, _ = fresh_schema recorder = _recorder(fresh_dsn, auto_migrate=True) try: - await _record_minimal(recorder, call_id=_cid("nodup")) - await _record_minimal(recorder, call_id=_cid("nodup"), response="second") + await _record_minimal(recorder, call_id="nodup") + await _record_minimal(recorder, call_id="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") + fresh_dsn, "SELECT response FROM llm_calls WHERE call_id = $1", "nodup" ) assert [r["response"] for r in rows] == ["ok"] # 首行胜出,写入幂等 finally: @@ -926,14 +843,14 @@ class TestConflictTargetFreeInsert: part_dsn, _ = partitioned_schema recorder = _recorder(part_dsn, auto_migrate=True) try: - await _record_minimal(recorder, call_id=_cid("part"), tenant_id="tenant-p") + await _record_minimal(recorder, call_id="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"), + "part", ) - assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(_cid("part"), "tenant-p")] + assert [(r["call_id"], r["tenant_id"]) for r in rows] == [("part", "tenant-p")] finally: await recorder.aclose() @@ -958,7 +875,7 @@ class TestManualSchemaModeAcceptance: recorder = _recorder(schema_dsn, auto_migrate=False) try: recorded = await _record_minimal( - recorder, call_id=_cid("man"), tenant_id="tenant-a", meta='{"k": 1}' + recorder, call_id="man", tenant_id="tenant-a", meta='{"k": 1}' ) cols = await _fetch( schema_dsn, @@ -971,7 +888,7 @@ class TestManualSchemaModeAcceptance: names = ", ".join(_PRE_TENANT_WRITTEN_COLUMNS) rows = await _fetch( - schema_dsn, f"SELECT {names} FROM llm_calls WHERE call_id = $1", _cid("man") + schema_dsn, f"SELECT {names} FROM llm_calls WHERE call_id = $1", "man" ) assert len(rows) == 1 # 裁剪后的 INSERT 真写进去了,不是被 PG 拒收 # 其余 22 列逐列与提交值相等: 少写两列最容易引发的错是剩下的值整体错位 @@ -999,9 +916,9 @@ class TestManualSchemaModeAcceptance: recorder = _recorder(least_privilege_pre_tenant_dsn, auto_migrate=False) try: recorded = await _record_minimal( - recorder, call_id=_cid("manlp1"), tenant_id="tenant-b", meta='{"k": 2}' + recorder, call_id="manlp1", tenant_id="tenant-b", meta='{"k": 2}' ) - await _record_minimal(recorder, call_id=_cid("manlp2"), cost=2.5) + await _record_minimal(recorder, call_id="manlp2", cost=2.5) assert [m for m in captured_warnings if "补列失败" in m] == [] assert [m for m in captured_warnings if "写入失败" in m] == [] @@ -1031,10 +948,10 @@ class TestManualSchemaModeAcceptance: names = ", ".join(_PRE_TENANT_WRITTEN_COLUMNS) rows = await _fetch( least_privilege_pre_tenant_dsn, - f"SELECT {names} FROM llm_calls WHERE call_id LIKE $1 ORDER BY call_id", - f"{_RUN_PREFIX}-manlp%", + f"SELECT {names} FROM llm_calls WHERE call_id = ANY($1::text[]) ORDER BY call_id", + ["manlp1", "manlp2"], ) - assert [r["call_id"] for r in rows] == [_cid("manlp1"), _cid("manlp2")] + assert [r["call_id"] for r in rows] == ["manlp1", "manlp2"] assert dict(rows[0]) == {c: recorded[c] for c in _PRE_TENANT_WRITTEN_COLUMNS} assert rows[1]["cost"] == 2.5 finally: @@ -1184,14 +1101,20 @@ async def _drop_template_objects(dsn: str, schema: str, roles: dict[str, str]) - @pytest.fixture -async def production_template(dsn): +async def production_template(template_admin_dsn): """在临时 schema + 临时角色上跑完 README 的整套模板,产出可用的三条连接串。 - 隔离纪律(M4 事故教训)同 `least_privilege_dsn`: 共享的 `public.llm_calls` + **有意不收敛到 `pg_sandbox`**(设计 §7.1 末段): 它要建三个角色、跑 README + 解析出的整套模板 SQL、按月建分区,权限语义与失败期清理都是它自己的,工厂 + 强行接管会把这些语义压扁。故它是本文件唯一仍持管理连接的 fixture。 + + 隔离纪律(M4 事故教训)同 `least_privilege_dsn`: 共享表 `llm_calls` 一个字节都不碰,建的 schema / 角色 / 函数 / 分区在 teardown 里删净。 """ import asyncpg + dsn = template_admin_dsn + suffix = uuid4().hex[:8] schema = f"pgwtpl_{suffix}" roles = { @@ -1206,7 +1129,7 @@ async def production_template(dsn): f"README 的模板锚点与预期不符: {list(blocks)}" ) - seeded = (_cid("tpl-a"), _cid("tpl-b")) + seeded = ("tpl-a", "tpl-b") admin_dsn = _search_path_dsn(dsn, schema) admin = await asyncpg.connect(dsn, timeout=10) # 权限门放在建任何对象**之前**: `pytest.skip` 抛的是 BaseException, @@ -1228,9 +1151,9 @@ async def production_template(dsn): for call_id, tenant in zip(seeded, ("tenant-a", "tenant-b"), strict=True): await admin.execute(_TEMPLATE_INSERT, call_id, tenant) except BaseException: - # 模板 SQL 出错时也必须删净: 建到一半的 schema 会残留一张 llm_calls, - # 而 `TestSchema` 那条按 table_name 查 information_schema 的用例不带 - # schema 过滤,会被残留物在**下一次运行**里以列数不符的形态误伤 + # 模板 SQL 出错时也必须删净: 建到一半的 schema 会残留一张 llm_calls,而三个 + # 角色是**全局**对象,不随库消失。`TestSchema` 那条用例如今自带 table_schema + # 过滤已不再受残留影响,但残留本身仍是这个共享实例上的垃圾,该清还是要清。 await admin.close() await _drop_template_objects(dsn, schema, roles) raise @@ -1303,17 +1226,17 @@ class TestProductionTemplate: env = production_template conn = await asyncpg.connect(env.app_dsn, timeout=10) try: - await conn.execute(_TEMPLATE_INSERT, _cid("tpl-app"), "tenant-a") + await conn.execute(_TEMPLATE_INSERT, "tpl-app", "tenant-a") with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): - await conn.execute("DELETE FROM llm_calls WHERE call_id = $1", _cid("tpl-app")) + await conn.execute("DELETE FROM llm_calls WHERE call_id = $1", "tpl-app") with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): await conn.execute("UPDATE llm_calls SET response = 'x'") finally: await conn.close() rows = await _fetch( - env.admin_dsn, "SELECT call_id FROM llm_calls WHERE call_id = $1", _cid("tpl-app") + env.admin_dsn, "SELECT call_id FROM llm_calls WHERE call_id = $1", "tpl-app" ) - assert [r["call_id"] for r in rows] == [_cid("tpl-app")] # 写入真落库了 + assert [r["call_id"] for r in rows] == ["tpl-app"] # 写入真落库了 async def test_report_can_read_but_cannot_write(self, production_template): """报表角色: 带租户上下文读得到自己的行,任何写入都被拒。""" @@ -1323,7 +1246,7 @@ class TestProductionTemplate: conn = await asyncpg.connect(env.report_dsn, timeout=10) try: with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): - await conn.execute(_TEMPLATE_INSERT, _cid("tpl-rpt"), "tenant-a") + await conn.execute(_TEMPLATE_INSERT, "tpl-rpt", "tenant-a") async with conn.transaction(): await conn.execute("SELECT set_config('app.tenant_id', 'tenant-a', true)") rows = await conn.fetch("SELECT call_id, tenant_id FROM llm_calls")