.pouch/tests/test_ack_task_store.py

from __future__ import annotations

import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = REPO_ROOT / "skills" / "ack" / "scripts"
VALIDATOR = SCRIPTS / "validate_tasks.py"
SELECTOR = SCRIPTS / "select_tasks.py"
MIGRATOR = SCRIPTS / "migrate_tasks.py"
RECORDER = SCRIPTS / "record_task.py"
sys.path.insert(0, str(SCRIPTS))

from task_store import materialize_board  # noqa: E402


def manifest() -> dict:
    return {
        "version": 1,
        "ackVersion": "0.20.0",
        "taskStore": {"version": 1, "directory": "task-records"},
        "project": {"name": "demo", "orchestration": {
            "profileVersion": 1,
            "mode": "manual",
            "allowedWorktrees": [],
            "modelAllowlist": {},
            "profiles": {},
            "defaults": {},
        }},
        "tasks": [],
        "workerReceipts": [],
    }


def record(record_id: str, task_id: str, task: dict, parents: list[str] | None = None) -> dict:
    return {
        "recordVersion": 1,
        "recordId": record_id,
        "taskId": task_id,
        "parents": parents or [],
        "kind": "task.snapshot",
        "task": task,
    }


class AckTaskStoreTests(unittest.TestCase):
    def write_board(self, root: Path, records: list[dict]) -> Path:
        board = root / "tasks.yaml"
        board.write_text(json.dumps(manifest()), encoding="utf-8")
        record_dir = root / "task-records"
        record_dir.mkdir()
        for item in records:
            (record_dir / f"{item['recordId']}.yaml").write_text(
                json.dumps(item), encoding="utf-8"
            )
        return board

    def test_materializes_current_task_from_one_head(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            path = self.write_board(
                root,
                [record("EV-1", "T-1", {"id": "T-1", "title": "one", "status": "open"})],
            )
            from validate_tasks import load_document

            board = load_document(path)
            self.assertEqual([task["id"] for task in board["tasks"]], ["T-1"])
            self.assertEqual(board["summary"]["open"], ["T-1"])
            self.assertNotIn("taskStoreErrors", board)

    def test_two_heads_are_reported_instead_of_last_write_wins(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            path = self.write_board(
                root,
                [
                    record("EV-1", "T-1", {"id": "T-1", "title": "one", "status": "open"}),
                    record("EV-2", "T-1", {"id": "T-1", "title": "two", "status": "verified"}, ["EV-1"]),
                    record("EV-3", "T-1", {"id": "T-1", "title": "three", "status": "blocked"}, ["EV-1"]),
                ],
            )
            result = subprocess.run(
                [sys.executable, str(VALIDATOR), str(path)],
                cwd=REPO_ROOT,
                text=True,
                capture_output=True,
                check=False,
            )
            self.assertEqual(result.returncode, 1)
            self.assertIn("未合并 head", result.stderr)

    def test_selector_reads_records_without_changing_cli_shape(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            path = self.write_board(
                root,
                [record("EV-1", "T-1", {"id": "T-1", "title": "one", "status": "open"})],
            )
            result = subprocess.run(
                [sys.executable, str(SELECTOR), str(path)],
                cwd=REPO_ROOT,
                text=True,
                capture_output=True,
                check=False,
            )
            self.assertEqual(result.returncode, 0, result.stderr)
            payload = json.loads(result.stdout)
            self.assertEqual(payload["selection"]["selectedTaskIds"], ["T-1"])

    def test_migrator_dry_run_does_not_modify_legacy_board(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            path = Path(temp) / "tasks.yaml"
            original = {
                "version": 1,
                "project": {"name": "demo"},
                "tasks": [{"id": "T-1", "title": "one", "status": "open"}],
                "workerReceipts": [],
            }
            path.write_text(json.dumps(original), encoding="utf-8")
            result = subprocess.run(
                [sys.executable, str(MIGRATOR), str(path)],
                cwd=REPO_ROOT,
                text=True,
                capture_output=True,
                check=False,
            )
            self.assertEqual(result.returncode, 0, result.stderr)
            self.assertEqual(json.loads(path.read_text()), original)
            self.assertFalse((path.parent / "task-records").exists())

    def test_migrator_apply_keeps_backup_and_materializes_same_board(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            path = Path(temp) / "tasks.yaml"
            path.write_text(json.dumps({
                "version": 1,
                "project": {"name": "demo"},
                "tasks": [{"id": "T-1", "title": "one", "status": "open"}],
                "workerReceipts": [],
            }), encoding="utf-8")
            result = subprocess.run(
                [sys.executable, str(MIGRATOR), str(path), "--apply"],
                cwd=REPO_ROOT,
                text=True,
                capture_output=True,
                check=False,
            )
            self.assertEqual(result.returncode, 0, result.stderr)
            self.assertTrue((path.parent / "tasks.legacy.yaml").is_file())
            self.assertTrue((path.parent / "task-records" / "EV-T-1-initial.yaml").is_file())
            from validate_tasks import load_document

            board = load_document(path)
            self.assertEqual(board["tasks"][0]["title"], "one")

    def test_recorder_appends_exclusive_snapshot(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            path = self.write_board(root, [])
            snapshot = root / "task.json"
            snapshot.write_text(
                json.dumps({"id": "T-1", "title": "one", "status": "open"}),
                encoding="utf-8",
            )
            result = subprocess.run(
                [
                    sys.executable, str(RECORDER), str(path), "--task-id", "T-1",
                    "--task-json", str(snapshot), "--record-id", "EV-1",
                ],
                cwd=REPO_ROOT,
                text=True,
                capture_output=True,
                check=False,
            )
            self.assertEqual(result.returncode, 0, result.stderr)
            self.assertTrue((root / "task-records" / "EV-1.yaml").is_file())
            duplicate = subprocess.run(
                [
                    sys.executable, str(RECORDER), str(path), "--task-id", "T-1",
                    "--task-json", str(snapshot), "--record-id", "EV-1",
                ],
                cwd=REPO_ROOT,
                text=True,
                capture_output=True,
                check=False,
            )
            self.assertEqual(duplicate.returncode, 1)
            self.assertIn("拒绝覆盖", duplicate.stderr)


if __name__ == "__main__":
    unittest.main()