tegwick
f6cfc28c33
Map every State Hub/core-hub caller to a post-retirement owner. Keep the 15-minute sweep schedule here and point the engine at repo-manager (State Hub dual-run by default, REPO_MANAGER_URL when present). Publish GET /execution/semantics and 410 workplan launch routes so State Hub /execution/* is not re-homed as a task database. T03 still waits on HUB-WP-0004.
39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from activity_core.execution_api import router
|
|
|
|
|
|
def _client() -> TestClient:
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
return TestClient(app)
|
|
|
|
|
|
def test_execution_semantics_names_boundary() -> None:
|
|
client = _client()
|
|
response = client.get("/execution/semantics")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["port"] == "port.schedule"
|
|
assert any("ops_run" in item for item in body["activity_core_owns"])
|
|
assert any("workplan" in item.lower() for item in body["activity_core_does_not_own"])
|
|
assert body["replacements"]["GET /execution/launch-requests"] == "GET /ops-runs"
|
|
|
|
|
|
def test_execution_workplan_routes_are_gone() -> None:
|
|
client = _client()
|
|
gone = [
|
|
client.get("/execution/launch-requests"),
|
|
client.post("/execution/launch-requests", json={"workplan_id": "x"}),
|
|
client.get("/execution/workplan-stack"),
|
|
client.patch("/execution/workplans/abc/intent", json={}),
|
|
client.patch("/execution/workstreams/abc/intent", json={}),
|
|
]
|
|
for response in gone:
|
|
assert response.status_code == 410
|
|
payload = response.json()
|
|
assert "replacement_ref" in payload
|
|
assert "execution-queue-boundary" in payload["document"]
|