96 lines
3.7 KiB
Python
96 lines
3.7 KiB
Python
"""B5 首批冒烟：APS 服务能起来、能按契约返回、能力清单与错误分类对得上。
|
||
|
||
只读能力全部基于 main 的 build_demo_world() 导出的快照（aps/data/demo_world.json），
|
||
所以断言里的数量是确定的：7 张销售订单、13 个可用工位。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
from aps.api.app import CAPABILITIES, app, manifest
|
||
from aps.identity import store as identity_store
|
||
from aps.world import load_world
|
||
|
||
# 能力调用要带登录态（A5.7）：测试先按真实流程换一个 token
|
||
client = TestClient(app)
|
||
_session = identity_store.login("E1001", "aps-demo-pass")
|
||
assert _session is not None, "演示账号应能登录"
|
||
client.headers.update({"authorization": f"Bearer {_session.token}"})
|
||
|
||
|
||
def test_health() -> None:
|
||
response = client.get("/api/health")
|
||
|
||
assert response.status_code == 200
|
||
assert response.json()["status"] == "ok"
|
||
|
||
|
||
def test_manifest_shape_matches_platform_contract() -> None:
|
||
body = manifest()
|
||
|
||
assert body["id"] == "aps"
|
||
assert body["platformVersion"] == 1, "平台只接受 platformVersion = 1"
|
||
assert body["requiresIdentity"] is True, "APS 有账号体系（ADR-0012）"
|
||
assert body["auth"]["kind"] in {"oauth", "password-exchange", "apikey"}
|
||
assert len(body["capabilities"]) >= 7, "只读能力至少 7 个"
|
||
|
||
|
||
def test_every_capability_is_described_for_the_model() -> None:
|
||
"""能力描述规范七要素：id 命名、用途 ≤40 字、写清副作用、声明参数 schema。"""
|
||
for spec in manifest()["capabilities"]:
|
||
assert spec["id"].startswith("aps."), spec["id"]
|
||
assert 0 < len(spec["description"]) <= 40, f"{spec['id']} 用途超 40 字"
|
||
assert spec["effect"] in {"read", "draft", "write"}
|
||
assert spec["mode"] in {"sync", "async"}
|
||
assert spec["idempotency"] in {"none", "key", "natural"}
|
||
assert spec["inputSchema"], f"{spec['id']} 缺 inputSchema"
|
||
assert spec["errors"], f"{spec['id']} 缺错误分类"
|
||
|
||
|
||
def test_read_capabilities_return_data() -> None:
|
||
orders = client.post("/api/capabilities/aps.orders.query", json={"params": {"limit": 5}}).json()
|
||
assert orders["status"] == "succeeded"
|
||
assert orders["result"]["total"] == 7
|
||
|
||
capacity = client.post("/api/capabilities/aps.capacity.query", json={"params": {}}).json()
|
||
assert capacity["status"] == "succeeded"
|
||
assert capacity["result"]["total"] == 13
|
||
|
||
|
||
def test_every_registered_capability_invokes() -> None:
|
||
for capability_id in CAPABILITIES:
|
||
body = client.post(f"/api/capabilities/{capability_id}", json={"params": {"limit": 2}}).json()
|
||
assert body["status"] == "succeeded", f"{capability_id} 调用失败：{body}"
|
||
assert body["executionId"].startswith("exec-aps-"), "回执要带 executionId"
|
||
|
||
|
||
def test_unknown_capability_is_invalid_argument() -> None:
|
||
body = client.post("/api/capabilities/aps.not-exist", json={"params": {}}).json()
|
||
|
||
assert body["status"] == "failed"
|
||
assert body["errorKind"] == "invalid_argument"
|
||
|
||
|
||
def test_bad_params_map_to_invalid_argument() -> None:
|
||
body = client.post("/api/capabilities/aps.orders.query", json={"params": {"limit": "不是数字"}}).json()
|
||
|
||
assert body["status"] == "failed"
|
||
assert body["errorKind"] == "invalid_argument"
|
||
|
||
|
||
def test_execution_id_is_echoed_back() -> None:
|
||
body = client.post(
|
||
"/api/capabilities/aps.orders.query",
|
||
json={"params": {}, "executionId": "exec-from-platform-1"},
|
||
).json()
|
||
|
||
assert body["executionId"] == "exec-from-platform-1", "平台给的 executionId 必须原样带回"
|
||
|
||
|
||
def test_demo_world_is_loaded_from_snapshot() -> None:
|
||
world = load_world()
|
||
|
||
assert len(world["salesOrders"]) == 7
|
||
assert len(world["workstations"]) == 13
|