q792602257/agentic-mobile-control
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'worktree-runtime-console-jinja2-templates'

Browse Source 
Server-rendered Jinja2 Runtime console at /ui/, replacing the Vue/Vite SPA.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
q792602257
2026-07-15 08:05:02 +08:00
co-authored by Claude Opus 4.6
parent 0d944ec97d e00c50e703
commit 8300c3b6b7
39 changed files with 1890 additions and 2228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_console_api.py
+31 -6
View File
@@ -80,17 +80,18 @@ def test_console_status_endpoints_cover_empty_and_populated_states(tmp_path) ->
"task-old",
]
assert [
task["id"]
for task in client.get("/console/tasks?device_id=iphone-1").json()
task["id"] for task in client.get("/console/tasks?device_id=iphone-1").json()
] == ["task-old"]
assert [task["id"] for task in client.get("/console/tasks?status=running").json()] == [
"task-new"
]
assert [
task["id"] for task in client.get("/console/tasks?status=running").json()
] == ["task-new"]
assert client.get("/console/tasks/task-old").json()["goal"] == "open settings"
assert client.get("/console/tasks/missing").status_code == 404
def test_console_timeline_inlines_screenshot_and_handles_empty_history(tmp_path) -> None:
def test_console_timeline_inlines_screenshot_and_handles_empty_history(
tmp_path,
) -> None:
timeline = Timeline(ArtifactStore(tmp_path / "history"))
client, metadata_store = _client(tmp_path, timeline=timeline)
task = Task(id="task-1", goal="tap search", device_id="iphone-1")
@@ -193,3 +194,27 @@ def test_console_startup_reloads_persisted_devices_and_settings(tmp_path) -> Non
assert runner.config.max_steps == 31
assert [device.id for device in manager.list_devices()] == ["persisted-1"]
assert client.get("/console/devices").json()[0]["name"] == "Persisted iPhone"
def test_console_json_reflects_page_form_mutations(tmp_path) -> None:
"""A device registered via the /ui/ form must be visible through /console/* JSON."""
config_store = DeviceConfigStore(tmp_path / "device_config.sqlite3")
client, _ = _client(tmp_path, config_store=config_store)
response = client.post(
"/ui/config/devices",
data={
"name": "From Form",
"driver_type": "wda",
"server_url": "http://127.0.0.1:4723",
"udid": "form-udid",
"wda_local_port": "8100",
},
follow_redirects=False,
)
assert response.status_code == 303
json_devices = client.get("/console/devices").json()
assert len(json_devices) == 1
assert json_devices[0]["name"] == "From Form"
assert json_devices[0]["connection_info"]["udid"] == "form-udid"
tests/test_runtime_console_packaging.py
+79
View File
@@ -0,0 +1,79 @@
"""Smoke test that console templates and assets ship inside the Runtime wheel.
Builds ``device-agent-runtime`` into a temporary directory, installs it into an
isolated venv that cannot reach the source checkout, and asserts the packaged
``api`` package carries the Jinja2 templates and static assets needed by
``/ui/``. This guards against setuptools package-data regressions that would
let the console work from an editable checkout but break from a real install.
"""
from __future__ import annotations
import subprocess
import venv
from pathlib import Path
import pytest
def _run(cmd: list[str], *, cwd: Path | None = None) -> str:
return subprocess.check_output(
cmd,
cwd=cwd,
stderr=subprocess.STDOUT,
text=True,
)
@pytest.mark.integration
def test_runtime_wheel_packages_console_templates_and_assets(tmp_path: Path) -> None:
repo_root = Path(__file__).resolve().parent.parent
wheel_dir = tmp_path / "wheels"
wheel_dir.mkdir()
_run(
["uv", "build", "--package", "device-agent-runtime", "--wheel", "--no-sources"],
cwd=repo_root,
)
wheels = list(repo_root.glob("dist/*.whl"))
assert wheels, "uv build did not produce a wheel"
wheel_path = wheels[0]
venv_dir = tmp_path / "venv"
venv.create(venv_dir, with_pip=True, clear=True)
pip = str(venv_dir / "Scripts" / "pip.exe")
if not Path(pip).exists():
pip = str(venv_dir / "bin" / "pip")
_run([pip, "install", str(wheel_path)], cwd=tmp_path)
python = str(venv_dir / "Scripts" / "python.exe")
if not Path(python).exists():
python = str(venv_dir / "bin" / "python")
probe = _run(
[
python,
"-c",
(
"from importlib.resources import files; "
"api_root = files('api'); "
"templates = sorted(p.name for p in "
"(api_root / 'templates' / 'runtime_console').iterdir()); "
"assets = sorted(p.name for p in "
"(api_root / 'static' / 'runtime_console').iterdir()); "
"print(','.join(templates)); "
"print(','.join(assets))"
),
],
cwd=tmp_path,
)
template_names, asset_names = probe.strip().splitlines()
assert "base.html" in template_names
assert "dashboard.html" in template_names
assert "config.html" in template_names
assert "console.css" in asset_names
assert "dashboard.js" in asset_names
# Clean up the build artifact so it does not leak into the working tree.
for wheel in wheels:
wheel.unlink()
tests/test_runtime_console_web.py
+393
View File
@@ -0,0 +1,393 @@
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", "args": {"x": 1, "y": 2}},
result={"ok": True},
screenshot=PNG_10X20,
)
body = client.get("/ui/tasks/task-with-timeline").text
expected_data_uri = "data:image/png;base64," + base64.b64encode(PNG_10X20).decode(
"ascii"
)
assert expected_data_uri in body
assert "tap" 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_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}