aps-agent/aps/tests/test_api.py

"""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