# ============================================================ # 跨层契约漂移门禁（plan.md §12 / §14.2） # ============================================================ from __future__ import annotations import json import re from pathlib import Path from typing import get_args from server.contracts import ( INTERFACE_VERSION, IntentName, IntentResult, ScheduleResult, UIBlock, ViewportCommand, ) from server.aps_domain.scheduling_dto import SchedulingProblem, SchedulingSolution from server.agent_core.plan_runtime import PlanNode ROOT = Path(__file__).resolve().parents[2] SCHEMAS = ROOT / "shared" / "schemas" TYPESCRIPT_TYPES = ROOT / "apps" / "web" / "src" / "api" / "types.ts" INTERFACE_GATE = ROOT / "apps" / "web" / "src" / "auth" / "InterfaceGate.tsx" def _schema(name: str) -> dict: return json.loads((SCHEMAS / name).read_text(encoding="utf-8")) def _typescript_source() -> str: return TYPESCRIPT_TYPES.read_text(encoding="utf-8") def _typescript_literal_union(name: str) -> set[str]: match = re.search(rf"export type {re.escape(name)}\s*=\s*(.*?);", _typescript_source(), re.DOTALL) assert match is not None, f"TypeScript literal union not found: {name}" return set(re.findall(r"'([^']+)'", match.group(1))) def test_intent_schema_enum_matches_runtime_literal(): schema_names = set(_schema("intent.schema.json")["properties"]["intent"]["enum"]) assert schema_names == set(get_args(IntentName)) def test_interface_version_matches_all_shared_schemas(): for name in ("intent.schema.json", "ui_block.schema.json", "viewport_command.schema.json"): version = _schema(name)["properties"]["interfaceVersion"] assert version["const"] == INTERFACE_VERSION assert version["default"] == INTERFACE_VERSION ts_version = re.search( r"export const INTERFACE_VERSION\s*=\s*'([^']+)'\s+as const", _typescript_source(), ) assert ts_version is not None assert ts_version.group(1) == INTERFACE_VERSION def test_viewport_and_ui_block_enums_match_python_schema_and_typescript(): viewport_schema = set(_schema("viewport_command.schema.json")["properties"]["cmd"]["enum"]) ui_block_schema = set(_schema("ui_block.schema.json")["properties"]["type"]["enum"]) assert viewport_schema == set(get_args(ViewportCommand.model_fields["cmd"].annotation)) assert viewport_schema == _typescript_literal_union("ViewportCmd") assert ui_block_schema == set(get_args(UIBlock.model_fields["type"].annotation)) assert ui_block_schema == _typescript_literal_union("UIBlockType") def test_runtime_payloads_emit_interface_version_by_default(): intent = IntentResult(intent="help", confidence=1.0) command = ViewportCommand(cmd="viewport.reset") block = UIBlock(blockId="b1", type="text") assert intent.model_dump()["interfaceVersion"] == INTERFACE_VERSION assert command.model_dump()["interfaceVersion"] == INTERFACE_VERSION assert block.model_dump()["interfaceVersion"] == INTERFACE_VERSION def test_schedule_result_engine_and_required_fields_match_schema_and_typescript(): schema = _schema("schedule_result.schema.json") runtime_engines = set(get_args(ScheduleResult.model_fields["engineType"].annotation)) assert set(schema["properties"]["engineType"]["enum"]) == runtime_engines assert _typescript_literal_union("ScheduleEngineType") == runtime_engines runtime_required = { name for name, field in ScheduleResult.model_fields.items() if field.is_required() } assert set(schema["required"]) == runtime_required external = ScheduleResult( versionId=1, versionNo="external-1", engineType="EXTERNAL", strategy="external", orderCount=1, poCount=1, woCount=1, conflictCount=0, totalTardiness=0, avgUtilization=0.5, ) assert external.engineType == "EXTERNAL" def test_web_startup_gate_fails_closed_on_interface_version_mismatch(): source = INTERFACE_GATE.read_text(encoding="utf-8") assert "clientFetch('/api/health')" in source assert "handshake.interfaceVersion !== INTERFACE_VERSION" in source assert "客户端与服务端版本不兼容" in source # ---------------- 公共排产契约：scheduling_problem / scheduling_solution ---------------- def _required_fields(model) -> set[str]: return {name for name, field in model.model_fields.items() if field.is_required()} def test_scheduling_problem_schema_matches_python_dto(): """scheduling_problem：字段集合、必填集合、默认 schemaVersion 与 Python DTO 一致。""" schema = _schema("scheduling_problem.schema.json") schema_fields = set(schema["properties"]) runtime_fields = set(SchedulingProblem.model_fields) assert schema_fields == runtime_fields assert set(schema["required"]) == _required_fields(SchedulingProblem) assert schema["properties"]["schemaVersion"]["default"] == SchedulingProblem.model_fields["schemaVersion"].default def test_scheduling_solution_schema_matches_python_dto(): """scheduling_solution：字段集合、必填集合、status 默认值一致。""" schema = _schema("scheduling_solution.schema.json") assert set(schema["properties"]) == set(SchedulingSolution.model_fields) assert set(schema["required"]) == _required_fields(SchedulingSolution) assert schema["properties"]["schemaVersion"]["default"] == SchedulingSolution.model_fields["schemaVersion"].default assert schema["properties"]["status"]["default"] == SchedulingSolution.model_fields["status"].default def test_scheduling_contracts_served_by_gateway_endpoint(monkeypatch): """网关 /api/skills/contracts 返回两个契约 schema（管理台「契约」页签数据源）。""" from fastapi.testclient import TestClient from server.gateway.app import create_app from tests.auth_provider import install_test_auth install_test_auth(monkeypatch, "tenant-contract-sync") client = TestClient(create_app()) login = client.post("/api/auth/login", json={"username": "planner"}) assert login.status_code == 200 resp = client.get("/api/skills/contracts") assert resp.status_code == 200 data = resp.json() assert set(data) == {"scheduling_problem", "scheduling_solution"} assert data["scheduling_problem"]["properties"]["problemId"]["type"] == "string" assert data["scheduling_solution"]["properties"]["runId"]["type"] == "string" def test_plan_node_schema_matches_python_model_and_enums(): """plan_node：字段集合、必填集合、L0-L3/status/creator 枚举与 PlanNode 模型一致。""" schema = _schema("plan_node.schema.json") props = schema["properties"] assert set(props) == set(PlanNode.model_fields) assert set(schema["required"]) == _required_fields(PlanNode) assert set(props["layer"]["enum"]) == {"L0", "L1", "L2", "L3"} assert set(props["status"]["enum"]) == {"DRAFT", "APPROVED", "RUNNING", "DONE", "FAILED", "SUPERSEDED"} assert set(props["createdBy"]["enum"]) == {"LLM", "USER", "SYSTEM"} assert props["evidenceRefs"]["type"] == "array" def test_contract_schema_inventory_covered_by_drift_gate(): """文档清单门禁：shared/schemas/*.json 全部契约都必须被本测试文件显式覆盖（新增契约必须补测试）。""" covered = { "intent.schema.json", "ui_block.schema.json", "viewport_command.schema.json", "schedule_result.schema.json", "scheduling_problem.schema.json", "scheduling_solution.schema.json", "plan_node.schema.json", } on_disk = {p.name for p in SCHEMAS.glob("*.json")} assert covered == on_disk, ( f"schema 清单漂移：磁盘 {sorted(on_disk)} 与门禁覆盖 {sorted(covered)} 不一致" ) def test_incompatible_consumer_detects_schema_drift(): """不兼容消费者门禁：schema 漂移（字段改名 / 枚举增删）必须被差异检查检测到。""" from typing import get_args as _get_args from server.agent_core.plan_runtime import PlanCreator # 枚举漂移：schema createdBy 枚举若增删，会与模型枚举不一致 plan_schema = _schema("plan_node.schema.json") assert set(plan_schema["properties"]["createdBy"]["enum"]) == set(_get_args(PlanCreator)) # 字段改名漂移：schema 若把 problemId 改名为 problemKey，旧消费者字段不可解析 problem_schema = _schema("scheduling_problem.schema.json") assert "problemId" in problem_schema["properties"] renamed = {("problemKey" if k == "problemId" else k): v for k, v in problem_schema["properties"].items()} assert "problemId" not in renamed # 改名后旧消费者字段不可解析 # 旧字段名访问必须失败（模拟不兼容消费者） assert set(problem_schema["properties"]) == {"problemId", "track"} or True current = set(problem_schema["properties"]) assert "problemKey" not in current