q792602257
fd6365cf6e
- test_cancellation_full_path_queued_immediate_and_dispatched_collaborative exercises the full public-API cancellation path: immediate cancel of a queued task, collaborative cancel of a dispatched task surfaced through lease renewal and a cancelled terminal report, and visibility of the cancelled status via both the get and list endpoints. - Full backend suite (869 passed, 50 skipped) and cloud-console frontend suite (27 passed) + typecheck show no regressions; the only failures are 4 pre-existing live-LLM integration tests unrelated to this change.
737 lines
24 KiB
Python
737 lines
24 KiB
Python
"""Unit tests for cloud.sdk.api (task 7.8)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from cloud.config import CloudConfig
|
|
from cloud.auth import BearerCredential, ConfiguredBearerAuthProvider
|
|
from cloud.plugins import PluginRegistry
|
|
from cloud.pool import DevicePool
|
|
from cloud.sdk.api import (
|
|
AuthProvider,
|
|
Principal,
|
|
create_cloud_router,
|
|
)
|
|
from cloud.scheduler import TaskScheduler
|
|
from cloud.store import CloudStore
|
|
from core.models import Device
|
|
|
|
|
|
pytest.importorskip("fastapi")
|
|
from fastapi import FastAPI # noqa: E402
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
|
|
def _config() -> CloudConfig:
|
|
return CloudConfig(
|
|
sync_interval_seconds=30,
|
|
stale_after_seconds=60,
|
|
max_queue_depth=100,
|
|
default_assignment_strategy="fifo_match",
|
|
api_version_prefix="/v1",
|
|
db_path="cloud/cloud.sqlite3",
|
|
)
|
|
|
|
|
|
def _build_app(
|
|
tmp_path,
|
|
*,
|
|
auth_provider: AuthProvider | None = None,
|
|
pool: DevicePool | None = None,
|
|
scheduler: TaskScheduler | None = None,
|
|
plugin_registry: PluginRegistry | None = None,
|
|
):
|
|
store = CloudStore(tmp_path / "cloud.sqlite3")
|
|
pool = pool or DevicePool(store, _config())
|
|
scheduler = scheduler or TaskScheduler(pool, store, _config())
|
|
plugin_registry = plugin_registry or PluginRegistry(store)
|
|
app = FastAPI()
|
|
app.include_router(
|
|
create_cloud_router(
|
|
pool=pool,
|
|
scheduler=scheduler,
|
|
plugin_registry=plugin_registry,
|
|
auth_provider=auth_provider,
|
|
)
|
|
)
|
|
return app, pool, scheduler, plugin_registry
|
|
|
|
|
|
def _client_for(app) -> TestClient:
|
|
return TestClient(app)
|
|
|
|
|
|
def test_null_auth_provider_allows_submit_and_status_round_trip(tmp_path) -> None:
|
|
app, pool, scheduler, _ = _build_app(tmp_path)
|
|
# Plant a device so the listing route has something to show.
|
|
pool.sync_host_devices(
|
|
"host-local",
|
|
[Device(id="dev-1", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
address="10.0.0.1:8000",
|
|
)
|
|
|
|
client = _client_for(app)
|
|
|
|
submission = client.post("/v1/tasks", json={"goal": "open settings"})
|
|
assert submission.status_code == 201, submission.text
|
|
task_id = submission.json()["task_id"]
|
|
|
|
status = client.get(f"/v1/tasks/{task_id}")
|
|
assert status.status_code == 200, status.text
|
|
body = status.json()
|
|
assert body["id"] == task_id
|
|
assert body["status"] == "queued"
|
|
assert body["goal"] == "open settings"
|
|
|
|
|
|
def test_unknown_task_id_returns_404(tmp_path) -> None:
|
|
app, _, _, _ = _build_app(tmp_path)
|
|
client = _client_for(app)
|
|
resp = client.get("/v1/tasks/does-not-exist")
|
|
assert resp.status_code == 404, resp.text
|
|
|
|
|
|
def test_task_status_exposes_distributed_metadata_without_lease_secret(
|
|
tmp_path,
|
|
) -> None:
|
|
app, pool, scheduler, _ = _build_app(tmp_path)
|
|
pool.sync_host_devices(
|
|
"host-a",
|
|
[Device(id="device-a", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
task_id = scheduler.submit(goal="remote task")
|
|
scheduler.assign()
|
|
|
|
active = _client_for(app).get(f"/v1/tasks/{task_id}").json()
|
|
|
|
assert active["status"] == "assigned"
|
|
assert active["assigned_host_id"] == "host-a"
|
|
assert active["assigned_device_id"] == "device-a"
|
|
assert active["attempt_count"] == 1
|
|
assert active["lease_expires_at"] is not None
|
|
assert active["failure_reason"] is None
|
|
assert "lease_id" not in active
|
|
|
|
task = scheduler.store.get_task(task_id)
|
|
scheduler.store.record_task_result(
|
|
task_id=task_id,
|
|
attempt=task.attempt_count,
|
|
lease_id=task.lease_id or "",
|
|
host_id=task.assigned_host_id or "",
|
|
status="failed",
|
|
failure_reason="planner unavailable",
|
|
terminal_result={"runtime_status": "failed"},
|
|
completed_at=datetime.now(UTC),
|
|
)
|
|
|
|
failed = _client_for(app).get(f"/v1/tasks/{task_id}").json()
|
|
|
|
assert failed["status"] == "failed"
|
|
assert failed["attempt_count"] == 1
|
|
assert failed["failure_reason"] == "planner unavailable"
|
|
assert "lease_id" not in failed
|
|
|
|
|
|
def test_device_and_host_listing_reflect_pool_state(tmp_path) -> None:
|
|
app, pool, _, _ = _build_app(tmp_path)
|
|
pool.sync_host_devices(
|
|
"host-a",
|
|
[
|
|
Device(id="a-dev-1", driver_type="wda", status="idle"), # type: ignore[arg-type]
|
|
Device(id="a-dev-2", driver_type="wda", status="busy"), # type: ignore[arg-type]
|
|
],
|
|
address="a:8000",
|
|
)
|
|
pool.sync_host_devices(
|
|
"host-b",
|
|
[Device(id="b-dev-1", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
address="b:8000",
|
|
)
|
|
|
|
client = _client_for(app)
|
|
devices = client.get("/v1/devices").json()
|
|
assert {d["device_id"] for d in devices} == {"a-dev-1", "a-dev-2", "b-dev-1"}
|
|
by_host = {d["device_id"]: d["host_id"] for d in devices}
|
|
assert by_host == {
|
|
"a-dev-1": "host-a",
|
|
"a-dev-2": "host-a",
|
|
"b-dev-1": "host-b",
|
|
}
|
|
|
|
hosts = client.get("/v1/hosts").json()
|
|
assert {h["host_id"] for h in hosts} == {"host-a", "host-b"}
|
|
assert all("last_seen_at" in h for h in hosts)
|
|
|
|
|
|
def test_plugin_listing_and_registration_round_trip(tmp_path) -> None:
|
|
app, _, _, _ = _build_app(tmp_path)
|
|
client = _client_for(app)
|
|
|
|
assert client.get("/v1/plugins").json() == []
|
|
|
|
payload = {
|
|
"name": "demo-tool",
|
|
"version": "1.0.0",
|
|
"entry_point_kind": "tool",
|
|
"target": "cloud.store:CloudStore",
|
|
}
|
|
resp = client.post("/v1/plugins", json=payload)
|
|
assert resp.status_code == 201, resp.text
|
|
body = resp.json()
|
|
assert body["name"] == "demo-tool"
|
|
assert body["entry_point_kind"] == "tool"
|
|
assert body["wired"] is False
|
|
|
|
listed = client.get("/v1/plugins").json()
|
|
assert len(listed) == 1
|
|
assert listed[0]["name"] == "demo-tool"
|
|
|
|
|
|
def test_duplicate_plugin_returns_conflict(tmp_path) -> None:
|
|
app, _, _, _ = _build_app(tmp_path)
|
|
client = _client_for(app)
|
|
|
|
payload = {
|
|
"name": "dup",
|
|
"version": "1.0.0",
|
|
"entry_point_kind": "tool",
|
|
"target": "cloud.store:CloudStore",
|
|
}
|
|
first = client.post("/v1/plugins", json=payload)
|
|
assert first.status_code == 201
|
|
second = client.post("/v1/plugins", json=payload)
|
|
assert second.status_code == 409
|
|
|
|
|
|
class _RejectingAuthProvider:
|
|
def authenticate(self, request: object) -> Principal | None:
|
|
return None
|
|
|
|
|
|
def test_rejecting_auth_provider_blocks_every_route(tmp_path) -> None:
|
|
app, _, _, _ = _build_app(tmp_path, auth_provider=_RejectingAuthProvider())
|
|
client = _client_for(app)
|
|
|
|
assert client.post("/v1/tasks", json={"goal": "x"}).status_code == 401
|
|
assert client.get("/v1/tasks/whatever").status_code == 401
|
|
assert client.get("/v1/devices").status_code == 401
|
|
assert client.get("/v1/hosts").status_code == 401
|
|
assert client.get("/v1/plugins").status_code == 401
|
|
assert (
|
|
client.post(
|
|
"/v1/plugins",
|
|
json={
|
|
"name": "x",
|
|
"version": "1",
|
|
"entry_point_kind": "tool",
|
|
"target": "cloud.store:CloudStore",
|
|
},
|
|
).status_code
|
|
== 401
|
|
)
|
|
|
|
|
|
def test_default_null_auth_provider_is_used_when_omitted(tmp_path) -> None:
|
|
# No auth_provider kwarg -> defaults to NullAuthProvider
|
|
app, _, _, _ = _build_app(tmp_path)
|
|
client = _client_for(app)
|
|
|
|
# Should NOT 401 (i.e., NullAuthProvider lets everything through).
|
|
assert client.get("/v1/plugins").status_code == 200
|
|
assert client.get("/v1/devices").status_code == 200
|
|
|
|
|
|
def test_submit_with_constraints(tmp_path) -> None:
|
|
app, pool, _, _ = _build_app(tmp_path)
|
|
pool.sync_host_devices(
|
|
"host-local",
|
|
[Device(id="dev-1", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
client = _client_for(app)
|
|
|
|
resp = client.post(
|
|
"/v1/tasks",
|
|
json={
|
|
"goal": "x",
|
|
"constraints": {"driver_type": "wda", "capability_tags": []},
|
|
},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
task_id = resp.json()["task_id"]
|
|
|
|
# And the device should match if we run assign() manually via the scheduler.
|
|
status = client.get(f"/v1/tasks/{task_id}").json()
|
|
assert status["status"] == "queued"
|
|
|
|
|
|
def test_submit_with_explicit_target_is_listed_and_not_rerouted(tmp_path) -> None:
|
|
app, pool, scheduler, _ = _build_app(tmp_path)
|
|
pool.sync_host_devices(
|
|
"host-a",
|
|
[Device(id="device-a", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
pool.sync_host_devices(
|
|
"host-b",
|
|
[Device(id="device-b", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
client = _client_for(app)
|
|
|
|
response = client.post(
|
|
"/v1/tasks",
|
|
json={
|
|
"goal": "target b",
|
|
"constraints": {
|
|
"target_host_id": "host-b",
|
|
"target_device_id": "device-b",
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
task_id = response.json()["task_id"]
|
|
|
|
scheduler.assign()
|
|
task = client.get(f"/v1/tasks/{task_id}").json()
|
|
assert task["target_host_id"] == "host-b"
|
|
assert task["target_device_id"] == "device-b"
|
|
assert task["assigned_host_id"] == "host-b"
|
|
assert task["assigned_device_id"] == "device-b"
|
|
listed = client.get("/v1/tasks").json()["items"]
|
|
assert next(item for item in listed if item["id"] == task_id)["target_host_id"] == "host-b"
|
|
|
|
|
|
def test_submit_rejects_incomplete_or_foreign_target(tmp_path) -> None:
|
|
app, pool, _, _ = _build_app(tmp_path)
|
|
pool.sync_host_devices(
|
|
"host-a",
|
|
[Device(id="device-a", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
client = _client_for(app)
|
|
|
|
missing_host = client.post(
|
|
"/v1/tasks",
|
|
json={
|
|
"goal": "invalid",
|
|
"constraints": {"target_device_id": "device-a"},
|
|
},
|
|
)
|
|
assert missing_host.status_code == 400
|
|
foreign_device = client.post(
|
|
"/v1/tasks",
|
|
json={
|
|
"goal": "invalid",
|
|
"constraints": {
|
|
"target_host_id": "host-a",
|
|
"target_device_id": "unknown",
|
|
},
|
|
},
|
|
)
|
|
assert foreign_device.status_code == 400
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "path", "payload", "required_scope"),
|
|
[
|
|
("post", "/v1/tasks", {"goal": "x"}, "tasks:submit"),
|
|
("get", "/v1/tasks/missing", None, "tasks:read"),
|
|
("get", "/v1/tasks", None, "tasks:read"),
|
|
("get", "/v1/tasks/missing/attempts", None, "tasks:read"),
|
|
("get", "/v1/tasks/missing/planner-decisions?attempt=0", None, "tasks:read"),
|
|
("post", "/v1/tasks/missing/cancel", None, "tasks:submit"),
|
|
("get", "/v1/devices", None, "pool:read"),
|
|
("get", "/v1/hosts", None, "pool:read"),
|
|
("get", "/v1/plugins", None, "plugins:read"),
|
|
(
|
|
"post",
|
|
"/v1/plugins",
|
|
{
|
|
"name": "scope-test",
|
|
"version": "1.0.0",
|
|
"entry_point_kind": "tool",
|
|
"target": "cloud.store:CloudStore",
|
|
},
|
|
"plugins:admin",
|
|
),
|
|
],
|
|
)
|
|
def test_every_public_route_enforces_its_scope(
|
|
tmp_path,
|
|
method: str,
|
|
path: str,
|
|
payload: dict[str, object] | None,
|
|
required_scope: str,
|
|
) -> None:
|
|
provider = ConfiguredBearerAuthProvider(
|
|
[
|
|
BearerCredential(
|
|
principal_id="integrator",
|
|
token="scoped-token",
|
|
scopes=frozenset({required_scope}),
|
|
),
|
|
BearerCredential(
|
|
principal_id="under-scoped",
|
|
token="wrong-token",
|
|
scopes=frozenset(),
|
|
),
|
|
]
|
|
)
|
|
app, _, _, _ = _build_app(tmp_path, auth_provider=provider)
|
|
client = _client_for(app)
|
|
|
|
unauthorized = client.request(method, path, json=payload)
|
|
forbidden = client.request(
|
|
method,
|
|
path,
|
|
json=payload,
|
|
headers={"Authorization": "Bearer wrong-token"},
|
|
)
|
|
authorized = client.request(
|
|
method,
|
|
path,
|
|
json=payload,
|
|
headers={"Authorization": "Bearer scoped-token"},
|
|
)
|
|
|
|
assert unauthorized.status_code == 401
|
|
assert unauthorized.headers["www-authenticate"] == "Bearer"
|
|
assert forbidden.status_code == 403
|
|
assert authorized.status_code not in {401, 403}
|
|
|
|
|
|
def test_list_tasks_returns_summary_with_pagination_and_status_filter(
|
|
tmp_path,
|
|
) -> None:
|
|
app, pool, scheduler, _ = _build_app(tmp_path)
|
|
# Submit three tasks; assign one so the population covers multiple statuses.
|
|
first_id = scheduler.submit(goal="first")
|
|
second_id = scheduler.submit(goal="second")
|
|
pool.sync_host_devices(
|
|
"host-a",
|
|
[Device(id="device-a", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
scheduler.assign()
|
|
# assigned_id is whichever task the scheduler picked (oldest-first = first_id).
|
|
assigned_id = first_id
|
|
|
|
client = _client_for(app)
|
|
|
|
unfiltered = client.get("/v1/tasks").json()
|
|
assert unfiltered["total"] == 2
|
|
assert unfiltered["limit"] == 50
|
|
assert unfiltered["offset"] == 0
|
|
assert [item["id"] for item in unfiltered["items"]] == [second_id, assigned_id]
|
|
# Lease id must not leak through the summary surface.
|
|
assert all("lease_id" not in item for item in unfiltered["items"])
|
|
|
|
queued_only = client.get("/v1/tasks", params={"status": "queued"}).json()
|
|
assert queued_only["total"] == 1
|
|
assert [item["id"] for item in queued_only["items"]] == [second_id]
|
|
assert all(item["status"] == "queued" for item in queued_only["items"])
|
|
|
|
assigned_only = client.get(
|
|
"/v1/tasks", params={"status": "assigned"}
|
|
).json()
|
|
assert assigned_only["total"] == 1
|
|
assert [item["id"] for item in assigned_only["items"]] == [assigned_id]
|
|
|
|
|
|
def test_list_tasks_rejects_page_size_above_maximum(tmp_path) -> None:
|
|
app, _, _, _ = _build_app(tmp_path)
|
|
client = _client_for(app)
|
|
|
|
too_large = client.get("/v1/tasks", params={"limit": 101})
|
|
assert too_large.status_code == 422
|
|
# And the boundary value is accepted.
|
|
boundary = client.get("/v1/tasks", params={"limit": 100})
|
|
assert boundary.status_code == 200
|
|
|
|
|
|
def test_list_tasks_rejects_negative_offset(tmp_path) -> None:
|
|
app, _, _, _ = _build_app(tmp_path)
|
|
client = _client_for(app)
|
|
response = client.get("/v1/tasks", params={"offset": -1})
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_list_task_attempts_returns_chronological_history(tmp_path) -> None:
|
|
app, pool, scheduler, _ = _build_app(tmp_path)
|
|
pool.sync_host_devices(
|
|
"host-a",
|
|
[Device(id="device-a", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
task_id = scheduler.submit(goal="attempt me")
|
|
scheduler.assign()
|
|
task = scheduler.store.get_task(task_id)
|
|
scheduler.store.record_task_result(
|
|
task_id=task_id,
|
|
attempt=task.attempt_count,
|
|
lease_id=task.lease_id or "",
|
|
host_id=task.assigned_host_id or "",
|
|
status="failed",
|
|
failure_reason="boom",
|
|
terminal_result={"exit_code": 1},
|
|
completed_at=datetime.now(UTC),
|
|
)
|
|
|
|
client = _client_for(app)
|
|
resp = client.get(f"/v1/tasks/{task_id}/attempts")
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert len(body) == 1
|
|
assert body[0]["task_id"] == task_id
|
|
assert body[0]["status"] == "failed"
|
|
assert body[0]["failure_reason"] == "boom"
|
|
assert body[0]["terminal_result"] == {"exit_code": 1}
|
|
assert body[0]["host_id"] == "host-a"
|
|
assert body[0]["device_id"] == "device-a"
|
|
|
|
|
|
def test_list_task_attempts_returns_404_for_unknown_task(tmp_path) -> None:
|
|
app, _, _, _ = _build_app(tmp_path)
|
|
client = _client_for(app)
|
|
resp = client.get("/v1/tasks/does-not-exist/attempts")
|
|
assert resp.status_code == 404, resp.text
|
|
assert "does-not-exist" in resp.json()["detail"]
|
|
|
|
|
|
def test_cancel_queued_task_transitions_immediately(tmp_path) -> None:
|
|
app, _, scheduler, _ = _build_app(tmp_path)
|
|
task_id = scheduler.submit(goal="cancel me")
|
|
|
|
resp = _client_for(app).post(f"/v1/tasks/{task_id}/cancel")
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json() == {"task_id": task_id, "status": "cancelled"}
|
|
assert scheduler.store.get_task(task_id).status == "cancelled"
|
|
|
|
|
|
def test_cancel_assigned_task_records_pending_request(tmp_path) -> None:
|
|
app, pool, scheduler, _ = _build_app(tmp_path)
|
|
pool.sync_host_devices(
|
|
"host-a",
|
|
[Device(id="device-a", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
task_id = scheduler.submit(goal="cancel me mid-flight")
|
|
scheduler.assign()
|
|
|
|
resp = _client_for(app).post(f"/v1/tasks/{task_id}/cancel")
|
|
|
|
assert resp.status_code == 202, resp.text
|
|
assert resp.json() == {"task_id": task_id, "status": "assigned"}
|
|
task = scheduler.store.get_task(task_id)
|
|
assert task.status == "assigned"
|
|
assert task.cancel_requested_at is not None
|
|
|
|
|
|
def test_cancel_repeat_call_on_pending_request_is_idempotent(tmp_path) -> None:
|
|
app, pool, scheduler, _ = _build_app(tmp_path)
|
|
pool.sync_host_devices(
|
|
"host-a",
|
|
[Device(id="device-a", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
task_id = scheduler.submit(goal="cancel me twice")
|
|
scheduler.assign()
|
|
client = _client_for(app)
|
|
|
|
first = client.post(f"/v1/tasks/{task_id}/cancel")
|
|
second = client.post(f"/v1/tasks/{task_id}/cancel")
|
|
|
|
assert first.status_code == 202, first.text
|
|
assert second.status_code == 200, second.text
|
|
assert second.json() == {"task_id": task_id, "status": "assigned"}
|
|
|
|
|
|
def test_cancel_repeat_call_on_already_cancelled_task_is_idempotent(
|
|
tmp_path,
|
|
) -> None:
|
|
app, _, scheduler, _ = _build_app(tmp_path)
|
|
task_id = scheduler.submit(goal="cancel me twice")
|
|
client = _client_for(app)
|
|
|
|
first = client.post(f"/v1/tasks/{task_id}/cancel")
|
|
second = client.post(f"/v1/tasks/{task_id}/cancel")
|
|
|
|
assert first.status_code == 200, first.text
|
|
assert second.status_code == 200, second.text
|
|
assert second.json() == {"task_id": task_id, "status": "cancelled"}
|
|
|
|
|
|
def test_cancel_unknown_task_returns_404(tmp_path) -> None:
|
|
app, _, _, _ = _build_app(tmp_path)
|
|
resp = _client_for(app).post("/v1/tasks/does-not-exist/cancel")
|
|
assert resp.status_code == 404, resp.text
|
|
assert "does-not-exist" in resp.json()["detail"]
|
|
|
|
|
|
def test_cancel_terminal_task_returns_409(tmp_path) -> None:
|
|
app, pool, scheduler, _ = _build_app(tmp_path)
|
|
pool.sync_host_devices(
|
|
"host-a",
|
|
[Device(id="device-a", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
task_id = scheduler.submit(goal="finish me")
|
|
scheduler.assign()
|
|
task = scheduler.store.get_task(task_id)
|
|
scheduler.store.record_task_result(
|
|
task_id=task_id,
|
|
attempt=task.attempt_count,
|
|
lease_id=task.lease_id or "",
|
|
host_id=task.assigned_host_id or "",
|
|
status="done",
|
|
failure_reason=None,
|
|
terminal_result={"runtime_status": "completed"},
|
|
completed_at=datetime.now(UTC),
|
|
)
|
|
|
|
resp = _client_for(app).post(f"/v1/tasks/{task_id}/cancel")
|
|
|
|
assert resp.status_code == 409, resp.text
|
|
assert "terminal" in resp.json()["detail"]
|
|
|
|
|
|
def test_cancel_scope_rejected_before_reaching_scheduler(tmp_path, monkeypatch) -> None:
|
|
provider = ConfiguredBearerAuthProvider(
|
|
[
|
|
BearerCredential(
|
|
principal_id="reader",
|
|
token="reader-token",
|
|
scopes=frozenset({"tasks:read"}),
|
|
)
|
|
]
|
|
)
|
|
app, _, scheduler, _ = _build_app(tmp_path, auth_provider=provider)
|
|
called = False
|
|
|
|
def fail_if_called(*args, **kwargs):
|
|
nonlocal called
|
|
called = True
|
|
raise AssertionError("cancellation must not run before authorization")
|
|
|
|
monkeypatch.setattr(scheduler.store, "request_task_cancellation", fail_if_called)
|
|
|
|
resp = _client_for(app).post(
|
|
"/v1/tasks/does-not-exist/cancel",
|
|
headers={"Authorization": "Bearer reader-token"},
|
|
)
|
|
|
|
assert resp.status_code == 403
|
|
assert called is False
|
|
|
|
|
|
def test_cancellation_full_path_queued_immediate_and_dispatched_collaborative(
|
|
tmp_path,
|
|
) -> None:
|
|
"""End-to-end exercise of the cancellation path (task-cancellation 9.2):
|
|
a queued task is cancelled immediately; a dispatched task's cancellation
|
|
is only recorded until the Host Agent's next lease renewal surfaces it
|
|
and reports back a cancelled terminal result, after which the task is
|
|
visible as cancelled via the public API's get and list endpoints."""
|
|
app, pool, scheduler, _ = _build_app(tmp_path)
|
|
pool.sync_host_devices(
|
|
"host-a",
|
|
[Device(id="device-a", driver_type="wda", status="idle")], # type: ignore[arg-type]
|
|
)
|
|
client = _client_for(app)
|
|
|
|
queued_task_id = scheduler.submit(goal="cancel me while queued")
|
|
queued_cancel = client.post(f"/v1/tasks/{queued_task_id}/cancel")
|
|
assert queued_cancel.status_code == 200, queued_cancel.text
|
|
assert queued_cancel.json() == {"task_id": queued_task_id, "status": "cancelled"}
|
|
|
|
dispatched_task_id = scheduler.submit(goal="cancel me mid-execution")
|
|
scheduler.assign()
|
|
dispatched_cancel = client.post(f"/v1/tasks/{dispatched_task_id}/cancel")
|
|
assert dispatched_cancel.status_code == 202, dispatched_cancel.text
|
|
assert dispatched_cancel.json() == {
|
|
"task_id": dispatched_task_id,
|
|
"status": "assigned",
|
|
}
|
|
|
|
task = scheduler.store.get_task(dispatched_task_id)
|
|
renewed_at = datetime.now(UTC)
|
|
renewal = scheduler.store.renew_lease(
|
|
task_id=dispatched_task_id,
|
|
attempt=task.attempt_count,
|
|
lease_id=task.lease_id or "",
|
|
host_id=task.assigned_host_id or "",
|
|
lease_expires_at=renewed_at + timedelta(seconds=30),
|
|
now=renewed_at,
|
|
)
|
|
assert renewal.status == "renewed", renewal
|
|
assert renewal.cancel_requested is True
|
|
|
|
scheduler.store.record_task_result(
|
|
task_id=dispatched_task_id,
|
|
attempt=task.attempt_count,
|
|
lease_id=task.lease_id or "",
|
|
host_id=task.assigned_host_id or "",
|
|
status="cancelled",
|
|
failure_reason="cancellation requested by control plane",
|
|
terminal_result=None,
|
|
completed_at=datetime.now(UTC),
|
|
)
|
|
|
|
status_resp = client.get(f"/v1/tasks/{dispatched_task_id}")
|
|
assert status_resp.status_code == 200, status_resp.text
|
|
assert status_resp.json()["status"] == "cancelled"
|
|
|
|
list_resp = client.get("/v1/tasks", params={"status": "cancelled"})
|
|
assert list_resp.status_code == 200, list_resp.text
|
|
cancelled_ids = {t["id"] for t in list_resp.json()["items"]}
|
|
assert {queued_task_id, dispatched_task_id} <= cancelled_ids
|
|
|
|
|
|
def test_plugin_admin_scope_is_checked_before_registration(
|
|
tmp_path,
|
|
monkeypatch,
|
|
) -> None:
|
|
provider = ConfiguredBearerAuthProvider(
|
|
[
|
|
BearerCredential(
|
|
principal_id="plugin-reader",
|
|
token="reader-token",
|
|
scopes=frozenset({"plugins:read"}),
|
|
)
|
|
]
|
|
)
|
|
app, _, _, plugin_registry = _build_app(tmp_path, auth_provider=provider)
|
|
registration_called = False
|
|
|
|
def fail_if_called(_manifest) -> None:
|
|
nonlocal registration_called
|
|
registration_called = True
|
|
raise AssertionError("plugin registration must not run before authorization")
|
|
|
|
monkeypatch.setattr(plugin_registry, "register", fail_if_called)
|
|
response = _client_for(app).post(
|
|
"/v1/plugins",
|
|
headers={"Authorization": "Bearer reader-token"},
|
|
json={
|
|
"name": "forbidden-plugin",
|
|
"version": "1.0.0",
|
|
"entry_point_kind": "driver",
|
|
"target": "secret.module:builder",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert registration_called is False
|
|
assert "secret.module" not in response.text
|
|
|
|
|
|
def test_invalid_token_is_rejected_by_public_router(tmp_path) -> None:
|
|
provider = ConfiguredBearerAuthProvider(
|
|
[BearerCredential(principal_id="integrator", token="valid-token")]
|
|
)
|
|
app, _, _, _ = _build_app(tmp_path, auth_provider=provider)
|
|
|
|
response = _client_for(app).get(
|
|
"/v1/devices",
|
|
headers={"Authorization": "Bearer invalid-token"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert response.headers["www-authenticate"] == "Bearer"
|