iomgaa/PolyGateway
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases 9 Wiki Activity

test: move the Postgres tests off the table other projects write to

Browse Source 
Seven cases wrote straight into the shared table and told their rows
apart by a call_id prefix. Reading was never the problem; the prefix
did that correctly, and it was built for concurrent runs. What it could
not do was stop those writes and deletes from moving a row count that
another test was watching, which is how issue #18 turned red.

They now write into sandbox schemas, which also ends the orphan rows a
killed run used to leave in there. Six fixtures collapse into factory
calls; what they yield is unchanged, so the cases that consume them did
not have to be touched, which is what makes them worth anything as a
check on the move.

Two of the seven kept something. The pool footprint case needs a unique
application_name, since connections are an instance-wide resource that
schema isolation does not reach, so it generates its own uuid instead
of borrowing the run prefix. And the frozen-columns case was querying
information_schema without a schema filter, so any leftover table of
the same name anywhere in the database could fail it: the file already
knew this, in a comment explaining why another fixture cleans up so
carefully. It now filters, and gets checked against a leftover table
planted on purpose.

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