diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index 2db3f58..7888904 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -390,3 +390,237 @@ class TestLeastPrivilegeDeployment: 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', '[]', 'contract text', 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 warnings(): + """捕获库发出的 WARNING;loguru 不经标准 logging,pytest 的 caplog 抓不到。""" + 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, 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 warnings) + # 缺列的表上 INSERT 必然失败;逐行 warning 正是"INSERT 照发了"的证据 + assert any("写入失败" in m for m in warnings) + finally: + await recorder.aclose() diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index 5c3efcc..3906ee5 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -2,6 +2,7 @@ import asyncio import json +import os import sqlite3 import subprocess from pathlib import Path @@ -285,6 +286,131 @@ class TestSQLiteColumnBackfill: recorder.close() +# issue #11 之前的表形态: 22 个 recorder 字段 + created_at = 23 个物理列,没有任何租户维度 +_PRE_TENANT_DDL = """ +CREATE TABLE 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 REAL, + max_inter_token_ms REAL, + cache_hit INTEGER NOT NULL DEFAULT 0, + error TEXT, + cost REAL, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + cached_prompt_tokens INTEGER, + model_reported TEXT, + sampling TEXT, + reasoning_tokens INTEGER +); +""" + +_PRE_TENANT_INSERT = ( + "INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, " + "prompt_tokens, completion_tokens, usage_source, latency_ms) " + "VALUES ('old-row', 'm', 'p', 's1', '[]', 'contract text', 1, 2, 'measured', 10)" +) + + +def _make_pre_tenant_db(path: Path) -> None: + """造一个 issue #11 之前的库: 22 字段旧表 + 一行没有租户归属的历史数据。""" + conn = sqlite3.connect(path) + conn.execute(_PRE_TENANT_DDL) + conn.execute(_PRE_TENANT_INSERT) + conn.commit() + conn.close() + + +class TestSQLiteCallerDimensionsAcceptance: + """issue #11 的机械化验收(SQLite 侧,真实临时文件): 新建库 / 旧表补列 / 补列失败方向。""" + + async def test_fresh_db_round_trips_the_dimensions(self, tmp_path): + """新建库: 列齐全,且维度值原样读回——只验列存在会漏掉写错列位的错。""" + db = tmp_path / "fresh.db" + recorder = SQLiteRecorder(db) + await _record_minimal( + recorder, call_id="c-dim", tenant_id="tenant-a", meta='{"batch": "b7"}' + ) + recorder.close() + + conn = sqlite3.connect(db) + assert [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")] == _EXPECTED_COLUMNS + row = conn.execute( + "SELECT tenant_id, meta FROM llm_calls WHERE call_id = 'c-dim'" + ).fetchone() + assert row[0] == "tenant-a" + assert json.loads(row[1]) == {"batch": "b7"} + + async def test_pre_tenant_table_gains_columns_and_old_rows_stay_auditable(self, tmp_path): + """22 字段旧表补列后,新行带维度,而**老行的 tenant_id 是空串而非 NULL**。 + + 这条直接验收 issue #11 的核心论点(先启用落库、后加列,补列之前的行没有 + 租户归属)。断言方向必须是空串: PG 的 RLS `USING` 表达式对返回 false **或 + NULL** 的行一律隐藏且不报错,故 NULL 的 `tenant_id` 不是"未归属",而是对 + 所有人永久不可见的黑洞;哨兵空串则能被一条 `COUNT(*) WHERE tenant_id = ''` + 审计出来,历史欠账是可见、可量化、可补录的。 + """ + db = tmp_path / "pre_tenant.db" + _make_pre_tenant_db(db) + + recorder = SQLiteRecorder(db) + await _record_minimal(recorder, call_id="new-row", tenant_id="tenant-a", meta='{"k": 1}') + recorder.close() + + conn = sqlite3.connect(db) + cols = [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")] + assert cols == _EXPECTED_COLUMNS # 22 → 24 个 recorder 字段(+ created_at 共 25 物理列) + rows = dict(conn.execute("SELECT call_id, tenant_id FROM llm_calls").fetchall()) + assert rows["new-row"] == "tenant-a" + assert rows["old-row"] == "" # 不是 None: NULL 会被 RLS 静默吞掉 + assert ( + conn.execute("SELECT meta FROM llm_calls WHERE call_id = 'old-row'").fetchone()[0] + == "{}" + ) + + async def test_readonly_file_backfill_failure_keeps_the_recorder_alive(self, tmp_path): + """补列失败的降级方向(SQLite 等价构造: 文件只读)。 + + SQLite 没有角色权限模型,与 PG「只有 SELECT/INSERT 权限的角色」等价的构造 + 是文件本身只读。库文件必须**预先置为 WAL 且干净关闭**,否则 `__init__` 的 + `PRAGMA journal_mode=WAL` 会先撞上只读而让失败点跑到补列之前,测不到本用例 + 要测的那条分支(实测: 非 WAL 库 chmod 444 后该 PRAGMA 报 readonly database)。 + 只读库连 INSERT 都做不了,故这里**只断言**补列失败不清空 `_conn`、不抛出 + `__init__`(sqlite.py `_backfill_columns` 那条纪律),不断言"写入仍成功"。 + """ + if os.geteuid() == 0: + pytest.skip("root 无视文件权限位,只读构造不成立") + db = tmp_path / "readonly.db" + conn = sqlite3.connect(db) + conn.execute("PRAGMA journal_mode=WAL") # 预置 WAL: 让只读连接不必改日志模式 + conn.execute(_PRE_TENANT_DDL) + conn.execute(_PRE_TENANT_INSERT) + conn.commit() + conn.close() + db.chmod(0o444) + + recorder = SQLiteRecorder(db) # 不得抛 + assert recorder._conn is not None # 补列失败 ≠ recorder 失能 + await _record_minimal(recorder, call_id="doomed") # 只读库写不进,但不得抛 + recorder.close() + + db.chmod(0o644) # 还原,让 tmp_path 清理不受阻 + stale = sqlite3.connect(db) + assert [r[1] for r in stale.execute("PRAGMA table_info(llm_calls)")] == ( + _EXPECTED_COLUMNS[:-2] + ) # 补列确实没成功,用例不是在只读库上空转 + + class _FakePgConn: """记录执行过的语句;可让 ALTER/CREATE/探测抛错以模拟权限不足与抖动。