q792602257
6776ac2f2d
- POST /v1/tasks/{task_id}/cancel: tasks:submit scoped, 200 for
immediate/idempotent cancellation, 202 for newly recorded pending
cancellation, 404 for unknown task, 409 for terminal task.
- TaskCancellationResponse{task_id, status} model.
- Widen list_tasks status_filter Literal to include "cancelled".
- CloudClient.cancel_task(task_id).
- SDK-level tests covering queued/assigned/idempotent/404/409/scope
cases for both the router and CloudClient.
Task 5/9 of task-cancellation change.
315 lines
9.8 KiB
Python
315 lines
9.8 KiB
Python
"""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"),
|
|
)
|