"""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 = "" _ESCAPED_PROBE = "<script>alert(1)</script>" @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, }