agentic-mobile-control/tests/test_runtime_console_web.py

from __future__ import annotations

import base64
from datetime import UTC, datetime
from pathlib import Path

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


_TEMPLATE_NAMES = [
    "base.html",
    "dashboard.html",
    "_status_fragment.html",
    "tasks.html",
    "task_detail.html",
    "config.html",
]


# -- 5.2 Template tests ------------------------------------------------------


def test_module_jinja_environment_autoescapes_html_and_xml() -> None:
    from api.console_web import _ENV

    # select_autoescape(["html", "xml"]) returns a callable used by Jinja2.
    assert callable(_ENV.autoescape)
    assert _ENV.autoescape("foo.html") is True
    assert _ENV.autoescape("foo.xml") is True


def test_every_console_template_is_known_and_loadable() -> None:
    from api.console_web import _ENV

    for name in _TEMPLATE_NAMES:
        assert _ENV.get_template(name) is not None


def test_no_template_uses_safe_filter_bypass() -> None:
    template_dir = (
        Path(__file__).resolve().parent.parent / "api" / "templates" / "runtime_console"
    )
    for path in template_dir.glob("*.html"):
        source = path.read_text(encoding="utf-8")
        assert "| safe" not in source, f"{path.name} uses |safe bypass"
        assert "|safe" not in source, f"{path.name} uses |safe bypass"


def test_dashboard_escapes_untrusted_device_name(tmp_path) -> None:
    manager = DeviceManager()
    manager.register_device(
        "dev-xss",
        lambda: FakeDriver(),
        name="<script>alert(1)</script>",
        driver_type="wda",
    )
    client, _ = _client(tmp_path, manager=manager)

    body = client.get("/ui/").text
    assert "&lt;script&gt;" in body
    assert "<script>alert(1)</script>" not in body


def test_tasks_list_escapes_untrusted_goal(tmp_path) -> None:
    client, metadata_store = _client(tmp_path)
    metadata_store.create_task(
        Task(
            id="task-xss",
            goal="<script>alert('xss')</script>",
            device_id="dev-1",
        )
    )
    body = client.get("/ui/tasks").text
    assert "&lt;script&gt;" in body
    assert "<script>alert('xss')</script>" not in body


def test_task_detail_escapes_failure_reason_and_structured_output(tmp_path) -> None:
    timeline = Timeline(ArtifactStore(tmp_path / "history"))
    client, metadata_store = _client(tmp_path, timeline=timeline)
    metadata_store.create_task(
        Task(
            id="task-detail",
            goal="do thing",
            device_id="dev-1",
            failure_reason="<script>alert('fail')</script>",
        )
    )
    timeline.append(
        task_id="task-detail",
        scene={"screen": {"width": 10, "height": 20}, "elements": []},
        prompt="do thing",
        tool_call={"action": "<script>", "args": {"x": 1}},
        result={"err": "</script><script>alert(1)</script>"},
        screenshot=PNG_10X20,
    )
    body = client.get("/ui/tasks/task-detail").text
    assert "<script>alert('fail')</script>" not in body
    assert "</script><script>" not in body
    # Structured JSON output uses tojson which escapes angle brackets.
    assert "\\u003c" in body or "&lt;script&gt;" in body


def test_config_form_error_preserves_and_escapes_submitted_name(tmp_path) -> None:
    client, _ = _client(tmp_path)
    response = client.post(
        "/ui/config/devices",
        data={
            "name": "<script>alert(1)</script>",
            "driver_type": "bad-driver",
        },
    )
    assert response.status_code == 400
    assert "<script>alert(1)</script>" not in response.text
    assert "&lt;script&gt;" in response.text


# -- 5.3 TestClient page-route coverage --------------------------------------


def test_root_redirects_to_ui(tmp_path) -> None:
    client, _ = _client(tmp_path)
    response = client.get("/", follow_redirects=False)
    assert response.status_code == 307
    assert response.headers["location"] == "/ui/"


def test_dashboard_serves_html_with_empty_state(tmp_path) -> None:
    client, _ = _client(tmp_path)
    response = client.get("/ui/")
    assert response.status_code == 200
    assert "text/html" in response.headers["content-type"]
    assert "No devices registered" in response.text


def test_dashboard_shows_populated_metrics_and_devices(tmp_path) -> None:
    manager = DeviceManager()
    manager.register_device(
        "iphone-1",
        lambda: FakeDriver(),
        name="Desk iPhone",
        driver_type="wda",
    )
    client, metadata_store = _client(tmp_path, manager=manager)
    metadata_store.create_task(
        Task(id="t-running", goal="run", device_id="iphone-1", status="running")
    )
    metadata_store.create_task(
        Task(id="t-failed", goal="fail", device_id="iphone-1", status="failed")
    )
    body = client.get("/ui/").text
    assert "Desk iPhone" in body
    assert "iphone-1" in body


def test_status_fragment_endpoint_returns_html_partial(tmp_path) -> None:
    client, _ = _client(tmp_path)
    response = client.get("/ui/_status_fragment")
    assert response.status_code == 200
    assert "text/html" in response.headers["content-type"]
    assert "Device Status" in response.text


def test_tasks_page_supports_device_and_status_filters(tmp_path) -> None:
    manager = DeviceManager()
    manager.register_device(
        "iphone-1",
        lambda: FakeDriver(),
        name="Desk",
        driver_type="wda",
    )
    client, metadata_store = _client(tmp_path, manager=manager)
    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)

    body_all = client.get("/ui/tasks").text
    assert "task-old" in body_all
    assert "task-new" in body_all

    body_filtered = client.get("/ui/tasks?device_id=iphone-1").text
    assert "task-old" in body_filtered
    assert "task-new" not in body_filtered

    body_status = client.get("/ui/tasks?status=running").text
    assert "task-new" in body_status
    assert "task-old" not in body_status


def test_task_detail_renders_timeline_with_screenshot(tmp_path) -> None:
    timeline = Timeline(ArtifactStore(tmp_path / "history"))
    client, metadata_store = _client(tmp_path, timeline=timeline)
    metadata_store.create_task(
        Task(id="task-with-timeline", goal="tap search", device_id="iphone-1")
    )
    timeline.append(
        task_id="task-with-timeline",
        scene={"screen": {"width": 10, "height": 20}, "elements": []},
        prompt="tap search",
        tool_call={
            "action": "tap",
            "description": "tap search",
            "args": {"x": 1, "y": 2},
        },
        result={"ok": True},
        before_screenshot=PNG_10X20 + b"before",
        after_screenshot=PNG_10X20 + b"after",
        ocr_results=[
            {
                "text": "Search",
                "confidence": 0.98,
                "bounds": {"x": 1, "y": 2, "width": 3, "height": 4},
            }
        ],
    )
    body = client.get("/ui/tasks/task-with-timeline").text
    before_data_uri = "data:image/png;base64," + base64.b64encode(
        PNG_10X20 + b"before"
    ).decode("ascii")
    after_data_uri = "data:image/png;base64," + base64.b64encode(
        PNG_10X20 + b"after"
    ).decode("ascii")
    assert before_data_uri in body
    assert after_data_uri in body
    assert "Before action" in body
    assert "After action" in body
    assert "Operation" in body
    assert "OCR results" in body
    assert "Search" in body
    assert "UI tree" not in body


def test_task_detail_404_for_unknown_task(tmp_path) -> None:
    client, _ = _client(tmp_path)
    response = client.get("/ui/tasks/does-not-exist")
    assert response.status_code == 404


def test_task_detail_renders_normalized_ui_tree_result(tmp_path) -> None:
    timeline = Timeline(ArtifactStore(tmp_path / "history"))
    client, metadata_store = _client(tmp_path, timeline=timeline)
    metadata_store.create_task(
        Task(id="task-ui-tree", goal="inspect the screen", device_id="iphone-1")
    )
    timeline.append(
        task_id="task-ui-tree",
        scene={"screen": {"width": 10, "height": 20}, "elements": []},
        prompt="inspect the screen",
        tool_call={"action": "get_ui_tree", "description": "inspect UI tree"},
        result={
            "success": True,
            "result": [
                {
                    "id": "ui-000",
                    "type": "button",
                    "text": "Search",
                    "bounds": {"x": 1, "y": 2, "width": 3, "height": 4},
                    "confidence": 1.0,
                }
            ],
        },
    )

    body = client.get("/ui/tasks/task-ui-tree").text

    assert "UI tree" in body
    assert "1 normalized nodes" in body
    assert "button" in body
    assert "Search" in body


def test_config_page_lists_supported_drivers_and_current_max_steps(tmp_path) -> None:
    config_store = DeviceConfigStore(tmp_path / "device_config.sqlite3")
    config_store.set_setting("max_steps", 5)
    runner = TaskRunner(config=TaskRunnerConfig(max_steps=1))
    client, _ = _client(tmp_path, runner=runner, config_store=config_store)
    body = client.get("/ui/config").text
    assert "wda" in body
    assert 'value="5"' in body


def test_register_device_prg_redirects_and_persists(tmp_path) -> None:
    config_store = DeviceConfigStore(tmp_path / "device_config.sqlite3")
    client, _ = _client(tmp_path, config_store=config_store)
    response = client.post(
        "/ui/config/devices",
        data={
            "name": "Desk iPhone",
            "driver_type": "wda",
            "server_url": "http://127.0.0.1:4723",
            "udid": "abc123",
            "wda_local_port": "8100",
        },
        follow_redirects=False,
    )
    assert response.status_code == 303
    assert response.headers["location"].endswith("/ui/config")
    assert len(config_store.list()) == 1


def test_register_device_rejects_bad_driver_without_partial_mutation(tmp_path) -> None:
    config_store = DeviceConfigStore(tmp_path / "device_config.sqlite3")
    client, _ = _client(tmp_path, config_store=config_store)
    response = client.post(
        "/ui/config/devices",
        data={"name": "Bad", "driver_type": "android"},
    )
    assert response.status_code == 400
    assert config_store.list() == []
    assert "unsupported driver_type" in response.text


def test_register_device_rejects_non_numeric_port_without_partial_mutation(
    tmp_path,
) -> None:
    config_store = DeviceConfigStore(tmp_path / "device_config.sqlite3")
    client, _ = _client(tmp_path, config_store=config_store)
    response = client.post(
        "/ui/config/devices",
        data={
            "name": "Bad Port",
            "driver_type": "wda",
            "wda_local_port": "not-a-number",
        },
    )
    assert response.status_code == 400
    assert config_store.list() == []
    assert "wda_local_port must be a number" in response.text


def test_remove_device_prg_redirects_and_removes(tmp_path) -> None:
    config_store = DeviceConfigStore(tmp_path / "device_config.sqlite3")
    config_store.add(
        device_id="removable-1",
        name="To Remove",
        driver_type="wda",
        connection_info={"udid": "abc"},
    )
    client, _ = _client(tmp_path, config_store=config_store)
    response = client.post(
        "/ui/config/devices/removable-1/delete",
        follow_redirects=False,
    )
    assert response.status_code == 303
    assert config_store.list() == []


def test_update_max_steps_prg_redirects_and_applies(tmp_path) -> None:
    runner = TaskRunner(config=TaskRunnerConfig(max_steps=1))
    client, _ = _client(tmp_path, runner=runner)
    response = client.post(
        "/ui/config/max-steps",
        data={"max_steps": "25"},
        follow_redirects=False,
    )
    assert response.status_code == 303
    assert runner.config.max_steps == 25


def test_update_max_steps_rejects_non_positive_without_partial_mutation(
    tmp_path,
) -> None:
    config_store = DeviceConfigStore(tmp_path / "device_config.sqlite3")
    config_store.set_setting("max_steps", 10)
    runner = TaskRunner(config=TaskRunnerConfig(max_steps=1))
    client, _ = _client(tmp_path, runner=runner, config_store=config_store)
    response = client.post("/ui/config/max-steps", data={"max_steps": "0"})
    assert response.status_code == 400
    assert runner.config.max_steps == 10
    assert "max_steps must be positive" in response.text


def test_update_max_steps_rejects_non_integer(tmp_path) -> None:
    client, _ = _client(tmp_path)
    response = client.post("/ui/config/max-steps", data={"max_steps": "abc"})
    assert response.status_code == 400
    assert "max_steps must be an integer" in response.text


def test_static_assets_are_served(tmp_path) -> None:
    client, _ = _client(tmp_path)
    assert client.get("/ui/assets/console.css").status_code == 200
    assert client.get("/ui/assets/dashboard.js").status_code == 200


# -- 5.4 Regression: no SPA static dir, no wildcard CORS ---------------------


def test_ui_works_without_runtime_console_static_dir(tmp_path, monkeypatch) -> None:
    monkeypatch.delenv("RUNTIME_CONSOLE_STATIC_DIR", raising=False)
    client, _ = _client(tmp_path)
    assert client.get("/ui/").status_code == 200


def test_runtime_app_does_not_register_wildcard_cors(tmp_path) -> None:
    client, _ = _client(tmp_path)
    # A same-origin browser client must not require CORS preflight. If wildcard
    # CORS were still registered, an explicit Origin header would produce
    # access-control-allow-origin in the response; assert it is absent for an
    # arbitrary same-origin page request.
    response = client.get(
        "/ui/",
        headers={"Origin": "http://127.0.0.1:8000"},
    )
    assert response.status_code == 200
    assert "access-control-allow-origin" not in {k.lower() for k in response.headers}