diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index a801c7e..786fd50 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -89,8 +89,13 @@ async def dsn(): async def _record_minimal( recorder: PostgresRecorder, call_id: str | None = None, **overrides -) -> None: - fields = { +) -> dict[str, object]: + """记一行最小遥测,并**返回实际提交的字段**供调用方逐列比对回读结果。 + + 返回值不是顺手加的: 逐列断言若在测试里另抄一份期望值,抄错的那一列会以 + "库写错列位"的形态误报,而漏抄的列则悄悄不被验证。 + """ + fields: dict[str, object] = { "call_id": call_id if call_id is not None else _cid("c1"), "parent_call_id": None, "session_id": "sess-1", @@ -119,6 +124,7 @@ async def _record_minimal( } fields.update(overrides) await recorder.record_llm_call(**fields) + return fields async def _fetch(dsn: str, sql: str, *args): @@ -429,6 +435,15 @@ _PRE_TENANT_INSERT = ( ) +# `_PRE_TENANT_DDL` 的物理列(23 个): 由 `_EXPECTED_COLUMNS` 去掉 issue #11 的两个新维度 +# 派生而非另抄一份——两份常量必然漂移,而漂移的表现是"manual 档没补列"这条断言假绿。 +# 去掉后的顺序与 DDL 逐字一致(tenant_id/meta 在 DDL 里本就排在末尾)。 +_PRE_TENANT_COLUMNS = [c for c in _EXPECTED_COLUMNS if c not in ("tenant_id", "meta")] + +# 回读要逐列比对的字段: 物理列去掉库从不显式写的 created_at,恰好 22 个 +_PRE_TENANT_WRITTEN_COLUMNS = [c for c in _PRE_TENANT_COLUMNS if c != "created_at"] + + def _search_path_dsn(dsn: str, schema: str) -> str: sep = "&" if "?" in dsn else "?" return f"{dsn}{sep}options=-csearch_path%3D{schema}" @@ -749,3 +764,104 @@ class TestConflictTargetFreeInsert: assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(_cid("part"), "tenant-p")] finally: await recorder.aclose() + + +class TestManualSchemaModeAcceptance: + """issue #13 manual 档的真实实例验收: 旧表原样不动,写入照常,缺列只作提示。 + + manual 档的承诺是"库一条 DDL 都不发"——单元测试只能验"没调用 execute", + 真表上才验得了"表结构确实没变"。两条用例分别覆盖有权补列却不补(纪律)与 + 无权补列(现场),后者正是 auto 档会刷出 `补列失败` warning 的那张表。 + """ + + async def test_manual_leaves_the_stale_table_untouched( + self, pre_tenant_schema, captured_warnings + ): + """22 字段旧表 + manual: 列一个不加,行照常落库,缺的两维度静默不写。 + + 与 `test_pre_tenant_table_gains_columns_and_old_rows_stay_auditable` 恰成对照: + 同一张表、同一份负载,只有 `auto_migrate` 不同,列数就必须是 23 与 25 之别。 + """ + schema_dsn, schema = pre_tenant_schema + recorder = PostgresRecorder(schema_dsn, auto_migrate=False) + try: + recorded = await _record_minimal( + recorder, call_id=_cid("man"), 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, + ) + # 表结构逐字不动: 既没多出 tenant_id/meta,也没被顺手改了列序 + assert [r["column_name"] for r in cols] == _PRE_TENANT_COLUMNS + + names = ", ".join(_PRE_TENANT_WRITTEN_COLUMNS) + rows = await _fetch( + schema_dsn, f"SELECT {names} FROM llm_calls WHERE call_id = $1", _cid("man") + ) + assert len(rows) == 1 # 裁剪后的 INSERT 真写进去了,不是被 PG 拒收 + # 其余 22 列逐列与提交值相等: 少写两列最容易引发的错是剩下的值整体错位 + assert dict(rows[0]) == {c: recorded[c] for c in _PRE_TENANT_WRITTEN_COLUMNS} + + assert [m for m in captured_warnings if "写入失败" in m] == [] + assert [m for m in captured_warnings if "补列失败" in m] == [] + notices = [m for m in captured_warnings if "auto_migrate=False" in m] + assert len(notices) == 1 # 准备期一次讲清,不逐行刷屏 + assert "以下维度不会被记录: tenant_id, meta" in notices[0] + finally: + await recorder.aclose() + + async def test_manual_on_a_role_that_cannot_alter_emits_no_backfill_failure( + self, least_privilege_pre_tenant_dsn, captured_warnings + ): + """缺列旧表 + 只授 SELECT/INSERT 的角色 + manual: 补列失败的 warning 彻底消失。 + + auto 档在这张表上会刷出 `补列失败` 再刷 `写入失败`(见 + `test_backfill_failure_degrades_per_row_not_wholesale`)——那是 issue #13 要 + 消灭的噪声。manual 档下 ALTER 压根不发,取而代之的是一条点名缺列并附可直接 + 执行的 ALTER 的提示,而遥测照常落库。 + """ + recorder = PostgresRecorder(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}' + ) + await _record_minimal(recorder, call_id=_cid("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 recorder._failed is False + notices = [m for m in captured_warnings if "auto_migrate=False" in m] + assert len(notices) == 1 # 准备期一次,第二行不再重复 + assert "以下维度不会被记录: tenant_id, meta" in notices[0] + # 提示里的 SQL 必须可直接粘贴执行,而不是只报个列名 + assert ( + "ALTER TABLE llm_calls ADD COLUMN tenant_id TEXT NOT NULL DEFAULT '';" in notices[0] + ) + assert ( + "ALTER TABLE llm_calls ADD COLUMN meta JSONB NOT NULL DEFAULT '{}'::jsonb;" + in notices[0] + ) + + # 该角色无权 ALTER,表必然还是旧形态: 缺的两列确实没被写 + cols = await _fetch( + least_privilege_pre_tenant_dsn, + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = current_schema() AND table_name = 'llm_calls' " + "ORDER BY ordinal_position", + ) + assert [r["column_name"] for r in cols] == _PRE_TENANT_COLUMNS + + 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%", + ) + assert [r["call_id"] for r in rows] == [_cid("manlp1"), _cid("manlp2")] + assert dict(rows[0]) == {c: recorded[c] for c in _PRE_TENANT_WRITTEN_COLUMNS} + assert rows[1]["cost"] == 2.5 + finally: + await recorder.aclose()