Author SHA1 Message Date
张胜宇 c5adf01603 test: 补端点编号守卫与权限判定的 HTTP 层用例
补两道守卫,覆盖 2026-09-12 那次合并评审暴露出来的两个盲区。

一、`docs/05` §19 端点编号唯一性(新增 `tools/check_docs_endpoint_ids.py`)
- 背景:`tools/check_authoritative_docs.py` 只校验 `docs/` 的**文件名编号**,不校验
  §19 的**端点编号**。两条线各自新增端点时都占了 `A034`/`A035`,合并后 §19 同时
  存在两个 `A034` 与两个 `A035`,而文档守卫照样通过 —— 编号复用会静默遗留。
- 口径要点:**不枚举前缀白名单**,前缀从数据里归纳后连同数量一起输出。同一件事上
  还踩过一次"扫描正则写成 `[AMKCS]`,漏掉 `O`/`R` 两段,把 62 个端点报成 55 个"——
  漏掉的号段一旦被复用,脚本仍会报"重复 0"。所以覆盖报告会打印
   `62 个端点编号 / 6 个号段(A×40、C×7、K×4、M×4、O×3、R×4)`,让漏扫本身可见。
- 只读、不依赖任何环境变量(纯解析文档),可在裸检出环境直接跑。

二、端点权限判定的 HTTP 层用例(新增 `tests/unit/api/test_permission_enforcement.py`)
- 背景:既有用例都通过 `build_request_context` 覆写注入**已经带好权限**的上下文,
  只覆盖"有权限能通",覆盖不到"缺权限必须被拒"。而"权限码在库里根本不存在"这类
  环境数据问题(本次三个新权限码)恰恰只会在这一层暴露:权限判定发生在身份解析之后,
  服务层测试自己构造 `RequestContext`,权限字段由测试塞入,所以全绿也照样漏。
- 覆盖客服二期三个权限码对应的 5 个端点:
  `handover:read`(工单列表/详情)、`memory:candidate:review`(候选列表/审核)、
  `memory:candidate:confirm`(用户确认)。
- 正反双向断言:缺权限 → 必须 403 且错误码为 `AGENT_PERMISSION_DENIED`;
  带权限 → **不能**再是 403(这一条把端点要求的权限码钉住,改动即红)。
- 只替换两处边界:`build_request_context`(跳过 JWT 与身份库)与
  `AuthorizationService` 的审计落库(内存替身);真实路由、真实权限判定与真实 403 信封。
- 已做反向验证:给一个不存在的权限码时,5 个端点全部被拒(403),确认正向断言非空过。

验证(隔离 worktree,基线 `origin/qyqy_develop` = 4d8edb4):
pytest tests/unit tests/contract -> 1294 passed, 2 skipped, 0 failed;
ruff check app tests tools alembic 全通过;mypy app 244 源文件 0 错;
check_authoritative_docs(50 份文档无撞号)与本脚本均通过。
2026-09-12 12:57:16 +08:00
3 changed files with 450 additions and 0 deletions
@@ -0,0 +1,209 @@
"""端点权限判定的 HTTP 层用例(不连数据库)。
## 为什么单开这一组
既有用例(含 `tests/integration/test_customer_service_handover_admin_mysql.py`)都通过
`app.dependency_overrides[build_request_context]` 注入**已经带好权限**的上下文,
所以只覆盖了"有权限能通",**覆盖不到"缺权限必须被拒"**。
而 2026-09-12 出过的那类事故(客服二期三个权限码 `memory:candidate:confirm` /
`memory:candidate:review` / `handover:read` 没随代码合并进环境)恰好只会在这一层暴露:
权限判定发生在身份解析之后(`app/api/dependencies/auth.py` → `IdentityService.resolve`),
而服务层测试是自己构造 `RequestContext` 的、权限字段由测试塞进去,
所以"权限码在库里根本不存在"这类问题,单元/集成测试全绿也照样漏。
## 替换了什么、保留了哪些真实部分
- **保留**:真实路由、真实 `build_request_context` 依赖注入位、真实
`AuthorizationService.require` 判定与 `ForbiddenAgentError` → 403 信封。
- **替换**(只替换两处边界):
1. `build_request_context` → 直接返回指定权限集的上下文(跳过 JWT 与身份库查询);
2. `AuthorizationService` 的审计落库 → 内存替身(拒绝时会写一条 `permission.denied`)。
## 为什么正反两个方向都要断言
- **反向**(权限集里没有该权限 → 必须 403):防"端点忘了做权限校验"。
- **正向**(把该权限放进去 → **不能**再是 403):把"端点要求的权限码"钉住 ——
若有人改了端点要的权限码而没同步这里,正向会立刻变红。
正向之后的下游(MySQL / Redis)在单元环境不可用,返回 5xx 属正常;
本用例只关心"不再因权限被拒",因此断言 `status_code != 403` 而不是 `== 200`。
"""
from __future__ import annotations
from typing import Any
import pytest
from fastapi.testclient import TestClient
from app.api.dependencies.auth import build_request_context
from app.core.contracts import RequestContext
from app.main import create_app
from app.service import authorization_service
PERMISSION_DENIED_CODE = "AGENT_PERMISSION_DENIED"
#: (HTTP 方法, 路径, 该端点要求的权限码, 角色, 请求体)
CASES: tuple[tuple[str, str, str, tuple[str, ...], dict[str, Any] | None], ...] = (
(
"GET",
"/api/v1/admin/customer-service/handover-tickets",
"handover:read",
("admin",),
None,
),
(
"GET",
"/api/v1/admin/customer-service/handover-tickets/T-0001",
"handover:read",
("admin",),
None,
),
(
"GET",
"/api/v1/admin/customer-profile-candidates",
"memory:candidate:review",
("admin",),
None,
),
(
"POST",
"/api/v1/admin/customer-profile-candidates/1/reviews",
"memory:candidate:review",
("admin",),
{"decision": "approved"},
),
(
"POST",
"/api/v1/users/me/memory-candidates/1/decisions",
"memory:candidate:confirm",
("customer",),
{"decision": "confirmed"},
),
)
class _FakeSession:
"""只承载"拒绝时写一条审计"这一步,不碰数据库。"""
def add(self, _instance: object) -> None:
return None
def begin(self) -> _FakeSession:
return self
async def __aenter__(self) -> _FakeSession:
return self
async def __aexit__(self, *_exc: object) -> bool:
return False
class _FakeSessionFactory:
def __call__(self) -> _FakeSession:
return _FakeSession()
def _contains_code(payload: Any, code: str) -> bool:
"""在响应 JSON 里递归找错误码,不依赖信封的具体层级。"""
if isinstance(payload, dict):
return any(
(isinstance(value, str) and value == code) or _contains_code(value, code)
for value in payload.values()
)
if isinstance(payload, list):
return any(_contains_code(item, code) for item in payload)
return False
def _request(
monkeypatch: pytest.MonkeyPatch,
*,
method: str,
path: str,
permissions: tuple[str, ...],
roles: tuple[str, ...],
body: dict[str, Any] | None,
) -> Any:
app = create_app()
async def override_context() -> RequestContext:
return RequestContext(
user_id="9003",
trace_id="permission-enforcement-trace",
roles=roles,
permissions=permissions,
)
app.dependency_overrides[build_request_context] = override_context
monkeypatch.setattr(authorization_service, "SessionFactory", _FakeSessionFactory())
try:
# 正向用例会走到下游(MySQL / Redis),单元环境不可用会抛异常;
# 这里让它变成 5xx 响应,用例只关心"不再因权限被拒"。
with TestClient(app, raise_server_exceptions=False) as client:
return client.request(method, path, json=body)
finally:
app.dependency_overrides.clear()
@pytest.mark.parametrize(("method", "path", "permission", "roles", "body"), CASES)
def test_endpoint_denies_request_without_required_permission(
monkeypatch: pytest.MonkeyPatch,
method: str,
path: str,
permission: str,
roles: tuple[str, ...],
body: dict[str, Any] | None,
) -> None:
"""反向:权限集里没有该权限时,必须 403 且错误码是 `AGENT_PERMISSION_DENIED`。"""
response = _request(
monkeypatch,
method=method,
path=path,
permissions=(),
roles=roles,
body=body,
)
assert response.status_code == 403, (
f"{method} {path} 缺少 {permission} 时预期 403,实际 {response.status_code}:"
f"{response.text[:200]}"
)
assert _contains_code(response.json(), PERMISSION_DENIED_CODE), response.text
@pytest.mark.parametrize(("method", "path", "permission", "roles", "body"), CASES)
def test_endpoint_accepts_request_with_required_permission(
monkeypatch: pytest.MonkeyPatch,
method: str,
path: str,
permission: str,
roles: tuple[str, ...],
body: dict[str, Any] | None,
) -> None:
"""正向:放入该权限后不能再是 403 —— 同时把端点要求的权限码钉住。"""
response = _request(
monkeypatch,
method=method,
path=path,
permissions=(permission,),
roles=roles,
body=body,
)
assert response.status_code != 403, (
f"{method} {path} 带上 {permission} 后仍被拒;"
f"端点要求的权限码可能已改动:{response.text[:200]}"
)
assert not _contains_code(response.json(), PERMISSION_DENIED_CODE), response.text
def test_cases_cover_every_phase2_permission_code() -> None:
"""这组用例必须覆盖客服二期新增的三个权限码,缺一个就失去意义。"""
covered = {case[2] for case in CASES}
assert covered == {
"handover:read",
"memory:candidate:confirm",
"memory:candidate:review",
}
+109
View File
@@ -0,0 +1,109 @@
"""`docs/05-接口文档.md` §19 端点编号唯一性(只读,不连库)。
规则与动机见 `tools/check_docs_endpoint_ids.py` 的模块说明:`check_authoritative_docs.py`
只校验 `docs/` 的**文件名编号**,不校验 §19 的**端点编号** —— 2026-09-12 两条线各自
新增端点时都占了 `A034`/`A035`,合并后同时存在两个 `A034` 与两个 `A035` 却仍然通过守卫。
这里额外守住两件事:
1. **不能"空扫通过"**:真实文档必须真的扫到足量编号,否则脚本坏掉了也会显示"无重复";
2. **覆盖范围要可见**:扫描不得依赖前缀白名单 —— 同一件事上曾因正则写成 `[AMKCS]`
而漏掉 `O`/`R` 两段,把 62 个端点报成 55 个;漏掉的号段一旦被复用,脚本仍会报
"重复 0"。所以覆盖报告必须列出每个号段的数量。
"""
from __future__ import annotations
import importlib.util
from pathlib import Path
from types import ModuleType
import pytest
PROJECT_ROOT = Path(__file__).resolve().parents[3]
#: 真实 §19 的规模下限:远小于实际数量,只为拦住"扫到 0 条也算通过"。
MIN_REAL_ENDPOINTS = 40
def _load_tool_module() -> ModuleType:
path = PROJECT_ROOT / "tools" / "check_docs_endpoint_ids.py"
spec = importlib.util.spec_from_file_location("check_docs_endpoint_ids", path)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _write_doc(tmp_path: Path, body: str) -> Path:
doc = tmp_path / "05-接口文档.md"
doc.write_text(
"# 接口文档\n\n## 19. 接口总目录\n\n"
"| 编号 | 方法与路径 | 权限 | 幂等 | 成功状态 | 审计 |\n"
"|---|---|---|---|---|---|\n"
f"{body}\n\n## 20. 变更流程\n\n占位。\n",
encoding="utf-8",
)
return doc
def test_real_document_has_no_duplicate_endpoint_ids() -> None:
module = _load_tool_module()
assert module.collect_findings() == []
def test_real_document_actually_scans_enough_endpoints() -> None:
"""防止脚本坏掉后"什么都没扫到"却报通过。"""
module = _load_tool_module()
ids, _ = module.collect_rows(module.read_section())
assert len(ids) >= MIN_REAL_ENDPOINTS
assert len(set(ids)) == len(ids)
def test_duplicate_endpoint_id_is_detected(tmp_path: Path) -> None:
module = _load_tool_module()
doc = _write_doc(
tmp_path,
"| A034 | `POST /api/v1/auth/tokens` | 公开 | 否 | `200` | 否 |\n"
"| A034 | `GET /api/v1/admin/customer-profile-candidates` | `x:y` | 否 | `200` | 否 |\n",
)
findings = module.collect_findings(doc)
assert any("A034" in item and "重复" in item for item in findings)
def test_unrecognized_first_column_is_reported(tmp_path: Path) -> None:
module = _load_tool_module()
doc = _write_doc(tmp_path, "| A034(临时) | `GET /x` | `a:b` | 否 | `200` | 否 |\n")
findings = module.collect_findings(doc)
assert any("无法识别" in item for item in findings)
def test_coverage_report_lists_every_prefix_segment(tmp_path: Path) -> None:
"""覆盖报告必须暴露各号段数量 —— 漏扫一个前缀就会在这里显形。"""
module = _load_tool_module()
doc = _write_doc(
tmp_path,
"| A001 | `GET /a1` | `a:b` | 否 | `200` | 否 |\n"
"| A002 | `GET /a2` | `a:b` | 否 | `200` | 否 |\n"
"| M003 | `GET /m3` | `a:b` | 否 | `200` | 否 |\n"
"| O001 | `GET /o1` | `a:b` | 否 | `200` | 否 |\n"
"| R001 | `GET /r1` | `a:b` | 否 | `200` | 否 |\n",
)
report = module.describe_coverage(doc)
for expected in ("5 个端点编号", "4 个号段", "A×2", "M×1", "O×1", "R×1"):
assert expected in report, report
def test_missing_section_raises(tmp_path: Path) -> None:
module = _load_tool_module()
doc = tmp_path / "05-接口文档.md"
doc.write_text("# 接口文档\n\n## 1. 概述\n\n没有 §19。\n", encoding="utf-8")
with pytest.raises(module.DocumentStructureError):
module.collect_findings(doc)
def test_empty_checklist_prepares_no_false_success(tmp_path: Path) -> None:
"""§19 存在但一张表都没有时,必须报结构异常,而不是"无重复"。"""
module = _load_tool_module()
doc = _write_doc(tmp_path, "本节没有表格。")
findings = module.collect_findings(doc)
assert any("没有扫描到任何端点编号" in item for item in findings)
+132
View File
@@ -0,0 +1,132 @@
"""校验 `docs/05-接口文档.md` §19「接口总目录」的端点编号唯一性(只读,不连数据库)。
## 为什么需要它
`tools/check_authoritative_docs.py` 只校验 `docs/` 的**文件名编号**与权威性声明,
**不校验 §19 里每个端点的编号**。2026-09-12 出过一次真实事故:两条线各自新增端点时
都占用了 `A034`/`A035`,合并后 §19 同时存在两个 `A034` 与两个 `A035`,
而文档守卫照样通过 —— 编号复用比"编号不够"更麻烦,而且**静默遗留**。
## 口径(含覆盖自检)
- 扫描范围:§19 章节内的**所有**表格行首列;
- **不枚举前缀白名单**,前缀从数据里归纳后打印出来。同一件事上还踩过一次
"扫描正则写成 `[AMKCS]` 就漏掉 `O`/`R` 两段,把 62 个端点报成 55 个" ——
一旦被漏掉的号段将来被复用,脚本仍会报"重复 0"。所以这里把
**"扫到哪几个号段、各多少条、共计多少"** 一并输出,让覆盖范围本身可核对;
- 首列不是 `XXX###` 形状的行会被显式列出来(而不是静默跳过);
- 只读:不修改任何文件。
用法:
python tools/check_docs_endpoint_ids.py
退出码:0 = 无重复;1 = 有重复或文档结构异常。
"""
from __future__ import annotations
import re
from collections import Counter
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
INTERFACE_DOC = PROJECT_ROOT / "docs" / "05-接口文档.md"
#: §19 的起止标题(`: \\s*` 之间允许一个空格,避免依赖标题里的具体文字)。
SECTION_START = re.compile(r"^##\s*19\.")
SECTION_END = re.compile(r"^##\s*20\.")
#: 表格行:只取首列,容忍单元格两侧空白。
TABLE_ROW = re.compile(r"^\|(.+)\|\s*$")
#: Markdown 分隔行:| --- | :---: | 之类。
SEPARATOR = re.compile(r"^[\s\-:|]+$")
#: 端点编号形状:1~4 个大写字母 + 3~4 位数字(A034 / M003 / R001)。
ENDPOINT_ID = re.compile(r"^[A-Z]{1,4}\d{3,4}$")
class DocumentStructureError(RuntimeError):
"""`docs/05` 的 §19 结构不符合预期,无法安全校验。"""
def read_section(path: Path = INTERFACE_DOC) -> list[str]:
"""返回 §19 的正文行(不含起止标题)。"""
if not path.exists():
raise DocumentStructureError(f"缺少接口权威文档:{path}")
lines = path.read_text(encoding="utf-8").splitlines()
start = next((i for i, line in enumerate(lines) if SECTION_START.match(line)), None)
if start is None:
raise DocumentStructureError("接口文档里找不到 §19 章节标题")
end = next(
(i for i in range(start + 1, len(lines)) if SECTION_END.match(lines[i])),
len(lines),
)
return lines[start + 1 : end]
def collect_rows(lines: list[str]) -> tuple[list[str], list[str]]:
"""返回 `(首列编号列表, 无法识别的首列列表)`。
表头行、分隔行、非表格行都会被跳过;非空但形状不对的首列会被归类为
"无法识别",以便显式暴露而不是静默漏扫。
"""
ids: list[str] = []
unrecognized: list[str] = []
for line in lines:
match = TABLE_ROW.match(line.strip())
if match is None:
continue
first_cell = match.group(1).split("|", 1)[0].strip()
if not first_cell or SEPARATOR.match(first_cell):
continue
if first_cell == "编号": # 表头
continue
if ENDPOINT_ID.match(first_cell):
ids.append(first_cell)
else:
unrecognized.append(first_cell)
return ids, unrecognized
def collect_findings(path: Path = INTERFACE_DOC) -> list[str]:
"""返回问题列表;空列表表示一致。"""
lines = read_section(path)
ids, unrecognized = collect_rows(lines)
problems: list[str] = []
if not ids:
problems.append("§19 没有扫描到任何端点编号,文档结构可能已变化,请人工确认")
counter = Counter(ids)
for endpoint_id in sorted(code for code, count in counter.items() if count > 1):
problems.append(f"端点编号 {endpoint_id} 重复 {counter[endpoint_id]} 次")
for cell in sorted(set(unrecognized)):
problems.append(f"§19 首列无法识别为端点编号:{cell!r}")
return problems
def describe_coverage(path: Path = INTERFACE_DOC) -> str:
"""返回覆盖范围报告 —— 让"扫到了什么"可见,而不是只报"没问题"。"""
ids, _ = collect_rows(read_section(path))
prefixes = Counter(re.match(r"[A-Z]+", code).group() for code in ids)
detail = "、".join(f"{prefix}×{count}" for prefix, count in sorted(prefixes.items()))
return f"§19 覆盖:{len(ids)} 个端点编号 / {len(prefixes)} 个号段({detail})"
def main() -> int:
problems = collect_findings()
print(describe_coverage())
if problems:
for problem in problems:
print(f" ✗ {problem}")
print(f"docs/05 §19 端点编号校验失败:{len(problems)} 处问题")
return 1
print("docs/05 §19 端点编号无重复")
return 0
if __name__ == "__main__":
raise SystemExit(main())