from __future__ import annotations import json import subprocess import sys import tempfile import unittest from pathlib import Path from pouch.yaml_io import safe_dump, safe_load REPO_ROOT = Path(__file__).resolve().parents[1] CREATE = REPO_ROOT / "skills" / "ship" / "scripts" / "create_ship.py" VALIDATE = REPO_ROOT / "skills" / "ship" / "scripts" / "validate_ship.py" APPEND = REPO_ROOT / "skills" / "ship" / "scripts" / "append_evidence.py" NEXT = REPO_ROOT / "skills" / "ship" / "scripts" / "next_step.py" FILLED_SPEC = """# 导出按钮 ## 要做什么 列表页给登录用户一个导出按钮。 ## 不做什么 不改权限模型。 ## 可观察验收 - 登录后列表页可见导出按钮 ## 怎么验 - 浏览器打开 /list 看到按钮 """ FAULT_SPEC = FILLED_SPEC + """ ## 故障证据 线上列表页没有导出按钮。复现：登录后打开 /list。 """ def _run(script: Path, *args: str, cwd: Path | None = None) -> subprocess.CompletedProcess[str]: return subprocess.run( [sys.executable, str(script), *args], check=False, capture_output=True, text=True, cwd=cwd, ) class YamlEmptyCollectionsTests(unittest.TestCase): def test_roundtrip_empty_list_and_dict(self) -> None: from pouch.yaml_io import safe_dump, safe_load data = {"items": [], "meta": {}} loaded = safe_load(safe_dump(data)) self.assertEqual(loaded, data) class CreateAndValidateTests(unittest.TestCase): def test_create_short_ship_validates(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) result = _run( CREATE, "--project-root", str(root), "--ship-id", "add-export-button", "--kind", "short", "--title", "列表页增加导出按钮", ) self.assertEqual(result.returncode, 0, result.stderr) ship_dir = root / ".pouch" / "ship" / "add-export-button" self.assertTrue((ship_dir / "current.yaml").is_file()) self.assertTrue((ship_dir / "spec.md").is_file()) self.assertFalse((ship_dir / "spec").is_dir()) check = _run(VALIDATE, str(ship_dir)) self.assertEqual(check.returncode, 0, check.stderr) def test_create_long_ship_has_spec_dir(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) result = _run( CREATE, "--project-root", str(root), "--ship-id", "tenant-permissions", "--kind", "long", "--title", "多租户权限", ) self.assertEqual(result.returncode, 0, result.stderr) ship_dir = root / ".pouch" / "ship" / "tenant-permissions" self.assertTrue((ship_dir / "spec").is_dir()) check = _run(VALIDATE, str(ship_dir)) self.assertEqual(check.returncode, 0, check.stderr) def test_rejects_date_id(self) -> None: with tempfile.TemporaryDirectory() as tmp: result = _run( CREATE, "--project-root", tmp, "--ship-id", "20261013", "--kind", "short", "--title", "nope", ) self.assertNotEqual(result.returncode, 0) self.assertIn("日期", result.stderr) def test_append_test_run_pointer(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) created = _run( CREATE, "--project-root", str(root), "--ship-id", "add-export-button", "--kind", "short", "--title", "列表页增加导出按钮", ) self.assertEqual(created.returncode, 0, created.stderr) run_dir = root / ".pouch" / "tester" / "runs" run_dir.mkdir(parents=True) run_path = run_dir / "20261013T010000Z-smoke-ab12.yaml" run_path.write_text("runId: demo\n", encoding="utf-8") ship_dir = root / ".pouch" / "ship" / "add-export-button" appended = _run( APPEND, str(ship_dir), "--kind", "testRun", "--path", ".pouch/tester/runs/20261013T010000Z-smoke-ab12.yaml", "--result", "pass", "--suite", "smoke", ) self.assertEqual(appended.returncode, 0, appended.stderr) text = (ship_dir / "current.yaml").read_text(encoding="utf-8") self.assertIn("20261013T010000Z-smoke-ab12.yaml", text) def test_missing_pointer_file_fails(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) _run( CREATE, "--project-root", str(root), "--ship-id", "add-export-button", "--kind", "short", "--title", "列表页增加导出按钮", ) ship_dir = root / ".pouch" / "ship" / "add-export-button" appended = _run( APPEND, str(ship_dir), "--kind", "testRun", "--path", ".pouch/tester/runs/missing.yaml", "--result", "pass", ) self.assertNotEqual(appended.returncode, 0) self.assertIn("不存在", appended.stderr) def _create(root: Path, ship_id: str, kind: str, *, stop_at: str | None = None) -> Path: args = [ "--project-root", str(root), "--ship-id", ship_id, "--kind", kind, "--title", "列表页增加导出按钮", ] if stop_at: args.extend(["--stop-at", stop_at]) result = _run(CREATE, *args) if result.returncode != 0: raise AssertionError(result.stderr) return root / ".pouch" / "ship" / ship_id def _next(root: Path, ship_id: str, entry: str) -> dict: result = _run( NEXT, "--project-root", str(root), "--ship-id", ship_id, "--entry", entry, ) if result.returncode != 0: raise AssertionError(result.stderr or result.stdout) return json.loads(result.stdout) def _patch_current(ship_dir: Path, **fields) -> None: path = ship_dir / "current.yaml" data = safe_load(path.read_text(encoding="utf-8")) for key, value in fields.items(): if key == "delivery_state": data.setdefault("delivery", {})["state"] = value elif key == "manager": data["manager"] = value elif key == "source": data["source"] = value elif key == "spec": data["spec"].update(value) else: data[key] = value path.write_text(safe_dump(data), encoding="utf-8") def _fill_task(ship_dir: Path) -> None: path = ship_dir / "tasks.yaml" data = safe_load(path.read_text(encoding="utf-8")) data["tasks"][0]["expected"] = ["登录用户看到导出"] data["tasks"][0]["verification"] = ["打开 /list"] path.write_text(safe_dump(data), encoding="utf-8") class NextStepTests(unittest.TestCase): def _prepare_delivery_dispatch( self, root: Path, ship_id: str, stop_at: str, state: str ) -> None: ship_dir = _create(root, ship_id, "short", stop_at=stop_at) (ship_dir / "spec.md").write_text(FILLED_SPEC, encoding="utf-8") _fill_task(ship_dir) _patch_current(ship_dir, delivery_state=state) data = safe_load((ship_dir / "current.yaml").read_text(encoding="utf-8")) data["evidence"]["commits"] = [{"sha": "abc1234deadbeef", "role": "head"}] (ship_dir / "current.yaml").write_text(safe_dump(data), encoding="utf-8") def test_stop_at_mappings_cover_each_delivery_boundary(self) -> None: cases = ( ("develop_verified", "change_ready", "verify", "verified"), ("review_ready", "verified", "open_pr", "pr_opened"), ("release_ready", "merged", "regress", "regression_passed"), ("released", "artifact_built", "publish", "published"), ("deployed", "published", "deploy", "deployed"), ) for stop_at, state, step, target in cases: with self.subTest(stop_at=stop_at): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) ship_id = f"ship-{stop_at.replace('_', '-')}" self._prepare_delivery_dispatch(root, ship_id, stop_at, state) doc = _next(root, ship_id, "short") self.assertEqual(doc["step"], step) self.assertEqual(doc["targetState"], target) def test_legacy_stop_at_names_resolve_to_new_boundaries(self) -> None: cases = ( ("verified", "change_ready", "verify", "verified"), ("validation_ready", "merged", "regress", "regression_passed"), ) for stop_at, state, step, target in cases: with self.subTest(stop_at=stop_at): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) ship_id = f"legacy-{stop_at.replace('_', '-')}" self._prepare_delivery_dispatch(root, ship_id, stop_at, state) doc = _next(root, ship_id, "short") self.assertEqual(doc["step"], step) self.assertEqual(doc["targetState"], target) def test_invalid_stop_at_fails_ship_validation(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) result = _run( CREATE, "--project-root", str(root), "--ship-id", "invalid-stop-at", "--kind", "short", "--title", "无效停止点", "--stop-at", "not_a_stop_point", ) self.assertNotEqual(result.returncode, 0) self.assertIn("stopAt 无效", result.stderr) def test_new_short_ship_writes_spec(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) _create(root, "add-export-button", "short") doc = _next(root, "add-export-button", "short") self.assertEqual(doc["step"], "write_spec") self.assertEqual(doc["actor"], "manager") def test_filled_short_without_stop_asks_user(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) ship_dir = _create(root, "add-export-button", "short") (ship_dir / "spec.md").write_text(FILLED_SPEC, encoding="utf-8") _fill_task(ship_dir) doc = _next(root, "add-export-button", "short") self.assertEqual(doc["status"], "ask_user") def test_short_implement_then_verify(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) ship_dir = _create(root, "add-export-button", "short", stop_at="verified") (ship_dir / "spec.md").write_text(FILLED_SPEC, encoding="utf-8") _fill_task(ship_dir) doc = _next(root, "add-export-button", "short") self.assertEqual(doc["step"], "implement") self.assertEqual(doc["actor"], "worker") self.assertIsNone(doc["skill"]) _patch_current( ship_dir, manager={"lastStep": "implement", "lastResult": "pass", "inFlight": None}, ) data = safe_load((ship_dir / "current.yaml").read_text(encoding="utf-8")) data["evidence"]["commits"] = [{"sha": "abc1234deadbeef", "role": "head"}] (ship_dir / "current.yaml").write_text(safe_dump(data), encoding="utf-8") doc = _next(root, "add-export-button", "short") self.assertEqual(doc["step"], "verify") self.assertEqual(doc["skill"], "tester") _patch_current(ship_dir, delivery_state="verified") doc = _next(root, "add-export-button", "short") self.assertEqual(doc["status"], "done") def test_fault_spec_dispatches_hunt(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) ship_dir = _create(root, "add-export-button", "short", stop_at="verified") (ship_dir / "spec.md").write_text(FAULT_SPEC, encoding="utf-8") _fill_task(ship_dir) doc = _next(root, "add-export-button", "short") self.assertEqual(doc["step"], "hunt") self.assertEqual(doc["skill"], "hunt") def test_long_starts_with_specifier_worker(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) _create(root, "tenant-permissions", "long") doc = _next(root, "tenant-permissions", "long") self.assertEqual(doc["step"], "specifier_new") self.assertEqual(doc["skill"], "specifier") self.assertEqual(doc["actor"], "worker") def test_pr_check_then_merge(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) ship_dir = _create(root, "pr-export", "short", stop_at="released") _patch_current( ship_dir, source={"kind": "pr", "ref": "https://git.example.com/app/-/merge_requests/12"}, ) data = safe_load((ship_dir / "current.yaml").read_text(encoding="utf-8")) data["evidence"]["prs"] = [ {"url": "https://git.example.com/app/-/merge_requests/12", "status": "opened"} ] (ship_dir / "current.yaml").write_text(safe_dump(data), encoding="utf-8") doc = _next(root, "pr-export", "pr") self.assertEqual(doc["step"], "check_pr") self.assertEqual(doc["skill"], "check") _patch_current( ship_dir, manager={"lastStep": "check_pr", "lastResult": "pass", "inFlight": None}, ) doc = _next(root, "pr-export", "pr") self.assertEqual(doc["step"], "merge_pr") self.assertEqual(doc["skill"], "delivery-loop") def test_failed_worker_blocks_manager(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) ship_dir = _create(root, "add-export-button", "short", stop_at="verified") (ship_dir / "spec.md").write_text(FILLED_SPEC, encoding="utf-8") _fill_task(ship_dir) _patch_current( ship_dir, manager={"lastStep": "implement", "lastResult": "fail", "inFlight": None}, ) doc = _next(root, "add-export-button", "short") self.assertEqual(doc["status"], "blocked") self.assertIn("不要自己做", doc["summary"]) def test_entry_skills_are_managers(self) -> None: for name in ("short-ship", "long-ship", "pr-ship"): text = (REPO_ROOT / "skills" / name / "SKILL.md").read_text(encoding="utf-8") self.assertIn("开新 agent", text) self.assertNotIn("读取并执行", text) if __name__ == "__main__": unittest.main()