"""Tests for /api/ralph/* endpoints (dashboard live). Acoperă: - /api/ralph/status: list cards cu state + count + fetchedAt - /api/ralph//log: tail progress.txt - /api/ralph//prd: full prd.json - _ralph_validate_slug: path traversal protection - corrupt prd.json: graceful error (status='error', not 500) """ from __future__ import annotations import json import sys from pathlib import Path import pytest PROJECT_ROOT = Path(__file__).resolve().parents[1] DASH = PROJECT_ROOT / "dashboard" if str(DASH) not in sys.path: sys.path.insert(0, str(DASH)) @pytest.fixture(scope="module") def ralph_module(): from handlers import ralph as _r # type: ignore return _r @pytest.fixture def handler(ralph_module, tmp_path, monkeypatch): """Build a stubbed handler with a temp WORKSPACE_DIR.""" import constants # type: ignore # Re-route WORKSPACE_DIR la tmp pentru izolare monkeypatch.setattr(constants, "WORKSPACE_DIR", tmp_path) class _Stub(ralph_module.RalphHandlers): def __init__(self): self.captured = None self.captured_code = None self.path = "/api/ralph/status" def send_json(self, data, code=200): self.captured = data self.captured_code = code def send_error(self, code): self.captured = {"error_code": code} self.captured_code = code return _Stub() def _make_ralph_project(workspace: Path, slug: str, stories: list, progress: str = "init"): """Create a fake ralph project under workspace//scripts/ralph/.""" ralph_dir = workspace / slug / "scripts" / "ralph" ralph_dir.mkdir(parents=True, exist_ok=True) (ralph_dir / "prd.json").write_text(json.dumps({ "projectName": slug, "branchName": f"ralph/{slug}", "userStories": stories, }), encoding="utf-8") (ralph_dir / "progress.txt").write_text(progress, encoding="utf-8") return ralph_dir # ── /api/ralph/status ────────────────────────────────────────── class TestStatus: def test_empty_workspace_returns_empty(self, handler): handler.handle_ralph_status() assert handler.captured_code == 200 assert handler.captured["projects"] == [] assert "fetchedAt" in handler.captured def test_status_skips_non_ralph_projects(self, handler, tmp_path): # Create a project WITHOUT scripts/ralph (tmp_path / "regular-proj").mkdir() handler.handle_ralph_status() assert handler.captured["projects"] == [] def test_status_lists_ralph_projects(self, handler, tmp_path): _make_ralph_project(tmp_path, "proj-a", [ {"id": "US-001", "title": "a", "priority": 10, "passes": True, "failed": False, "blocked": False, "retries": 0, "tags": []}, {"id": "US-002", "title": "b", "priority": 20, "passes": False, "failed": False, "blocked": False, "retries": 0, "tags": ["ui"]}, ]) handler.handle_ralph_status() projects = handler.captured["projects"] assert len(projects) == 1 p = projects[0] assert p["slug"] == "proj-a" assert p["storiesTotal"] == 2 assert p["storiesComplete"] == 1 assert p["storiesFailed"] == 0 assert p["storiesBlocked"] == 0 assert p["status"] == "idle" # not running (no .ralph.pid) def test_status_corrupt_prd_returns_error_not_500(self, handler, tmp_path): ralph_dir = tmp_path / "broken" / "scripts" / "ralph" ralph_dir.mkdir(parents=True) (ralph_dir / "prd.json").write_text("{not valid json", encoding="utf-8") handler.handle_ralph_status() assert handler.captured_code == 200 assert any(p.get("status") == "error" for p in handler.captured["projects"]) def test_status_count_matches_projects(self, handler, tmp_path): _make_ralph_project(tmp_path, "p1", []) _make_ralph_project(tmp_path, "p2", []) handler.handle_ralph_status() assert handler.captured["count"] == 2 def test_complete_status_when_all_pass(self, handler, tmp_path): _make_ralph_project(tmp_path, "donezo", [ {"id": "US-001", "passes": True, "failed": False, "blocked": False, "retries": 0, "tags": [], "title": "x", "priority": 10}, ]) handler.handle_ralph_status() p = handler.captured["projects"][0] assert p["status"] == "complete" def test_failed_status_propagation(self, handler, tmp_path): _make_ralph_project(tmp_path, "broken-proj", [ {"id": "US-001", "passes": False, "failed": True, "blocked": False, "retries": 3, "tags": [], "title": "x", "priority": 10, "failureReason": "max_retries"}, ]) handler.handle_ralph_status() p = handler.captured["projects"][0] assert p["status"] == "failed" assert p["storiesFailed"] == 1 # ── /api/ralph//log ────────────────────────────────────── class TestLog: def test_log_returns_progress_lines(self, handler, tmp_path): _make_ralph_project(tmp_path, "p1", [], progress="line1\nline2\nline3") handler.path = "/api/ralph/p1/log" handler.handle_ralph_log("p1") assert handler.captured_code == 200 assert handler.captured["lines"] == ["line1", "line2", "line3"] assert handler.captured["total"] == 3 def test_log_invalid_slug_400(self, handler): handler.handle_ralph_log("../etc/passwd") assert handler.captured_code == 400 def test_log_path_traversal_blocked(self, handler): handler.handle_ralph_log("..") assert handler.captured_code == 400 def test_log_missing_progress_returns_empty(self, handler, tmp_path): ralph_dir = tmp_path / "noprogress" / "scripts" / "ralph" ralph_dir.mkdir(parents=True) (ralph_dir / "prd.json").write_text("{}") # no progress.txt handler.path = "/api/ralph/noprogress/log" handler.handle_ralph_log("noprogress") assert handler.captured_code == 200 assert handler.captured["lines"] == [] # ── /api/ralph//prd ────────────────────────────────────── class TestPrd: def test_prd_returns_full_json(self, handler, tmp_path): stories = [{"id": "US-001", "passes": False, "title": "t", "priority": 10}] _make_ralph_project(tmp_path, "p1", stories) handler.handle_ralph_prd("p1") assert handler.captured_code == 200 assert handler.captured["projectName"] == "p1" assert len(handler.captured["userStories"]) == 1 def test_prd_404_when_missing(self, handler, tmp_path): (tmp_path / "ghost").mkdir() # exists, but no prd.json handler.handle_ralph_prd("ghost") assert handler.captured_code == 404 def test_prd_invalid_slug_400(self, handler): handler.handle_ralph_prd("/etc/passwd") assert handler.captured_code == 400 # ── /api/ralph/usage ──────────────────────────────────────────── class TestUsageEndpoint: def test_usage_empty_workspace(self, handler): handler.path = "/api/ralph/usage" handler.handle_ralph_usage() assert handler.captured_code == 200 assert handler.captured["today_runs"] == 0 assert handler.captured["total_runs"] == 0 assert handler.captured["by_project"] == {} def test_usage_aggregates_across_projects(self, handler, tmp_path): # Create two projects, each with usage.jsonl for slug, cost, ts in [("proj-a", 0.5, "2026-04-26T10:00:00+00:00"), ("proj-b", 0.3, "2026-04-26T11:00:00+00:00")]: ralph_dir = tmp_path / slug / "scripts" / "ralph" ralph_dir.mkdir(parents=True) (ralph_dir / "usage.jsonl").write_text( json.dumps({"slug": slug, "ts": ts, "total_cost_usd": cost, "input_tokens": 100, "output_tokens": 50, "cache_read": 0}) + "\n", encoding="utf-8", ) handler.path = "/api/ralph/usage?days=30" handler.handle_ralph_usage() assert handler.captured_code == 200 # Should have both projects assert "proj-a" in handler.captured["by_project"] assert "proj-b" in handler.captured["by_project"] assert handler.captured["total_runs"] == 2 assert handler.captured["window_runs"] == 2 def test_usage_invalid_days_falls_back(self, handler): handler.path = "/api/ralph/usage?days=abc" handler.handle_ralph_usage() assert handler.captured_code == 200 assert handler.captured["window_days"] == 7 def test_usage_handles_corrupt_jsonl(self, handler, tmp_path): # Project with corrupt usage.jsonl shouldn't 500 ralph_dir = tmp_path / "broken" / "scripts" / "ralph" ralph_dir.mkdir(parents=True) (ralph_dir / "usage.jsonl").write_text("not json\n", encoding="utf-8") handler.path = "/api/ralph/usage" handler.handle_ralph_usage() assert handler.captured_code == 200 # ── _ralph_validate_slug ─────────────────────────────────────── class TestValidateSlug: def test_valid_slug_returns_path(self, handler, tmp_path): (tmp_path / "good-slug").mkdir() result = handler._ralph_validate_slug("good-slug") assert result is not None assert result.name == "good-slug" def test_slash_rejected(self, handler): assert handler._ralph_validate_slug("a/b") is None def test_dotdot_rejected(self, handler): assert handler._ralph_validate_slug("..") is None def test_empty_rejected(self, handler): assert handler._ralph_validate_slug("") is None def test_nonexistent_returns_none(self, handler): assert handler._ralph_validate_slug("does-not-exist") is None