281 lines
9.1 KiB
Python
281 lines
9.1 KiB
Python
"""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,
|
|
source_task_id="cloud-task-visible",
|
|
source_attempt=2,
|
|
)
|
|
|
|
client, _ = _build_client(
|
|
tmp_path, metadata_store=metadata_store, timeline=timeline
|
|
)
|
|
_login(client)
|
|
|
|
response = client.get("/tasks")
|
|
assert response.status_code == 200
|
|
assert "Executed tasks on this Host" in response.text
|
|
assert "task-visible" in response.text
|
|
assert "cloud-task-visible" in response.text
|
|
assert "2" 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
|
|
assert "Before action" in response.text
|
|
assert "After action" in response.text
|
|
|
|
|
|
def test_authenticated_task_detail_renders_complete_step_evidence(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-evidence",
|
|
goal="inspect 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,
|
|
source_task_id="cloud-evidence",
|
|
source_attempt=3,
|
|
)
|
|
timeline.append(
|
|
task_id=task.id,
|
|
scene={"screen": {"width": 10, "height": 20}, "elements": []},
|
|
prompt="inspect the current UI tree",
|
|
tool_call={"action": "get_ui_tree", "description": "inspect UI tree"},
|
|
result={
|
|
"result": [
|
|
{
|
|
"id": "search",
|
|
"type": "button",
|
|
"text": "Search",
|
|
"bounds": {"x": 1, "y": 2, "width": 3, "height": 4},
|
|
"confidence": 0.98,
|
|
}
|
|
]
|
|
},
|
|
before_screenshot=PNG_10X20,
|
|
after_screenshot=PNG_10X20 + b"after",
|
|
ocr_results=[
|
|
{
|
|
"text": "Search",
|
|
"bounds": {"x": 1, "y": 2, "width": 3, "height": 4},
|
|
"confidence": 0.95,
|
|
}
|
|
],
|
|
)
|
|
|
|
client, _ = _build_client(
|
|
tmp_path, metadata_store=metadata_store, timeline=timeline
|
|
)
|
|
_login(client)
|
|
|
|
response = client.get(f"/tasks/{task.id}")
|
|
|
|
assert response.status_code == 200
|
|
assert "cloud-evidence" in response.text
|
|
assert "Cloud attempt" in response.text
|
|
assert "OCR results (1)" in response.text
|
|
assert "UI tree (1 normalized nodes)" in response.text
|
|
assert "inspect the current UI tree" in response.text
|
|
assert response.text.count("data:image/png;base64,") == 2
|
|
|
|
|
|
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 executions recorded yet" in response.text
|