44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from device.manager import DeviceManager
|
|
from core.models import Task
|
|
from storage.task_metadata import TaskMetadataStore
|
|
from tests.fakes import FakeDriver
|
|
|
|
|
|
def test_rest_start_task_and_poll_until_complete(tmp_path) -> None:
|
|
pytest.importorskip("fastapi")
|
|
from fastapi.testclient import TestClient
|
|
|
|
from api.rest import create_app
|
|
|
|
driver = FakeDriver()
|
|
manager = DeviceManager()
|
|
manager.register_device("iphone-1", lambda: driver)
|
|
manager.connect("iphone-1", max_retries=1)
|
|
store = TaskMetadataStore(tmp_path / "tasks.sqlite3")
|
|
|
|
class CompletingRunner:
|
|
def run(self, task: Task) -> None:
|
|
store.update_task(task.id, status="completed", completed=True)
|
|
|
|
app = create_app(
|
|
manager=manager,
|
|
metadata_store=store,
|
|
task_runner=CompletingRunner(),
|
|
)
|
|
client = TestClient(app)
|
|
|
|
response = client.post(
|
|
"/agent/task",
|
|
json={"goal": "search", "device_id": "iphone-1"},
|
|
)
|
|
assert response.status_code == 200
|
|
task_id = response.json()["task_id"]
|
|
|
|
status_response = client.get(f"/task/{task_id}")
|
|
assert status_response.status_code == 200
|
|
assert status_response.json()["status"] == "completed"
|