#!/usr/bin/env python3 """Unit tests for nightly advance plan builder (no repo side effects). Runs against fixtures in scripts/tests/fixtures/nightly/. Covers: - CURRENT row parsing (skip header/separators, keyed by task id) - plan ordering: integrate_closeout -> feature_advance -> cleanup_verify - done-but-blocked worktrees stay out of the plan - registry fallback when CURRENT has no row """ from __future__ import annotations import json import subprocess import sys import unittest from pathlib import Path from tempfile import TemporaryDirectory from unittest import mock SCRIPTS = Path(__file__).resolve().parent.parent # scripts/ sys.path.insert(0, str(SCRIPTS)) import importlib.util spec = importlib.util.spec_from_file_location( "nightly_plan", SCRIPTS / "nightly" / "plan.py" ) nightly_plan = importlib.util.module_from_spec(spec) sys.modules["nightly_plan"] = nightly_plan spec.loader.exec_module(nightly_plan) FIXTURES = Path(__file__).parent / "fixtures" / "nightly" CURRENT_MD = """# 当前焦点 ## 并行认领 | owner | task | branch | worktree | mutex | status | |---|---|---|---|---|---| | Cursor | feat-a-2026-09-28 | feature/feat-a-2026-09-28 | .worktrees/feat-a-2026-09-28 | m1 | in_progress | | pi | feat-b-2026-09-27 | feature/feat-b-2026-09-27 | .worktrees/feat-b-2026-09-27 | m2 | done | | Codex | feat-c-2026-09-26 | fix/feat-c-2026-09-26 | （已清理） | m3 | done | """ CLEANUP_REPORT = { "registry": "initialized", "status": "ok", "worktrees": [ { "path": "/tmp/repo/.worktrees/integrate-feat-b-2026-09-27", "branch": "feature/integrate-feat-b-2026-09-27", "blockers": ["registry:TASK_ACTIVE"], "notes": ["Close the registered task before removing its worktree"], "tasks": [{"id": "integrate-feat-b-2026-09-27", "owner": "Cursor", "state": "active"}], "verdict": "blocked", }, { "path": "/tmp/repo/.worktrees/feat-a-2026-09-28", "branch": "feature/feat-a-2026-09-28", "blockers": ["registry:TASK_ACTIVE", "dirty_or_ignored_files"], "notes": ["ancestor_of_dev"], "tasks": [{"id": "feat-a-2026-09-28", "owner": "Cursor", "state": "active"}], "verdict": "blocked", }, { "path": "/tmp/repo/.worktrees/feat-d-2026-09-25", "branch": "feature/feat-d-2026-09-25", "blockers": [], "notes": ["ancestor_of_dev", "registry_clear", "lifecycle_lock_idle"], "tasks": [{"id": "feat-d-2026-09-25", "owner": "Cursor", "state": "done"}], "verdict": "ready", }, { "path": "/tmp/repo/.worktrees/feat-done-blocked-2026-09-24", "branch": "feature/feat-done-blocked-2026-09-24", "blockers": ["dirty_or_ignored_files"], "notes": ["ancestor_of_dev"], "tasks": [{"id": "feat-done-blocked-2026-09-24", "owner": "Cursor", "state": "done"}], "verdict": "blocked", }, ], } class ParseCurrentRows(unittest.TestCase): def test_parses_and_skips_non_rows(self): with TemporaryDirectory() as td: (Path(td) / ".doc/ops").mkdir(parents=True) (Path(td) / ".doc/ops/CURRENT.md").write_text(CURRENT_MD, encoding="utf-8") rows = nightly_plan.parse_current_rows(Path(td)) self.assertIn("feat-a-2026-09-28", rows) self.assertEqual(rows["feat-a-2026-09-28"]["status"], "in_progress") self.assertEqual(rows["feat-b-2026-09-27"]["owner"], "pi") self.assertEqual(rows["feat-c-2026-09-26"]["worktree"], "（已清理）") def test_separator_and_header_rejected(self): lines = CURRENT_MD.splitlines() rows = {} for line in lines: cells = [c.strip() for c in line.strip().strip("|").split("|")] if len(cells) < 6 or cells[0] in {"owner", "---"} or set(cells[0]) <= {"-", " "}: continue rows[cells[1]] = cells self.assertNotIn("task", rows) self.assertNotIn("owner", rows) class BuildPlan(unittest.TestCase): def _plan(self, cleanup=CLEANUP_REPORT, current=CURRENT_MD, registry=None): with mock.patch.object(nightly_plan, "_agentctl") as ac, \ mock.patch.object(nightly_plan, "_git") as g: ac.side_effect = lambda repo, *a: ( cleanup if a[:2] == ("cleanup", "report") else (registry if a and a[0] == "status" else {}) ) g.return_value = "sha" with TemporaryDirectory() as td: (Path(td) / ".doc/ops").mkdir(parents=True) (Path(td) / ".doc/ops/CURRENT.md").write_text(current, encoding="utf-8") entries = nightly_plan.build_plan(Path(td)) return entries def test_ordering_and_phases(self): entries = self._plan() phases = [e.phase for e in entries] self.assertEqual( phases, ["integrate_closeout", "feature_advance", "cleanup_verify"], "expected integrate -> feature -> cleanup ordering", ) def test_integrate_not_ancestor_gets_integrate_phase(self): entries = self._plan() integ = [e for e in entries if e.phase == "integrate_closeout"] self.assertEqual(len(integ), 1) self.assertEqual(integ[0].task_id, "integrate-feat-b-2026-09-27") def test_done_ready_worktree_is_cleanup_verify(self): entries = self._plan() clean = [e for e in entries if e.phase == "cleanup_verify"] self.assertEqual(clean[0].task_id, "feat-d-2026-09-25") def test_done_blocked_out_of_plan(self): entries = self._plan() ids = [e.task_id for e in entries] self.assertNotIn("feat-done-blocked-2026-09-24", ids) def test_current_fallback_to_registry_status(self): # feat-d has no CURRENT row here; registry state done + verdict ready entries = self._plan() clean = [e for e in entries if e.phase == "cleanup_verify"] self.assertEqual(clean[0].status, "done") def test_registry_unavailable_still_plans(self): with mock.patch.object(nightly_plan, "registry_tasks", return_value={}): entries = self._plan(registry={}) self.assertTrue(entries) class Cli(unittest.TestCase): def test_max_tasks_cap(self): # Subprocess run: patch nothing, but point --repo at a fixture tree so # _agentctl/_git never touch the real repo; build_plan must still work. with TemporaryDirectory() as td: (Path(td) / ".doc/ops").mkdir(parents=True) (Path(td) / ".doc/ops/CURRENT.md").write_text(CURRENT_MD, encoding="utf-8") out = Path(td) / "plan.json" # No entries can be planned without a real registry; assert rc=0 # and empty plan rather than touching production. rc = subprocess.run( [sys.executable, "-B", str(SCRIPTS / "nightly/plan.py"), "--repo", td, "--out", str(out), "--max-tasks", "1"], capture_output=True, text=True, ).returncode self.assertEqual(rc, 0) data = json.loads(out.read_text()) self.assertEqual(data["count"], 0) if __name__ == "__main__": unittest.main()