143 lines
5.3 KiB
Python
143 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SELECTOR = REPO_ROOT / "skills" / "ack" / "scripts" / "select_tasks.py"
|
|
EXAMPLE = REPO_ROOT / "skills" / "ack" / "examples" / "tasks.example.yaml"
|
|
SPEC = importlib.util.spec_from_file_location("ack_select_tasks", SELECTOR)
|
|
assert SPEC and SPEC.loader
|
|
SELECT_TASKS = importlib.util.module_from_spec(SPEC)
|
|
sys.path.insert(0, str(SELECTOR.parent))
|
|
SPEC.loader.exec_module(SELECT_TASKS)
|
|
|
|
|
|
def board() -> dict:
|
|
return {
|
|
"version": 1,
|
|
"updatedAt": "2026-08-01T20:00:00+08:00",
|
|
"source": "Coordinator",
|
|
"project": {"name": "demo"},
|
|
"summary": {"open": ["T-OPEN"], "verified": ["T-DONE"]},
|
|
"tasks": [
|
|
{"id": "T-OPEN", "title": "open", "status": "open"},
|
|
{"id": "T-DEV", "title": "dev", "status": "fixed_by_dev"},
|
|
{"id": "T-BLOCKED", "title": "blocked", "status": "blocked"},
|
|
{"id": "T-DONE", "title": "done", "status": "verified"},
|
|
],
|
|
}
|
|
|
|
|
|
class AckTaskSelectorTests(unittest.TestCase):
|
|
def run_selector(
|
|
self,
|
|
value: dict,
|
|
*args: str,
|
|
no_site_packages: bool = False,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
path = Path(temp_dir) / "tasks.json"
|
|
path.write_text(json.dumps(value), encoding="utf-8")
|
|
command = [sys.executable]
|
|
if no_site_packages:
|
|
command.append("-S")
|
|
command.extend((str(SELECTOR), str(path), *args))
|
|
return subprocess.run(
|
|
command,
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
def test_default_selects_only_actionable_tasks_in_all_modes(self) -> None:
|
|
for no_site_packages in (False, True):
|
|
with self.subTest(no_site_packages=no_site_packages):
|
|
result = self.run_selector(board(), no_site_packages=no_site_packages)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertEqual(
|
|
payload["selection"]["selectedTaskIds"],
|
|
["T-OPEN", "T-DEV"],
|
|
)
|
|
self.assertEqual(payload["selection"]["totalTasks"], 4)
|
|
self.assertEqual(payload["project"]["name"], "demo")
|
|
self.assertEqual(payload["summary"]["open"], ["T-OPEN"])
|
|
|
|
def test_example_yaml_runs_without_site_packages(self) -> None:
|
|
result = subprocess.run(
|
|
[sys.executable, "-S", str(SELECTOR), str(EXAMPLE), "--compact"],
|
|
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["project"]["name"], "notes-web")
|
|
self.assertEqual(payload["selection"]["mode"], "statuses")
|
|
|
|
def test_explicit_task_id_selects_terminal_task(self) -> None:
|
|
result = self.run_selector(board(), "--task-id", "T-DONE")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertEqual(payload["selection"]["mode"], "task_ids")
|
|
self.assertEqual(payload["selection"]["selectedTaskIds"], ["T-DONE"])
|
|
self.assertEqual([task["id"] for task in payload["tasks"]], ["T-DONE"])
|
|
|
|
def test_missing_task_and_over_budget_fail_closed(self) -> None:
|
|
missing = self.run_selector(board(), "--task-id", "T-MISSING")
|
|
self.assertEqual(missing.returncode, 1)
|
|
self.assertIn("找不到任务", missing.stderr)
|
|
|
|
over_budget = self.run_selector(board(), "--limit", "1")
|
|
self.assertEqual(over_budget.returncode, 1)
|
|
self.assertIn("超过 --limit=1", over_budget.stderr)
|
|
|
|
def test_invalid_board_is_rejected_before_selection(self) -> None:
|
|
invalid = board()
|
|
invalid["tasks"].append(
|
|
{"id": "T-OPEN", "title": "duplicate", "status": "open"}
|
|
)
|
|
result = self.run_selector(invalid)
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertIn("任务板无效", result.stderr)
|
|
self.assertIn("id 重复", result.stderr)
|
|
|
|
def test_receipts_and_delivery_runs_are_reduced_to_selected_tasks(self) -> None:
|
|
selected = [
|
|
{
|
|
"id": "T-1",
|
|
"dispatch": {
|
|
"developer": {"receiptId": "WR-1"},
|
|
"test": {"receiptId": None},
|
|
},
|
|
}
|
|
]
|
|
data = {
|
|
"workerReceipts": [{"id": "WR-1"}, {"id": "WR-2"}],
|
|
"deliveryRuns": [
|
|
{"id": "DR-1", "taskIds": ["T-1"]},
|
|
{"id": "DR-2", "taskIds": ["T-2"]},
|
|
],
|
|
}
|
|
self.assertEqual(
|
|
SELECT_TASKS.select_referenced_receipts(data, selected),
|
|
[{"id": "WR-1"}],
|
|
)
|
|
self.assertEqual(
|
|
SELECT_TASKS.select_delivery_runs(data, selected),
|
|
[{"id": "DR-1", "taskIds": ["T-1"]}],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|