zhifu-platform/tests/test_api_documentation.py

import json
from pathlib import Path

import httpx
import pytest
from jsonschema import Draft202012Validator, FormatChecker
from test_after_sales import scenario  # noqa: F401
from test_conversations import setup as conversation_setup
from test_fulfillment import connected  # noqa: F401
from test_knowledge_memory import cleanup_vectors, document

from zhifu.app import create_app
from zhifu.modules.knowledge.inference import app as inference_app
from zhifu.monitoring.api import app as operations_app
from zhifu.oms.api import create_app as create_oms


@pytest.mark.parametrize(
    "name,app",
    [
        ("openapi.json", create_app()),
        ("oms-openapi.json", create_oms()),
        ("inference-openapi.json", inference_app),
        ("operations-openapi.json", operations_app),
    ],
)
def test_published_definitions_match_routes_and_describe_responses(name, app):
    schema = app.openapi()
    assert json.loads(Path("docs/api", name).read_text(encoding="utf-8")) == schema
    identifiers = []
    for path, methods in schema["paths"].items():
        for method, operation in methods.items():
            assert operation["summary"] and operation.get("description"), (name, path, method)
            identifiers.append(operation["operationId"])
            public = path in {
                "/openapi/v1/auth/sessions",
                "/openapi/v1/platform/auth/sessions",
                "/openapi/v1/stores/{store_id}/catalog",
                "/openapi/v1/stores/{store_id}/products/{product_id}",
                "/health",
                "/health/ready",
                "/embeddings",
                "/metrics",
            }
            assert bool(operation.get("security")) != public, (name, path, method)
            for status, response in operation["responses"].items():
                if status == "204":
                    assert "content" not in response
                elif status.startswith("2"):
                    assert all(value.get("schema") for value in response["content"].values())
            if name == "openapi.json" and "422" in operation["responses"]:
                assert operation["responses"]["422"]["content"]["application/json"]["schema"] == {
                    "$ref": "#/components/schemas/ApiError"
                }
    assert len(identifiers) == len(set(identifiers))
    for model in schema.get("components", {}).get("schemas", {}).values():
        if model.get("title") in {"HTTPValidationError", "ValidationError"}:
            continue
        for field, value in model.get("properties", {}).items():
            assert value.get("description"), (name, model["title"], field)


def validate_response(schema, path, method, response, expected_status):
    assert response.status_code == expected_status, response.text
    body_schema = schema["paths"][path][method]["responses"][str(expected_status)]["content"][
        "application/json"
    ]["schema"]
    Draft202012Validator(
        {"components": schema["components"], **body_schema}, format_checker=FormatChecker()
    ).validate(response.json())


@pytest.mark.integration
async def test_real_platform_responses_match_documented_schema(scenario):  # noqa: F811
    schema = scenario["app"].openapi()
    call = scenario["call"]
    path = "/openapi/v1/auth/me"
    validate_response(schema, path, "get", await call("GET", path), 200)
    path = "/openapi/v1/orders/{order_id}"
    validate_response(
        schema, path, "get", await call("GET", path.format(order_id=scenario["order_id"])), 200
    )
    path = "/openapi/v1/after-sales/proposals"
    validate_response(schema, path, "post", await call("POST", path, payload={}), 422)
    response = await call("POST", path, payload=scenario["proposal"])
    validate_response(schema, path, "post", response, 201)
    path = "/openapi/v1/tickets/{ticket_id}"
    validate_response(
        schema,
        path,
        "get",
        await call("GET", path.format(ticket_id=response.json()["ticket_id"])),
        200,
    )
    path = "/openapi/v1/agent/runs"
    validate_response(
        schema,
        path,
        "post",
        await call(
            "POST",
            path,
            payload={"store_id": str(scenario["store"]), "message": "查询订单进度"},
        ),
        202,
    )
    conversation = await conversation_setup(scenario)
    validate_response(
        schema,
        "/openapi/v1/conversations/{conversation_id}",
        "get",
        await call("GET", conversation),
        200,
    )
    response = await call(
        "POST",
        conversation + "/messages",
        payload={"expected_control_version": 1, "text": "请查询订单"},
    )
    validate_response(
        schema, "/openapi/v1/conversations/{conversation_id}/messages", "post", response, 201
    )
    path = "/openapi/v1/me/memory"
    validate_response(schema, path, "get", await call("GET", path), 200)
    await call("PUT", path + "/consent", payload={"expected_revision": 0, "enabled": True})
    response = await call(
        "PUT",
        path,
        payload={"expected_revision": 1, "communication": {"response_style": "concise"}},
    )
    validate_response(schema, path, "put", response, 200)
    knowledge = await document(scenario)
    try:
        response = await call("POST", knowledge + "/publish", "agent", {"expected_version": 1})
        validate_response(
            schema,
            "/admin/v1/stores/{store_id}/knowledge/{document_id}/{action}",
            "post",
            response,
            200,
        )
        path = "/openapi/v1/stores/{store_id}/knowledge/search"
        response = await call(
            "POST", path.format(store_id=scenario["store"]), payload={"query": "键盘清洁"}
        )
        validate_response(schema, path, "post", response, 200)
        assert response.json()["items"]
    finally:
        await cleanup_vectors(scenario["tenant"])


@pytest.mark.integration
async def test_real_oms_and_inference_response_formats(connected):  # noqa: F811
    app = create_oms()
    async with app.router.lifespan_context(app):
        async with httpx.AsyncClient(
            transport=httpx.ASGITransport(app=app), base_url="http://oms"
        ) as client:
            path = "/orders/{order_id}"
            url = path.format(order_id=connected["order_id"])
            response = await client.get(url, headers=connected["warehouse"].headers)
            validate_response(app.openapi(), path, "get", response, 200)
            response = await client.get(url)
            validate_response(app.openapi(), path, "get", response, 401)
    async with httpx.AsyncClient(base_url="http://127.0.0.1:8200", timeout=30) as client:
        response = await client.post("/embeddings", json={"input": ["售后服务说明"]})
        validate_response(inference_app.openapi(), "/embeddings", "post", response, 200)
        assert len(response.json()["vectors"][0]) == 1024
        for payload in ({"input": [""]}, {"input": []}):
            response = await client.post("/embeddings", json=payload)
            validate_response(inference_app.openapi(), "/embeddings", "post", response, 422)