Lap Tran
4a2301b077
CI / build (push) Has been cancelled
- Archive 79 completed framework-dev tasks from tasks/ -> tasks/complete/ - status.py: add --cleanup-done and --install-cleanup-schedule commands - Add scripts/automaton-cleanup.sh for periodic task archiving - Dashboard: rename 'Background' tab -> 'Agent', 'Cleanup' agent -> 'Completed Task Archiver', remove redundant group headers and pill badges, dim inactive agent placeholders - .rules.md: add Self-Documenting UI Names rule - New tests: test_cleanup_done.py, expanded test_app.py and test_task.py
320 lines
13 KiB
Python
320 lines
13 KiB
Python
"""Tests for automaton.dashboard.ui.app."""
|
|
|
|
from pathlib import Path
|
|
|
|
import io
|
|
|
|
import pytest
|
|
|
|
from automaton.dashboard.ui.app import DashboardHandler
|
|
|
|
|
|
def test_validate_task_name() -> None:
|
|
assert DashboardHandler._validate_task_name("good-task") is True
|
|
assert DashboardHandler._validate_task_name("bad/../task") is False
|
|
assert DashboardHandler._validate_task_name("bad\\task") is False
|
|
assert DashboardHandler._validate_task_name("") is False
|
|
|
|
|
|
def test_find_tasks_dir(tmp_path: Path) -> None:
|
|
(tmp_path / ".automaton" / "tasks").mkdir(parents=True)
|
|
tasks_dir = DashboardHandler._find_tasks_dir(tmp_path)
|
|
assert tasks_dir == tmp_path / ".automaton" / "tasks"
|
|
|
|
|
|
def test_find_tasks_dir_missing(tmp_path: Path) -> None:
|
|
tasks_dir = DashboardHandler._find_tasks_dir(tmp_path)
|
|
assert tasks_dir == tmp_path / ".automaton" / "tasks"
|
|
|
|
|
|
def test_path_traversal_attempt() -> None:
|
|
"""Task names with path traversal should be rejected."""
|
|
assert DashboardHandler._validate_task_name("../etc/passwd") is False
|
|
assert DashboardHandler._validate_task_name("task%2f..%2fetc") is False
|
|
|
|
|
|
def test_static_path_traversal_symlink(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Static file serving must reject symlinks that resolve outside the html directory."""
|
|
html_dir = tmp_path / "html"
|
|
html_dir.mkdir()
|
|
outside = tmp_path / "secret.txt"
|
|
outside.write_text("secret")
|
|
symlink = html_dir / "link.txt"
|
|
symlink.symlink_to(outside)
|
|
|
|
monkeypatch.setattr(DashboardHandler, "dashboard_path", html_dir)
|
|
|
|
handler = DashboardHandler.__new__(DashboardHandler)
|
|
handler.path = "/link.txt"
|
|
errors: list[tuple[int, str]] = []
|
|
|
|
def capture_error(code: int, message: str) -> None:
|
|
errors.append((code, message))
|
|
|
|
handler._send_error = capture_error
|
|
handler._serve_static()
|
|
assert errors == [(403, "Forbidden")]
|
|
|
|
|
|
def test_static_valid_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Static file serving returns a valid html file."""
|
|
html_dir = tmp_path / "html"
|
|
html_dir.mkdir()
|
|
(html_dir / "index.html").write_text("<html></html>")
|
|
|
|
monkeypatch.setattr(DashboardHandler, "dashboard_path", html_dir)
|
|
|
|
handler = DashboardHandler.__new__(DashboardHandler)
|
|
handler.path = "/"
|
|
|
|
response_status: list[int] = []
|
|
response_headers: list[tuple[str, str]] = []
|
|
|
|
def fake_send_response(code: int) -> None:
|
|
response_status.append(code)
|
|
|
|
def fake_send_header(key: str, value: str) -> None:
|
|
response_headers.append((key, value))
|
|
|
|
handler.send_response = fake_send_response
|
|
handler.send_header = fake_send_header
|
|
handler.end_headers = lambda: None
|
|
handler.wfile = io.BytesIO()
|
|
handler._send_error = lambda code, msg: None
|
|
|
|
handler._serve_static()
|
|
assert response_status == [200]
|
|
assert any(h[0] == "Content-Type" and h[1] == "text/html" for h in response_headers)
|
|
assert handler.wfile.getvalue() == b"<html></html>"
|
|
|
|
|
|
class TestCORSAndSecurityHeaders:
|
|
"""Tests for security headers on API responses (no CORS wildcard)."""
|
|
|
|
def test_send_json_no_cors_wildcard(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
html_dir = tmp_path / "html"
|
|
html_dir.mkdir()
|
|
monkeypatch.setattr(DashboardHandler, "dashboard_path", html_dir)
|
|
|
|
handler = DashboardHandler.__new__(DashboardHandler)
|
|
response_headers: list[tuple[str, str]] = []
|
|
handler.send_response = lambda code: None
|
|
handler.send_header = lambda k, v: response_headers.append((k, v))
|
|
handler.end_headers = lambda: None
|
|
handler.wfile = io.BytesIO()
|
|
|
|
handler._send_json({"test": True})
|
|
header_dict = dict(response_headers)
|
|
assert "Access-Control-Allow-Origin" not in header_dict
|
|
assert header_dict.get("X-Content-Type-Options") == "nosniff"
|
|
|
|
def test_send_error_no_cors_wildcard(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
html_dir = tmp_path / "html"
|
|
html_dir.mkdir()
|
|
monkeypatch.setattr(DashboardHandler, "dashboard_path", html_dir)
|
|
|
|
handler = DashboardHandler.__new__(DashboardHandler)
|
|
response_headers: list[tuple[str, str]] = []
|
|
handler.send_response = lambda code: None
|
|
handler.send_header = lambda k, v: response_headers.append((k, v))
|
|
handler.end_headers = lambda: None
|
|
handler.wfile = io.BytesIO()
|
|
|
|
handler._send_error(404, "Not found")
|
|
header_dict = dict(response_headers)
|
|
assert "Access-Control-Allow-Origin" not in header_dict
|
|
assert header_dict.get("X-Content-Type-Options") == "nosniff"
|
|
|
|
|
|
class TestContentLengthBound:
|
|
"""Tests for POST content-length limits."""
|
|
|
|
def test_max_post_body_constant(self) -> None:
|
|
from automaton.dashboard.ui.app import MAX_POST_BODY, MAX_REVIEW_COMMENT_LENGTH
|
|
assert MAX_POST_BODY == 65536
|
|
assert MAX_REVIEW_COMMENT_LENGTH == 4096
|
|
|
|
|
|
class TestTaskCache:
|
|
"""Tests for server-side task caching."""
|
|
|
|
def test_cache_returns_tasks(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from automaton.dashboard.ui.app import _get_cached_tasks, _task_cache
|
|
tasks_dir = tmp_path / ".automaton" / "tasks"
|
|
task_dir = tasks_dir / "my-task"
|
|
task_dir.mkdir(parents=True)
|
|
(task_dir / "SPEC.md").write_text("# Spec")
|
|
_task_cache["timestamp"] = 0.0
|
|
_task_cache["tasks"] = []
|
|
result = _get_cached_tasks(tmp_path)
|
|
assert len(result) == 1
|
|
assert result[0].name == "my-task"
|
|
|
|
def test_cache_uses_ttl(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import time
|
|
from automaton.dashboard.ui.app import _get_cached_tasks, _task_cache, CACHE_TTL
|
|
tasks_dir = tmp_path / ".automaton" / "tasks"
|
|
task_dir = tasks_dir / "cached-task"
|
|
task_dir.mkdir(parents=True)
|
|
(task_dir / "SPEC.md").write_text("# Spec")
|
|
_task_cache["timestamp"] = 0.0
|
|
_task_cache["tasks"] = []
|
|
result1 = _get_cached_tasks(tmp_path)
|
|
assert len(result1) == 1
|
|
_task_cache["timestamp"] = time.time() + CACHE_TTL + 10
|
|
new_task = tasks_dir / "new-task"
|
|
new_task.mkdir()
|
|
(new_task / "SPEC.md").write_text("# New")
|
|
result2 = _get_cached_tasks(tmp_path)
|
|
assert len(result2) == 1
|
|
_task_cache["timestamp"] = 0.0
|
|
result3 = _get_cached_tasks(tmp_path)
|
|
assert len(result3) == 2
|
|
|
|
def test_invalidate_cache(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import time
|
|
from automaton.dashboard.ui.app import _invalidate_task_cache, _get_cached_tasks, _task_cache
|
|
tasks_dir = tmp_path / ".automaton" / "tasks"
|
|
task_dir = tasks_dir / "inv-task"
|
|
task_dir.mkdir(parents=True)
|
|
(task_dir / "SPEC.md").write_text("# Spec")
|
|
_task_cache["timestamp"] = time.time() + 9999
|
|
_task_cache["tasks"] = []
|
|
_invalidate_task_cache()
|
|
assert _task_cache["timestamp"] == 0.0
|
|
|
|
|
|
class TestConfigEndpoint:
|
|
"""Tests for GET/PUT /api/config."""
|
|
|
|
def _make_handler(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, config=None):
|
|
from automaton.dashboard.ui.app import DashboardHandler
|
|
from automaton.dashboard.config import DashboardConfig
|
|
html_dir = tmp_path / "html"
|
|
html_dir.mkdir()
|
|
monkeypatch.setattr(DashboardHandler, "dashboard_path", html_dir)
|
|
handler = DashboardHandler.__new__(DashboardHandler)
|
|
handler.config = config or DashboardConfig()
|
|
handler.path = "/api/config"
|
|
return handler
|
|
|
|
def test_serve_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from automaton.dashboard.config import DashboardConfig
|
|
handler = self._make_handler(tmp_path, monkeypatch)
|
|
response_data = {}
|
|
handler._send_json = lambda d: response_data.update(d)
|
|
handler._serve_config()
|
|
assert response_data["theme"] == "default"
|
|
assert response_data["auto_refresh_interval"] == 2
|
|
assert response_data["default_view"] == "board"
|
|
|
|
def test_serve_config_not_available(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
handler = self._make_handler(tmp_path, monkeypatch)
|
|
handler.config = None
|
|
errors = []
|
|
handler._send_error = lambda c, m: errors.append((c, m))
|
|
handler._serve_config()
|
|
assert errors[0][0] == 503
|
|
|
|
def test_handle_config_update(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from automaton.dashboard.config import DashboardConfig
|
|
from automaton.dashboard.ui.app import DashboardHandler
|
|
tasks_dir = tmp_path / ".automaton" / "tasks"
|
|
tasks_dir.mkdir(parents=True)
|
|
monkeypatch.setattr("automaton.dashboard.ui.app.get_config_path",
|
|
lambda pr: tmp_path / ".automaton" / "dashboard-config.json")
|
|
handler = self._make_handler(tmp_path, monkeypatch)
|
|
handler.project_root = tmp_path
|
|
handler.headers = {"Content-Length": "19"}
|
|
handler.rfile = io.BytesIO(b'{"theme": "dark"}')
|
|
response_data = {}
|
|
handler._send_json = lambda d: response_data.update(d)
|
|
handler._handle_config_update()
|
|
assert response_data["theme"] == "dark"
|
|
assert handler.config.theme == "dark"
|
|
|
|
def test_handle_config_update_invalid(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from automaton.dashboard.config import DashboardConfig
|
|
handler = self._make_handler(tmp_path, monkeypatch)
|
|
handler.headers = {"Content-Length": "39"}
|
|
handler.rfile = io.BytesIO(b'{"auto_refresh_interval": 999}')
|
|
errors = []
|
|
handler._send_error = lambda c, m: errors.append((c, m))
|
|
handler._handle_config_update()
|
|
assert errors[0][0] == 400
|
|
|
|
|
|
def test_serve_scheduled_returns_jobs(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""`/api/scheduled` returns jobs from `_list_scheduled_jobs`."""
|
|
import automaton.dashboard.ui.app as app_mod
|
|
|
|
class _FakeStatusMod:
|
|
def _list_scheduled_jobs(self=None, project=None):
|
|
return [
|
|
{"label": "com.automaton.cleanup", "kind": "cleanup", "name": "",
|
|
"status": "enabled", "next_run_seconds": 86400,
|
|
"stub_path": "/x/stub.sh", "runtime_state": None},
|
|
{"label": "com.automaton.loop.ci", "kind": "loop", "name": "ci",
|
|
"status": "disabled", "next_run_seconds": 60,
|
|
"stub_path": None, "runtime_state": {"status": "paused"},
|
|
"agent_type": "single"},
|
|
]
|
|
|
|
monkeypatch.setattr(app_mod, "_get_status_module", lambda: _FakeStatusMod())
|
|
handler = DashboardHandler.__new__(DashboardHandler)
|
|
handler.project_root = tmp_path
|
|
response = {}
|
|
handler._send_json = lambda d: response.update(d)
|
|
handler._serve_scheduled()
|
|
assert response["available"] is True
|
|
assert len(response["jobs"]) == 2
|
|
assert response["jobs"][0]["kind"] == "cleanup"
|
|
assert response["jobs"][1]["name"] == "ci"
|
|
assert response["jobs"][1]["agent_type"] == "single"
|
|
|
|
|
|
def test_serve_scheduled_unavailable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""When status.py can't be loaded, `/api/scheduled` reports unavailable."""
|
|
import automaton.dashboard.ui.app as app_mod
|
|
|
|
monkeypatch.setattr(app_mod, "_get_status_module", lambda: None)
|
|
handler = DashboardHandler.__new__(DashboardHandler)
|
|
handler.project_root = tmp_path
|
|
response = {}
|
|
handler._send_json = lambda d: response.update(d)
|
|
handler._serve_scheduled()
|
|
assert response["available"] is False
|
|
assert response["jobs"] == []
|
|
|
|
|
|
def test_serve_scheduled_handles_errors(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Exceptions from `_list_scheduled_jobs` surface as `error` in the response."""
|
|
import automaton.dashboard.ui.app as app_mod
|
|
|
|
class _FakeStatusMod:
|
|
def _list_scheduled_jobs(self=None, project=None):
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr(app_mod, "_get_status_module", lambda: _FakeStatusMod())
|
|
handler = DashboardHandler.__new__(DashboardHandler)
|
|
handler.project_root = tmp_path
|
|
response = {}
|
|
handler._send_json = lambda d: response.update(d)
|
|
handler._serve_scheduled()
|
|
assert response["available"] is True
|
|
assert response["jobs"] == []
|
|
assert "boom" in response["error"]
|
|
|
|
|
|
def test_get_status_module_returns_none_when_missing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""`_get_status_module` returns None when status.py isn't found in ~/.automaton."""
|
|
import automaton.dashboard.ui.app as app_mod
|
|
|
|
# Reset the cached module, and point HOME at an empty dir so
|
|
# `Path.home() / .automaton / scripts / status.py` doesn't exist.
|
|
monkeypatch.setattr(app_mod, "_status_module", None)
|
|
monkeypatch.setenv("HOME", str(tmp_path))
|
|
# pathlib.Path.home() respects $HOME on every platform.
|
|
result = app_mod._get_status_module()
|
|
assert result is None
|