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)