docs: ship a production deployment template with its own test

README 的多租户 RLS 段扩为完整的"生产部署 DDL 模板"一节: 三角色、 REVOKE + 触发器兜底、created_at RANGE 分区与 pg_partman retention、库需要的最小权限、合规下游的推荐配置、截断覆盖面的诚实声明,以及 SQLite 侧按天轮转库文件的保留期建议。模板 SQL 只有 README 里这一份: 集成测试用 HTML 注释锚点 (``)把它解析出来,做受控标识符替换后在真实 PG 的临时 schema + 临时角色上逐条执行(doctest 同款范式)。测试里另抄一份就会与 README 各自漂移,而"README 的 SQL 能跑"这个承诺只在同源时才成立;解析不到必须当场红,故块名与占位符都显式钉死。新增 5 条真实 PG 用例: app 能 INSERT 不能 UPDATE/DELETE(拿到的是权限错而非触发器错)、report 只读、未设 app.tenant_id 时读为零行且设了只见本租户、行落进当月分区、触发器拦得住 DELETE 却拦不住 DROP PARTITION(这是"清理只能走分区"的机械化依据)。写侧 policy 定为 WITH CHECK (true) 而非等值比较: 库用一个连接池给所有租户写遥测且从不发 set_config,把写侧绑到 GUC 上会让每条 INSERT 被拒,而遥测的失败方向是静默降级——表现是整表零行。
2026-08-19 14:42:36 -04:00
parent 511aa4899c
commit daf7ab3268
2 changed files with 468 additions and 24 deletions
@@ -14,7 +14,9 @@ import asyncio
 import json
 import os
 import re
+from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
+from pathlib import Path
 from uuid import uuid4

 import pytest
@@ -913,3 +915,297 @@ class TestPublishedSchemaScript:
        await _execute_script(fresh_dsn, script)  # 可重复执行: 第二遍不得抛
        rerun = [r["column_name"] for r in await _fetch(fresh_dsn, _PHYSICAL_COLUMNS_SQL, schema)]
        assert rerun == actual  # 且第二遍没有偷偷改动表结构
+
+
+# ---------------------------------------------------------------------------
+# issue #12 Task 4: README 的生产部署 DDL 模板,逐条在真实 PG 上执行
+# ---------------------------------------------------------------------------
+
+# 模板 SQL **只有一份**,在 README 里。测试从 README 解析出来跑,而不是在这里另抄
+# 一份: 抄一份就是两份会各自漂移的东西,而"README 里的 SQL 能跑"这个承诺恰恰只在
+# 同源时才成立(doctest / Rust doc tests / mdbook test 都是这个范式)。
+_README = Path(__file__).resolve().parents[2] / "README.md"
+
+# 锚点写成 HTML 注释,渲染时不可见,比按章节标题或代码块序号定位稳固得多。
+_TEMPLATE_BLOCK = re.compile(r"<!-- pg-template:([a-z_]+) -->\s*\n```sql\n(.*?)\n```", re.DOTALL)
+
+# 顺序即执行顺序;数量与名字都钉死——解析不到或多出一块必须当场红,
+# 绝不能退化成空列表让这条测试变成永远绿的摆设。
+_EXPECTED_TEMPLATE_BLOCKS = (
+    "roles",
+    "table",
+    "partition",
+    "grants",
+    "immutable",
+    "rls",
+    "index",
+)
+
+# README 里必须原样保留、由本测试做受控替换的标识符。README 那份是给下游照抄的,
+# 故占位符是**合法可执行的具体值**而不是 `<schema>` 之类的尖括号洞。
+_TEMPLATE_PLACEHOLDERS = (
+    "polygateway_owner",
+    "polygateway_app",
+    "polygateway_report",
+    "CHANGE_ME_APP",
+    "CHANGE_ME_REPORT",
+    "SCHEMA public",
+    "llm_calls_2026_01",
+    "'2026-01-01 00:00:00+00'",
+    "'2026-02-01 00:00:00+00'",
+)
+
+# 应用角色在生产里能发的唯一一类写语句(与库的 INSERT 同形,只列 NOT NULL 列)
+_TEMPLATE_INSERT = (
+    "INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, "
+    "prompt_tokens, completion_tokens, usage_source, latency_ms, tenant_id) "
+    "VALUES ($1, 'm', 'p', 's1', '[]', 'ok', 1, 2, 'measured', 10, $2)"
+)
+
+
+def _template_blocks() -> dict[str, str]:
+    """从 README 解析带锚点的 SQL 块;顺序即文中出现顺序。"""
+    return dict(_TEMPLATE_BLOCK.findall(_README.read_text(encoding="utf-8")))
+
+
+@dataclass(frozen=True)
+class _TemplateEnv:
+    """模板部署完成后的现场句柄:三个角色各自的连接串 + 当月分区名。"""
+
+    admin_dsn: str
+    app_dsn: str
+    report_dsn: str
+    schema: str
+    partition: str
+    seeded: tuple[str, str]  # (tenant-a 的行, tenant-b 的行)
+
+
+def _localize(sql: str, schema: str, roles: dict[str, str], month: datetime) -> str:
+    """把 README 里给下游照抄的标识符换成本次运行专属的临时对象。
+
+    替换规则写在测试里而不是让 README 变得不可直接复制: README 里那份必须是
+    下游 `pip install` 后照抄就能用的,占位符因此都是合法 SQL 值。
+    """
+    start = month.strftime("%Y-%m-%d %H:%M:%S%z")
+    end = (month + timedelta(days=32)).replace(day=1).strftime("%Y-%m-%d %H:%M:%S%z")
+    for placeholder, actual in (
+        # 长名在前: 三个角色名互不为前缀,但顺序稳定便于排查
+        ("polygateway_owner", roles["owner"]),
+        ("polygateway_report", roles["report"]),
+        ("polygateway_app", roles["app"]),
+        ("CHANGE_ME_APP", _PROBE_PASSWORD),
+        ("CHANGE_ME_REPORT", _PROBE_PASSWORD),
+        ("SCHEMA public", f"SCHEMA {schema}"),
+        ("llm_calls_2026_01", f"llm_calls_{month:%Y_%m}"),
+        ("'2026-01-01 00:00:00+00'", f"'{start}'"),
+        ("'2026-02-01 00:00:00+00'", f"'{end}'"),
+    ):
+        sql = sql.replace(placeholder, actual)
+    return sql
+
+
+def _role_dsn(dsn: str, role: str, schema: str) -> str:
+    low = re.sub(r"//[^@/]+@", f"//{role}:{_PROBE_PASSWORD}@", dsn, count=1)
+    return _search_path_dsn(low, schema)
+
+
+async def _drop_template_objects(dsn: str, schema: str, roles: dict[str, str]) -> None:
+    """删净临时 schema 与三个角色(角色是**全局**对象,漏删会跨 run 残留)。"""
+    import asyncpg
+
+    admin = await asyncpg.connect(dsn, timeout=10)
+    try:
+        await admin.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE")
+        for role in roles.values():
+            await admin.execute(f"DROP OWNED BY {role}")
+            await admin.execute(f"DROP ROLE IF EXISTS {role}")
+    finally:
+        await admin.close()
+
+
+@pytest.fixture
+async def production_template(dsn):
+    """在临时 schema + 临时角色上跑完 README 的整套模板,产出可用的三条连接串。
+
+    隔离纪律(M4 事故教训)同 `least_privilege_dsn`: 共享的 `public.llm_calls`
+    一个字节都不碰,建的 schema / 角色 / 函数 / 分区在 teardown 里删净。
+    """
+    import asyncpg
+
+    suffix = uuid4().hex[:8]
+    schema = f"pgwtpl_{suffix}"
+    roles = {
+        "owner": f"pgwtpl_owner_{suffix}",
+        "app": f"pgwtpl_app_{suffix}",
+        "report": f"pgwtpl_report_{suffix}",
+    }
+    month = datetime.now(UTC).replace(day=1, hour=0, minute=0, second=0, microsecond=0)
+    blocks = _template_blocks()
+    # 解析不到就地红: 空 dict 会让下面的 for 一句不执行,测试变成"只验证了能连上库"
+    assert list(blocks) == list(_EXPECTED_TEMPLATE_BLOCKS), (
+        f"README 的模板锚点与预期不符: {list(blocks)}"
+    )
+
+    seeded = (_cid("tpl-a"), _cid("tpl-b"))
+    admin_dsn = _search_path_dsn(dsn, schema)
+    admin = await asyncpg.connect(dsn, timeout=10)
+    # 权限门放在建任何对象**之前**: `pytest.skip` 抛的是 BaseException,
+    # 若它在下面的清理块内触发,清理会去 DROP 从未建过的角色而把 skip 盖掉
+    can_create = await admin.fetchval(
+        "SELECT rolcreaterole OR rolsuper FROM pg_roles WHERE rolname = current_user"
+    )
+    if not can_create:
+        await admin.close()
+        pytest.skip("当前账号无权建临时角色,跳过生产模板用例")
+    try:
+        await admin.execute(f"CREATE SCHEMA {schema}")
+        await admin.execute(f"SET search_path = {schema}")
+        # README §2 写明的前置步骤: 先用库自带脚本建出普通表当模子
+        await admin.execute(telemetry_schema_sql("postgres"))
+        for name in _EXPECTED_TEMPLATE_BLOCKS:
+            await admin.execute(_localize(blocks[name], schema, roles, month))
+        # 种两个租户的行(超级用户绕过 RLS,属于布景不属于被测行为)
+        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 过滤,会被残留物在**下一次运行**里以列数不符的形态误伤
+        await admin.close()
+        await _drop_template_objects(dsn, schema, roles)
+        raise
+    finally:
+        if not admin.is_closed():
+            await admin.close()
+
+    yield _TemplateEnv(
+        admin_dsn=admin_dsn,
+        app_dsn=_role_dsn(dsn, roles["app"], schema),
+        report_dsn=_role_dsn(dsn, roles["report"], schema),
+        schema=schema,
+        partition=f"llm_calls_{month:%Y_%m}",
+        seeded=seeded,
+    )
+
+    admin = await asyncpg.connect(dsn, timeout=10)
+    try:
+        await admin.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE")
+        for role in roles.values():
+            await admin.execute(f"DROP OWNED BY {role}")
+            await admin.execute(f"DROP ROLE IF EXISTS {role}")
+    finally:
+        await admin.close()
+
+
+class TestProductionTemplate:
+    """issue #12: README 的生产部署 DDL 模板必须逐条可执行,且行为与文中描述一致。
+
+    模板出错的代价全部落在下游身上(照抄就中招),而人工核对不构成回归保护——
+    改一次 README 就会悄悄失去它。故这里从 README **直接解析** SQL 来执行。
+    """
+
+    def test_readme_exposes_exactly_the_expected_template_blocks(self):
+        """先钉死解析本身: 锚点没了、改名了、块数变了,这条当场红。
+
+        没有它,`production_template` 里解析出空 dict 时下面每条用例都会以
+        "表不存在"之类的间接形态失败,真因(README 结构变了)要靠猜。
+        """
+        blocks = _template_blocks()
+        assert list(blocks) == list(_EXPECTED_TEMPLATE_BLOCKS)
+        assert all(sql.strip() for sql in blocks.values())
+        joined = "\n".join(blocks.values())
+        for placeholder in _TEMPLATE_PLACEHOLDERS:
+            # 占位符没了 = 受控替换静默失效,测试会去打真实的 polygateway_* 角色
+            assert placeholder in joined, f"README 模板缺占位符 {placeholder!r}"
+
+    async def test_app_can_insert_but_cannot_mutate(self, production_template):
+        """应用角色: INSERT 通过,UPDATE / DELETE 被权限层拒绝(不是被触发器拒)。
+
+        权限检查早于行级触发器,故这里拿到的必须是 InsufficientPrivilegeError——
+        若换成触发器的 RaiseError,说明 REVOKE 那一块没生效,而"不可变"就只剩
+        一层属主随手可关的兜底。
+        """
+        import asyncpg
+
+        env = production_template
+        conn = await asyncpg.connect(env.app_dsn, timeout=10)
+        try:
+            await conn.execute(_TEMPLATE_INSERT, _cid("tpl-app"), "tenant-a")
+            with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError):
+                await conn.execute("DELETE FROM llm_calls WHERE call_id = $1", _cid("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")
+        )
+        assert [r["call_id"] for r in rows] == [_cid("tpl-app")]  # 写入真落库了
+
+    async def test_report_can_read_but_cannot_write(self, production_template):
+        """报表角色: 带租户上下文读得到自己的行,任何写入都被拒。"""
+        import asyncpg
+
+        env = production_template
+        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")
+            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")
+            assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(env.seeded[0], "tenant-a")]
+        finally:
+            await conn.close()
+
+    async def test_reads_are_fail_closed_until_the_tenant_guc_is_set(self, production_template):
+        """未设 `app.tenant_id` → 零行(fail-closed);设了 → 只看得到本租户。
+
+        两个断言缺一不可: 只验"设了能看到自己的"漏掉了 GUC 未设时全表泄露,
+        只验"未设是零行"则一条永远返回 false 的 policy 也能通过。
+        """
+        import asyncpg
+
+        env = production_template
+        conn = await asyncpg.connect(env.app_dsn, timeout=10)
+        try:
+            async with conn.transaction():
+                assert await conn.fetch("SELECT call_id FROM llm_calls") == []
+            async with conn.transaction():
+                await conn.execute("SELECT set_config('app.tenant_id', 'tenant-b', true)")
+                rows = await conn.fetch("SELECT call_id, tenant_id FROM llm_calls")
+            assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(env.seeded[1], "tenant-b")]
+        finally:
+            await conn.close()
+
+    async def test_rows_land_in_the_current_month_partition(self, production_template):
+        """分区表写入成功,且行确实落进当月分区(不是落进某个兜底分区)。"""
+        env = production_template
+        rows = await _fetch(
+            env.admin_dsn,
+            "SELECT tableoid::regclass::text AS part FROM llm_calls WHERE call_id = $1",
+            env.seeded[0],
+        )
+        assert [r["part"].split(".")[-1] for r in rows] == [env.partition]
+
+    async def test_trigger_blocks_delete_while_drop_partition_still_works(
+        self, production_template
+    ):
+        """兜底触发器拦得住 DELETE(连超级用户也拦),却拦不住 DROP PARTITION。
+
+        这正是 README 说"清理只能走 DROP PARTITION 而不是 DELETE"的机械化依据:
+        既要对应用角色 REVOKE DELETE、又要能清理过期数据,分区是唯一不冲突的解。
+        """
+        import asyncpg
+
+        env = production_template
+        conn = await asyncpg.connect(env.admin_dsn, timeout=10)
+        try:
+            with pytest.raises(asyncpg.exceptions.RaiseError) as exc:
+                await conn.execute("DELETE FROM llm_calls WHERE call_id = $1", env.seeded[0])
+            assert "不可变审计表" in str(exc.value)
+            await conn.execute(f"ALTER TABLE llm_calls DETACH PARTITION {env.partition}")
+            await conn.execute(f"DROP TABLE {env.partition}")
+            assert await conn.fetchval("SELECT count(*) FROM llm_calls") == 0
+        finally:
+            await conn.close()