Merge remote-tracking branch 'origin/qyqy_develop' into nl-merge-colleague
This commit is contained in:
@@ -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",
|
||||
}
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user