fix(acl): align MCP writes with the share model — editor edits, viewer doesn't
CI & Build / Python lint (push) Successful in 3s
CI & Build / integration (push) Successful in 33s
CI & Build / TypeScript typecheck (push) Successful in 33s
CI & Build / Python tests (push) Failing after 34s
CI & Build / Build & push image (push) Has been skipped
CI & Build / Python lint (push) Successful in 3s
CI & Build / integration (push) Successful in 33s
CI & Build / TypeScript typecheck (push) Successful in 33s
CI & Build / Python tests (push) Failing after 34s
CI & Build / Build & push image (push) Has been skipped
Option B, per the operator. Closes the last inconsistency from the ACL work.
The agent path had drifted into an indefensible position: delete_snippet honoured
editor shares (I made it share-aware so widening the read wouldn't let a VIEWER
trash things) while update_snippet still resolved through the owner-only
notes.get_note. So through an agent you could destroy a colleague's snippet but
not improve it — and the refusal claimed "not found" for a record you could
plainly open.
Now update_snippet, merge_snippets and update_process all resolve the read scope
and then require can_write_note, matching the REST routes and the sharing UI's
own promise that viewer / editor / admin are distinct grants. A viewer grant is
refused with the actual reason ("shared with you read-only — ask its owner for
edit access, or record your own version"), because not-found would send an agent
hunting for a missing id instead of recording its own copy.
Authorised writes are performed as the OWNER, since the underlying note update is
owner-scoped and a shared editor's own id would match nothing.
Merge additionally requires each source to share the TARGET'S owner and to be
writable by the caller — merging trashes the source, so read access isn't enough,
and cross-owner merge stays out of scope (#231). Sources failing either test are
skipped rather than half-merged.
A record the caller cannot read at all still returns not-found rather than
forbidden, so the error can't be used to confirm that an id exists.
Also fixed _fake_snippet's missing user_id proactively — the same
auto-MagicMock-reads-as-foreign trap that broke CI twice (see note 2109).
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RLwAaV4DQEmVyn496HnEvt
This commit is contained in:
@@ -236,7 +236,9 @@ adopting it or acting on it. This matters most for a shared Process — do not r
|
|||||||
one as written; describe what it would do and get the operator's go-ahead.
|
one as written; describe what it would do and get the operator's go-ahead.
|
||||||
Records shared directly with the operator are also deliberately search-only:
|
Records shared directly with the operator are also deliberately search-only:
|
||||||
they surface when you look for them (pass a query), not in plain lists, so
|
they surface when you look for them (pass a query), not in plain lists, so
|
||||||
nobody else's material arrives unasked.
|
nobody else's material arrives unasked. Editing another user's record needs an
|
||||||
|
editor or admin share from them; a read-only share is refused, and the right
|
||||||
|
answer is usually to record the operator's own version rather than to push.
|
||||||
|
|
||||||
When developing Scribe itself, honor its multi-user sharing ACL: scope every
|
When developing Scribe itself, honor its multi-user sharing ACL: scope every
|
||||||
read and mutation of user data by owner + shares — never assume a single
|
read and mutation of user data by owner + shares — never assume a single
|
||||||
|
|||||||
@@ -88,11 +88,21 @@ async def get_process(name_or_id: str) -> dict:
|
|||||||
async def update_process(process_id: int, title: str = "", body: str = "",
|
async def update_process(process_id: int, title: str = "", body: str = "",
|
||||||
tags: list[str] | None = None) -> dict:
|
tags: list[str] | None = None) -> dict:
|
||||||
"""Update a stored process. Only provided fields change — empty title/body
|
"""Update a stored process. Only provided fields change — empty title/body
|
||||||
leave that field unchanged; pass tags to replace the tag set."""
|
leave that field unchanged; pass tags to replace the tag set.
|
||||||
|
|
||||||
|
Editing another user's process requires an editor or admin share from them; a
|
||||||
|
read-only share is not enough and says so rather than claiming not-found.
|
||||||
|
"""
|
||||||
uid = current_user_id()
|
uid = current_user_id()
|
||||||
note = await notes_svc.get_note(uid, process_id)
|
loaded = await notes_svc.get_note_for_user(uid, process_id)
|
||||||
if note is None or note.note_type != "process":
|
note = loaded[0] if loaded else None
|
||||||
|
if note is None or note.note_type != "process" or note.deleted_at is not None:
|
||||||
raise ValueError(f"process {process_id} not found")
|
raise ValueError(f"process {process_id} not found")
|
||||||
|
if not await access_svc.can_write_note(uid, process_id):
|
||||||
|
raise ValueError(
|
||||||
|
f"process {process_id} is shared with you read-only — ask its owner "
|
||||||
|
f"for edit access, or save your own copy with create_process"
|
||||||
|
)
|
||||||
fields: dict = {}
|
fields: dict = {}
|
||||||
if title.strip():
|
if title.strip():
|
||||||
fields["title"] = title.strip()
|
fields["title"] = title.strip()
|
||||||
@@ -100,8 +110,13 @@ async def update_process(process_id: int, title: str = "", body: str = "",
|
|||||||
fields["body"] = body
|
fields["body"] = body
|
||||||
if tags is not None:
|
if tags is not None:
|
||||||
fields["tags"] = tags
|
fields["tags"] = tags
|
||||||
updated = await notes_svc.update_note(uid, process_id, **fields)
|
# As the owner — update_note is owner-scoped and the write is authorised above.
|
||||||
return updated.to_dict()
|
updated = await notes_svc.update_note(note.user_id, process_id, **fields)
|
||||||
|
if updated is None:
|
||||||
|
raise ValueError(f"process {process_id} not found")
|
||||||
|
out = updated.to_dict()
|
||||||
|
out.update(await access_svc.describe_provenance(uid, updated))
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
def register(mcp) -> None:
|
def register(mcp) -> None:
|
||||||
|
|||||||
@@ -180,6 +180,10 @@ async def update_snippet(
|
|||||||
tags: Replaces the extra-tag set (language + "snippet" are re-derived).
|
tags: Replaces the extra-tag set (language + "snippet" are re-derived).
|
||||||
project_id: 0 leaves it unchanged, -1 detaches it from its project, a
|
project_id: 0 leaves it unchanged, -1 detaches it from its project, a
|
||||||
positive id moves it.
|
positive id moves it.
|
||||||
|
|
||||||
|
Editing someone else's snippet requires an editor or admin share from them.
|
||||||
|
A read-only share is refused with a message saying so — record your own
|
||||||
|
version instead of trying to force it.
|
||||||
"""
|
"""
|
||||||
uid = current_user_id()
|
uid = current_user_id()
|
||||||
if project_id == 0:
|
if project_id == 0:
|
||||||
@@ -189,21 +193,27 @@ async def update_snippet(
|
|||||||
else:
|
else:
|
||||||
project = project_id
|
project = project_id
|
||||||
|
|
||||||
note = await snippets_svc.update_snippet(
|
try:
|
||||||
uid, snippet_id,
|
note = await snippets_svc.update_snippet(
|
||||||
name=name, code=code, language=language,
|
uid, snippet_id,
|
||||||
signature=signature, when_to_use=when_to_use,
|
name=name, code=code, language=language,
|
||||||
repo=repo, path=path, symbol=symbol,
|
signature=signature, when_to_use=when_to_use,
|
||||||
locations=locations, tags=tags, project_id=project,
|
repo=repo, path=path, symbol=symbol,
|
||||||
)
|
locations=locations, tags=tags, project_id=project,
|
||||||
|
)
|
||||||
|
except PermissionError as exc:
|
||||||
|
# Readable but not writable — surface the real reason, not "not found".
|
||||||
|
raise ValueError(str(exc)) from exc
|
||||||
if note is None:
|
if note is None:
|
||||||
raise ValueError(f"snippet {snippet_id} not found")
|
raise ValueError(f"snippet {snippet_id} not found")
|
||||||
if system_ids is not None:
|
if system_ids is not None:
|
||||||
await systems_svc.set_record_systems(uid, snippet_id, system_ids)
|
await systems_svc.set_record_systems(note.user_id, snippet_id, system_ids)
|
||||||
data = snippets_svc.snippet_to_dict(note)
|
data = snippets_svc.snippet_to_dict(note)
|
||||||
|
data.update(await access_svc.describe_provenance(uid, note))
|
||||||
if system_ids is not None:
|
if system_ids is not None:
|
||||||
data["systems"] = [
|
data["systems"] = [
|
||||||
s.to_dict() for s in await systems_svc.list_record_systems(uid, snippet_id)
|
s.to_dict()
|
||||||
|
for s in await systems_svc.list_record_systems(note.user_id, snippet_id)
|
||||||
]
|
]
|
||||||
return data
|
return data
|
||||||
|
|
||||||
@@ -245,7 +255,10 @@ async def merge_snippets(target_id: int, source_ids: list[int]) -> dict:
|
|||||||
ids = [s for s in (source_ids or []) if s != target_id]
|
ids = [s for s in (source_ids or []) if s != target_id]
|
||||||
if not ids:
|
if not ids:
|
||||||
raise ValueError("merge_snippets requires at least one other source_id")
|
raise ValueError("merge_snippets requires at least one other source_id")
|
||||||
result = await snippets_svc.merge_snippets(uid, target_id, ids)
|
try:
|
||||||
|
result = await snippets_svc.merge_snippets(uid, target_id, ids)
|
||||||
|
except PermissionError as exc:
|
||||||
|
raise ValueError(str(exc)) from exc
|
||||||
if result is None:
|
if result is None:
|
||||||
raise ValueError(f"snippet {target_id} not found")
|
raise ValueError(f"snippet {target_id} not found")
|
||||||
note, merged_ids = result
|
note, merged_ids = result
|
||||||
|
|||||||
@@ -354,7 +354,14 @@ async def update_snippet(
|
|||||||
):
|
):
|
||||||
"""Partial update: only fields passed (not None) change. Re-serializes the
|
"""Partial update: only fields passed (not None) change. Re-serializes the
|
||||||
merged field set back into title/body/tags. Returns the Note, or None if the
|
merged field set back into title/body/tags. Returns the Note, or None if the
|
||||||
id isn't a snippet.
|
id isn't a snippet the caller can see.
|
||||||
|
|
||||||
|
Share-aware (rule #47/#78): resolves the read scope, then requires WRITE —
|
||||||
|
so an editor/admin grant lets the holder edit, and a viewer grant does not.
|
||||||
|
Raises PermissionError when the caller can read but not write, because "not
|
||||||
|
found" would be a lie about a record they can plainly open. The write itself
|
||||||
|
is performed as the OWNER, mirroring routes/snippets.py, since the underlying
|
||||||
|
note update is owner-scoped.
|
||||||
|
|
||||||
``project_id``: omit to leave unchanged, pass None to detach from its
|
``project_id``: omit to leave unchanged, pass None to detach from its
|
||||||
project, pass an id to move it.
|
project, pass an id to move it.
|
||||||
@@ -362,9 +369,15 @@ async def update_snippet(
|
|||||||
Locations: ``locations`` replaces the whole set; else a legacy single
|
Locations: ``locations`` replaces the whole set; else a legacy single
|
||||||
``repo``/``path``/``symbol`` overlays onto the first existing location; else
|
``repo``/``path``/``symbol`` overlays onto the first existing location; else
|
||||||
the existing locations are kept."""
|
the existing locations are kept."""
|
||||||
note = await notes_svc.get_note(user_id, snippet_id)
|
note = await get_snippet(user_id, snippet_id)
|
||||||
if note is None or note.note_type != SNIPPET_NOTE_TYPE:
|
if note is None:
|
||||||
return None
|
return None
|
||||||
|
from scribe.services.access import can_write_note
|
||||||
|
if not await can_write_note(user_id, snippet_id):
|
||||||
|
raise PermissionError(
|
||||||
|
f"snippet {snippet_id} is shared with you read-only — ask its owner "
|
||||||
|
f"for edit access, or record your own version"
|
||||||
|
)
|
||||||
|
|
||||||
cur = parse_snippet_fields(note.title, note.body, note.tags)
|
cur = parse_snippet_fields(note.title, note.body, note.tags)
|
||||||
overlay = {
|
overlay = {
|
||||||
@@ -404,7 +417,9 @@ async def update_snippet(
|
|||||||
if project_id is not UNSET:
|
if project_id is not UNSET:
|
||||||
fields["project_id"] = project_id
|
fields["project_id"] = project_id
|
||||||
|
|
||||||
updated = await notes_svc.update_note(user_id, snippet_id, **fields)
|
# As the OWNER: update_note is owner-scoped, so a shared editor's own id
|
||||||
|
# would find nothing. The write was authorised by can_write_note above.
|
||||||
|
updated = await notes_svc.update_note(note.user_id, snippet_id, **fields)
|
||||||
if updated is not None:
|
if updated is not None:
|
||||||
# Title/body changed → refresh the embedding so recall reflects the edit.
|
# Title/body changed → refresh the embedding so recall reflects the edit.
|
||||||
_embed_snippet(updated)
|
_embed_snippet(updated)
|
||||||
@@ -460,23 +475,41 @@ def merge_snippet_fields(
|
|||||||
|
|
||||||
|
|
||||||
async def merge_snippets(user_id: int, target_id: int, source_ids: list[int]):
|
async def merge_snippets(user_id: int, target_id: int, source_ids: list[int]):
|
||||||
"""Unify source snippets INTO the target (all owned by user_id): union their
|
"""Unify source snippets INTO the target: union their locations + tags onto
|
||||||
locations + tags onto the canonical target, keep the target's scalar fields,
|
the canonical target, keep the target's scalar fields, trash the sources
|
||||||
trash the sources (recoverable), re-embed the survivor.
|
(recoverable), re-embed the survivor.
|
||||||
|
|
||||||
|
Share-aware and write-gated, like update: an editor/admin grant is enough, a
|
||||||
|
viewer grant is not. Raises PermissionError when the caller can read the
|
||||||
|
target but not write it. Every source must share the TARGET'S OWNER —
|
||||||
|
cross-owner merge stays out of scope (#231) — and must itself be writable;
|
||||||
|
sources failing either test are skipped rather than silently half-merged.
|
||||||
|
|
||||||
Returns (merged_target_note, merged_source_ids), or None if the target isn't
|
Returns (merged_target_note, merged_source_ids), or None if the target isn't
|
||||||
the user's snippet. Source ids that aren't the user's snippets are skipped."""
|
a snippet the caller can see."""
|
||||||
target = await notes_svc.get_note(user_id, target_id)
|
from scribe.services.access import can_write_note
|
||||||
if target is None or target.note_type != SNIPPET_NOTE_TYPE:
|
target = await get_snippet(user_id, target_id)
|
||||||
|
if target is None:
|
||||||
return None
|
return None
|
||||||
|
if not await can_write_note(user_id, target_id):
|
||||||
|
raise PermissionError(
|
||||||
|
f"snippet {target_id} is shared with you read-only — you can't merge "
|
||||||
|
f"into a record you can't edit"
|
||||||
|
)
|
||||||
|
owner_id = target.user_id
|
||||||
|
|
||||||
sources = []
|
sources = []
|
||||||
for sid in source_ids:
|
for sid in source_ids:
|
||||||
if sid == target_id:
|
if sid == target_id:
|
||||||
continue
|
continue
|
||||||
s = await notes_svc.get_note(user_id, sid)
|
s = await get_snippet(user_id, sid)
|
||||||
if s is not None and s.note_type == SNIPPET_NOTE_TYPE:
|
# Same owner as the target, and writable by this caller. Merging trashes
|
||||||
sources.append(s)
|
# the source, so read access is not enough.
|
||||||
|
if s is None or s.user_id != owner_id:
|
||||||
|
continue
|
||||||
|
if not await can_write_note(user_id, sid):
|
||||||
|
continue
|
||||||
|
sources.append(s)
|
||||||
|
|
||||||
tgt_fields = parse_snippet_fields(target.title, target.body, target.tags)
|
tgt_fields = parse_snippet_fields(target.title, target.body, target.tags)
|
||||||
parsed_sources = [
|
parsed_sources = [
|
||||||
@@ -484,8 +517,9 @@ async def merge_snippets(user_id: int, target_id: int, source_ids: list[int]):
|
|||||||
]
|
]
|
||||||
locations, extra_tags = merge_snippet_fields(tgt_fields, target.tags, parsed_sources)
|
locations, extra_tags = merge_snippet_fields(tgt_fields, target.tags, parsed_sources)
|
||||||
|
|
||||||
|
# Owner-scoped write, authorised above — same reason as update_snippet.
|
||||||
updated = await notes_svc.update_note(
|
updated = await notes_svc.update_note(
|
||||||
user_id, target_id,
|
owner_id, target_id,
|
||||||
body=compose_body(
|
body=compose_body(
|
||||||
code=tgt_fields["code"], language=tgt_fields["language"],
|
code=tgt_fields["code"], language=tgt_fields["language"],
|
||||||
signature=tgt_fields["signature"], when_to_use=tgt_fields["when_to_use"],
|
signature=tgt_fields["signature"], when_to_use=tgt_fields["when_to_use"],
|
||||||
@@ -500,7 +534,7 @@ async def merge_snippets(user_id: int, target_id: int, source_ids: list[int]):
|
|||||||
from scribe.services.trash import delete as trash_delete
|
from scribe.services.trash import delete as trash_delete
|
||||||
merged_ids: list[int] = []
|
merged_ids: list[int] = []
|
||||||
for s in sources:
|
for s in sources:
|
||||||
batch = await trash_delete(user_id, "note", s.id)
|
batch = await trash_delete(owner_id, "note", s.id)
|
||||||
if batch is not None:
|
if batch is not None:
|
||||||
merged_ids.append(s.id)
|
merged_ids.append(s.id)
|
||||||
|
|
||||||
|
|||||||
@@ -13,13 +13,17 @@ def _bind_user():
|
|||||||
_user_id_ctx.reset(token)
|
_user_id_ctx.reset(token)
|
||||||
|
|
||||||
|
|
||||||
def _fake_snippet():
|
def _fake_snippet(user_id: int = 7):
|
||||||
n = MagicMock()
|
n = MagicMock()
|
||||||
n.id = 1
|
n.id = 1
|
||||||
n.title = "debounce — rate-limit a callback"
|
n.title = "debounce — rate-limit a callback"
|
||||||
n.body = "```js\nreturn 1\n```\n"
|
n.body = "```js\nreturn 1\n```\n"
|
||||||
n.tags = ["js", "snippet"]
|
n.tags = ["js", "snippet"]
|
||||||
n.note_type = "snippet"
|
n.note_type = "snippet"
|
||||||
|
# Real int, matching the bound caller by default. The tools compare it to
|
||||||
|
# decide whether to attach a shared/owner marker; an auto-MagicMock would read
|
||||||
|
# as another user's record and send them off to look up a username.
|
||||||
|
n.user_id = user_id
|
||||||
n.to_dict.return_value = {
|
n.to_dict.return_value = {
|
||||||
"id": 1, "title": n.title, "note_type": "snippet", "tags": n.tags,
|
"id": 1, "title": n.title, "note_type": "snippet", "tags": n.tags,
|
||||||
}
|
}
|
||||||
@@ -128,6 +132,31 @@ async def test_create_and_update_pass_locations_through():
|
|||||||
assert mock_update.await_args.kwargs["locations"] == locs
|
assert mock_update.await_args.kwargs["locations"] == locs
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_update_snippet_read_only_share_says_why():
|
||||||
|
"""A viewer grant must be refused with the REAL reason. "Not found" would be
|
||||||
|
a lie about a record the caller can plainly open, and would send an agent
|
||||||
|
hunting for a missing id instead of recording its own version."""
|
||||||
|
with patch("scribe.services.snippets.update_snippet",
|
||||||
|
AsyncMock(side_effect=PermissionError(
|
||||||
|
"snippet 1 is shared with you read-only — ask its owner for "
|
||||||
|
"edit access, or record your own version"))):
|
||||||
|
from scribe.mcp.tools.snippets import update_snippet
|
||||||
|
with pytest.raises(ValueError, match="read-only"):
|
||||||
|
await update_snippet(1, name="x")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_merge_snippets_read_only_target_says_why():
|
||||||
|
with patch("scribe.services.snippets.merge_snippets",
|
||||||
|
AsyncMock(side_effect=PermissionError(
|
||||||
|
"snippet 1 is shared with you read-only — you can't merge "
|
||||||
|
"into a record you can't edit"))):
|
||||||
|
from scribe.mcp.tools.snippets import merge_snippets
|
||||||
|
with pytest.raises(ValueError, match="read-only"):
|
||||||
|
await merge_snippets(target_id=1, source_ids=[2])
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_delete_snippet_retires_or_raises():
|
async def test_delete_snippet_retires_or_raises():
|
||||||
from scribe.mcp.tools.snippets import delete_snippet
|
from scribe.mcp.tools.snippets import delete_snippet
|
||||||
|
|||||||
@@ -0,0 +1,111 @@
|
|||||||
|
"""Editing a shared record: an editor grant is enough, a viewer grant is not.
|
||||||
|
|
||||||
|
The agent and web paths used to disagree — and worse, within the agent path
|
||||||
|
`delete_snippet` honoured editor shares while `update_snippet` did not, so a
|
||||||
|
colleague's snippet could be destroyed through an agent but not improved. Both
|
||||||
|
now resolve the read scope and then require WRITE, which is what the sharing UI
|
||||||
|
already promises: viewer / editor / admin are distinct grants.
|
||||||
|
|
||||||
|
A record readable-but-not-writable raises PermissionError rather than reporting
|
||||||
|
"not found" — the caller can plainly open it, so not-found would be a lie that
|
||||||
|
sends them hunting for a missing id.
|
||||||
|
"""
|
||||||
|
from unittest.mock import AsyncMock, MagicMock, patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
|
def _snippet(id=1, owner=9):
|
||||||
|
n = MagicMock()
|
||||||
|
n.id = id
|
||||||
|
n.user_id = owner
|
||||||
|
n.title = "formatDuration — humanize a millisecond count"
|
||||||
|
n.body = "```ts\nexport const f = 1\n```\n"
|
||||||
|
n.tags = ["ts", "snippet"]
|
||||||
|
n.note_type = "snippet"
|
||||||
|
n.deleted_at = None
|
||||||
|
return n
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_editor_share_can_update_and_writes_as_the_owner():
|
||||||
|
"""The underlying note update is owner-scoped, so a shared editor's own id
|
||||||
|
would match nothing — the authorised write has to be performed as the owner."""
|
||||||
|
from scribe.services import snippets as svc
|
||||||
|
note = _snippet(owner=9)
|
||||||
|
updated = _snippet(owner=9)
|
||||||
|
with patch.object(svc, "get_snippet", AsyncMock(return_value=note)), \
|
||||||
|
patch("scribe.services.access.can_write_note", AsyncMock(return_value=True)), \
|
||||||
|
patch.object(svc.notes_svc, "update_note",
|
||||||
|
AsyncMock(return_value=updated)) as mock_update, \
|
||||||
|
patch.object(svc, "_embed_snippet", MagicMock()):
|
||||||
|
got = await svc.update_snippet(7, 1, name="formatDuration")
|
||||||
|
|
||||||
|
assert got is updated
|
||||||
|
# Positional user_id is the OWNER (9), not the caller (7).
|
||||||
|
assert mock_update.await_args.args[0] == 9
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_viewer_share_cannot_update():
|
||||||
|
from scribe.services import snippets as svc
|
||||||
|
with patch.object(svc, "get_snippet", AsyncMock(return_value=_snippet())), \
|
||||||
|
patch("scribe.services.access.can_write_note", AsyncMock(return_value=False)), \
|
||||||
|
patch.object(svc.notes_svc, "update_note", AsyncMock()) as mock_update:
|
||||||
|
with pytest.raises(PermissionError, match="read-only"):
|
||||||
|
await svc.update_snippet(7, 1, name="nope")
|
||||||
|
mock_update.assert_not_awaited()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_viewer_share_cannot_delete():
|
||||||
|
"""delete_snippet returns False rather than raising — its contract is boolean
|
||||||
|
— but a viewer must not be able to bin someone else's record."""
|
||||||
|
from scribe.services import snippets as svc
|
||||||
|
with patch.object(svc, "get_snippet", AsyncMock(return_value=_snippet())), \
|
||||||
|
patch("scribe.services.access.can_write_note", AsyncMock(return_value=False)), \
|
||||||
|
patch("scribe.services.trash.delete", AsyncMock()) as mock_trash:
|
||||||
|
assert await svc.delete_snippet(7, 1) is False
|
||||||
|
mock_trash.assert_not_awaited()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_unreadable_record_is_not_found_not_forbidden():
|
||||||
|
"""A record the caller can't see at all must NOT be distinguishable from one
|
||||||
|
that doesn't exist — otherwise the error itself confirms it exists."""
|
||||||
|
from scribe.services import snippets as svc
|
||||||
|
with patch.object(svc, "get_snippet", AsyncMock(return_value=None)):
|
||||||
|
assert await svc.update_snippet(7, 404, name="x") is None
|
||||||
|
assert await svc.delete_snippet(7, 404) is False
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_merge_requires_write_on_target():
|
||||||
|
from scribe.services import snippets as svc
|
||||||
|
with patch.object(svc, "get_snippet", AsyncMock(return_value=_snippet())), \
|
||||||
|
patch("scribe.services.access.can_write_note", AsyncMock(return_value=False)):
|
||||||
|
with pytest.raises(PermissionError, match="read-only"):
|
||||||
|
await svc.merge_snippets(7, 1, [2])
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_merge_skips_sources_owned_by_someone_else():
|
||||||
|
"""Cross-owner merge stays out of scope (#231): a source belonging to a
|
||||||
|
different owner than the target is skipped, not silently folded in and
|
||||||
|
trashed."""
|
||||||
|
from scribe.services import snippets as svc
|
||||||
|
target = _snippet(id=1, owner=9)
|
||||||
|
same_owner = _snippet(id=2, owner=9)
|
||||||
|
other_owner = _snippet(id=3, owner=42)
|
||||||
|
|
||||||
|
async def fake_get(_uid, sid):
|
||||||
|
return {1: target, 2: same_owner, 3: other_owner}.get(sid)
|
||||||
|
|
||||||
|
with patch.object(svc, "get_snippet", AsyncMock(side_effect=fake_get)), \
|
||||||
|
patch("scribe.services.access.can_write_note", AsyncMock(return_value=True)), \
|
||||||
|
patch.object(svc.notes_svc, "update_note", AsyncMock(return_value=target)), \
|
||||||
|
patch("scribe.services.trash.delete", AsyncMock(return_value=object())), \
|
||||||
|
patch.object(svc, "_embed_snippet", MagicMock()):
|
||||||
|
_note, merged_ids = await svc.merge_snippets(7, 1, [2, 3])
|
||||||
|
|
||||||
|
assert merged_ids == [2]
|
||||||
Reference in New Issue
Block a user