119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
"""openapi.json 导出测试：守住「一份文档覆盖三个服务」的合并结果
|
|
|
|
这份文档是外部（Apifox / Postman / 上游联调）唯一的接口来源，合并逻辑错了就会
|
|
把人引到不存在的接口、或漏标鉴权。下面的用例全部在内存里 build_spec() 检查合并
|
|
结果，不读仓库里的 openapi.json——那个文件是 gitignore 的生成物，CI 的全新 clone
|
|
里根本不存在，断言「文件内容 == 当前导出结果」在 CI 上必然失败。
|
|
|
|
因此「改了接口忘了重新导出」没有自动兜底，得手动跑：
|
|
.venv/Scripts/python.exe scripts/export_openapi.py --check
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
SCRIPT = ROOT / "scripts" / "export_openapi.py"
|
|
|
|
_METHODS = ("get", "put", "post", "delete", "options", "head", "patch", "trace")
|
|
|
|
|
|
def _load_exporter():
|
|
"""按路径加载 scripts/export_openapi.py（scripts 不是包）
|
|
|
|
必须先塞进 sys.modules 再 exec：dataclass 解析类型注解时会去
|
|
sys.modules[cls.__module__] 找命名空间，没登记就直接 AttributeError。
|
|
"""
|
|
spec = importlib.util.spec_from_file_location("export_openapi", SCRIPT)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def exporter():
|
|
return _load_exporter()
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def spec(exporter):
|
|
return exporter.build_spec()
|
|
|
|
|
|
def _operations(document: dict) -> list[tuple[str, str, dict]]:
|
|
return [
|
|
(path, method, operation)
|
|
for path, item in document["paths"].items()
|
|
for method, operation in item.items()
|
|
if method in _METHODS
|
|
]
|
|
|
|
|
|
def test_covers_all_three_services(spec):
|
|
paths = spec["paths"]
|
|
# 抓取
|
|
assert "/api/search" in paths
|
|
assert "/api/rakuma/search" in paths
|
|
# 交易（有状态端）
|
|
assert "/api/auth/login" in paths
|
|
assert "/api/cart/add" in paths
|
|
# 网关
|
|
assert "/api/orders" in paths
|
|
assert "/api/orders/lease" in paths
|
|
|
|
|
|
def test_each_operation_points_at_its_own_service(spec):
|
|
"""operation 级 servers：导入后不必手动切 base URL"""
|
|
by_path = {path: operation["servers"][0]["url"] for path, _, operation in _operations(spec)}
|
|
assert by_path["/api/search"].endswith(":31107")
|
|
assert by_path["/api/cart/add"].endswith(":31108")
|
|
assert by_path["/api/orders/lease"].endswith(":31109")
|
|
# /health 三个服务都有，合成一条并列出三个 server
|
|
health_servers = {s["url"] for s in spec["paths"]["/health"]["get"]["servers"]}
|
|
assert len(health_servers) == 3
|
|
|
|
|
|
def test_only_health_is_public(spec):
|
|
"""除 /health 外都必须标 Bearer 鉴权——漏标会让调用方以为能匿名调"""
|
|
unprotected = [
|
|
f"{method.upper()} {path}"
|
|
for path, method, operation in _operations(spec)
|
|
if "security" not in operation
|
|
]
|
|
assert unprotected == ["GET /health"]
|
|
assert "BearerAuth" in spec["components"]["securitySchemes"]
|
|
|
|
|
|
def test_operation_ids_are_unique(spec):
|
|
"""三个服务的 /health 原本都叫 health_health_get，撞了会让导入工具丢接口"""
|
|
ids = [operation["operationId"] for _, _, operation in _operations(spec)]
|
|
assert len(ids) == len(set(ids))
|
|
|
|
|
|
def test_all_refs_resolve(spec):
|
|
"""合并 schema 时如果重名处理错了，$ref 会指向不存在的定义"""
|
|
names = set(spec["components"]["schemas"])
|
|
refs = set(re.findall(r"#/components/schemas/([^\"]+)", json.dumps(spec)))
|
|
assert not refs - names
|
|
|
|
|
|
def test_submit_order_intent_schema_documents_multi_item_and_legacy_fields(spec):
|
|
"""intent 保持透传对象，同时在 OpenAPI 中明确展示新旧两种商品格式。"""
|
|
intent = spec["components"]["schemas"]["SubmitOrderRequest"]["properties"]["intent"]
|
|
properties = intent["properties"]
|
|
assert intent["additionalProperties"] is True
|
|
assert properties["items"]["type"] == "array"
|
|
assert properties["items"]["minItems"] == 1
|
|
item_object = properties["items"]["items"]["oneOf"][0]
|
|
assert item_object["required"] == ["item_url"]
|
|
assert "item_url" in properties
|
|
assert "quantity" in properties
|