"""合成游标与窄投影验收；记录真实存储层发出的SQL，不连接数据库。""" import base64 import json from contextlib import contextmanager from datetime import UTC, datetime, timedelta from types import SimpleNamespace from uuid import UUID import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from muse.作品规划.接口 import 作品错误 from muse.共享.调用身份 import 内容用途, 用途, 调用身份 from muse.接入.http.作者会话 import 要求作者 from muse.接入.http.路由.作者审阅 import 路由 from muse.接入.http.错误响应 import 安装错误响应 from muse.正式变更.接口 import 变更错误 from muse.正文写作.候选摘要 import 编码游标, 解码游标, 读取摘要页 from muse.正文写作.接口 import 正文服务 身份 = 调用身份("author-local", None, 用途.生产, 内容用途.规划) 时间 = datetime(2026, 9, 17, tzinfo=UTC) class 合成连接: def __init__(self, 数量): self.行集 = [ { "candidate_id": str(UUID(int=i + 1)), "revision": 1, "candidate_hash": f"hash-{i}", "created_at": 时间, "origin": "author", "base_revision": 2, "source_stale": False, } for i in range(数量) ] self.语句 = [] self.结果 = [] self.作者 = "author-local" self.当前版本 = 2 self.决定 = {} self.借用 = 0 @contextmanager def 连接(self, *, 只读): assert 只读 self.借用 += 1 yield self def cursor(self, **_): return self def execute(self, sql, args): self.语句.append((sql, args)) if "FROM muse_chapter c" in sql: self.结果 = [{"chapter_id": "chapter", "work_id": "work", "author_id": self.作者}] elif "FROM muse_writing_candidate c" in sql: assert "first.revision=1" in sql assert "v.revision=c.current_revision" in sql assert "first.created_at DESC,c.candidate_id DESC LIMIT %s" in sql assert "OFFSET" not in sql # 真实SQL须只选择窄字段；不允许把全文丢到Python后再裁剪。 projection = sql.split("FROM muse_writing_candidate c")[0] assert not any(v in projection for v in ("v.*", "v.document,", "checks", "merge_basis")) assert "LEFT JOIN muse_branch_merge_source" in sql assert "IS DISTINCT FROM" in sql 上界 = (args[3], args[4]) if args[2] else None 最后 = (args[6], args[7]) if args[5] else None self.结果 = sorted( [ r.copy() for r in self.行集 if (上界 is None or (r["created_at"], r["candidate_id"]) <= 上界) and (最后 is None or (r["created_at"], r["candidate_id"]) < 最后) ], key=lambda r: (r["created_at"], r["candidate_id"]), reverse=True, )[: args[-1]] elif "current_revision AS revision" in sql: assert "document_hash" not in sql and "document_version" not in sql self.结果 = [{"revision": self.当前版本}] if self.当前版本 else [] elif "FROM muse_candidate_decision" in sql: assert args[0] == "author-local" and "author_id=%s" in sql self.结果 = [ {"candidate_id": key[0], "candidate_revision": key[1], "decision": value} for key, value in self.决定.items() if key in zip(args[1], args[2], strict=True) ] else: raise AssertionError(f"摘要发出了未知或全文SQL: {sql}") return self def fetchone(self): return self.结果[0] if self.结果 else None def fetchall(self): return self.结果 @pytest.mark.case_id("TC-O08-SUMMARY-001") @pytest.mark.parametrize("数量", [0, 1, 20, 200, 2000]) @pytest.mark.parametrize("limit", [20, 50]) def test_摘要有界且查询数不随候选数增长__o08001(数量, limit): 连 = 合成连接(数量) 页 = 正文服务(连, None).列出候选摘要(身份, "chapter", limit=limit) assert len(页["items"]) == min(数量, limit) assert len(连.语句) == (4 if 数量 else 3) assert 连.借用 == 1 assert bool(页["next_cursor"]) == (数量 > limit) for 项 in 页["items"]: assert set(项) == { "candidate_id", "revision", "candidate_hash", "document_id", "chapter_id", "branch_id", "origin", "base_revision", "current_document_revision", "decision", "stale", } @pytest.mark.case_id("TC-O08-SUMMARY-002") def test_同时间跨页不重漏且新增刷新可见__o08002(): 连 = 合成连接(45) 第一页 = 读取摘要页(连, 身份, "chapter", "main", 20, None) ids = [r["candidate_id"] for r in 第一页["items"]] 新行 = { **连.行集[0], "candidate_id": str(UUID(int=999)), "created_at": 时间 + timedelta(seconds=1), } 连.行集.append(新行) # 首版本创建时间不变；编辑当前候选也不改变排序身份。 连.行集[0]["revision"] = 2 cursor = 第一页["next_cursor"] while cursor: 页 = 读取摘要页(连, 身份, "chapter", "main", 20, cursor) ids.extend(r["candidate_id"] for r in 页["items"]) cursor = 页["next_cursor"] assert len(ids) == len(set(ids)) == 45 assert 新行["candidate_id"] not in ids assert ( 读取摘要页(连, 身份, "chapter", "main", 20, None)["items"][0]["candidate_id"] == 新行["candidate_id"] ) @pytest.mark.case_id("NC-O08-SUMMARY-003") @pytest.mark.parametrize( "字段,值", [ ("version", True), ("version", 2), ("chapter", "other"), ("branch", "other"), ("order", "asc"), ("upper", []), ("upper", ["2026-09-17", str(UUID(int=10))]), ("last", [时间.isoformat(), "not-a-uuid"]), ("last", [9, str(UUID(int=1))]), ("last", [时间.isoformat(), str(UUID(int=11))]), ("extra", 1), ], ) def test_游标严格校验作用域类型顺序__o08003(字段, 值): cursor = 编码游标("chapter", "main", (时间, str(UUID(int=10))), (时间, str(UUID(int=5)))) 数据 = json.loads(base64.urlsafe_b64decode(cursor)) 数据[字段] = 值 无效 = base64.urlsafe_b64encode(json.dumps(数据).encode()).decode() with pytest.raises(变更错误, match="候选摘要游标无效"): 解码游标(无效, "chapter", "main") @pytest.mark.case_id("NC-O08-SUMMARY-004") @pytest.mark.parametrize("cursor", ["", "!invalid", "a" * 2049, "W10=", "bnVsbA==", "e30="]) def test_损坏游标拒绝__o08004(cursor): with pytest.raises(变更错误): 解码游标(cursor, "chapter", "main") @pytest.mark.case_id("NC-O08-SUMMARY-005") @pytest.mark.parametrize("limit", [0, 51, -1, True, "20"]) def test_服务拒绝无效页尺寸__o08005(limit): 连 = 合成连接(20) with pytest.raises(变更错误, match="数量必须"): 读取摘要页(连, 身份, "chapter", "main", limit, None) assert 连.语句 == [] @pytest.mark.case_id("TC-O08-SUMMARY-006") def test_只读当前版本一次且合并来源与决定独立__o08006(): 连 = 合成连接(3) 连.行集[0]["source_stale"] = True 连.行集[1]["base_revision"] = 1 连.决定[(连.行集[2]["candidate_id"], 1)] = "reject" 页 = 读取摘要页(连, 身份, "chapter", "main", 20, None) assert [r["stale"] for r in 页["items"]] == [False, True, True] assert [r["decision"] for r in 页["items"]] == ["reject", "undecided", "undecided"] assert sum("current_revision AS revision" in sql for sql, _ in 连.语句) == 1 连.作者 = "other-author" with pytest.raises(作品错误, match="不属于当前作者"): 读取摘要页(连, 身份, "chapter", "main", 20, None) assert "FROM muse_chapter c" in 连.语句[-1][0] @pytest.mark.case_id("TC-O08-SUMMARY-007") def test_HTTP摘要页尺寸与准确响应合同__o08007(): 连 = 合成连接(55) 服务 = 正文服务(连, None) app = FastAPI() app.include_router(路由) 安装错误响应(app) app.dependency_overrides[要求作者] = lambda: "author-local" app.state.装配 = SimpleNamespace( 要求正文=lambda: 服务, 配置=SimpleNamespace(运行用途=用途.生产) ) with TestClient(app) as 客户端: 路径 = "/api/v1/chapters/chapter/writing-candidate-summaries" 响应 = 客户端.get(路径) assert 响应.status_code == 200 assert len(响应.json()["items"]) == 20 assert 客户端.get(路径, params={"limit": 50}).status_code == 200 assert 客户端.get(路径, params={"limit": 51}).status_code == 422 assert 客户端.get(路径, params={"limit": 0}).status_code == 422 assert 客户端.get(路径, params={"cursor": "malformed"}).json()["code"] == "INVALID_CURSOR" 跨支 = 客户端.get(路径, params={"cursor": 响应.json()["next_cursor"], "branch_id": "other"}) assert 跨支.status_code == 422 and 跨支.json()["code"] == "INVALID_CURSOR" 下一页 = 客户端.get(路径, params={"cursor": 响应.json()["next_cursor"]}) assert 下一页.status_code == 200 assert not set(r["candidate_id"] for r in 响应.json()["items"]) & set( r["candidate_id"] for r in 下一页.json()["items"] ) schema = app.openapi() 摘要 = next( s["properties"] for s in schema["components"]["schemas"].values() if s.get("title") == "正文候选摘要" ) assert not {"document", "visible_text", "changes"} & set(摘要) assert "/api/v1/chapters/{chapter_id}/writing-candidates" in schema["paths"] @pytest.mark.case_id("NC-O08-SUMMARY-008") def test_未认证调用不得读取候选__o08008(): 连 = 合成连接(20) 未认证 = 调用身份(None, None, 用途.生产, 内容用途.规划) with pytest.raises(变更错误, match="需要已认证作者"): 读取摘要页(连, 未认证, "chapter", "main", 20, None) assert 连.语句 == []