bvandeusen
9fa474b3c4
CI & Build / Python lint (push) Successful in 3s
CI & Build / integration (push) Successful in 31s
CI & Build / TypeScript typecheck (push) Successful in 42s
CI & Build / Python tests (push) Successful in 56s
CI & Build / Build & push image (push) Successful in 1m0s
The injected menu tells the agent to open any hit with get_note(id), and that menu can list a collaborator's record reached through a shared project. The fetch was still owner-only, so those lines answered "not found" — for a record the same user opens fine in the browser. The agent path was strictly narrower than the web path for the same id. Same boundary miss as #2093: the list side was widened for sharing, the fetch side wasn't. Both tools now resolve through get_note_for_user, apply the trash filter themselves (permission resolution says nothing about liveness), and attach describe_provenance so a shared record arrives marked as someone else's rather than passing as the caller's. routes/tasks.py's parent-title lookup had the same narrowness: a shared subtask rendered as an orphan when its parent was equally shared. Four fetch tests across three files were patching notes_svc.get_note and had to be retargeted — note 2109's third sub-case, caught by grepping tests/ for the old name before pushing rather than by CI (#2159). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RLwAaV4DQEmVyn496HnEvt
255 lines
8.9 KiB
Python
255 lines
8.9 KiB
Python
"""Tests for fable_*_note tools."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from scribe.mcp._context import _user_id_ctx
|
|
from scribe.mcp.tools.notes import (
|
|
list_notes, get_note, create_note,
|
|
update_note, delete_note,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _bind_user():
|
|
token = _user_id_ctx.set(7)
|
|
yield
|
|
_user_id_ctx.reset(token)
|
|
|
|
|
|
def _fake_note(*, user_id: int = 7, **overrides) -> MagicMock:
|
|
note = MagicMock()
|
|
base = {"id": 1, "title": "t", "body": "b", "tags": [], "is_task": False}
|
|
base.update(overrides)
|
|
note.to_dict.return_value = base
|
|
# Real values, not auto-attributes: get_note reads deleted_at, and compares
|
|
# user_id against the bound caller to decide whether to attach a shared/owner
|
|
# marker. A MagicMock is truthy on both, so it would read as another user's
|
|
# trashed note and reach for the DB (note 2109).
|
|
note.id = base["id"]
|
|
note.user_id = user_id
|
|
note.deleted_at = None
|
|
return note
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_note_blocked_by_duplicate_gate():
|
|
from scribe.services.dedup import DuplicateMatch
|
|
dup = DuplicateMatch(id=88, title="Embeddings notes", similarity=0.94, reason="semantic")
|
|
create_mock = AsyncMock()
|
|
with patch("scribe.mcp.tools.notes.dedup_svc.find_duplicate_note",
|
|
AsyncMock(return_value=dup)), \
|
|
patch("scribe.mcp.tools.notes.notes_svc.create_note", create_mock):
|
|
out = await create_note(title="embeddings", body="notes about embeddings")
|
|
assert out["duplicate"] is True
|
|
assert out["existing_id"] == 88
|
|
assert out["match"] == "semantic"
|
|
create_mock.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_note_force_bypasses_duplicate_gate():
|
|
find_mock = AsyncMock()
|
|
with patch("scribe.mcp.tools.notes.dedup_svc.find_duplicate_note", find_mock), \
|
|
patch("scribe.mcp.tools.notes.notes_svc.create_note",
|
|
AsyncMock(return_value=_fake_note(id=3))):
|
|
out = await create_note(title="dup", force=True)
|
|
assert out["id"] == 3
|
|
find_mock.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_notes_repackages_tuple_into_dict():
|
|
rows = [_fake_note(id=1), _fake_note(id=2)]
|
|
with patch(
|
|
"scribe.mcp.tools.notes.notes_svc.list_notes",
|
|
AsyncMock(return_value=(rows, 2)),
|
|
):
|
|
out = await list_notes()
|
|
assert out["total"] == 2
|
|
assert len(out["notes"]) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_notes_passes_is_task_false():
|
|
mock = AsyncMock(return_value=([], 0))
|
|
with patch("scribe.mcp.tools.notes.notes_svc.list_notes", mock):
|
|
await list_notes()
|
|
assert mock.call_args.kwargs["is_task"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_notes_tag_filter_maps_to_list():
|
|
"""The single-tag string param maps to a one-element list at the service layer."""
|
|
mock = AsyncMock(return_value=([], 0))
|
|
with patch("scribe.mcp.tools.notes.notes_svc.list_notes", mock):
|
|
await list_notes(tag="ops")
|
|
assert mock.call_args.kwargs["tags"] == ["ops"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_notes_empty_tag_means_no_filter():
|
|
mock = AsyncMock(return_value=([], 0))
|
|
with patch("scribe.mcp.tools.notes.notes_svc.list_notes", mock):
|
|
await list_notes(tag="")
|
|
assert mock.call_args.kwargs["tags"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_notes_search_text_maps_to_q():
|
|
mock = AsyncMock(return_value=([], 0))
|
|
with patch("scribe.mcp.tools.notes.notes_svc.list_notes", mock):
|
|
await list_notes(search_text="kafka")
|
|
assert mock.call_args.kwargs["q"] == "kafka"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_notes_limit_clamped():
|
|
mock = AsyncMock(return_value=([], 0))
|
|
with patch("scribe.mcp.tools.notes.notes_svc.list_notes", mock):
|
|
await list_notes(limit=9999)
|
|
assert mock.call_args.kwargs["limit"] == 100
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_note_returns_dict():
|
|
fake = _fake_note(id=5, title="found")
|
|
with patch(
|
|
"scribe.mcp.tools.notes.notes_svc.get_note_for_user",
|
|
AsyncMock(return_value=(fake, "owner")),
|
|
):
|
|
out = await get_note(note_id=5)
|
|
assert out["id"] == 5
|
|
assert out["title"] == "found"
|
|
# Own record: no provenance noise.
|
|
assert "shared" not in out
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_note_opens_a_shared_record_and_says_whose_it_is():
|
|
"""The injected menu tells the agent to open any hit with get_note(id), and
|
|
that menu can list a collaborator's note reached through a shared project.
|
|
An owner-only fetch would answer "not found" for a record the agent was just
|
|
handed — and the web UI opens the same note fine."""
|
|
theirs = _fake_note(id=5, title="Their note", user_id=9)
|
|
with patch(
|
|
"scribe.mcp.tools.notes.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_note(note_id=5)
|
|
assert out["shared"] is True
|
|
assert out["owner"] == "alex"
|
|
assert out["permission"] == "viewer"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_note_raises_when_not_found():
|
|
with patch(
|
|
"scribe.mcp.tools.notes.notes_svc.get_note_for_user",
|
|
AsyncMock(return_value=None),
|
|
):
|
|
with pytest.raises(ValueError, match="note 999 not found"):
|
|
await get_note(note_id=999)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_note_treats_a_trashed_note_as_missing():
|
|
"""get_note_for_user resolves permission, not liveness — the trash filter is
|
|
the caller's to apply."""
|
|
trashed = _fake_note(id=5)
|
|
trashed.deleted_at = "2026-07-01T00:00:00Z"
|
|
with patch(
|
|
"scribe.mcp.tools.notes.notes_svc.get_note_for_user",
|
|
AsyncMock(return_value=(trashed, "owner")),
|
|
):
|
|
with pytest.raises(ValueError, match="note 5 not found"):
|
|
await get_note(note_id=5)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_note_passes_through():
|
|
fake = _fake_note(id=10, title="new")
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.notes.notes_svc.create_note", mock):
|
|
out = await create_note(title="new", body="x", tags=["a"], project_id=5)
|
|
assert out["id"] == 10
|
|
assert mock.call_args.kwargs["title"] == "new"
|
|
assert mock.call_args.kwargs["project_id"] == 5
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_note_project_zero_becomes_none():
|
|
"""project_id=0 sentinel must become None at the service layer (orphan note)."""
|
|
fake = _fake_note()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.notes.notes_svc.create_note", mock):
|
|
await create_note(title="t", project_id=0)
|
|
assert mock.call_args.kwargs["project_id"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_note_only_sends_non_default_fields():
|
|
"""Omitted (default) fields must NOT reach the service — otherwise they'd
|
|
overwrite real data with empty strings."""
|
|
fake = _fake_note()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.notes.notes_svc.update_note", mock):
|
|
await update_note(note_id=1, title="new title")
|
|
# Service got user_id, note_id positional + only the title kwarg
|
|
args, kwargs = mock.call_args
|
|
assert args == (7, 1)
|
|
assert kwargs == {"title": "new title"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_note_empty_tags_clears_explicitly():
|
|
"""tags=[] is an explicit clear, distinct from tags=None (omit)."""
|
|
fake = _fake_note()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.notes.notes_svc.update_note", mock):
|
|
await update_note(note_id=1, tags=[])
|
|
assert mock.call_args.kwargs == {"tags": []}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_note_tags_none_means_omit():
|
|
fake = _fake_note()
|
|
mock = AsyncMock(return_value=fake)
|
|
with patch("scribe.mcp.tools.notes.notes_svc.update_note", mock):
|
|
await update_note(note_id=1, tags=None)
|
|
assert "tags" not in mock.call_args.kwargs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_note_raises_when_not_found():
|
|
with patch(
|
|
"scribe.mcp.tools.notes.notes_svc.update_note",
|
|
AsyncMock(return_value=None),
|
|
):
|
|
with pytest.raises(ValueError, match="note 999 not found"):
|
|
await update_note(note_id=999, title="x")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_note_soft_deletes_and_returns_batch():
|
|
with patch(
|
|
"scribe.mcp.tools.notes.trash_svc.delete",
|
|
AsyncMock(return_value="batch-1"),
|
|
):
|
|
result = await delete_note(note_id=7)
|
|
assert result["deleted_batch_id"] == "batch-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_note_raises_when_not_found():
|
|
with patch(
|
|
"scribe.mcp.tools.notes.trash_svc.delete",
|
|
AsyncMock(return_value=None),
|
|
):
|
|
with pytest.raises(ValueError, match="note 999 not found"):
|
|
await delete_note(note_id=999)
|