195 lines
8.8 KiB
Python
195 lines
8.8 KiB
Python
# ============================================================
|
||
# 跨层契约漂移门禁（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
|