408 lines
18 KiB
Python
408 lines
18 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def tracker(prefix=None, epics=True, number=1):
|
|
prefix_line = f"task-id-prefix: '{prefix}'\n" if prefix is not None else ""
|
|
epic_frontmatter = """epics:
|
|
- value: 'poc'
|
|
label: 'Proof of Concept'
|
|
description: 'Validate the architecture.'
|
|
- value: 'release'
|
|
label: "Team's Release"
|
|
description: 'Ship the product.'
|
|
""" if epics else ""
|
|
anchor_prefix = prefix.lower() if prefix else "ex"
|
|
return f"""---
|
|
{prefix_line}projects:
|
|
- value: 'shared'
|
|
label: 'Shared'
|
|
subprojects:
|
|
- value: 'engine'
|
|
label: 'Engine'
|
|
{epic_frontmatter}task-statuses:
|
|
- value: 'pending'
|
|
label: 'Pending'
|
|
- value: 'in_progress'
|
|
label: 'In Progress'
|
|
task-types:
|
|
- value: 'feature'
|
|
label: 'Feature'
|
|
---
|
|
# Tracker
|
|
|
|
| ID | Title | Project | Status | Type |
|
|
|:---|:------|:--------|:-------|:-----|
|
|
| {anchor_prefix.upper()}-{number:03d} | Original | Shared | Pending | Feature |
|
|
|
|
---
|
|
|
|
## Detailed Tasks
|
|
|
|
<a id="{anchor_prefix}-{number:03d}" class="task" data-project="shared" data-subproject="engine" data-status="pending" data-task-type="feature"></a>
|
|
### Original
|
|
**ID:** {anchor_prefix.upper()}-{number:03d}
|
|
**Project:** Shared
|
|
**Subproject:** Engine
|
|
**Status:** Pending
|
|
**Type:** Feature
|
|
|
|
**Description:**
|
|
Keep this description.
|
|
- [ ] First requirement
|
|
- [x] Second requirement
|
|
|
|
---
|
|
|
|
## Notes
|
|
|
|
See [original](#{anchor_prefix}-{number:03d}).
|
|
|
|
#### Projects (Rendered from frontmatter projects)
|
|
Placeholder
|
|
<!-- Table Sort Injection -->
|
|
"""
|
|
|
|
|
|
class TaskScriptsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.path = Path(self.temp.name) / "task tracker's file.md"
|
|
self.path.write_text(tracker(), encoding="utf-8")
|
|
self.env = os.environ.copy()
|
|
self.env.pop("TASK_ID_PREFIX", None)
|
|
self.env.pop("TASKS_FILE", None)
|
|
|
|
def run_cli(self, command, *args, success=True, env=None):
|
|
executable = ["bash"] if command == "get-tasks" else [sys.executable]
|
|
result = subprocess.run(
|
|
[*executable, str(SCRIPTS / command), *args, "-f", str(self.path)],
|
|
env=self.env if env is None else env,
|
|
text=True, capture_output=True,
|
|
)
|
|
if success:
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
else:
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
return result
|
|
|
|
def get_task(self, task_id="1"):
|
|
return json.loads(self.run_cli("get-task", task_id, "--json").stdout)
|
|
|
|
def test_epic_add_update_clear_and_preserve(self):
|
|
self.run_cli("add-task", "New task", "--epic", "Proof of Concept", "--subproject", "engine")
|
|
task = self.get_task("EX-002")
|
|
self.assertEqual((task["epic"], task["epic_label"]), ("poc", "Proof of Concept"))
|
|
self.assertEqual(task["subproject"], "engine")
|
|
self.assertIn('data-epic="poc"', task["raw_markdown"])
|
|
self.assertIn("**Epic:** Proof of Concept", task["raw_markdown"])
|
|
self.assertEqual(self.run_cli("get-task", "2", "--epic").stdout.strip(), "poc")
|
|
self.assertIn("Epic: Proof of Concept", self.run_cli("get-task", "2", "--metadata-only").stdout)
|
|
self.run_cli("update-task", "1", "--epic", "poc", "--status", "in_progress", "--check", "1")
|
|
self.run_cli("update-task", "1", "--title", "Renamed")
|
|
original = self.get_task()
|
|
self.assertEqual(original["epic"], "poc")
|
|
self.assertIn("Keep this description.", original["description"])
|
|
self.assertIn("- [x] First requirement", original["description"])
|
|
self.run_cli("update-task", "1", "--epic", "Team's Release", "--no-sync")
|
|
self.assertEqual(self.get_task()["epic"], "release")
|
|
self.run_cli("update-tasks")
|
|
synced = self.path.read_text()
|
|
self.run_cli("update-tasks")
|
|
self.assertEqual(synced, self.path.read_text())
|
|
self.assertIn("#### Epics (Rendered from frontmatter epics)", synced)
|
|
self.run_cli("update-task", "1", "--clear-epic")
|
|
self.assertEqual(self.get_task()["epic"], "")
|
|
self.assertNotIn("**Epic:**", self.get_task()["raw_markdown"])
|
|
self.run_cli("update-task", "2", "--epic", "")
|
|
self.assertEqual(self.get_task("2")["epic"], "")
|
|
|
|
def test_invalid_epic_does_not_write(self):
|
|
before = self.path.read_text()
|
|
for command, args in (
|
|
("add-task", ("Invalid", "--epic", "missing")),
|
|
("update-task", ("1", "--epic", "missing")),
|
|
):
|
|
with self.subTest(command=command):
|
|
result = self.run_cli(command, *args, success=False)
|
|
self.assertIn("Unknown epic", result.stderr)
|
|
self.assertEqual(before, self.path.read_text())
|
|
|
|
def test_frontmatter_prefix_with_digits_and_large_ids(self):
|
|
self.path.write_text(tracker(prefix="ab2-work", number=999))
|
|
self.run_cli("add-task", "Thousandth", "--epic", "release")
|
|
task = self.get_task("ab2-work-1000")
|
|
self.assertEqual(task["number"], 1000)
|
|
self.assertEqual(task["id"], "AB2-WORK-1000")
|
|
self.assertEqual(task["anchor_id"], "ab2-work-1000")
|
|
self.assertEqual(self.get_task("999")["anchor_id"], "ab2-work-0999")
|
|
self.assertEqual(self.get_task("999")["id"], "AB2-WORK-0999")
|
|
self.assertEqual(self.get_task("AB2-WORK-00999")["number"], 999)
|
|
self.run_cli("update-task", "999", "--title", "Still padded", "--no-sync")
|
|
self.assertIn("**ID:** AB2-WORK-0999", self.get_task("999")["raw_markdown"])
|
|
self.run_cli("update-task", "AB2-WORK-1000", "--status", "pending")
|
|
self.assertEqual(self.get_task("1000")["status"], "pending")
|
|
env = {**self.env, "PATH": "/usr/bin:/bin"}
|
|
output = self.run_cli("get-tasks", "--all", env=env).stdout
|
|
self.assertLess(output.index("AB2-WORK-1000"), output.index("AB2-WORK-0999"))
|
|
self.run_cli("get-task", "WRONG-1000", success=False)
|
|
for invalid in ("abc", "1junk", "something 1000"):
|
|
self.run_cli("get-task", invalid, success=False)
|
|
|
|
def test_environment_override_migrates_ids_and_links(self):
|
|
self.path.write_text(
|
|
tracker(prefix="doc")
|
|
+ '\n[reference]: #doc-001\n<a href="#doc-001">Task</a>\n'
|
|
+ '[external](https://example.com/#doc-001)\n'
|
|
)
|
|
env = {**self.env, "TASK_ID_PREFIX": "env"}
|
|
self.run_cli("update-tasks", env=env)
|
|
text = self.path.read_text()
|
|
self.assertIn('id="env-001"', text)
|
|
self.assertIn("**ID:** ENV-001", text)
|
|
self.assertIn("[original](#env-001)", text)
|
|
self.assertIn("[reference]: #env-001", text)
|
|
self.assertIn('href="#env-001"', text)
|
|
self.assertIn("[external](https://example.com/#doc-001)", text)
|
|
self.run_cli("add-task", "Next", env=env)
|
|
task = json.loads(self.run_cli("get-task", "ENV-002", "--json", env=env).stdout)
|
|
self.assertEqual(task["id"], "ENV-002")
|
|
|
|
def test_default_prefix_without_epics_and_environment_file(self):
|
|
self.path.write_text(tracker(epics=False))
|
|
self.run_cli("update-tasks")
|
|
self.assertEqual(self.get_task()["id"], "EX-001")
|
|
self.assertEqual(self.get_task()["epic"], "")
|
|
self.assertNotIn("| Epic", self.path.read_text())
|
|
env = {**self.env, "TASKS_FILE": str(self.path)}
|
|
result = subprocess.run(
|
|
[sys.executable, str(SCRIPTS / "get-task"), "1", "--id"],
|
|
env=env, text=True, capture_output=True,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout.strip(), "EX-001")
|
|
|
|
def test_invalid_prefix_and_duplicate_numbers(self):
|
|
for prefix in ("", "1BAD", "BAD SPACE", "BAD/"):
|
|
with self.subTest(prefix=prefix):
|
|
before = self.path.read_text()
|
|
self.run_cli("update-tasks", env={**self.env, "TASK_ID_PREFIX": prefix}, success=False)
|
|
self.assertEqual(before, self.path.read_text())
|
|
text = self.path.read_text()
|
|
chunk = text.split("## Detailed Tasks\n", 1)[1].split("\n---\n\n## Notes", 1)[0]
|
|
self.path.write_text(text.replace("\n---\n\n## Notes", chunk + "\n---\n\n## Notes"))
|
|
self.assertIn("Duplicate numeric task IDs", self.run_cli("update-tasks", success=False).stderr)
|
|
|
|
def test_summary_table_conflict_handling(self):
|
|
text = self.path.read_text()
|
|
# Inject conflict markers surrounding and within the summary table
|
|
conflicted = text.replace(
|
|
"| ID | Title |",
|
|
"<<<<<<< HEAD\n| ID | Title |\n=======\n| ID | Title | Project |\n>>>>>>> branch",
|
|
)
|
|
self.path.write_text(conflicted)
|
|
self.run_cli("update-tasks")
|
|
updated = self.path.read_text()
|
|
self.assertNotIn("<<<<<<<", updated)
|
|
self.assertNotIn("=======", updated)
|
|
self.assertNotIn(">>>>>>>", updated)
|
|
self.assertIn("EX-001", updated)
|
|
self.assertIn("Original", updated)
|
|
|
|
|
|
def test_listing_epic_filters_and_selection_fallback(self):
|
|
self.run_cli("update-task", "1", "--epic", "release")
|
|
self.run_cli("add-task", "Second", "--epic", "poc")
|
|
env = {**self.env, "PATH": "/usr/bin:/bin"}
|
|
output = self.run_cli("get-tasks", "--all", "--sort", "asc", env=env).stdout
|
|
self.assertIn("Epic", output)
|
|
self.assertLess(output.index("EX-001"), output.index("EX-002"))
|
|
output = self.run_cli(
|
|
"get-tasks", "--all", "--accept", "Epic:Team's Release", "--select", "ID,Epic", env=env,
|
|
).stdout
|
|
self.assertIn("EX-001", output)
|
|
self.assertNotIn("EX-002", output)
|
|
self.assertNotIn("Title", output)
|
|
output = self.run_cli("get-tasks", "--all", "--reject", "Epic:Proof of Concept", env=env).stdout
|
|
self.assertNotIn("EX-002", output)
|
|
self.assertIn("EX-001", output)
|
|
for value in ("poc", "POC", "Proof of Concept"):
|
|
with self.subTest(value=value):
|
|
output = self.run_cli("get-tasks", "--accept", f"Epic:{value}", env=env).stdout
|
|
self.assertIn("EX-002", output)
|
|
self.assertNotIn("EX-001", output)
|
|
output = self.run_cli("get-tasks", "--all", "--reject", "Epic:poc", env=env).stdout
|
|
self.assertNotIn("EX-002", output)
|
|
self.assertIn("EX-001", output)
|
|
output = self.run_cli(
|
|
"get-tasks", "--accept", "Epic:poc,release", "--accept", "Status:Triage", env=env,
|
|
).stdout
|
|
self.assertIn("EX-002", output)
|
|
self.assertNotIn("EX-001", output)
|
|
self.assertNotIn("EX-001", self.run_cli("get-tasks", "--accept", "Epic:missing", env=env).stdout)
|
|
|
|
def test_listing_external_filter_table_selects_epic(self):
|
|
self.run_cli("update-tasks")
|
|
stub = Path(self.temp.name) / "filter-table"
|
|
stub.write_text("#!/bin/sh\nprintf '%s\\n' \"$@\"\n")
|
|
stub.chmod(0o755)
|
|
env = {**self.env, "PATH": f"{self.temp.name}:/usr/bin:/bin"}
|
|
output = self.run_cli("get-tasks", "--all", "--accept", "Epic:Proof of Concept", env=env).stdout
|
|
self.assertIn("ID,Title,Project,Subproject,Epic,Status", output)
|
|
self.assertIn("Epic:Proof of Concept", output)
|
|
output = self.run_cli(
|
|
"get-tasks", "--all", "--accept", "epic:POC,release", "--reject", "Epic:release",
|
|
"--accept", "Status:Pending", env=env,
|
|
).stdout
|
|
self.assertIn("Epic:Proof of Concept,Team's Release", output)
|
|
self.assertIn("Epic:Team's Release", output)
|
|
self.assertIn("Status:Pending", output)
|
|
self.assertNotIn("Epic:release", output)
|
|
|
|
def test_filter_frontmatter_errors_are_reported(self):
|
|
self.path.write_text("No frontmatter\n")
|
|
env = {**self.env, "PATH": "/usr/bin:/bin"}
|
|
result = self.run_cli("get-tasks", "--accept", "Epic:poc", env=env, success=False)
|
|
self.assertIn("Error normalizing task filters: YAML frontmatter not found", result.stderr)
|
|
|
|
def test_workflow_surfaces(self):
|
|
self.run_cli("update-task", "1", "--epic", "poc")
|
|
output = self.run_cli("work-on-task", "1", "--start").stdout
|
|
self.assertIn("Proof of Concept", output)
|
|
self.assertEqual(self.get_task()["status"], "in_progress")
|
|
with patch.dict(os.environ, self.env, clear=True):
|
|
sys.path.insert(0, str(SCRIPTS / "lib"))
|
|
try:
|
|
import task_context
|
|
task = task_context.fetch_task("1", tasks_file=str(self.path))
|
|
prompt = task_context.build_workflow_prompt(task, "testing", "Verify the task")
|
|
finally:
|
|
sys.path.pop(0)
|
|
self.assertIn("- Epic: Proof of Concept (poc)", prompt)
|
|
self.assertIn("--epic=poc", prompt)
|
|
|
|
def test_ready_task_can_be_started(self):
|
|
self.run_cli("update-task", "1", "--status", "ready")
|
|
self.run_cli("work-on-task", "1", "--start")
|
|
self.assertEqual(self.get_task()["status"], "in_progress")
|
|
self.run_cli("work-on-task", "1")
|
|
|
|
def test_hyphenated_in_progress_is_already_started(self):
|
|
self.run_cli("update-task", "1", "--status", "in-progress")
|
|
before = self.path.read_text()
|
|
self.run_cli("work-on-task", "1", "--start")
|
|
self.assertEqual(before, self.path.read_text())
|
|
|
|
def test_repository_sample_and_config_are_consistent(self):
|
|
config = json.loads((SCRIPTS.parent / "project-config.json").read_text())
|
|
sample = SCRIPTS.parent / "docs" / "tasks.md"
|
|
result = subprocess.run(
|
|
[sys.executable, str(SCRIPTS / "get-task"), "1", "-f", str(sample), "--json"],
|
|
env=self.env, text=True, capture_output=True,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
task = json.loads(result.stdout)
|
|
self.assertIsInstance(config["name"], str)
|
|
sys.path.insert(0, str(SCRIPTS / "lib"))
|
|
try:
|
|
import task_context
|
|
update = task_context.load_get_task_module().load_update_tasks_module()
|
|
finally:
|
|
sys.path.pop(0)
|
|
frontmatter = update.read_frontmatter(sample.read_text())
|
|
project = next(item for item in frontmatter["projects"] if item["value"] == task["project"])
|
|
self.assertIn(f"**Project:** {project['label']}", task["raw_markdown"])
|
|
self.assertEqual(task["id"].rsplit("-", 1)[0], update.get_task_id_prefix(frontmatter))
|
|
copy = Path(self.temp.name) / "sample.md"
|
|
copy.write_text(sample.read_text())
|
|
result = subprocess.run(
|
|
[sys.executable, str(SCRIPTS / "update-tasks"), "-f", str(copy)],
|
|
env=self.env, text=True, capture_output=True,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(sample.read_text(), copy.read_text())
|
|
|
|
def test_environment_preserves_external_tracker(self):
|
|
env = {**self.env, "TASKS_FILE": str(self.path)}
|
|
result = subprocess.run(
|
|
["bash", "-c", 'source "$1" >/dev/null; printf "%s\\n" "$TASKS_FILE"; command -v get-task',
|
|
"bash", str(SCRIPTS / "env.sh")],
|
|
env=env, text=True, capture_output=True,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout.splitlines(), [str(self.path), str(SCRIPTS / "get-task")])
|
|
|
|
|
|
class ProjectConfigTest(unittest.TestCase):
|
|
def setUp(self):
|
|
sys.path.insert(0, str(SCRIPTS / "lib"))
|
|
try:
|
|
import task_context
|
|
self.context = task_context
|
|
finally:
|
|
sys.path.pop(0)
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.path = Path(self.temp.name) / "project-config.json"
|
|
self.config = {
|
|
"name": "The Lighthouse Novel",
|
|
"description": "A novel about a coastal community.",
|
|
"commands": {},
|
|
}
|
|
self.path.write_text(json.dumps(self.config))
|
|
|
|
def test_nonsoftware_configuration_in_prompt(self):
|
|
task = {
|
|
"id": "BOOK-001", "title": "Outline chapter one", "project": "manuscript",
|
|
"status": "triage", "type": "task", "description": "Review the narrative arc.",
|
|
}
|
|
with patch.object(self.context, "load_project_config", return_value=self.context.load_project_config(self.path)):
|
|
prompt = self.context.build_workflow_prompt(task, "triaging", "Outline the chapter")
|
|
self.assertIn("for The Lighthouse Novel.", prompt)
|
|
self.assertIn(self.config["description"], prompt)
|
|
self.assertIn("No commands configured", prompt)
|
|
self.assertIn("Do not assume this is software", prompt)
|
|
self.assertIn("AGENTS.md", prompt)
|
|
|
|
def test_arbitrary_named_commands_are_guidance(self):
|
|
self.config["commands"] = {"word-count": "wc -w manuscript.txt"}
|
|
self.path.write_text(json.dumps(self.config))
|
|
config = self.context.load_project_config(self.path)
|
|
task = {"id": "1", "title": "Review", "project": "manuscript", "status": "triage", "type": "task"}
|
|
with patch.object(self.context, "load_project_config", return_value=config):
|
|
prompt = self.context.build_workflow_prompt(task, "reviewing", "Review")
|
|
self.assertIn("- word-count: `wc -w manuscript.txt`", prompt)
|
|
self.assertIn("not automatically executed", prompt)
|
|
|
|
def test_invalid_configuration_is_reported(self):
|
|
invalid_configs = (
|
|
[], {}, {**self.config, "name": ""}, {**self.config, "description": None},
|
|
{**self.config, "commands": []}, {**self.config, "commands": {"test": None}},
|
|
{**self.config, "commands": {"": "echo ok"}},
|
|
{**self.config, "commands": {"review": " "}},
|
|
)
|
|
for config in invalid_configs:
|
|
with self.subTest(config=config):
|
|
self.path.write_text(json.dumps(config))
|
|
with self.assertRaisesRegex(ValueError, "Project configuration"):
|
|
self.context.load_project_config(self.path)
|
|
self.path.write_text("{")
|
|
with self.assertRaisesRegex(ValueError, "Cannot read project configuration"):
|
|
self.context.load_project_config(self.path)
|
|
with self.assertRaisesRegex(ValueError, "Cannot read project configuration"):
|
|
self.context.load_project_config(self.path.parent / "missing.json")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|