agentic-mobile-control/tests/test_host_agent_console_tasks.py

"""Tests for Host Agent console /tasks routes (task 5.4)."""

from __future__ import annotations

import re
from datetime import UTC, datetime

from fastapi.testclient import TestClient

from device.manager import DeviceManager
from host_agent.config import HostAgentConfig
from host_agent.history import ConsoleHistoryStore
from host_agent.identity import HostIdentityStore
from host_agent.local_account import LocalAccountStore
from host_agent.status import AgentStatusTracker
from host_agent.web.app import create_console_app
from host_agent.web.auth import SessionManager
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

CSRF_PATTERN = re.compile(r'name="csrf_token" value="([^"]+)"')


def _build_client(
    tmp_path,
    *,
    metadata_store: TaskMetadataStore | None = None,
    timeline: Timeline | None = None,
    create_account: bool = True,
) -> tuple[TestClient, dict]:
    config = HostAgentConfig(
        control_plane_url="https://control.example",
        host_id="host-a",
        token="secret",
        enrollment_managed=False,
        console_session_ttl_seconds=3600.0,
    )
    manager = DeviceManager()
    config_store = DeviceConfigStore(tmp_path / "devices.sqlite3")
    local_account_store = LocalAccountStore(tmp_path / "host_local_account.json")
    if create_account:
        local_account_store.create("operator", "correct horse battery staple")
    identity_store = HostIdentityStore(tmp_path / "host_identity.json")
    history_store = ConsoleHistoryStore(tmp_path / "history.sqlite3")
    status_tracker = AgentStatusTracker()
    session_manager = SessionManager(ttl_seconds=3600.0)

    if metadata_store is None:
        metadata_store = TaskMetadataStore(tmp_path / "task_progress.sqlite3")
    if timeline is None:
        timeline = Timeline(ArtifactStore(tmp_path / "history"))

    app = create_console_app(
        config=config,
        manager=manager,
        config_store=config_store,
        local_account_store=local_account_store,
        identity_store=identity_store,
        history_store=history_store,
        status_tracker=status_tracker,
        session_manager=session_manager,
        enrollment_client=None,
        metadata_store=metadata_store,
        timeline=timeline,
    )
    client = TestClient(app)
    context = {
        "metadata_store": metadata_store,
        "timeline": timeline,
        "session_manager": session_manager,
    }
    return client, context


def _login(client: TestClient) -> str:
    response = client.post(
        "/login",
        data={"username": "operator", "password": "correct horse battery staple"},
    )
    assert response.status_code == 200
    match = CSRF_PATTERN.search(response.text)
    assert match is not None
    return match.group(1)


# --------------------------------------------------------------------------- #
# Unauthenticated requests rejected
# --------------------------------------------------------------------------- #


def test_unauthenticated_tasks_list_redirects_to_login(tmp_path) -> None:
    client, _ = _build_client(tmp_path)

    response = client.get("/tasks", follow_redirects=False)

    assert response.status_code == 303
    assert response.headers["location"] == "/login"


def test_unauthenticated_task_detail_redirects_to_login(tmp_path) -> None:
    client, _ = _build_client(tmp_path)

    response = client.get("/tasks/some-id", follow_redirects=False)

    assert response.status_code == 303
    assert response.headers["location"] == "/login"


def test_unauthenticated_devices_page_also_redirects_for_parity(tmp_path) -> None:
    """Confirm /tasks has the same auth behavior as the existing /devices route."""
    client, _ = _build_client(tmp_path)

    response = client.get("/devices", follow_redirects=False)

    assert response.status_code == 303
    assert response.headers["location"] == "/login"


# --------------------------------------------------------------------------- #
# Authenticated list/detail render
# --------------------------------------------------------------------------- #


def test_authenticated_tasks_list_renders_completed_task(tmp_path) -> None:
    from core.models import Task

    metadata_store = TaskMetadataStore(tmp_path / "task_progress.sqlite3")
    timeline = Timeline(ArtifactStore(tmp_path / "history"))

    task = Task(
        id="task-visible",
        goal="open settings",
        device_id="iphone-1",
        status="completed",
        created_at=datetime(2026, 7, 10, tzinfo=UTC),
        updated_at=datetime(2026, 7, 10, 1, tzinfo=UTC),
        completed_at=datetime(2026, 7, 10, 1, tzinfo=UTC),
    )
    metadata_store.create_task(task)

    client, _ = _build_client(
        tmp_path, metadata_store=metadata_store, timeline=timeline
    )
    _login(client)

    response = client.get("/tasks")
    assert response.status_code == 200
    assert "task-visible" in response.text
    assert "completed" in response.text


def test_authenticated_task_detail_renders_timeline_with_screenshot(tmp_path) -> None:
    from core.models import Task

    metadata_store = TaskMetadataStore(tmp_path / "task_progress.sqlite3")
    timeline = Timeline(ArtifactStore(tmp_path / "history"))

    task = Task(
        id="task-detail",
        goal="tap search",
        device_id="iphone-1",
        status="completed",
        created_at=datetime(2026, 7, 10, tzinfo=UTC),
        updated_at=datetime(2026, 7, 10, 1, tzinfo=UTC),
    )
    metadata_store.create_task(task)

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

    client, _ = _build_client(
        tmp_path, metadata_store=metadata_store, timeline=timeline
    )
    _login(client)

    response = client.get("/tasks/task-detail")
    assert response.status_code == 200
    assert "task-detail" in response.text
    # Timeline step visible.
    assert "Step 1" in response.text
    assert "find search button" in response.text
    # Screenshot inlined as base64 data URI.
    assert "data:image/png;base64," in response.text


def test_task_detail_404_for_unknown_task(tmp_path) -> None:
    client, _ = _build_client(tmp_path)
    _login(client)

    response = client.get("/tasks/nonexistent")
    assert response.status_code == 404


def test_tasks_list_shows_empty_state(tmp_path) -> None:
    metadata_store = TaskMetadataStore(tmp_path / "task_progress.sqlite3")
    client, _ = _build_client(tmp_path, metadata_store=metadata_store)
    _login(client)

    response = client.get("/tasks")
    assert response.status_code == 200
    assert "No tasks recorded" in response.text