bvandeusenandClaude Fable 5
77bb3729a3
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 8s
CI & Build / integration (push) Successful in 24s
CI & Build / TypeScript typecheck (push) Successful in 33s
CI & Build / Python tests (push) Failing after 36s
CI & Build / Build & push image (push) Skipped
Second pass over the tests/ ledger after bbee0d0. fake_record(**attrs) is the
one MagicMock-with-real-attributes builder (to_dict mirrors them; the
note-2109 hazard documented once); fake_note/fake_task/fake_snippet/
fake_project/fake_milestone/fake_system/fake_rulebook/fake_topic/fake_rule
carry each model's ordinary defaults on top of it, replacing 14 per-file
factories (two rulebook trios in tool-vs-service wordings, _fake_task, _fake_ms,
_fake_project, _plan_note, _fake_snippet, two _snippet adapters now one-liners
over fake_snippet). FakeMCP replaces the five closure-over-a-list registrar
fakes (+ _Recorder); loc() and design_token_stub() replace the paired _loc /
_token / _T stand-ins; every hand-built async_session mock (9 helper defs and
14 inline copies) now starts from make_mock_session(). Call sites rewritten by
AST with each file's former defaults made explicit, so behaviour is unchanged.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
274 lines
10 KiB
Python
274 lines
10 KiB
Python
"""Tests for fable_*_task tools."""
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from scribe.mcp.tools.tasks import (
|
|
list_tasks, get_task, create_task,
|
|
update_task, add_task_log,
|
|
)
|
|
from tests.helpers import fake_task
|
|
|
|
|
|
pytestmark = pytest.mark.usefixtures("_bind_user")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tasks_passes_is_task_true_and_repackages():
|
|
rows = [fake_task(id=1), fake_task(id=2)]
|
|
mock = AsyncMock(return_value=(rows, 2))
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.list_notes", mock):
|
|
out = await list_tasks()
|
|
assert out["total"] == 2
|
|
assert len(out["tasks"]) == 2
|
|
assert mock.call_args.kwargs["is_task"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tasks_status_filter():
|
|
mock = AsyncMock(return_value=([], 0))
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.list_notes", mock):
|
|
await list_tasks(status="in_progress")
|
|
assert mock.call_args.kwargs["status"] == "in_progress"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tasks_empty_status_means_no_filter():
|
|
mock = AsyncMock(return_value=([], 0))
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.list_notes", mock):
|
|
await list_tasks(status="")
|
|
assert mock.call_args.kwargs["status"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_task_with_no_parent_returns_null_parent_title():
|
|
fake = fake_task(id=5, title="solo", parent_id=None)
|
|
with patch(
|
|
"scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
|
|
AsyncMock(return_value=(fake, "owner")),
|
|
):
|
|
out = await get_task(task_id=5)
|
|
assert out["parent_title"] is None
|
|
assert out["title"] == "solo"
|
|
assert "shared" not in out
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_task_enriches_with_parent_title():
|
|
"""When parent_id is set, get_task fetches the parent and adds parent_title."""
|
|
child = fake_task(id=10, title="child", parent_id=5)
|
|
parent = fake_task(id=5, title="parent of 10", parent_id=None)
|
|
# fetched twice: once for the child, once for the parent
|
|
mock_get = AsyncMock(side_effect=[(child, "owner"), (parent, "owner")])
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user", mock_get):
|
|
out = await get_task(task_id=10)
|
|
assert out["parent_title"] == "parent of 10"
|
|
assert mock_get.await_count == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_task_parent_missing_returns_null():
|
|
"""If parent_id is set but the parent is gone (orphaned), parent_title is None."""
|
|
child = fake_task(id=10, parent_id=5)
|
|
mock_get = AsyncMock(side_effect=[(child, "owner"), None])
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user", mock_get):
|
|
out = await get_task(task_id=10)
|
|
assert out["parent_title"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_task_opens_a_shared_task_and_says_whose_it_is():
|
|
"""A task in a shared project opens in the web UI, so the agent path must not
|
|
answer "not found" for the same id — and must say it isn't the caller's."""
|
|
theirs = fake_task(id=5, title="Their task", user_id=9)
|
|
with patch(
|
|
"scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
|
|
AsyncMock(return_value=(theirs, "viewer")),
|
|
), patch(
|
|
"scribe.services.access.describe_provenance",
|
|
AsyncMock(return_value={"shared": True, "owner": "alex",
|
|
"permission": "viewer"}),
|
|
):
|
|
out = await get_task(task_id=5)
|
|
assert out["shared"] is True
|
|
assert out["owner"] == "alex"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_task_raises_when_not_found():
|
|
with patch(
|
|
"scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
|
|
AsyncMock(return_value=None),
|
|
):
|
|
with pytest.raises(ValueError, match="task 999 not found"):
|
|
await get_task(task_id=999)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_passes_status():
|
|
fake = fake_task()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.create_note", mock):
|
|
await create_task(title="do x", status="todo")
|
|
assert mock.call_args.kwargs["status"] == "todo"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_blocked_by_duplicate_gate():
|
|
"""A near-duplicate blocks creation and returns the existing id (no insert)."""
|
|
from scribe.services.dedup import DuplicateMatch
|
|
dup = DuplicateMatch(id=42, title="Set up CI", similarity=1.0, reason="title")
|
|
create_mock = AsyncMock()
|
|
with patch("scribe.mcp.tools.tasks.dedup_svc.find_duplicate_note",
|
|
AsyncMock(return_value=dup)), \
|
|
patch("scribe.mcp.tools.tasks.notes_svc.create_note", create_mock):
|
|
out = await create_task(title="set up ci")
|
|
assert out["duplicate"] is True
|
|
assert out["existing_id"] == 42
|
|
create_mock.assert_not_called() # nothing was created
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_force_bypasses_duplicate_gate():
|
|
"""force=true skips the gate entirely and creates."""
|
|
find_mock = AsyncMock()
|
|
with patch("scribe.mcp.tools.tasks.dedup_svc.find_duplicate_note", find_mock), \
|
|
patch("scribe.mcp.tools.tasks.notes_svc.create_note",
|
|
AsyncMock(return_value=fake_task(id=9))):
|
|
out = await create_task(title="dup", force=True)
|
|
assert out["id"] == 9
|
|
find_mock.assert_not_called() # gate not even consulted
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_rejects_retired_plan_kind():
|
|
"""kind=plan is hard-retired — plans are milestones (start_planning)."""
|
|
mock = AsyncMock()
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.create_note", mock):
|
|
with pytest.raises(ValueError, match="kind=plan is retired"):
|
|
await create_task(title="x", kind="plan")
|
|
assert not mock.called # never reached the create
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_priority_empty_becomes_none():
|
|
fake = fake_task()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.create_note", mock):
|
|
await create_task(title="x", priority="")
|
|
assert mock.call_args.kwargs["priority"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_zero_id_sentinels_become_none():
|
|
fake = fake_task()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.create_note", mock):
|
|
await create_task(title="x", project_id=0, milestone_id=0, parent_id=0)
|
|
assert mock.call_args.kwargs["project_id"] is None
|
|
assert mock.call_args.kwargs["milestone_id"] is None
|
|
assert mock.call_args.kwargs["parent_id"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_only_sends_non_default_fields():
|
|
fake = fake_task()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.update_note", mock):
|
|
await update_task(task_id=1, status="done")
|
|
args, kwargs = mock.call_args
|
|
assert args == (7, 1)
|
|
assert kwargs == {"status": "done"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_empty_priority_is_omitted():
|
|
"""Priority="" is "leave unchanged" — must not reach service as empty string."""
|
|
fake = fake_task()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.update_note", mock):
|
|
await update_task(task_id=1, status="done", priority="")
|
|
assert "priority" not in mock.call_args.kwargs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_raises_when_not_found():
|
|
with patch(
|
|
"scribe.mcp.tools.tasks.notes_svc.update_note",
|
|
AsyncMock(return_value=None),
|
|
):
|
|
with pytest.raises(ValueError, match="task 999 not found"):
|
|
await update_task(task_id=999, status="done")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_milestone_zero_is_omitted():
|
|
"""milestone_id=0 is 'leave unchanged' — must not reach the service."""
|
|
fake = fake_task()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.update_note", mock):
|
|
await update_task(task_id=1, milestone_id=0)
|
|
assert "milestone_id" not in mock.call_args.kwargs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_milestone_positive_is_set():
|
|
fake = fake_task()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.update_note", mock):
|
|
await update_task(task_id=1, milestone_id=42)
|
|
assert mock.call_args.kwargs["milestone_id"] == 42
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_milestone_negative_one_clears():
|
|
"""milestone_id=-1 clears the milestone (sets the column NULL)."""
|
|
fake = fake_task()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.update_note", mock):
|
|
await update_task(task_id=1, milestone_id=-1)
|
|
assert mock.call_args.kwargs["milestone_id"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_clearing_project_also_clears_milestone():
|
|
"""project_id=-1 clears the project and, with it, the milestone."""
|
|
fake = fake_task()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.update_note", mock):
|
|
await update_task(task_id=1, project_id=-1)
|
|
assert mock.call_args.kwargs["project_id"] is None
|
|
assert mock.call_args.kwargs["milestone_id"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_task_log_returns_log_dict():
|
|
log = MagicMock()
|
|
log.id = 5
|
|
log.task_id = 1
|
|
log.content = "checkpoint"
|
|
log.created_at = datetime(2026, 5, 26, 12, tzinfo=timezone.utc)
|
|
# No to_dict on TaskLog model (per task_logs.py inspection) — fall through to manual dict
|
|
del log.to_dict
|
|
with patch(
|
|
"scribe.mcp.tools.tasks.task_logs_svc.create_log",
|
|
AsyncMock(return_value=log),
|
|
):
|
|
out = await add_task_log(task_id=1, content="checkpoint")
|
|
assert out["id"] == 5
|
|
assert out["task_id"] == 1
|
|
assert out["content"] == "checkpoint"
|
|
assert out["created_at"].startswith("2026-05-26")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_task_log_propagates_value_error_when_task_missing():
|
|
"""create_log raises ValueError if the task doesn't exist; we let it through."""
|
|
with patch(
|
|
"scribe.mcp.tools.tasks.task_logs_svc.create_log",
|
|
AsyncMock(side_effect=ValueError("Task 999 not found")),
|
|
):
|
|
with pytest.raises(ValueError):
|
|
await add_task_log(task_id=999, content="x")
|