agentic-mobile-control/apps/device-host-agent/tests/host_agent/web/conftest.py

"""Shared fixtures and context factories for Jinja2 template tests."""

from __future__ import annotations

from datetime import UTC, datetime
from typing import Any

import pytest

from host_agent.config import HostAgentConfig
from host_agent.web.app import _ENV
from host_agent.web.auth import SessionState

XSS_PROBE = "<script>alert(1)</script>"
_ESCAPED_PROBE = "&lt;script&gt;alert(1)&lt;/script&gt;"


@pytest.fixture
def env() -> Any:
    """The module-level Jinja2 Environment from host_agent.web.app."""
    return _ENV


@pytest.fixture
def sample_session() -> SessionState:
    """A representative logged-in session."""
    return SessionState(
        username="operator",
        csrf_token="test-csrf-token",
        expires_at=datetime(2030, 1, 1, tzinfo=UTC),
    )


@pytest.fixture
def xss_probe() -> str:
    return XSS_PROBE


# ---------------------------------------------------------------------------
# Context factories
# ---------------------------------------------------------------------------


def make_login_context(
    *, account: Any = None, error: str | None = None
) -> dict[str, Any]:
    return {
        "title": "Login",
        "session": None,
        "account": account,
        "error": error,
    }


def make_dashboard_context(
    session: SessionState,
    *,
    devices: list[dict[str, Any]] | None = None,
    identity: Any = None,
    heartbeat_text: str = "never",
    assignment_text: str = "none",
    progress_text: str = "",
    policy_text: str = "no Cloud policy cached",
    config: HostAgentConfig | None = None,
) -> dict[str, Any]:
    if devices is None:
        devices = [
            {
                "id": "dev-1",
                "name": "Pixel 8",
                "driver_type": "wda",
                "display_status": "connected",
            },
            {
                "id": "dev-2",
                "name": "iPhone 15",
                "driver_type": "wda",
                "display_status": "busy",
            },
        ]
    if config is None:
        config = HostAgentConfig(control_plane_url="http://localhost:8080")
    return {
        "title": "Status",
        "session": session,
        "identity": identity,
        "devices": devices,
        "config": config,
        "heartbeat_text": heartbeat_text,
        "assignment_text": assignment_text,
        "progress_text": progress_text,
        "policy_text": policy_text,
    }


def make_devices_context(
    session: SessionState,
    *,
    devices: list[dict[str, Any]] | None = None,
    edit_record: dict[str, Any] | None = None,
    connection_info_json: str = "{}",
    error: str | None = None,
) -> dict[str, Any]:
    if devices is None:
        devices = [
            {
                "device_id": "dev-1",
                "name": "Pixel 8",
                "driver_type": "wda",
                "cloud_device_id": "cloud-1",
                "connection_info": {"port": 8100},
            },
        ]
    if edit_record is None:
        edit_record = {
            "device_id": "dev-1",
            "name": "Pixel 8",
            "driver_type": "wda",
            "connection_info": {"port": 8100},
        }
        connection_info_json = '{"port": 8100}'
    return {
        "title": "Devices",
        "session": session,
        "devices": devices,
        "csrf_token": session.csrf_token,
        "edit_record": edit_record,
        "connection_info_json": connection_info_json,
        "error": error,
    }


def make_account_context(
    session: SessionState,
    *,
    message: str | None = None,
    error: str | None = None,
) -> dict[str, Any]:
    return {
        "title": "Account",
        "session": session,
        "csrf_token": session.csrf_token,
        "message": message,
        "error": error,
    }


def make_history_context(
    session: SessionState,
    *,
    entries: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    if entries is None:
        entries = [
            {
                "occurred_at": "2026-01-01T00:00:00Z",
                "kind": "assignment",
                "summary": "Task abc-123 started on dev-1",
            },
        ]
    return {
        "title": "History",
        "session": session,
        "entries": entries,
    }


def make_tasks_list_context(
    session: SessionState,
    *,
    tasks: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    if tasks is None:
        tasks = [
            {
                "id": "task-001",
                "status": "completed",
                "device_id": "dev-1",
                "created_at": "2026-01-01T00:00:00Z",
                "updated_at": "2026-01-01T00:05:00Z",
            },
        ]
    return {
        "title": "Tasks",
        "session": session,
        "tasks": tasks,
    }


def make_task_detail_context(
    session: SessionState,
    *,
    task: dict[str, Any] | None = None,
    task_rows: list[tuple[str, Any]] | None = None,
    timeline_steps: list[dict[str, Any]] | None = None,
    can_cancel: bool = False,
    cancel_notice: str | None = None,
    cancel_error: str | None = None,
) -> dict[str, Any]:
    if task is None:
        task = {
            "id": "task-001",
            "goal": "Open settings",
            "device_id": "dev-1",
            "status": "completed",
            "created_at": "2026-01-01T00:00:00Z",
            "updated_at": "2026-01-01T00:05:00Z",
        }
    if task_rows is None:
        task_rows = [
            ("id", task["id"]),
            ("goal", task["goal"]),
            ("device_id", task["device_id"]),
            ("status", task["status"]),
        ]
    if timeline_steps is None:
        timeline_steps = [
            {
                "index": 0,
                "timestamp": "2026-01-01T00:01:00Z",
                "prompt": "Tap the Settings icon",
                "tool_call": {"action": "tap", "x": 100, "y": 200},
                "result": {"ok": True},
                "before_screenshot_src": None,
                "after_screenshot_src": None,
                "ocr_results": [],
                "ui_tree_nodes": [],
            },
        ]
    return {
        "title": "Task task-001",
        "session": session,
        "csrf_token": session.csrf_token,
        "task": task,
        "task_rows": task_rows,
        "timeline_steps": timeline_steps,
        "can_cancel": can_cancel,
        "cancel_notice": cancel_notice,
        "cancel_error": cancel_error,
    }