mirror of
https://github.com/shareAI-lab/learn-claude-code.git
synced 2026-08-25 16:22:41 +00:00
238 lines
8.4 KiB
Python
238 lines
8.4 KiB
Python
import importlib.util
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
LESSON = ROOT / "s10_task_system" / "code.py"
|
|
|
|
|
|
def load_lesson(workdir: Path):
|
|
fake_anthropic = types.ModuleType("anthropic")
|
|
fake_dotenv = types.ModuleType("dotenv")
|
|
|
|
class FakeAnthropic:
|
|
def __init__(self, *args, **kwargs):
|
|
self.messages = types.SimpleNamespace(create=None)
|
|
|
|
fake_anthropic.Anthropic = FakeAnthropic
|
|
fake_dotenv.load_dotenv = lambda override=True: None
|
|
|
|
previous_modules = {
|
|
"anthropic": sys.modules.get("anthropic"),
|
|
"dotenv": sys.modules.get("dotenv"),
|
|
}
|
|
previous_cwd = Path.cwd()
|
|
previous_model = os.environ.get("MODEL_ID")
|
|
|
|
module_name = f"s10_task_system_test_{id(workdir)}"
|
|
spec = importlib.util.spec_from_file_location(module_name, LESSON)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
|
|
sys.modules["anthropic"] = fake_anthropic
|
|
sys.modules["dotenv"] = fake_dotenv
|
|
sys.modules[module_name] = module
|
|
try:
|
|
os.chdir(workdir)
|
|
os.environ["MODEL_ID"] = "test-model"
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
finally:
|
|
os.chdir(previous_cwd)
|
|
sys.modules.pop(module_name, None)
|
|
if previous_model is None:
|
|
os.environ.pop("MODEL_ID", None)
|
|
else:
|
|
os.environ["MODEL_ID"] = previous_model
|
|
for name, previous in previous_modules.items():
|
|
if previous is None:
|
|
sys.modules.pop(name, None)
|
|
else:
|
|
sys.modules[name] = previous
|
|
|
|
|
|
def tool_call(name: str, **arguments):
|
|
return types.SimpleNamespace(name=name, input=arguments, id="tool-1")
|
|
|
|
|
|
def test_s10_keeps_the_s04_kernel_and_adds_task_tools() -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
workdir = Path(tmp)
|
|
lesson = load_lesson(workdir)
|
|
|
|
assert [tool["name"] for tool in lesson.TOOLS] == [
|
|
"bash",
|
|
"read_file",
|
|
"write_file",
|
|
"edit_file",
|
|
"glob",
|
|
"create_task",
|
|
"update_task",
|
|
"list_tasks",
|
|
"get_task",
|
|
"claim_task",
|
|
"complete_task",
|
|
]
|
|
assert lesson.permission_hook in lesson.HOOKS["PreToolUse"]
|
|
assert hasattr(lesson, "execute_tool")
|
|
assert not hasattr(lesson, "MEMORY_DIR")
|
|
assert not (workdir / ".tasks").exists()
|
|
|
|
tools = {tool["name"]: tool for tool in lesson.TOOLS}
|
|
create_schema = tools["create_task"]["input_schema"]
|
|
update_schema = tools["update_task"]["input_schema"]
|
|
assert "blockedBy" not in create_schema["properties"]
|
|
assert create_schema["additionalProperties"] is False
|
|
assert update_schema["required"] == ["task_id", "addBlockedBy"]
|
|
|
|
|
|
def test_dependencies_gate_claim_and_completion_checks_owner() -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
workdir = Path(tmp)
|
|
lesson = load_lesson(workdir)
|
|
|
|
schema = lesson.create_task("create schema")
|
|
api = lesson.create_task("write API")
|
|
lesson.update_task(api.id, [schema.id])
|
|
|
|
assert lesson.claim_task(api.id) == f"Blocked by: ['{schema.id}']"
|
|
assert "Claimed" in lesson.claim_task(schema.id)
|
|
assert "Unblocked: write API" in lesson.complete_task(schema.id)
|
|
assert "Claimed" in lesson.claim_task(api.id)
|
|
assert "owned by agent, not other" in lesson.complete_task(
|
|
api.id, owner="other"
|
|
)
|
|
assert "Completed" in lesson.complete_task(api.id)
|
|
assert lesson.load_task(api.id).status == "completed"
|
|
|
|
|
|
def test_dependencies_are_added_after_create_returns_runtime_ids() -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
|
|
create_results = [
|
|
lesson.execute_tool(tool_call("create_task", subject=subject))
|
|
for subject in (
|
|
"create schema",
|
|
"write API",
|
|
"write tests",
|
|
"write docs",
|
|
)
|
|
]
|
|
task_ids = [result.split()[1].rstrip(":") for result in create_results]
|
|
schema_id, api_id, tests_id, docs_id = task_ids
|
|
|
|
update_results = [
|
|
lesson.execute_tool(tool_call(
|
|
"update_task", task_id=api_id, addBlockedBy=[schema_id]
|
|
)),
|
|
lesson.execute_tool(tool_call(
|
|
"update_task", task_id=tests_id, addBlockedBy=[api_id]
|
|
)),
|
|
lesson.execute_tool(tool_call(
|
|
"update_task", task_id=docs_id, addBlockedBy=[schema_id]
|
|
)),
|
|
]
|
|
|
|
assert all(not result.startswith("Error:") for result in update_results)
|
|
assert lesson.load_task(schema_id).blockedBy == []
|
|
assert lesson.load_task(api_id).blockedBy == [schema_id]
|
|
assert lesson.load_task(tests_id).blockedBy == [api_id]
|
|
assert lesson.load_task(docs_id).blockedBy == [schema_id]
|
|
|
|
|
|
def test_invalid_and_missing_task_ids_become_tool_results() -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
|
|
invalid = lesson.execute_tool(tool_call("get_task", task_id="../outside"))
|
|
missing = lesson.execute_tool(
|
|
tool_call("claim_task", task_id="task_00000000")
|
|
)
|
|
|
|
assert invalid.startswith("Error: Invalid task ID")
|
|
assert missing.startswith("Error:")
|
|
|
|
|
|
def test_create_retries_instead_of_overwriting_an_existing_id(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
values = iter(["deadbeef", "deadbeef", "cafebabe"])
|
|
monkeypatch.setattr(lesson.secrets, "token_hex", lambda _size: next(values))
|
|
|
|
first = lesson.create_task("first")
|
|
second = lesson.create_task("second")
|
|
|
|
assert first.id == "task_deadbeef"
|
|
assert second.id == "task_cafebabe"
|
|
assert [task.subject for task in lesson.list_tasks()] == ["second", "first"]
|
|
|
|
|
|
def test_update_rejects_invalid_graph_changes_without_partial_mutation() -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
dependency = lesson.create_task("create schema")
|
|
target = lesson.create_task("write API")
|
|
|
|
missing = lesson.execute_tool(tool_call(
|
|
"update_task",
|
|
task_id=target.id,
|
|
addBlockedBy=[dependency.id, "task_00000000"],
|
|
))
|
|
self_dependency = lesson.execute_tool(tool_call(
|
|
"update_task", task_id=target.id, addBlockedBy=[target.id]
|
|
))
|
|
|
|
assert missing == "Error: Dependency not found: task_00000000"
|
|
assert self_dependency == "Error: Task cannot depend on itself"
|
|
assert lesson.load_task(target.id).blockedBy == []
|
|
|
|
|
|
def test_update_is_idempotent_and_rejects_cycles_or_started_tasks() -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
first = lesson.create_task("first")
|
|
second = lesson.create_task("second")
|
|
third = lesson.create_task("third")
|
|
|
|
lesson.update_task(second.id, [first.id, first.id])
|
|
lesson.update_task(second.id, [first.id])
|
|
lesson.update_task(third.id, [second.id])
|
|
|
|
cycle = lesson.execute_tool(tool_call(
|
|
"update_task", task_id=first.id, addBlockedBy=[third.id]
|
|
))
|
|
assert cycle.startswith("Error: Dependency cycle detected")
|
|
assert lesson.load_task(first.id).blockedBy == []
|
|
assert lesson.load_task(second.id).blockedBy == [first.id]
|
|
|
|
assert "Claimed" in lesson.claim_task(first.id)
|
|
started = lesson.execute_tool(tool_call(
|
|
"update_task", task_id=first.id, addBlockedBy=[second.id]
|
|
))
|
|
assert "only be updated while pending and unowned" in started
|
|
|
|
|
|
def test_task_store_rejects_a_symlink_outside_the_workspace() -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
with tempfile.TemporaryDirectory() as outside:
|
|
workdir = Path(tmp)
|
|
(workdir / ".tasks").symlink_to(
|
|
Path(outside), target_is_directory=True
|
|
)
|
|
lesson = load_lesson(workdir)
|
|
|
|
output = lesson.execute_tool(
|
|
tool_call("create_task", subject="unsafe")
|
|
)
|
|
|
|
assert output == "Error: Task store escapes the workspace"
|
|
assert list(Path(outside).iterdir()) == []
|