bvandeusen
322cbc3b5e
#755 Phase 5. create_note / create_task now BLOCK a near-duplicate instead of silently inserting: they return {"duplicate": true, "existing_id", message} pointing at the record to UPDATE. Fights store bloat and stale competing copies that semantic search (RAG) would otherwise resurface for reconciliation. A force=true override creates anyway for genuinely-distinct records. - services/dedup.py: find_duplicate_note — two signals, scoped to owner + same project + same kind: (1) normalized-title exact match (cheap, always); (2) semantic cosine ≥ 0.90 but ONLY when body ≥ 200 chars (short/title-only embeddings false-positive — the pre-pivot lesson). Project-less (orphan) records compare only to other orphans on BOTH signals (orphan_only on the semantic call) — they're not matched across every project. - Gate wired into the MCP create_note/create_task tools (the LLM write path) with force override; _INSTRUCTIONS documents the duplicate response + force. - Opt-in by design: the service helper is only called from the interactive create tools. Internal/programmatic creates (recurrence spawn, imports) go straight through services.create_note and are NOT gated — a recurring task spawning its next same-titled instance must not be blocked. - Scope v1: MCP tools only. REST/web (human CRUD, needs a UI affordance) and create_rule (not a RAG surface; _INSTRUCTIONS already steer it) are follow-ups. - tests: dedup service (title/semantic/body-gate/type-filter) + tool gate (blocks, force bypasses) for notes and tasks. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
273 lines
9.8 KiB
Python
273 lines
9.8 KiB
Python
"""Tests for fable_*_task tools."""
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from scribe.mcp._context import _user_id_ctx
|
|
from scribe.mcp.tools.tasks import (
|
|
list_tasks, get_task, create_task,
|
|
update_task, add_task_log,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _bind_user():
|
|
token = _user_id_ctx.set(7)
|
|
yield
|
|
_user_id_ctx.reset(token)
|
|
|
|
|
|
def _fake_task(*, parent_id: int | None = None, **overrides) -> MagicMock:
|
|
n = MagicMock()
|
|
n.parent_id = parent_id
|
|
base = {
|
|
"id": 1, "title": "t", "body": "", "status": "todo",
|
|
"priority": "none", "tags": [], "parent_id": parent_id,
|
|
"is_task": True,
|
|
}
|
|
base.update(overrides)
|
|
n.to_dict.return_value = base
|
|
n.title = base["title"]
|
|
return n
|
|
|
|
|
|
@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",
|
|
AsyncMock(return_value=fake),
|
|
):
|
|
out = await get_task(task_id=5)
|
|
assert out["parent_title"] is None
|
|
assert out["title"] == "solo"
|
|
|
|
|
|
@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)
|
|
# get_note is called twice: once for child, once for parent
|
|
mock_get = AsyncMock(side_effect=[child, parent])
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.get_note", 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, None])
|
|
with patch("scribe.mcp.tools.tasks.notes_svc.get_note", mock_get):
|
|
out = await get_task(task_id=10)
|
|
assert out["parent_title"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_task_raises_when_not_found():
|
|
with patch(
|
|
"scribe.mcp.tools.tasks.notes_svc.get_note",
|
|
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")
|