mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-13 07:22:24 -06:00
Patrick Buckley
7a06f5e8bc
* refactor(session): make ModelLane the provider boundary (#979) ## Summary This closes the model-lane ownership gap left by #832: `ChatSession` no longer stores raw provider/client handles. `ResolvedModelBinding` now carries the provider, client, model, capabilities, registry generation, and backend-auth configuration as one coherent snapshot. - Atomically rebind existing sessions after model-registry changes while pinning each in-flight send, fallback, judge, output guard, task agent, title, compaction, perception, and voice operation to its initiating principal and binding. - Fence UI publication, canonical trajectory folds, durable writes, streams, retries, child scopes, and judge work by generation. Stop can hand off to a successor without accepting late state; cancelled tools retain typed effect receipts, and concurrent approval batches resolve by exact cycle or call. - Make create, fork, open, close, and delete race-safe with hidden `creating` reservations, incarnation-aware state tails, and an ACL-rechecked transaction that clones checkpoint-bounded history, configuration, project/persona state, and attachment references. - Extend REST/OpenAPI and Python/TypeScript SDK contracts for create/fork inputs, routed-create metadata, live-workstream probes, targeted approvals, and structured cancellation results. - Update architecture, storage, authentication, judge, channel, console, API, and SDK documentation, including regenerated architecture diagrams and OpenAPI artifacts. ## Validation - SQLite suite: 11,188 passed, 9 skipped, 10 deselected - PostgreSQL suite: 11,195 passed, 2 skipped, 10 deselected - Live backend: 3 passed - SSE recovery: 6 passed; browser recovery harness passed all scenarios - Ruff: clean; 595 files correctly formatted - mypy: 243 source files clean - TypeScript: typecheck/build and 35 tests passed - OpenAPI artifacts fresh; all 14 changed diagrams reproduce byte-for-byte - `git diff --check` and Git LFS integrity clean Closes #979. * fix(deps): update nanoid for GHSA-2v37-7h3g-55p8 Refresh the transitive lock entry admitted by PostCSS so the TypeScript security gate no longer resolves the vulnerable custom-generator implementation. Validation: - npm ci - npm audit --audit-level=moderate: 0 vulnerabilities - TypeScript typecheck and build - TypeScript tests: 35 passed * fix(test): assert canonical model registry URLs Replace prefix checks with exact canonical base URL assertions so the tests do not model incomplete URL validation. Validation: tests/test_model_registry.py (185 passed); Ruff check/format; mypy.
1134 lines
42 KiB
Python
1134 lines
42 KiB
Python
"""Tests for console routing proxy endpoints (route_create, route_proxy, route_lookup)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
import httpx
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
|
|
from turnstone.console.collector import ClusterCollector
|
|
from turnstone.console.router import ConsoleRouter, NodeRef
|
|
from turnstone.core.rendezvous import NoAvailableNodeError
|
|
|
|
# Shared test auth — JWT-based
|
|
_TEST_JWT_SECRET = "test-jwt-secret-minimum-32-chars!"
|
|
|
|
|
|
def _test_jwt() -> str:
|
|
from turnstone.core.auth import JWT_AUD_CONSOLE, create_jwt
|
|
|
|
return create_jwt(
|
|
user_id="test-routing",
|
|
scopes=frozenset({"read", "write", "approve", "service"}),
|
|
source="test",
|
|
secret=_TEST_JWT_SECRET,
|
|
audience=JWT_AUD_CONSOLE,
|
|
)
|
|
|
|
|
|
_TEST_AUTH_HEADERS: dict[str, str] = {"Authorization": f"Bearer {_test_jwt()}"}
|
|
_DEST_WS_ID = "a" * 32
|
|
_FORK_DEST_WS_ID = "b" * 32
|
|
_RETRY_DEST_WS_ID = "c" * 32
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_mock_collector() -> MagicMock:
|
|
collector = MagicMock(spec=ClusterCollector)
|
|
collector.get_overview.return_value = {
|
|
"nodes": 1,
|
|
"workstreams": 0,
|
|
"states": {"running": 0, "thinking": 0, "attention": 0, "idle": 0, "error": 0},
|
|
"aggregate": {"total_tokens": 0, "total_tool_calls": 0},
|
|
}
|
|
return collector
|
|
|
|
|
|
def _make_mock_router(ready: bool = True) -> MagicMock:
|
|
router = MagicMock(spec=ConsoleRouter)
|
|
router.is_ready.return_value = ready
|
|
router.route.return_value = NodeRef("node-a", "http://a:8080")
|
|
router.generate_ws_id_for_node.return_value = "00ff" + "0" * 28
|
|
return router
|
|
|
|
|
|
def _make_app(
|
|
collector: Any = None,
|
|
router: Any = None,
|
|
auth_storage: Any = None,
|
|
) -> Any:
|
|
from turnstone.console.server import _load_static, create_app
|
|
|
|
_load_static()
|
|
return create_app(
|
|
collector=collector or _make_mock_collector(),
|
|
jwt_secret=_TEST_JWT_SECRET,
|
|
router=router,
|
|
auth_storage=auth_storage,
|
|
)
|
|
|
|
|
|
def _make_proxy_post(
|
|
status_code: int = 200,
|
|
json_data: dict[str, Any] | None = None,
|
|
) -> MagicMock:
|
|
"""Create a mock for httpx.AsyncClient.post that returns a fixed response."""
|
|
data = json_data or {"ws_id": "abc123", "name": "test"}
|
|
|
|
async def _mock_post(*args: Any, **kwargs: Any) -> httpx.Response:
|
|
return httpx.Response(
|
|
status_code,
|
|
json=data,
|
|
request=httpx.Request("POST", args[0] if args else "http://test"),
|
|
)
|
|
|
|
mock_post = MagicMock(side_effect=_mock_post)
|
|
return mock_post
|
|
|
|
|
|
def _wire_proxy(app: Any, mock_post: MagicMock | None = None) -> None:
|
|
"""Attach a mock proxy_client to the app (lifespan doesn't run in TestClient)."""
|
|
if mock_post is None:
|
|
mock_post = _make_proxy_post()
|
|
mock_proxy = MagicMock(spec=httpx.AsyncClient)
|
|
mock_proxy.post = mock_post
|
|
|
|
# route_proxy uses ``client.request(method, url, ...)`` for path-keyed
|
|
# routes (so DELETE on /send proxies through correctly). Wire a
|
|
# request-shim that drops the leading method positional and forwards
|
|
# to the same mock_post for compatibility.
|
|
async def _request_shim(method: str, *args: Any, **kwargs: Any) -> httpx.Response:
|
|
return await mock_post(*args, **kwargs)
|
|
|
|
mock_proxy.request = MagicMock(side_effect=_request_shim)
|
|
app.state.proxy_client = mock_proxy
|
|
|
|
|
|
def _wire_proxy_get(
|
|
app: Any,
|
|
*,
|
|
status_code: int = 200,
|
|
json_data: dict[str, Any] | None = None,
|
|
raw_content: bytes | None = None,
|
|
) -> MagicMock:
|
|
"""Attach a proxy client whose GET returns one deterministic response."""
|
|
|
|
async def _mock_get(*args: Any, **kwargs: Any) -> httpx.Response:
|
|
request = httpx.Request("GET", args[0] if args else "http://test")
|
|
if raw_content is not None:
|
|
return httpx.Response(status_code, content=raw_content, request=request)
|
|
return httpx.Response(status_code, json=json_data or {}, request=request)
|
|
|
|
mock_get = MagicMock(side_effect=_mock_get)
|
|
mock_proxy = MagicMock(spec=httpx.AsyncClient)
|
|
mock_proxy.get = mock_get
|
|
app.state.proxy_client = mock_proxy
|
|
return mock_get
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests — route_create
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRouteCreate:
|
|
"""POST /v1/api/route/workstreams/new — create via rendezvous routing."""
|
|
|
|
@pytest.fixture()
|
|
def client(self):
|
|
router = _make_mock_router()
|
|
app = _make_app(router=router)
|
|
_wire_proxy(app, _make_proxy_post(json_data={"ws_id": _DEST_WS_ID, "name": "test"}))
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
yield client
|
|
client.close()
|
|
|
|
def test_route_create_proxies_to_node(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"name": "test-ws"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["ws_id"] == _DEST_WS_ID
|
|
|
|
def test_route_create_injects_node_url(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"name": "test-ws"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["node_url"] == "http://a:8080"
|
|
assert data["node_id"] == "node-a"
|
|
|
|
def test_route_create_resume_ws(self):
|
|
"""resume_ws should route to the node that owns the old workstream."""
|
|
router = _make_mock_router()
|
|
router.route.return_value = NodeRef("node-b", "http://b:8080")
|
|
storage = MagicMock()
|
|
storage.resolve_workstream.return_value = "d" * 32
|
|
app = _make_app(router=router, auth_storage=storage)
|
|
_wire_proxy(
|
|
app,
|
|
_make_proxy_post(json_data={"ws_id": _FORK_DEST_WS_ID, "name": "resumed"}),
|
|
)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"resume_ws": "saved-alias"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["node_url"] == "http://b:8080"
|
|
assert data["node_id"] == "node-b"
|
|
storage.resolve_workstream.assert_called_once_with("saved-alias")
|
|
router.route.assert_called_with("d" * 32)
|
|
router.remember_override.assert_called_once_with(
|
|
_FORK_DEST_WS_ID,
|
|
NodeRef("node-b", "http://b:8080"),
|
|
)
|
|
assert app.state.proxy_client.post.call_args.kwargs["json"]["resume_ws"] == "d" * 32
|
|
client.close()
|
|
|
|
def test_route_create_target_node(self):
|
|
"""target_node should generate a ws_id that hashes to that node."""
|
|
router = _make_mock_router()
|
|
router.generate_ws_id_for_node.return_value = "00ff" + "0" * 28
|
|
router.route.return_value = NodeRef("node-c", "http://c:8080")
|
|
app = _make_app(router=router)
|
|
_wire_proxy(
|
|
app,
|
|
_make_proxy_post(json_data={"ws_id": "00ff" + "0" * 28, "name": "pinned"}),
|
|
)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"target_node": "node-c"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["node_id"] == "node-c"
|
|
router.generate_ws_id_for_node.assert_called_with("node-c")
|
|
client.close()
|
|
|
|
def test_route_create_routing_strategy_rendezvous(self, client):
|
|
"""Default fan-out (no resume_ws / no target_node) reports
|
|
routing_strategy='rendezvous' so the coordinator's spawn tool
|
|
can explain why the node was chosen."""
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"name": "test-ws"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["routing_strategy"] == "rendezvous"
|
|
|
|
def test_route_create_routing_strategy_target_node(self):
|
|
router = _make_mock_router()
|
|
router.generate_ws_id_for_node.return_value = "00ff" + "0" * 28
|
|
router.route.return_value = NodeRef("node-c", "http://c:8080")
|
|
app = _make_app(router=router)
|
|
_wire_proxy(app, _make_proxy_post(json_data={"ws_id": "00ff" + "0" * 28, "name": "pinned"}))
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"target_node": "node-c"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["routing_strategy"] == "target_node"
|
|
client.close()
|
|
|
|
def test_route_create_routing_strategy_resume(self):
|
|
router = _make_mock_router()
|
|
router.route.return_value = NodeRef("node-b", "http://b:8080")
|
|
storage = MagicMock()
|
|
storage.resolve_workstream.return_value = "d" * 32
|
|
app = _make_app(router=router, auth_storage=storage)
|
|
_wire_proxy(
|
|
app,
|
|
_make_proxy_post(json_data={"ws_id": _FORK_DEST_WS_ID, "name": "resumed"}),
|
|
)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"resume_ws": "source-alias"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["routing_strategy"] == "resume"
|
|
client.close()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_python_sdk_decodes_live_route_create_response(self):
|
|
"""Exercise the SDK against the actual ASGI route, not a mock transport."""
|
|
from turnstone.sdk.console import AsyncTurnstoneConsole
|
|
|
|
router = _make_mock_router()
|
|
app = _make_app(router=router)
|
|
_wire_proxy(app, _make_proxy_post(json_data={"ws_id": _DEST_WS_ID, "name": "sdk"}))
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="http://test",
|
|
headers=_TEST_AUTH_HEADERS,
|
|
) as http_client:
|
|
sdk = AsyncTurnstoneConsole(httpx_client=http_client)
|
|
result = await sdk.route_create_workstream(name="sdk")
|
|
|
|
assert result.ws_id == _DEST_WS_ID
|
|
assert result.node_id == "node-a"
|
|
assert result.node_url == "http://a:8080"
|
|
assert result.routing_strategy == "rendezvous"
|
|
|
|
@pytest.mark.parametrize("payload", [[], "text", 7])
|
|
def test_route_create_rejects_non_object_json(self, client, payload):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json=payload,
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 400
|
|
assert resp.json() == {"error": "Request body must be a JSON object"}
|
|
|
|
def test_route_create_rejects_json_null(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
content=b"null",
|
|
headers={**_TEST_AUTH_HEADERS, "Content-Type": "application/json"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert resp.json() == {"error": "Request body must be a JSON object"}
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value", "error"),
|
|
[
|
|
("resume_ws", 3, "resume_ws must be a string"),
|
|
("resume_ws", "x" * 257, "resume_ws must be at most 256 characters"),
|
|
("target_node", ["node-a"], "target_node must be a string"),
|
|
("target_node", "bad/node", "invalid target_node format"),
|
|
("ws_id", None, "ws_id must be a string"),
|
|
("ws_id", "abc", "invalid ws_id format"),
|
|
],
|
|
)
|
|
def test_route_create_validates_placement_field_shapes(self, client, field, value, error):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={field: value},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 400
|
|
assert resp.json() == {"error": error}
|
|
|
|
def test_explicit_json_ws_id_is_preserved_and_routed_by_rendezvous(self):
|
|
router = _make_mock_router()
|
|
app = _make_app(router=router)
|
|
post = _make_proxy_post(json_data={"ws_id": _DEST_WS_ID, "name": "fixed"})
|
|
_wire_proxy(app, post)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"ws_id": _DEST_WS_ID, "target_node": "node-a"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["routing_strategy"] == "rendezvous"
|
|
router.route.assert_called_once_with(_DEST_WS_ID)
|
|
router.generate_ws_id_for_node.assert_not_called()
|
|
assert post.call_args.kwargs["json"]["ws_id"] == _DEST_WS_ID
|
|
|
|
def test_resume_alias_missing_and_storage_uncertainty_are_bounded(self):
|
|
router = _make_mock_router()
|
|
storage = MagicMock()
|
|
storage.resolve_workstream.return_value = None
|
|
app = _make_app(router=router, auth_storage=storage)
|
|
_wire_proxy(app)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
missing = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"resume_ws": "missing-alias"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert missing.status_code == 404
|
|
assert missing.json() == {"error": "Workstream not found"}
|
|
|
|
storage.resolve_workstream.side_effect = RuntimeError("database details")
|
|
unavailable = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"resume_ws": "source-alias"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
assert unavailable.status_code == 503
|
|
assert unavailable.json() == {"error": "Storage not available"}
|
|
|
|
def test_full_resume_id_wins_over_alias_shadow(self):
|
|
source_id = "a" * 32
|
|
shadow_id = "b" * 32
|
|
router = _make_mock_router()
|
|
storage = MagicMock()
|
|
storage.get_workstream.return_value = {"ws_id": source_id, "state": "idle"}
|
|
storage.resolve_workstream.return_value = shadow_id
|
|
app = _make_app(router=router, auth_storage=storage)
|
|
post = _make_proxy_post(json_data={"ws_id": _FORK_DEST_WS_ID, "name": "fork"})
|
|
_wire_proxy(app, post)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"resume_ws": source_id},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
storage.get_workstream.assert_any_call(source_id)
|
|
storage.resolve_workstream.assert_not_called()
|
|
router.route.assert_called_once_with(source_id)
|
|
assert post.call_args.kwargs["json"]["resume_ws"] == source_id
|
|
|
|
@pytest.mark.parametrize(
|
|
"upstream",
|
|
[
|
|
httpx.Response(200, content=b"not json"),
|
|
httpx.Response(200, json=[]),
|
|
httpx.Response(200, json={"name": "missing id"}),
|
|
httpx.Response(200, json={"ws_id": "not-a-workstream-id"}),
|
|
httpx.Response(200, json={"ws_id": _DEST_WS_ID}),
|
|
httpx.Response(200, json={"ws_id": _DEST_WS_ID, "name": 42}),
|
|
],
|
|
)
|
|
def test_malformed_upstream_success_returns_bounded_502(self, upstream):
|
|
router = _make_mock_router()
|
|
app = _make_app(router=router)
|
|
|
|
async def _post(*args: Any, **kwargs: Any) -> httpx.Response:
|
|
upstream.request = httpx.Request("POST", args[0])
|
|
return upstream
|
|
|
|
mock_proxy = MagicMock(spec=httpx.AsyncClient)
|
|
mock_proxy.post = MagicMock(side_effect=_post)
|
|
app.state.proxy_client = mock_proxy
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 502
|
|
assert resp.json() == {"error": "Dispatch to node node-a failed"}
|
|
|
|
def test_returned_destination_is_binding_and_audit_authority(self, monkeypatch):
|
|
router = _make_mock_router()
|
|
storage = MagicMock()
|
|
storage.resolve_workstream.return_value = "d" * 32
|
|
storage.get_workstream.return_value = {"node_id": "stored-node"}
|
|
app = _make_app(router=router, auth_storage=storage)
|
|
_wire_proxy(
|
|
app,
|
|
_make_proxy_post(json_data={"ws_id": _FORK_DEST_WS_ID, "name": "fork"}),
|
|
)
|
|
audit = MagicMock()
|
|
monkeypatch.setattr("turnstone.console.server._emit_route_audit", audit)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"resume_ws": "source-alias"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["node_id"] == "stored-node"
|
|
storage.get_workstream.assert_called_once_with(_FORK_DEST_WS_ID)
|
|
audit.assert_called_once()
|
|
assert audit.call_args.args[2] == _FORK_DEST_WS_ID
|
|
|
|
def test_multipart_preallocated_id_reports_rendezvous(self):
|
|
router = _make_mock_router()
|
|
app = _make_app(router=router)
|
|
_wire_proxy(app, _make_proxy_post(json_data={"ws_id": _DEST_WS_ID, "name": "upload"}))
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
resp = client.post(
|
|
f"/v1/api/route/workstreams/new?ws_id={_DEST_WS_ID}",
|
|
data={"meta": json.dumps({"ws_id": _DEST_WS_ID})},
|
|
files={"file": ("a.txt", b"hello", "text/plain")},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["routing_strategy"] == "rendezvous"
|
|
router.route.assert_called_once_with(_DEST_WS_ID)
|
|
|
|
|
|
class TestRouteCreate503Retry:
|
|
"""503 retry logic in route_create."""
|
|
|
|
def test_route_create_503_retries_on_different_node(self):
|
|
"""If the first node returns 503, retry with a new ws_id targeting a different node."""
|
|
router = _make_mock_router()
|
|
call_count = 0
|
|
|
|
def side_effect_route(ws_id: str) -> NodeRef:
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count <= 1:
|
|
# First call returns node-a
|
|
return NodeRef("node-a", "http://a:8080")
|
|
# Subsequent calls return node-b (different node for retry)
|
|
return NodeRef("node-b", "http://b:8080")
|
|
|
|
router.route.side_effect = side_effect_route
|
|
app = _make_app(router=router)
|
|
|
|
post_count = 0
|
|
|
|
async def _mock_post(*args: Any, **kwargs: Any) -> httpx.Response:
|
|
nonlocal post_count
|
|
post_count += 1
|
|
if post_count == 1:
|
|
return httpx.Response(
|
|
503,
|
|
json={"error": "overloaded"},
|
|
request=httpx.Request("POST", args[0] if args else "http://test"),
|
|
)
|
|
return httpx.Response(
|
|
200,
|
|
json={"ws_id": _RETRY_DEST_WS_ID, "name": "retry"},
|
|
request=httpx.Request("POST", args[0] if args else "http://test"),
|
|
)
|
|
|
|
mock_proxy = MagicMock(spec=httpx.AsyncClient)
|
|
mock_proxy.post = MagicMock(side_effect=_mock_post)
|
|
app.state.proxy_client = mock_proxy
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"name": "test-ws"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["ws_id"] == _RETRY_DEST_WS_ID
|
|
assert data["node_id"] == "node-b"
|
|
assert post_count == 2
|
|
client.close()
|
|
|
|
def test_explicit_ws_id_is_not_replaced_or_retried_on_503(self):
|
|
router = _make_mock_router()
|
|
app = _make_app(router=router)
|
|
post = _make_proxy_post(status_code=503, json_data={"error": "overloaded"})
|
|
_wire_proxy(app, post)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"ws_id": _DEST_WS_ID},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 503
|
|
assert post.call_count == 1
|
|
assert post.call_args.kwargs["json"]["ws_id"] == _DEST_WS_ID
|
|
router.route.assert_called_once_with(_DEST_WS_ID)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests — cluster create (capacity-routed proxy)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestClusterCreate:
|
|
"""POST /v1/api/cluster/workstreams/new — the launcher's create proxy.
|
|
|
|
Create-with-attachments rides multipart (a ``meta`` JSON field + ``file``
|
|
parts) and must forward to the node AS multipart — not collapse to JSON,
|
|
which would silently drop the files (the pre-fix behaviour, gated in the UI
|
|
as "Attachments aren't supported for interactive sessions yet")."""
|
|
|
|
def _app_with_node(self, mock_post: MagicMock) -> Any:
|
|
collector = _make_mock_collector()
|
|
collector.get_node_detail.return_value = {"server_url": "http://a:8080"}
|
|
app = _make_app(collector=collector)
|
|
_wire_proxy(app, mock_post)
|
|
return app
|
|
|
|
def test_cluster_create_json_forwards_json(self):
|
|
mock_post = _make_proxy_post(json_data={"ws_id": "abc123"})
|
|
client = TestClient(self._app_with_node(mock_post), raise_server_exceptions=False)
|
|
resp = client.post(
|
|
"/v1/api/cluster/workstreams/new",
|
|
json={"node_id": "node-a", "name": "j", "initial_message": "hi"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["correlation_id"] == "abc123"
|
|
kwargs = mock_post.call_args.kwargs
|
|
assert "json" in kwargs and "files" not in kwargs, "no-file create must stay JSON"
|
|
assert kwargs["json"]["initial_message"] == "hi"
|
|
client.close()
|
|
|
|
def test_cluster_create_multipart_forwards_files(self):
|
|
mock_post = _make_proxy_post(json_data={"ws_id": "withfile"})
|
|
client = TestClient(self._app_with_node(mock_post), raise_server_exceptions=False)
|
|
meta = {"node_id": "node-a", "name": "i", "initial_message": "describe"}
|
|
resp = client.post(
|
|
"/v1/api/cluster/workstreams/new",
|
|
data={"meta": json.dumps(meta)},
|
|
files={"file": ("a.txt", b"hello world", "text/plain")},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["correlation_id"] == "withfile"
|
|
kwargs = mock_post.call_args.kwargs
|
|
# Forwarded as multipart: a `meta` JSON field + `file` parts, never json=.
|
|
assert "json" not in kwargs, "a multipart create must not collapse to JSON"
|
|
assert kwargs.get("files"), "the blob must be forwarded to the node"
|
|
forwarded_meta = json.loads(kwargs["data"]["meta"])
|
|
assert forwarded_meta["initial_message"] == "describe"
|
|
assert "user_id" in forwarded_meta, "the proxy must inject the owner uid"
|
|
# The file part carries our blob unchanged: ("file", (name, bytes, ctype)).
|
|
name, payload = kwargs["files"][0]
|
|
assert name == "file"
|
|
assert payload[0] == "a.txt" and payload[1] == b"hello world"
|
|
client.close()
|
|
|
|
def test_cluster_create_forwards_project_id(self) -> None:
|
|
# Phase 6: the launcher's project picker sends project_id; the proxy
|
|
# selectively REBUILDS the forwarded body (it doesn't pass it through),
|
|
# so project_id must be explicitly carried or the node never scopes the
|
|
# session to its project.
|
|
mock_post = _make_proxy_post(json_data={"ws_id": "p1ws"})
|
|
client = TestClient(self._app_with_node(mock_post), raise_server_exceptions=False)
|
|
resp = client.post(
|
|
"/v1/api/cluster/workstreams/new",
|
|
json={"node_id": "node-a", "name": "j", "project_id": "proj-42"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert mock_post.call_args.kwargs["json"]["project_id"] == "proj-42"
|
|
client.close()
|
|
|
|
def test_cluster_create_forwards_persona(self) -> None:
|
|
# The launcher's persona picker sends persona; the proxy selectively
|
|
# REBUILDS the forwarded body (it doesn't pass it through), so persona
|
|
# must be explicitly carried or the receiving node stamps its kind
|
|
# default instead of the operator's choice.
|
|
mock_post = _make_proxy_post(json_data={"ws_id": "p1ws"})
|
|
client = TestClient(self._app_with_node(mock_post), raise_server_exceptions=False)
|
|
resp = client.post(
|
|
"/v1/api/cluster/workstreams/new",
|
|
json={"node_id": "node-a", "name": "j", "persona": "scribe"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert mock_post.call_args.kwargs["json"]["persona"] == "scribe"
|
|
client.close()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_python_sdk_forwards_schema_contract_fields(self) -> None:
|
|
"""Run the SDK through the live handler and inspect its upstream request."""
|
|
from turnstone.sdk.console import AsyncTurnstoneConsole
|
|
|
|
mock_post = _make_proxy_post(json_data={"ws_id": "contract-ws"})
|
|
app = self._app_with_node(mock_post)
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="http://test",
|
|
headers=_TEST_AUTH_HEADERS,
|
|
) as http_client:
|
|
sdk = AsyncTurnstoneConsole(httpx_client=http_client)
|
|
result = await sdk.create_workstream(
|
|
node_id="node-a",
|
|
name="contract",
|
|
project_id="project-42",
|
|
judge_model="judge-fast",
|
|
)
|
|
|
|
assert result.correlation_id == "contract-ws"
|
|
forwarded = mock_post.call_args.kwargs["json"]
|
|
assert forwarded["project_id"] == "project-42"
|
|
assert forwarded["judge_model"] == "judge-fast"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests — route_proxy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRouteProxy:
|
|
"""POST /v1/api/route/workstreams/{ws_id}/<verb> (and the surviving
|
|
body-keyed plan/command routes)."""
|
|
|
|
@pytest.fixture()
|
|
def client(self):
|
|
router = _make_mock_router()
|
|
app = _make_app(router=router)
|
|
_wire_proxy(app, _make_proxy_post(json_data={"status": "ok"}))
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
yield client
|
|
client.close()
|
|
|
|
def test_route_proxy_send(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/send",
|
|
json={"message": "hello"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
# Verify upstream URL was /v1/api/workstreams/abc123/send
|
|
# (not /v1/api/route/workstreams/abc123/send).
|
|
mock_request = client.app.state.proxy_client.request
|
|
call_args = mock_request.call_args
|
|
# request is called as ``request(method, url, ...)`` — url is the
|
|
# second positional arg.
|
|
upstream_url = call_args[0][1]
|
|
assert "/v1/api/workstreams/abc123/send" in upstream_url
|
|
assert "/route/" not in upstream_url
|
|
|
|
def test_route_proxy_approve(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/approve",
|
|
json={"approved": True},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
def test_route_proxy_cancel(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/cancel",
|
|
json={},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
@pytest.mark.parametrize(
|
|
("content", "content_type"),
|
|
[
|
|
(b"", None),
|
|
(b"{", "application/json"),
|
|
(b"[]", "application/json"),
|
|
],
|
|
ids=["empty", "malformed", "non-object"],
|
|
)
|
|
def test_route_proxy_cancel_normalizes_unusable_body(self, client, content, content_type):
|
|
headers = dict(_TEST_AUTH_HEADERS)
|
|
if content_type is not None:
|
|
headers["Content-Type"] = content_type
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/cancel",
|
|
content=content,
|
|
headers=headers,
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert client.app.state.proxy_client.request.call_args.kwargs["json"] == {}
|
|
|
|
def test_route_proxy_non_cancel_rejects_non_object_json(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/send",
|
|
json=[],
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
|
|
assert resp.status_code == 400
|
|
assert resp.json() == {"error": "Request body must be a JSON object"}
|
|
|
|
def test_route_proxy_command(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/command",
|
|
json={"ws_id": "abc123", "command": "status"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
def test_route_proxy_close(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/close",
|
|
json={},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
|
|
class TestRouteProxyPermissionGates:
|
|
"""``route_proxy`` was pre-existing infra that forwarded blindly —
|
|
any authenticated caller could send/approve/cancel/close. PR
|
|
adding 057_role_permission_overrides added verb-scoped gates on
|
|
approve + close (the verbs that had vestigial perms in
|
|
``_VALID_PERMISSIONS`` with no enforcement site). These tests
|
|
pin the new shape and the OR fallback to ``admin.coordinator``."""
|
|
|
|
@pytest.fixture()
|
|
def client(self):
|
|
router = _make_mock_router()
|
|
app = _make_app(router=router)
|
|
_wire_proxy(app, _make_proxy_post(json_data={"status": "ok"}))
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
yield client
|
|
client.close()
|
|
|
|
@staticmethod
|
|
def _hdr(*, perms: frozenset[str] = frozenset()) -> dict[str, str]:
|
|
# Plain user — no service scope, so the bypass doesn't kick in;
|
|
# just the perms passed by the test.
|
|
from turnstone.core.auth import JWT_AUD_CONSOLE, create_jwt
|
|
|
|
return {
|
|
"Authorization": (
|
|
"Bearer "
|
|
+ create_jwt(
|
|
user_id="test-user",
|
|
scopes=frozenset({"read", "write", "approve"}),
|
|
source="test",
|
|
secret=_TEST_JWT_SECRET,
|
|
audience=JWT_AUD_CONSOLE,
|
|
permissions=perms,
|
|
)
|
|
)
|
|
}
|
|
|
|
def test_approve_without_perm_returns_403(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/approve",
|
|
json={"approved": True},
|
|
headers=self._hdr(),
|
|
)
|
|
assert resp.status_code == 403
|
|
assert "tools.approve" in resp.json()["error"]
|
|
|
|
def test_close_without_perm_returns_403(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/close",
|
|
json={},
|
|
headers=self._hdr(),
|
|
)
|
|
assert resp.status_code == 403
|
|
assert "workstreams.close" in resp.json()["error"]
|
|
|
|
def test_approve_with_tools_approve_passes(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/approve",
|
|
json={"approved": True},
|
|
headers=self._hdr(perms=frozenset({"tools.approve"})),
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
def test_close_with_admin_coordinator_passes(self, client):
|
|
# The OR fallback: coord sessions can drive close on
|
|
# interactive children without holding workstreams.close.
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/close",
|
|
json={},
|
|
headers=self._hdr(perms=frozenset({"admin.coordinator"})),
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
def test_send_remains_authenticated_only(self, client):
|
|
# send/cancel/dequeue/command/plan are unchanged — no new gate.
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc123/send",
|
|
json={"message": "hi"},
|
|
headers=self._hdr(),
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests — route_lookup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRouteLookup:
|
|
"""GET /v1/api/route — look up which node owns a workstream."""
|
|
|
|
@pytest.fixture()
|
|
def client(self):
|
|
router = _make_mock_router()
|
|
app = _make_app(router=router)
|
|
app.state.proxy_client = MagicMock(spec=httpx.AsyncClient)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
yield client
|
|
client.close()
|
|
|
|
def test_route_lookup(self, client):
|
|
resp = client.get("/v1/api/route?ws_id=abc123", headers=_TEST_AUTH_HEADERS)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["node_url"] == "http://a:8080"
|
|
assert data["node_id"] == "node-a"
|
|
|
|
def test_route_lookup_missing_ws_id(self, client):
|
|
resp = client.get("/v1/api/route", headers=_TEST_AUTH_HEADERS)
|
|
assert resp.status_code == 400
|
|
assert "ws_id" in resp.json()["error"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests — route_workstream_live
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRouteWorkstreamLive:
|
|
"""GET routed live probe reads the owner node's active manager list."""
|
|
|
|
def test_reports_exact_visible_active_row(self):
|
|
router = _make_mock_router()
|
|
app = _make_app(router=router)
|
|
mock_get = _wire_proxy_get(
|
|
app,
|
|
json_data={
|
|
"workstreams": [
|
|
{"ws_id": "other", "state": "idle"},
|
|
{"ws_id": "ws-live", "state": "running"},
|
|
]
|
|
},
|
|
)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
resp = client.get(
|
|
"/v1/api/route/workstreams/ws-live/live",
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"ws_id": "ws-live", "live": True}
|
|
router.route.assert_called_once_with("ws-live")
|
|
assert mock_get.call_args.args[0] == "http://a:8080/v1/api/workstreams"
|
|
|
|
def test_false_miss_refreshes_stale_override_and_reprobes_new_owner(self):
|
|
router = _make_mock_router()
|
|
stale_ref = NodeRef("node-b", "http://b:8080")
|
|
owner_ref = NodeRef("node-a", "http://a:8080")
|
|
router.route.side_effect = [stale_ref, owner_ref]
|
|
app = _make_app(router=router)
|
|
|
|
async def _get(url: str, **_kwargs: Any) -> httpx.Response:
|
|
payload = (
|
|
{"workstreams": []}
|
|
if url.startswith(stale_ref.url)
|
|
else {"workstreams": [{"ws_id": "ws-live", "state": "idle"}]}
|
|
)
|
|
return httpx.Response(200, json=payload, request=httpx.Request("GET", url))
|
|
|
|
proxy = MagicMock(spec=httpx.AsyncClient)
|
|
proxy.get = MagicMock(side_effect=_get)
|
|
app.state.proxy_client = proxy
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
resp = client.get(
|
|
"/v1/api/route/workstreams/ws-live/live",
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"ws_id": "ws-live", "live": True}
|
|
router.force_refresh.assert_called_once_with()
|
|
assert [call.args[0] for call in proxy.get.call_args_list] == [
|
|
"http://b:8080/v1/api/workstreams",
|
|
"http://a:8080/v1/api/workstreams",
|
|
]
|
|
|
|
@pytest.mark.parametrize(
|
|
"rows",
|
|
[
|
|
[],
|
|
[{"ws_id": "other", "state": "idle"}],
|
|
[{"ws_id": "ws-live", "state": "creating"}],
|
|
],
|
|
ids=["missing-or-private", "different-row", "creating"],
|
|
)
|
|
def test_reports_false_without_exposing_non_live_rows(self, rows):
|
|
app = _make_app(router=_make_mock_router())
|
|
_wire_proxy_get(app, json_data={"workstreams": rows})
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
resp = client.get(
|
|
"/v1/api/route/workstreams/ws-live/live",
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"ws_id": "ws-live", "live": False}
|
|
|
|
def test_propagates_upstream_acl_failure(self):
|
|
app = _make_app(router=_make_mock_router())
|
|
_wire_proxy_get(
|
|
app,
|
|
status_code=403,
|
|
json_data={"error": "Forbidden"},
|
|
)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
resp = client.get(
|
|
"/v1/api/route/workstreams/ws-live/live",
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 403
|
|
assert resp.json() == {"error": "Forbidden"}
|
|
|
|
def test_malformed_active_list_fails_closed(self):
|
|
app = _make_app(router=_make_mock_router())
|
|
_wire_proxy_get(app, json_data={"unexpected": []})
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
resp = client.get(
|
|
"/v1/api/route/workstreams/ws-live/live",
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
client.close()
|
|
|
|
assert resp.status_code == 502
|
|
assert "invalid active list" in resp.json()["error"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests — not ready / no router -> 503
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRouteNotReady:
|
|
"""When router is None or empty cache, all routing endpoints return 503."""
|
|
|
|
@pytest.fixture()
|
|
def client_no_router(self):
|
|
app = _make_app(router=None)
|
|
app.state.proxy_client = MagicMock(spec=httpx.AsyncClient)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
yield client
|
|
client.close()
|
|
|
|
@pytest.fixture()
|
|
def client_empty_cache(self):
|
|
router = _make_mock_router(ready=False)
|
|
app = _make_app(router=router)
|
|
app.state.proxy_client = MagicMock(spec=httpx.AsyncClient)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
yield client
|
|
client.close()
|
|
|
|
def test_route_create_no_router_503(self, client_no_router):
|
|
resp = client_no_router.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"name": "test"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 503
|
|
|
|
def test_route_create_empty_cache_503(self, client_empty_cache):
|
|
resp = client_empty_cache.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"name": "test"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 503
|
|
|
|
def test_route_proxy_no_router_503(self, client_no_router):
|
|
resp = client_no_router.post(
|
|
"/v1/api/route/workstreams/abc/send",
|
|
json={"message": "hello"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 503
|
|
|
|
def test_route_lookup_no_router_503(self, client_no_router):
|
|
resp = client_no_router.get("/v1/api/route?ws_id=abc", headers=_TEST_AUTH_HEADERS)
|
|
assert resp.status_code == 503
|
|
|
|
def test_route_live_no_router_503(self, client_no_router):
|
|
resp = client_no_router.get(
|
|
"/v1/api/route/workstreams/abc/live",
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 503
|
|
|
|
def test_route_proxy_empty_cache_503(self, client_empty_cache):
|
|
resp = client_empty_cache.post(
|
|
"/v1/api/route/workstreams/abc/send",
|
|
json={"message": "hello"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 503
|
|
|
|
def test_route_lookup_empty_cache_503(self, client_empty_cache):
|
|
resp = client_empty_cache.get("/v1/api/route?ws_id=abc", headers=_TEST_AUTH_HEADERS)
|
|
assert resp.status_code == 503
|
|
|
|
def test_route_live_empty_cache_503(self, client_empty_cache):
|
|
resp = client_empty_cache.get(
|
|
"/v1/api/route/workstreams/abc/live",
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 503
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests — NoAvailableNodeError handling
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRouteNoNode:
|
|
"""When router.route() raises NoAvailableNodeError, endpoints return 503."""
|
|
|
|
@pytest.fixture()
|
|
def client(self):
|
|
router = _make_mock_router()
|
|
router.route.side_effect = NoAvailableNodeError("bucket 0 not assigned")
|
|
app = _make_app(router=router)
|
|
app.state.proxy_client = MagicMock(spec=httpx.AsyncClient)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
yield client
|
|
client.close()
|
|
|
|
def test_route_create_no_node_503(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/new",
|
|
json={"name": "test"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 503
|
|
assert "No available node" in resp.json()["error"]
|
|
|
|
def test_route_proxy_no_node_503(self, client):
|
|
resp = client.post(
|
|
"/v1/api/route/workstreams/abc/send",
|
|
json={"message": "hello"},
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 503
|
|
|
|
def test_route_lookup_no_node_503(self, client):
|
|
resp = client.get("/v1/api/route?ws_id=abc", headers=_TEST_AUTH_HEADERS)
|
|
assert resp.status_code == 503
|
|
|
|
def test_route_live_no_node_503(self, client):
|
|
resp = client.get(
|
|
"/v1/api/route/workstreams/abc/live",
|
|
headers=_TEST_AUTH_HEADERS,
|
|
)
|
|
assert resp.status_code == 503
|