"""Unit tests for cloud.sdk.api (task 7.8).""" from __future__ import annotations from datetime import UTC, datetime 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" @pytest.mark.parametrize( ("method", "path", "payload", "required_scope"), [ ("post", "/v1/tasks", {"goal": "x"}, "tasks:submit"), ("get", "/v1/tasks/missing", None, "tasks:read"), ("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_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"