518 lines
20 KiB
Python
518 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Regression tests for the Todo Skill Golden Routes and deterministic scripts."""
|
|
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
sys.dont_write_bytecode = True
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
TODO_ROOT = ROOT / "skills" / "multi" / "dingtalk-todo"
|
|
|
|
|
|
def load_script(filename):
|
|
path = TODO_ROOT / "scripts" / filename
|
|
spec = importlib.util.spec_from_file_location(f"todo_test_{path.stem}", path)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"cannot load {path}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
DAILY = load_script("todo_daily_summary.py")
|
|
OVERDUE = load_script("todo_overdue_check.py")
|
|
BATCH = load_script("todo_batch_create.py")
|
|
|
|
|
|
class TodoSkillAlignmentTest(unittest.TestCase):
|
|
def test_golden_routes_prefer_verified_shortcuts(self):
|
|
skill = (TODO_ROOT / "SKILL.md").read_text(encoding="utf-8")
|
|
for route in (
|
|
"todo +remind",
|
|
"todo +assign",
|
|
"todo +create",
|
|
"todo +get-my-tasks",
|
|
"todo +search",
|
|
"todo +complete",
|
|
"todo +update",
|
|
"todo +reminder",
|
|
):
|
|
with self.subTest(route=route):
|
|
self.assertIn(route, skill)
|
|
self.assertIn("## Golden Routes", skill)
|
|
self.assertIn("只有当前 leaf 的 flag 或安全语义确实不明时才查精确 leaf", skill)
|
|
self.assertLessEqual(len(skill.encode("utf-8")), 16000)
|
|
|
|
def test_composite_lifecycle_starts_with_atomic_create(self):
|
|
skill = (TODO_ROOT / "SKILL.md").read_text(encoding="utf-8")
|
|
lifecycle = (TODO_ROOT / "references" / "02-task.md").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("组合生命周期从原子创建开始", skill)
|
|
self.assertIn("不要用创建 Shortcut 代替第一步", skill)
|
|
self.assertIn("dws todo task create --title", lifecycle)
|
|
self.assertIn("dws contact user get-self --format json", lifecycle)
|
|
self.assertIn(
|
|
'task create-sub --parent-id <PARENT_ID> --title "<标题>" --executors <USER_ID>',
|
|
lifecycle,
|
|
)
|
|
self.assertNotIn("dws contact me --format json", lifecycle)
|
|
|
|
def test_step_routing_keeps_shortcuts_and_dynamic_id_boundaries(self):
|
|
lifecycle = (TODO_ROOT / "references" / "02-task.md").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
for route in (
|
|
"+get-my-tasks",
|
|
"+get-related-tasks",
|
|
"+due-today",
|
|
"+overdue",
|
|
"+search",
|
|
"+get",
|
|
"+complete",
|
|
"+reopen",
|
|
"+update",
|
|
"+comment",
|
|
"+reminder",
|
|
"+list-sub",
|
|
"+list-attachment",
|
|
):
|
|
with self.subTest(route=route):
|
|
self.assertIn(route, lifecycle)
|
|
for stable_id in (
|
|
"taskId",
|
|
"commentId",
|
|
"attachmentId",
|
|
"tagCode",
|
|
"userId",
|
|
):
|
|
with self.subTest(stable_id=stable_id):
|
|
self.assertIn(stable_id, lifecycle)
|
|
self.assertIn("禁止运行 `git tag`", lifecycle)
|
|
|
|
def test_golden_route_table_keeps_exactly_three_columns(self):
|
|
skill = (TODO_ROOT / "SKILL.md").read_text(encoding="utf-8")
|
|
table = skill.split("## Golden Routes", 1)[1].split("## 低频原子能力", 1)[0]
|
|
for row in (line for line in table.splitlines() if line.startswith("|")):
|
|
with self.subTest(row=row):
|
|
self.assertEqual(5, len(re.split(r"(?<!\\)\|", row)), row)
|
|
|
|
def test_all_markdown_tables_keep_consistent_column_counts(self):
|
|
for document in TODO_ROOT.rglob("*.md"):
|
|
expected_columns = None
|
|
for line_number, line in enumerate(
|
|
document.read_text(encoding="utf-8").splitlines(), start=1
|
|
):
|
|
if line.startswith("|") and line.endswith("|"):
|
|
columns = len(re.split(r"(?<!\\)\|", line))
|
|
if expected_columns is None:
|
|
expected_columns = columns
|
|
with self.subTest(document=document.name, line=line_number):
|
|
self.assertEqual(expected_columns, columns, line)
|
|
else:
|
|
expected_columns = None
|
|
|
|
def test_references_are_todo_only_and_reminder_contract_is_consistent(self):
|
|
combined = "\n".join(
|
|
path.read_text(encoding="utf-8")
|
|
for path in (TODO_ROOT / "references").rglob("*.md")
|
|
)
|
|
self.assertNotIn("## #7 听记与会后", combined)
|
|
self.assertNotIn("minutes list all", combined)
|
|
self.assertNotIn("当前不支持单独的 `reminder`", combined)
|
|
self.assertIn("提醒写入目前没有对应的查询接口", combined)
|
|
|
|
def test_markdown_links_resolve(self):
|
|
missing = []
|
|
pattern = re.compile(r"\[[^\]]+\]\(([^)#]+)(?:#[^)]+)?\)")
|
|
for document in TODO_ROOT.rglob("*.md"):
|
|
for target in pattern.findall(document.read_text(encoding="utf-8")):
|
|
if "://" not in target and not (document.parent / target).resolve().exists():
|
|
missing.append(f"{document.relative_to(ROOT)} -> {target}")
|
|
self.assertEqual([], missing)
|
|
|
|
|
|
class TodoDailySummaryTest(unittest.TestCase):
|
|
def test_uses_bounded_shortcut_and_excludes_missing_due(self):
|
|
start, _ = DAILY.date_range("today")
|
|
inside = int((start + timedelta(hours=9)).timestamp() * 1000)
|
|
outside = int((start + timedelta(days=2)).timestamp() * 1000)
|
|
calls = []
|
|
|
|
def fake_run(args, dws="dws"):
|
|
calls.append(args)
|
|
return {
|
|
"ok": True,
|
|
"outcome": "success",
|
|
"data": {
|
|
"complete": True,
|
|
"count": 3,
|
|
"todos": [
|
|
{"taskId": "in", "subject": "inside", "dueTime": inside},
|
|
{"taskId": "none", "subject": "no due"},
|
|
{"taskId": "out", "subject": "outside", "dueTime": outside},
|
|
],
|
|
},
|
|
}
|
|
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(DAILY, "run_dws_json", side_effect=fake_run):
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = DAILY.run(["today"])
|
|
self.assertEqual(0, code)
|
|
self.assertEqual(1, len(calls))
|
|
self.assertEqual(["todo", "+get-my-tasks"], calls[0][:2])
|
|
self.assertIn("--all", calls[0])
|
|
self.assertIn("--plan-finish-start", calls[0])
|
|
self.assertEqual(
|
|
["in"], [item["taskId"] for item in json.loads(stdout.getvalue())["todos"]]
|
|
)
|
|
|
|
def test_incomplete_traversal_fails_closed(self):
|
|
payload = {"ok": True, "data": {"complete": False, "todos": []}}
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(DAILY, "run_dws_json", return_value=payload):
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = DAILY.run(["today"])
|
|
self.assertEqual(2, code)
|
|
self.assertFalse(json.loads(stdout.getvalue())["complete"])
|
|
|
|
|
|
class TodoOverdueTest(unittest.TestCase):
|
|
def test_uses_overdue_shortcut_and_empty_is_success(self):
|
|
calls = []
|
|
|
|
def fake_run(args, dws="dws"):
|
|
calls.append(args)
|
|
return {"ok": True, "outcome": "success", "data": {"overdue": []}}
|
|
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(OVERDUE, "run_dws_json", side_effect=fake_run):
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = OVERDUE.run([])
|
|
self.assertEqual(0, code)
|
|
self.assertEqual(["todo", "+overdue", "--format", "json"], calls[0])
|
|
self.assertEqual(0, json.loads(stdout.getvalue())["count"])
|
|
|
|
|
|
class TodoBatchCreateTest(unittest.TestCase):
|
|
def test_batch_uses_iso_due_captures_id_and_reads_back(self):
|
|
with tempfile.TemporaryDirectory() as raw:
|
|
source = Path(raw) / "todos.json"
|
|
source.write_text(
|
|
json.dumps(
|
|
[
|
|
{
|
|
"title": "reviewed task",
|
|
"executors": "user1",
|
|
"priority": 40,
|
|
"due": "2026-08-18",
|
|
}
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
calls = []
|
|
|
|
def fake_subprocess_run(argv, **kwargs):
|
|
calls.append((argv, kwargs))
|
|
if argv[1:4] == ["todo", "task", "create"]:
|
|
payload = {"result": {"taskId": "task-1"}}
|
|
else:
|
|
payload = {
|
|
"ok": True,
|
|
"data": {
|
|
"todoDetailModel": {
|
|
"taskId": "task-1",
|
|
"subject": "reviewed task",
|
|
}
|
|
},
|
|
}
|
|
return mock.Mock(
|
|
returncode=0,
|
|
stdout=json.dumps(payload),
|
|
stderr="",
|
|
)
|
|
|
|
stdout = io.StringIO()
|
|
preview_stdout = io.StringIO()
|
|
with mock.patch.object(BATCH.subprocess, "run") as preview_run_dws:
|
|
with contextlib.redirect_stdout(preview_stdout):
|
|
preview_code = BATCH.run(
|
|
[str(source), "--dry-run", "--dws", "fake-dws"]
|
|
)
|
|
self.assertEqual(0, preview_code)
|
|
preview_run_dws.assert_not_called()
|
|
plan_digest = json.loads(preview_stdout.getvalue())["planDigest"]
|
|
|
|
with mock.patch.object(
|
|
BATCH.subprocess, "run", side_effect=fake_subprocess_run
|
|
):
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = BATCH.run(
|
|
[
|
|
str(source),
|
|
"--yes",
|
|
"--confirm-digest",
|
|
plan_digest,
|
|
"--dws",
|
|
"fake-dws",
|
|
]
|
|
)
|
|
self.assertEqual(0, code)
|
|
self.assertEqual(2, len(calls))
|
|
self.assertEqual(
|
|
[
|
|
"fake-dws",
|
|
"todo",
|
|
"task",
|
|
"create",
|
|
"--title",
|
|
"reviewed task",
|
|
"--executors",
|
|
"user1",
|
|
"--priority",
|
|
"40",
|
|
"--due",
|
|
"2026-08-18T23:59:59+08:00",
|
|
"--format",
|
|
"json",
|
|
"--yes",
|
|
],
|
|
calls[0][0],
|
|
)
|
|
self.assertEqual(
|
|
[
|
|
"fake-dws",
|
|
"todo",
|
|
"task",
|
|
"get",
|
|
"--task-id",
|
|
"task-1",
|
|
"--format",
|
|
"json",
|
|
],
|
|
calls[1][0],
|
|
)
|
|
for _, kwargs in calls:
|
|
self.assertTrue(kwargs["capture_output"])
|
|
self.assertTrue(kwargs["text"])
|
|
self.assertIs(BATCH.subprocess.DEVNULL, kwargs["stdin"])
|
|
self.assertEqual(120, kwargs["timeout"])
|
|
payload = json.loads(stdout.getvalue())
|
|
self.assertTrue(payload["complete"])
|
|
self.assertEqual(plan_digest, payload["planDigest"])
|
|
self.assertEqual("verified", payload["ledger"][0]["status"])
|
|
|
|
def test_unconfirmed_batch_stops_before_first_dws_call(self):
|
|
with tempfile.TemporaryDirectory() as raw:
|
|
source = Path(raw) / "todos.json"
|
|
source.write_text(
|
|
'[{"title":"reviewed task","executors":"user1"}]',
|
|
encoding="utf-8",
|
|
)
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(BATCH.subprocess, "run") as run_dws:
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = BATCH.run([str(source)])
|
|
self.assertEqual(2, code)
|
|
run_dws.assert_not_called()
|
|
payload = json.loads(stdout.getvalue())
|
|
self.assertFalse(payload["complete"])
|
|
self.assertEqual("confirmation_required", payload["reason"])
|
|
self.assertFalse(payload["executionStarted"])
|
|
|
|
def test_yes_without_confirmed_digest_stops_before_first_dws_call(self):
|
|
with tempfile.TemporaryDirectory() as raw:
|
|
source = Path(raw) / "todos.json"
|
|
source.write_text(
|
|
'[{"title":"reviewed task","executors":"user1"}]',
|
|
encoding="utf-8",
|
|
)
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(BATCH.subprocess, "run") as run_dws:
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = BATCH.run([str(source), "--yes"])
|
|
self.assertEqual(2, code)
|
|
run_dws.assert_not_called()
|
|
payload = json.loads(stdout.getvalue())
|
|
self.assertEqual("confirmation_required", payload["reason"])
|
|
self.assertFalse(payload["executionStarted"])
|
|
|
|
def test_dry_run_previews_exact_batch_without_confirmation_bypass(self):
|
|
with tempfile.TemporaryDirectory() as raw:
|
|
source = Path(raw) / "todos.json"
|
|
source.write_text(
|
|
'[{"title":"reviewed task","executors":"user1"}]',
|
|
encoding="utf-8",
|
|
)
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(BATCH.subprocess, "run") as run_dws:
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = BATCH.run([str(source), "--dry-run"])
|
|
self.assertEqual(0, code)
|
|
run_dws.assert_not_called()
|
|
payload = json.loads(stdout.getvalue())
|
|
self.assertTrue(payload["complete"])
|
|
self.assertTrue(payload["dryRun"])
|
|
self.assertRegex(payload["planDigest"], r"^sha256:[0-9a-f]{64}$")
|
|
self.assertEqual(
|
|
[
|
|
"dws",
|
|
"todo",
|
|
"task",
|
|
"create",
|
|
"--title",
|
|
"reviewed task",
|
|
"--executors",
|
|
"user1",
|
|
"--format",
|
|
"json",
|
|
],
|
|
payload["ledger"][0]["command"],
|
|
)
|
|
self.assertNotIn("--yes", payload["ledger"][0]["command"])
|
|
|
|
def test_plan_digest_is_stable_for_equivalent_validated_content(self):
|
|
first = BATCH.validate(
|
|
[
|
|
{
|
|
"title": " reviewed task ",
|
|
"executors": "user1",
|
|
"priority": "40",
|
|
"due": "2026-08-18",
|
|
}
|
|
]
|
|
)
|
|
second = BATCH.validate(
|
|
[
|
|
{
|
|
"due": "2026-08-18T23:59:59+08:00",
|
|
"priority": 40,
|
|
"executors": "user1",
|
|
"title": "reviewed task",
|
|
}
|
|
]
|
|
)
|
|
self.assertEqual(
|
|
BATCH.batch_plan_digest(first), BATCH.batch_plan_digest(second)
|
|
)
|
|
|
|
def test_changed_file_rejects_confirmed_digest_before_first_dws_call(self):
|
|
with tempfile.TemporaryDirectory() as raw:
|
|
source = Path(raw) / "todos.json"
|
|
source.write_text(
|
|
'[{"title":"reviewed task","executors":"user1"}]',
|
|
encoding="utf-8",
|
|
)
|
|
preview_stdout = io.StringIO()
|
|
with mock.patch.object(BATCH.subprocess, "run") as preview_run_dws:
|
|
with contextlib.redirect_stdout(preview_stdout):
|
|
preview_code = BATCH.run([str(source), "--dry-run"])
|
|
self.assertEqual(0, preview_code)
|
|
preview_run_dws.assert_not_called()
|
|
confirmed_digest = json.loads(preview_stdout.getvalue())["planDigest"]
|
|
|
|
source.write_text(
|
|
'[{"title":"changed task","executors":"user1"}]',
|
|
encoding="utf-8",
|
|
)
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(BATCH.subprocess, "run") as run_dws:
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = BATCH.run(
|
|
[
|
|
str(source),
|
|
"--yes",
|
|
"--confirm-digest",
|
|
confirmed_digest,
|
|
]
|
|
)
|
|
self.assertEqual(2, code)
|
|
run_dws.assert_not_called()
|
|
payload = json.loads(stdout.getvalue())
|
|
self.assertEqual("plan_mismatch", payload["reason"])
|
|
self.assertFalse(payload["executionStarted"])
|
|
self.assertEqual(confirmed_digest, payload["confirmedPlanDigest"])
|
|
self.assertNotEqual(confirmed_digest, payload["actualPlanDigest"])
|
|
|
|
def test_short_numeric_due_stops_before_first_dws_call(self):
|
|
for due in ("0", "123", "2026", "0000000000123"):
|
|
with self.subTest(due=due), tempfile.TemporaryDirectory() as raw:
|
|
source = Path(raw) / "todos.json"
|
|
source.write_text(
|
|
json.dumps(
|
|
[
|
|
{
|
|
"title": "must not be created",
|
|
"executors": "user1",
|
|
"due": due,
|
|
}
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(BATCH.subprocess, "run") as run_dws:
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = BATCH.run(
|
|
[
|
|
str(source),
|
|
"--yes",
|
|
"--confirm-digest",
|
|
"sha256:" + "0" * 64,
|
|
]
|
|
)
|
|
self.assertEqual(2, code)
|
|
run_dws.assert_not_called()
|
|
payload = json.loads(stdout.getvalue())
|
|
self.assertFalse(payload["complete"])
|
|
self.assertIn("13-digit", payload["error"])
|
|
|
|
def test_modern_epoch_millisecond_due_is_accepted(self):
|
|
self.assertEqual(
|
|
"2025-01-01T08:00:00+08:00",
|
|
BATCH.normalize_due("1735689600000"),
|
|
)
|
|
|
|
def test_possible_commit_is_preserved_as_unknown(self):
|
|
with tempfile.TemporaryDirectory() as raw:
|
|
source = Path(raw) / "todos.json"
|
|
source.write_text(
|
|
'[{"title":"x","executors":"u"}]', encoding="utf-8"
|
|
)
|
|
failure = BATCH.ScriptError("timeout", commit_unknown=True)
|
|
items = BATCH.validate(json.loads(source.read_text(encoding="utf-8")))
|
|
plan_digest = BATCH.batch_plan_digest(items)
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(BATCH, "run_dws_json", side_effect=failure):
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = BATCH.run(
|
|
[
|
|
str(source),
|
|
"--yes",
|
|
"--confirm-digest",
|
|
plan_digest,
|
|
]
|
|
)
|
|
self.assertEqual(2, code)
|
|
payload = json.loads(stdout.getvalue())
|
|
self.assertEqual(1, payload["unknownCount"])
|
|
self.assertEqual("unknown", payload["ledger"][0]["status"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|