"""Tests for W3 DAG execution + retry guard. Acoperă: - topological_eligible: alegere story DAG-aware (passes/failed/blocked propagation) - cmd_incr_retry: 3-retry guard cu auto-fail la max_retries - cmd_mark_failed: propagare blocked la dependenți - _normalize_story: validează schema extinsă (tags, dependsOn, retries, blocked) """ from __future__ import annotations import json import sys from pathlib import Path import pytest PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from tools import ralph_dag # noqa: E402 from tools.ralph_prd_generator import _normalize_story # noqa: E402 # ── topological_eligible ─────────────────────────────────────── def _stories(*specs): """Helper: build minimal stories list. Each spec = (id, priority, dependsOn, flags...)""" out = [] for spec in specs: sid, prio, deps, *rest = spec s = { "id": sid, "priority": prio, "dependsOn": list(deps), "passes": False, "failed": False, "blocked": False, "retries": 0, } for flag in rest: s[flag] = True out.append(s) return out class TestTopologicalEligible: def test_no_deps_lowest_priority_picked(self): stories = _stories( ("US-002", 20, []), ("US-001", 10, []), ) chosen = ralph_dag.topological_eligible(stories) assert chosen["id"] == "US-001" def test_dependent_skipped_until_dep_passes(self): stories = _stories( ("US-001", 10, []), ("US-002", 20, ["US-001"]), ) # US-001 not done yet → US-001 picked assert ralph_dag.topological_eligible(stories)["id"] == "US-001" # Mark US-001 passes → US-002 eligible stories[0]["passes"] = True assert ralph_dag.topological_eligible(stories)["id"] == "US-002" def test_failed_dep_propagates_blocked(self): stories = _stories( ("US-001", 10, []), ("US-002", 20, ["US-001"]), ) stories[0]["failed"] = True chosen = ralph_dag.topological_eligible(stories) assert chosen is None # US-002 marcat blocked, nimic eligibil assert stories[1]["blocked"] is True assert stories[1]["failureReason"] == "blocked_by:US-001" def test_independent_runs_when_other_chain_failed(self): # US-001 failed → US-002 blocked, dar US-003 e independent → eligibil stories = _stories( ("US-001", 10, []), ("US-002", 20, ["US-001"]), ("US-003", 30, []), ) stories[0]["failed"] = True chosen = ralph_dag.topological_eligible(stories) assert chosen["id"] == "US-003" assert stories[1]["blocked"] is True def test_chain_blocking_propagates_transitively(self): # US-001 → US-002 → US-003. US-001 failed → US-002 blocked → US-003 blocked. stories = _stories( ("US-001", 10, []), ("US-002", 20, ["US-001"]), ("US-003", 30, ["US-002"]), ) stories[0]["failed"] = True chosen = ralph_dag.topological_eligible(stories) assert chosen is None assert stories[1]["blocked"] is True assert stories[2]["blocked"] is True def test_all_complete_returns_none(self): stories = _stories(("US-001", 10, [])) stories[0]["passes"] = True assert ralph_dag.topological_eligible(stories) is None def test_already_blocked_story_skipped(self): stories = _stories( ("US-001", 10, []), ("US-002", 20, []), ) stories[0]["blocked"] = True chosen = ralph_dag.topological_eligible(stories) assert chosen["id"] == "US-002" # ── cmd_incr_retry / cmd_mark_failed (file-based) ────────────── @pytest.fixture def prd_path(tmp_path): """Construiește un prd.json minimal pentru test.""" data = { "projectName": "test-proj", "branchName": "ralph/test-proj", "userStories": [ { "id": "US-001", "title": "first", "priority": 10, "dependsOn": [], "tags": [], "acceptanceCriteria": ["a"], "passes": False, "failed": False, "blocked": False, "retries": 0, }, { "id": "US-002", "title": "second", "priority": 20, "dependsOn": ["US-001"], "tags": [], "acceptanceCriteria": ["a"], "passes": False, "failed": False, "blocked": False, "retries": 0, }, ], } p = tmp_path / "prd.json" p.write_text(json.dumps(data), encoding="utf-8") return p class TestRetryGuard: def test_incr_retry_increments_count(self, prd_path): rc = ralph_dag.cmd_incr_retry(prd_path, "US-001") assert rc == 0 data = json.loads(prd_path.read_text()) assert data["userStories"][0]["retries"] == 1 assert data["userStories"][0]["failed"] is False def test_three_retries_marks_failed_max_retries(self, prd_path): # incr 3 times for _ in range(3): ralph_dag.cmd_incr_retry(prd_path, "US-001") data = json.loads(prd_path.read_text()) assert data["userStories"][0]["retries"] == 3 assert data["userStories"][0]["failed"] is True assert data["userStories"][0]["failureReason"] == "max_retries" def test_max_retries_propagates_blocked_to_dependent(self, prd_path): for _ in range(3): ralph_dag.cmd_incr_retry(prd_path, "US-001") data = json.loads(prd_path.read_text()) # US-002 depinde de US-001 → blocked assert data["userStories"][1]["blocked"] is True def test_unknown_story_returns_error(self, prd_path): rc = ralph_dag.cmd_incr_retry(prd_path, "US-999") assert rc == 1 class TestMarkFailed: def test_mark_failed_sets_flags(self, prd_path): rc = ralph_dag.cmd_mark_failed(prd_path, "US-001", "rate_limited") assert rc == 0 data = json.loads(prd_path.read_text()) assert data["userStories"][0]["failed"] is True assert data["userStories"][0]["failureReason"] == "rate_limited" def test_mark_failed_propagates_blocked(self, prd_path): ralph_dag.cmd_mark_failed(prd_path, "US-001", "rate_limited") data = json.loads(prd_path.read_text()) assert data["userStories"][1]["blocked"] is True assert data["userStories"][1]["failureReason"] == "blocked_by:US-001" class TestNextStory: def test_next_story_prints_id(self, prd_path, capsys): rc = ralph_dag.cmd_next_story(prd_path) assert rc == 0 captured = capsys.readouterr() assert captured.out.strip() == "US-001" def test_next_story_returns_1_when_none_eligible(self, prd_path, capsys): # Mark all complete data = json.loads(prd_path.read_text()) for s in data["userStories"]: s["passes"] = True prd_path.write_text(json.dumps(data)) rc = ralph_dag.cmd_next_story(prd_path) assert rc == 1 # ── _normalize_story (PRD generator schema) ──────────────────── class TestNormalizeStory: def test_default_fields_populated(self): s = _normalize_story({"title": "x"}, idx=0) # Schema W3 fields trebuie să existe toate for key in ("id", "title", "description", "priority", "acceptanceCriteria", "tags", "dependsOn", "passes", "failed", "blocked", "retries", "failureReason", "notes"): assert key in s, f"Missing schema field: {key}" assert s["passes"] is False assert s["failed"] is False assert s["blocked"] is False assert s["retries"] == 0 def test_invalid_tags_filtered(self): s = _normalize_story({"title": "x", "tags": ["frontend", "ui", "made-up"]}, idx=0) assert s["tags"] == ["ui"] # frontend & made-up nu sunt în VALID_TAGS def test_empty_acceptance_gets_default(self): s = _normalize_story({"title": "x"}, idx=0) assert len(s["acceptanceCriteria"]) >= 1 def test_ui_tag_implies_browser_check(self): s = _normalize_story({"title": "x", "tags": ["ui"]}, idx=0) assert s["requiresBrowserCheck"] is True def test_explicit_browser_check_preserved(self): s = _normalize_story({"title": "x", "tags": [], "requiresBrowserCheck": True}, idx=0) assert s["requiresBrowserCheck"] is True def test_id_auto_generated_from_idx(self): s = _normalize_story({"title": "x"}, idx=4) assert s["id"] == "US-005" def test_id_preserved_when_provided(self): s = _normalize_story({"id": "US-042", "title": "x"}, idx=0) assert s["id"] == "US-042" def test_depends_on_preserves_strings_filters_garbage(self): s = _normalize_story({"title": "x", "dependsOn": ["US-001", "", None, "US-002"]}, idx=0) assert s["dependsOn"] == ["US-001", "US-002"]