107 lines
4.9 KiB
Python
107 lines
4.9 KiB
Python
"""Contract tests for the sub-mode runtime API surfaces."""
|
|
|
|
from pathlib import Path
|
|
|
|
from api.status_projection import RuntimeDisplayStatus, project_deployment_status, project_runtime_run_status
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _read(relative_path: str) -> str:
|
|
"""Load a repository file as text."""
|
|
return (ROOT / relative_path).read_text(encoding="utf-8")
|
|
|
|
|
|
def test_primary_and_compat_runtime_routes_are_registered():
|
|
"""The new agent surface and both compatibility surfaces must coexist."""
|
|
app_source = _read("app.py")
|
|
agent_router_source = _read("api/agent/router.py")
|
|
callbacks_source = _read("api/agnet/callbacks.py")
|
|
swarm_router_source = _read("api/swarm/router.py")
|
|
|
|
assert "from api.agent.router import router as agent_router" in app_source
|
|
assert 'prefix="/api/agent"' in agent_router_source
|
|
assert 'prefix="/sub-agile"' in agent_router_source
|
|
assert '@router.post("/runtime-events")' in callbacks_source
|
|
assert '@compat_router.post("/swarm-events")' in callbacks_source
|
|
assert '@router.post("/deployments/{deployment_id}/artifact-edits"' in _read("api/agnet/deployments.py")
|
|
assert 'prefix="/api/swarms"' in swarm_router_source
|
|
assert '@swarms_router.post("/{swarm_id}/approvals/{approval_id}")' in swarm_router_source
|
|
assert '@swarms_router.get("/{swarm_id}/artifacts/{artifact_id}/manifest")' in swarm_router_source
|
|
assert '@swarms_router.get("/{swarm_id}/artifacts/{artifact_id}/archive.zip")' in swarm_router_source
|
|
assert '@swarms_router.get("/{swarm_id}/artifacts/{artifact_id}/files/{file_path:path}")' in swarm_router_source
|
|
|
|
|
|
def test_projected_statuses_cover_manager_facing_contract():
|
|
"""Projected deployment statuses should match the Manager contract."""
|
|
assert project_deployment_status("pending") == RuntimeDisplayStatus.ACCEPTED.value
|
|
assert (
|
|
project_deployment_status(
|
|
"running",
|
|
events=[{"event_type": "approval.requested", "payload": {"approval_id": "appr_1"}}],
|
|
)
|
|
== RuntimeDisplayStatus.WAITING_APPROVAL.value
|
|
)
|
|
assert (
|
|
project_deployment_status(
|
|
"running",
|
|
phase="done",
|
|
events=[{"event_type": "task.completed", "payload": {"status": "completed"}}],
|
|
)
|
|
== RuntimeDisplayStatus.COMPLETED.value
|
|
)
|
|
assert (
|
|
project_deployment_status(
|
|
"running",
|
|
events=[{"event_type": "task.failed", "payload": {"status": "failed"}}],
|
|
)
|
|
== RuntimeDisplayStatus.FAILED.value
|
|
)
|
|
assert project_deployment_status("stopped") == RuntimeDisplayStatus.STOPPED.value
|
|
|
|
|
|
def test_runtime_run_projection_remains_sub_mode_compatible():
|
|
"""Legacy /api/swarms runs should expose the shared projected status set."""
|
|
assert project_runtime_run_status("initializing") == RuntimeDisplayStatus.ACCEPTED.value
|
|
assert project_runtime_run_status("running") == RuntimeDisplayStatus.RUNNING.value
|
|
assert project_runtime_run_status("completed") == RuntimeDisplayStatus.COMPLETED.value
|
|
assert project_runtime_run_status("failed") == RuntimeDisplayStatus.FAILED.value
|
|
assert project_runtime_run_status("stopped") == RuntimeDisplayStatus.STOPPED.value
|
|
|
|
|
|
def test_project_folder_contract_is_documented_and_flagged():
|
|
"""Structural code artifacts should be documented and tagged distinctly."""
|
|
doc_source = _read("docs/HEICODE_SUB_MODE_RUNTIME_INTEGRATION.md")
|
|
orchestrator_source = _read("api/swarm/orchestrator.py")
|
|
swarm_router_source = _read("api/swarm/router.py")
|
|
|
|
assert "project_folder artifact（结构性代码强制要求）" in doc_source
|
|
assert "manifest_uri" in doc_source
|
|
assert "archive_uri" in doc_source
|
|
assert "files/{path}" in doc_source
|
|
assert '"artifact_layout": "project_folder"' in orchestrator_source
|
|
assert '"primary_read_path": "manifest"' in orchestrator_source
|
|
assert '"summary_only": True' in swarm_router_source
|
|
|
|
|
|
def test_runtime_rich_workflow_contract_is_present():
|
|
"""Workflow/work outputs should expose richer runtime fields."""
|
|
swarm_models_source = _read("api/swarm/models.py")
|
|
swarm_router_source = _read("api/swarm/router.py")
|
|
orchestrator_source = _read("api/swarm/orchestrator.py")
|
|
|
|
assert "tokens: int = 0" in swarm_models_source
|
|
assert "tools: int = 0" in swarm_models_source
|
|
assert "elapsed_seconds: int = 0" in swarm_models_source
|
|
assert "artifact_ids: List[str] = Field(default_factory=list)" in swarm_models_source
|
|
assert "class SwarmPhaseInfo" in swarm_models_source
|
|
assert "phases: List[SwarmPhaseInfo]" in swarm_models_source
|
|
assert "source_agent_role" in orchestrator_source
|
|
assert "project_revision" in orchestrator_source
|
|
assert "delivery_ref" in orchestrator_source
|
|
assert "@swarms_router.post(\"/{swarm_id}/artifact-edits\"" in swarm_router_source
|
|
callbacks_source = _read("api/agnet/callbacks.py")
|
|
assert "artifact.local_edit_applied" in callbacks_source
|
|
assert '"project_folder"' in callbacks_source
|