agentic-mobile-control/tests/test_console_api.py

from __future__ import annotations

import base64
from datetime import UTC, datetime

import pytest

from core.models import Task
from device.manager import DeviceManager
from runtime.task import TaskRunner, TaskRunnerConfig
from storage.artifact_store import ArtifactStore
from storage.device_config import DeviceConfigStore
from storage.task_metadata import TaskMetadataStore
from storage.timeline import Timeline
from tests.fakes import PNG_10X20, FakeDriver


def _client(tmp_path, *, manager=None, runner=None, config_store=None, timeline=None):
    pytest.importorskip("fastapi")
    from fastapi.testclient import TestClient

    from api.rest import create_app

    metadata_store = TaskMetadataStore(tmp_path / "tasks.sqlite3")
    app = create_app(
        manager=manager or DeviceManager(),
        metadata_store=metadata_store,
        task_runner=runner,
        device_config_store=config_store
        or DeviceConfigStore(tmp_path / "device_config.sqlite3"),
        timeline=timeline or Timeline(ArtifactStore(tmp_path / "history")),
    )
    return TestClient(app), metadata_store


def test_console_status_endpoints_cover_empty_and_populated_states(tmp_path) -> None:
    manager = DeviceManager()
    client, metadata_store = _client(tmp_path, manager=manager)

    assert client.get("/console/devices").json() == []
    assert client.get("/console/tasks").json() == []

    manager.register_device(
        "iphone-1",
        lambda: FakeDriver(),
        name="Desk iPhone",
        driver_type="wda",
    )
    older = Task(
        id="task-old",
        goal="open settings",
        device_id="iphone-1",
        created_at=datetime(2026, 1, 1, tzinfo=UTC),
        updated_at=datetime(2026, 1, 1, tzinfo=UTC),
    )
    newer = Task(
        id="task-new",
        goal="search",
        device_id="iphone-2",
        status="running",
        created_at=datetime(2026, 1, 2, tzinfo=UTC),
        updated_at=datetime(2026, 1, 2, tzinfo=UTC),
    )
    metadata_store.create_task(older)
    metadata_store.create_task(newer)

    devices = client.get("/console/devices").json()
    assert devices == [
        {
            "id": "iphone-1",
            "name": "Desk iPhone",
            "status": "idle",
            "driver_type": "wda",
            "connection_info": {},
            "capability_tags": [],
        }
    ]
    assert [task["id"] for task in client.get("/console/tasks").json()] == [
        "task-new",
        "task-old",
    ]
    assert [
        task["id"]
        for task in client.get("/console/tasks?device_id=iphone-1").json()
    ] == ["task-old"]
    assert [task["id"] for task in client.get("/console/tasks?status=running").json()] == [
        "task-new"
    ]
    assert client.get("/console/tasks/task-old").json()["goal"] == "open settings"
    assert client.get("/console/tasks/missing").status_code == 404


def test_console_timeline_inlines_screenshot_and_handles_empty_history(tmp_path) -> None:
    timeline = Timeline(ArtifactStore(tmp_path / "history"))
    client, metadata_store = _client(tmp_path, timeline=timeline)
    task = Task(id="task-1", goal="tap search", device_id="iphone-1")
    metadata_store.create_task(task)

    assert client.get("/console/tasks/task-1/timeline").json() == []

    timeline.append(
        task_id="task-1",
        scene={"screen": {"width": 10, "height": 20}, "elements": []},
        prompt="tap search",
        tool_call={"action": "tap", "args": {"x": 1, "y": 2}},
        result={"ok": True},
        screenshot=PNG_10X20,
    )

    records = client.get("/console/tasks/task-1/timeline").json()
    assert records[0]["index"] == 1
    assert records[0]["image_base64"] == base64.b64encode(PNG_10X20).decode("ascii")
    assert client.get("/console/tasks/missing/timeline").status_code == 404


def test_console_device_registration_and_unregistration(tmp_path) -> None:
    config_store = DeviceConfigStore(tmp_path / "device_config.sqlite3")
    manager = DeviceManager()
    client, _ = _client(tmp_path, manager=manager, config_store=config_store)

    rejected = client.post(
        "/console/devices",
        json={"driver_type": "android", "connection_info": {}},
    )
    assert rejected.status_code == 400
    assert config_store.list() == []
    assert manager.list_devices() == []

    response = client.post(
        "/console/devices",
        json={
            "driver_type": "wda",
            "name": "Desk iPhone",
            "connection_info": {
                "server_url": "http://127.0.0.1:4723",
                "udid": "abc123",
            },
        },
    )
    assert response.status_code == 201
    device_id = response.json()["id"]
    assert response.json()["status"] == "idle"
    assert config_store.get(device_id)["connection_info"]["udid"] == "abc123"
    assert [device.id for device in manager.list_devices()] == [device_id]

    delete_response = client.delete(f"/console/devices/{device_id}")
    assert delete_response.status_code == 204
    assert config_store.get(device_id) is None
    assert manager.list_devices() == []
    assert client.delete("/console/devices/missing").status_code == 404


def test_console_config_get_update_and_validation(tmp_path) -> None:
    config_store = DeviceConfigStore(tmp_path / "device_config.sqlite3")
    config_store.set_setting("max_steps", 7)
    runner = TaskRunner(config=TaskRunnerConfig(max_steps=1))
    client, _ = _client(tmp_path, runner=runner, config_store=config_store)

    assert runner.config.max_steps == 7
    assert client.get("/console/config").json() == {"max_steps": 7}

    response = client.put("/console/config", json={"max_steps": 30})
    assert response.status_code == 200
    assert response.json() == {"max_steps": 30}
    assert runner.config.max_steps == 30
    assert config_store.get_setting("max_steps") == "30"

    rejected = client.put("/console/config", json={"max_steps": 0})
    assert rejected.status_code == 400
    assert runner.config.max_steps == 30
    assert config_store.get_setting("max_steps") == "30"


def test_console_startup_reloads_persisted_devices_and_settings(tmp_path) -> None:
    config_store = DeviceConfigStore(tmp_path / "device_config.sqlite3")
    config_store.add(
        device_id="persisted-1",
        name="Persisted iPhone",
        driver_type="wda",
        connection_info={"udid": "abc123"},
    )
    config_store.set_setting("max_steps", 31)
    manager = DeviceManager()
    runner = TaskRunner(config=TaskRunnerConfig(max_steps=1))

    client, _ = _client(
        tmp_path,
        manager=manager,
        runner=runner,
        config_store=config_store,
    )

    assert runner.config.max_steps == 31
    assert [device.id for device in manager.list_devices()] == ["persisted-1"]
    assert client.get("/console/devices").json()[0]["name"] == "Persisted iPhone"