"""Unit tests for cloud.sdk.client.CloudClient (task 8.2).""" from __future__ import annotations import pytest import httpx from cloud.auth import BearerCredential, ConfiguredBearerAuthProvider from cloud.config import CloudConfig from cloud.plugins import PluginRegistry from cloud.pool import DevicePool from cloud.sdk.api import create_cloud_router from cloud.sdk.client import CloudAuthorizationError, CloudClient 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 _client_and_pool(tmp_path, *, auth_provider=None, token: str | None = None): store = CloudStore(tmp_path / "cloud.sqlite3") pool = DevicePool(store, _config()) scheduler = TaskScheduler(pool, store, _config()) plugin_registry = PluginRegistry(store) app = FastAPI() app.include_router( create_cloud_router( pool=pool, scheduler=scheduler, plugin_registry=plugin_registry, auth_provider=auth_provider, ) ) test_client = TestClient(app) cloud_client = CloudClient( "http://testserver", http_client=test_client, token=token, ) return cloud_client, pool def test_client_submit_and_get_status_round_trip(tmp_path) -> None: client, _ = _client_and_pool(tmp_path) submission = client.submit_task(goal="open settings") assert "task_id" in submission task_id = submission["task_id"] status = client.get_task_status(task_id) assert status["id"] == task_id assert status["status"] == "queued" assert status["goal"] == "open settings" def test_client_list_devices_and_hosts(tmp_path) -> None: client, pool = _client_and_pool(tmp_path) pool.sync_host_devices( "host-a", [Device(id="dev-1", driver_type="wda", status="idle")], # type: ignore[arg-type] address="a:8000", ) devices = client.list_devices() assert [d["device_id"] for d in devices] == ["dev-1"] assert devices[0]["host_id"] == "host-a" hosts = client.list_hosts() assert [h["host_id"] for h in hosts] == ["host-a"] def test_client_plugin_listing_and_registration(tmp_path) -> None: client, _ = _client_and_pool(tmp_path) assert client.list_plugins() == [] registered = client.register_plugin( name="demo", version="1.0.0", entry_point_kind="tool", target="cloud.store:CloudStore", ) assert registered["name"] == "demo" assert registered["wired"] is False listed = client.list_plugins() assert [p["name"] for p in listed] == ["demo"] def test_client_submit_with_constraints(tmp_path) -> None: client, pool = _client_and_pool(tmp_path) pool.sync_host_devices( "host-a", [Device(id="dev-1", driver_type="wda", status="idle")], # type: ignore[arg-type] ) submission = client.submit_task( goal="x", driver_type="wda", capability_tags=[], ) task_id = submission["task_id"] status = client.get_task_status(task_id) assert status["status"] == "queued" def test_client_unknown_task_raises(tmp_path) -> None: client, _ = _client_and_pool(tmp_path) with pytest.raises(httpx.HTTPStatusError): client.get_task_status("does-not-exist") def test_client_cancel_task_round_trip(tmp_path) -> None: client, _ = _client_and_pool(tmp_path) task_id = client.submit_task(goal="cancel me")["task_id"] cancelled = client.cancel_task(task_id) assert cancelled == {"task_id": task_id, "status": "cancelled"} assert client.get_task_status(task_id)["status"] == "cancelled" def test_client_cancel_unknown_task_raises(tmp_path) -> None: client, _ = _client_and_pool(tmp_path) with pytest.raises(httpx.HTTPStatusError): client.cancel_task("does-not-exist") def test_client_applies_bearer_token_to_every_public_method(tmp_path) -> None: token = "sdk-secret" provider = ConfiguredBearerAuthProvider( [ BearerCredential( principal_id="sdk", token=token, scopes=frozenset( { "tasks:submit", "tasks:read", "pool:read", "plugins:read", "plugins:admin", } ), ) ] ) client, _ = _client_and_pool( tmp_path, auth_provider=provider, token=token, ) task_id = client.submit_task(goal="authenticated")["task_id"] assert client.get_task_status(task_id)["status"] == "queued" assert client.list_tasks()["total"] == 1 assert client.get_task_attempts(task_id) == [] assert client.cancel_task(task_id) == {"task_id": task_id, "status": "cancelled"} assert client.list_devices() == [] assert client.list_hosts() == [] assert client.list_plugins() == [] assert ( client.register_plugin( name="authenticated-plugin", version="1.0.0", entry_point_kind="tool", target="cloud.store:CloudStore", )["name"] == "authenticated-plugin" ) def test_client_list_tasks_and_get_attempts_match_direct_http(tmp_path) -> None: from datetime import UTC, datetime from core.models import Device client, pool = _client_and_pool(tmp_path) pool.sync_host_devices( "host-a", [Device(id="dev-a", driver_type="wda", status="idle")], # type: ignore[arg-type] ) first_id = client.submit_task(goal="first")["task_id"] second_id = client.submit_task(goal="second")["task_id"] # Drive one task through an attempt so get_task_attempts has data. scheduler = TaskScheduler(pool, CloudStore(tmp_path / "cloud.sqlite3"), _config()) scheduler.assign() task = scheduler.store.get_task(first_id) if task is not None and task.status == "assigned": scheduler.store.record_task_result( task_id=first_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), ) # The TestClient is the HTTP boundary — issuing direct calls through it # exercises the same FastAPI routes the client does, which is the parity # contract platform-sdk already relies on. http_client = client._http # type: ignore[attr-defined] direct_list = http_client.get( client._url("/tasks"), # type: ignore[attr-defined] params={"limit": 50, "offset": 0}, headers=client._headers, # type: ignore[attr-defined] ).json() direct_attempts = http_client.get( f"{client._url('/tasks')}/{first_id}/attempts", # type: ignore[attr-defined] headers=client._headers, # type: ignore[attr-defined] ).json() via_client = client.list_tasks() assert via_client == direct_list assert via_client["total"] == 2 # Most-recent-first: second (the newer) before first. assert [item["id"] for item in via_client["items"]] == [second_id, first_id] failed_only = client.list_tasks(status="failed") assert failed_only["total"] == 1 assert failed_only["items"][0]["id"] == first_id attempts = client.get_task_attempts(first_id) assert attempts == direct_attempts assert len(attempts) == 1 assert attempts[0]["status"] == "failed" assert attempts[0]["failure_reason"] == "boom" def test_client_get_attempts_raises_for_unknown_task(tmp_path) -> None: import httpx client, _ = _client_and_pool(tmp_path) with pytest.raises(httpx.HTTPStatusError): client.get_task_attempts("does-not-exist") def test_client_raises_typed_authorization_error_without_exposing_token( tmp_path, ) -> None: provider = ConfiguredBearerAuthProvider( [BearerCredential(principal_id="sdk", token="valid-token")] ) client, _ = _client_and_pool( tmp_path, auth_provider=provider, token="invalid-secret-token", ) with pytest.raises(CloudAuthorizationError) as error: client.list_devices() assert error.value.status_code == 401 assert "invalid-secret-token" not in str(error.value) def test_client_supports_injected_httpx_auth(tmp_path) -> None: class StaticAuth(httpx.Auth): def auth_flow(self, request): request.headers["Authorization"] = "Bearer injected-token" yield request provider = ConfiguredBearerAuthProvider( [ BearerCredential( principal_id="sdk", token="injected-token", scopes=frozenset({"pool:read"}), ) ] ) store = CloudStore(tmp_path / "auth.sqlite3") pool = DevicePool(store, _config()) app = FastAPI() app.include_router( create_cloud_router( pool=pool, scheduler=TaskScheduler(pool, store, _config()), plugin_registry=PluginRegistry(store), auth_provider=provider, ) ) client = CloudClient( "http://testserver", http_client=TestClient(app), auth=StaticAuth(), ) assert client.list_devices() == [] def test_client_rejects_token_and_auth_together() -> None: with pytest.raises(ValueError, match="mutually exclusive"): CloudClient( "https://cloud.example", token="secret", auth=httpx.BasicAuth("user", "password"), )