laily
a15a046e4e
Promote ship to the user-facing delivery skill. Kind is short or long from an explicit mode, then self-judgment, then a question, defaulting to short. Old names remain deprecation stubs.
440 lines
16 KiB
Python
440 lines
16 KiB
Python
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 | None = None) -> dict:
|
|
args = [
|
|
"--project-root",
|
|
str(root),
|
|
"--ship-id",
|
|
ship_id,
|
|
]
|
|
if entry is not None:
|
|
args.extend(["--entry", entry])
|
|
result = _run(NEXT, *args)
|
|
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 ("ship", "pr-ship"):
|
|
text = (REPO_ROOT / "skills" / name / "SKILL.md").read_text(encoding="utf-8")
|
|
self.assertIn("开新 agent", text)
|
|
self.assertNotIn("读取并执行", text)
|
|
|
|
def test_next_step_infers_entry_from_kind(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
_create(root, "add-export-button", "short", stop_at="verified")
|
|
doc = _next(root, "add-export-button")
|
|
self.assertEqual(doc["entry"], "short")
|
|
self.assertEqual(doc["step"], "write_spec")
|
|
|
|
_create(root, "tenant-permissions", "long")
|
|
doc = _next(root, "tenant-permissions")
|
|
self.assertEqual(doc["entry"], "long")
|
|
self.assertEqual(doc["step"], "specifier_new")
|
|
|
|
missing = _run(
|
|
NEXT,
|
|
"--project-root",
|
|
str(root),
|
|
"--ship-id",
|
|
"does-not-exist",
|
|
)
|
|
self.assertEqual(missing.returncode, 0, missing.stderr)
|
|
blocked = json.loads(missing.stdout)
|
|
self.assertEqual(blocked["entry"], "short")
|
|
self.assertEqual(blocked["step"], "ensure_record")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|