bvandeusen
dd1fc2d506
Completes the Phase 5 follow-up: rules now get the same update-over-create gate. Title-based only (rules aren't a semantic-retrieval/RAG surface), scoped to the same topic (rulebook rule) or same project (project rule). force=true overrides; fail-open like the note/task gate. Deferred-item decisions (operator): REST/web gating SKIPPED (kept MCP-only — humans rarely double-create and a hard block needs UI affordance); orphan scope kept orphan↔orphan (no change). So this rule gate is the only remaining build. - services/dedup.py: find_duplicate_rule(title, topic_id|project_id). - create_rule + create_project_rule: force param + gate. - tests: rule title match, scope-required guard, tool gate (block + force). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
"""Unit tests for the write-time near-duplicate gate (services/dedup.py)."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from scribe.services.dedup import (
|
|
DuplicateMatch,
|
|
duplicate_response,
|
|
find_duplicate_note,
|
|
find_duplicate_rule,
|
|
)
|
|
|
|
|
|
def _session_returning(note):
|
|
"""A mocked async_session() whose single execute() yields `note` (or None)."""
|
|
s = AsyncMock()
|
|
s.__aenter__ = AsyncMock(return_value=s)
|
|
s.__aexit__ = AsyncMock(return_value=False)
|
|
result = MagicMock()
|
|
result.scalars.return_value.first.return_value = note
|
|
s.execute = AsyncMock(return_value=result)
|
|
return s
|
|
|
|
|
|
def _fake_note(id=1, title="T", note_type="note"):
|
|
n = MagicMock()
|
|
n.id, n.title, n.note_type = id, title, note_type
|
|
return n
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_title_exact_match_returns_title_duplicate():
|
|
note = _fake_note(id=10, title="Setup CI")
|
|
with patch("scribe.services.dedup.async_session",
|
|
return_value=_session_returning(note)):
|
|
# whitespace/case differences are normalized away
|
|
dup = await find_duplicate_note(7, " setup ci ", project_id=2, is_task=True)
|
|
assert dup is not None
|
|
assert dup.id == 10
|
|
assert dup.reason == "title"
|
|
assert dup.similarity == 1.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_short_body_skips_semantic_check():
|
|
sem = AsyncMock()
|
|
with patch("scribe.services.dedup.async_session",
|
|
return_value=_session_returning(None)), \
|
|
patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
|
|
dup = await find_duplicate_note(7, "Unique", body="too short", project_id=2)
|
|
assert dup is None
|
|
sem.assert_not_called() # body under _MIN_BODY_FOR_SEMANTIC
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_semantic_match_when_body_substantial():
|
|
hit = _fake_note(id=20, title="Existing", note_type="note")
|
|
sem = AsyncMock(return_value=[(0.93, hit)])
|
|
with patch("scribe.services.dedup.async_session",
|
|
return_value=_session_returning(None)), \
|
|
patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
|
|
dup = await find_duplicate_note(
|
|
7, "Title", body="x" * 250, project_id=2, is_task=False, note_type="note",
|
|
)
|
|
assert dup is not None
|
|
assert dup.id == 20
|
|
assert dup.reason == "semantic"
|
|
assert dup.similarity == 0.93
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_semantic_match_of_other_note_type_is_ignored():
|
|
other = _fake_note(id=21, title="X", note_type="process")
|
|
sem = AsyncMock(return_value=[(0.97, other)])
|
|
with patch("scribe.services.dedup.async_session",
|
|
return_value=_session_returning(None)), \
|
|
patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
|
|
dup = await find_duplicate_note(7, "Title", body="x" * 250, note_type="note")
|
|
assert dup is None # type mismatch must not block
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rule_title_match_in_topic():
|
|
rule = _fake_note(id=47, title="Honor the multi-user sharing ACL")
|
|
with patch("scribe.services.dedup.async_session",
|
|
return_value=_session_returning(rule)):
|
|
dup = await find_duplicate_rule(
|
|
"honor the multi-user sharing acl", topic_id=7,
|
|
)
|
|
assert dup is not None
|
|
assert dup.id == 47
|
|
assert dup.reason == "title"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rule_requires_a_scope():
|
|
# No topic_id and no project_id → nothing to scope to → no match, no query.
|
|
sess = AsyncMock()
|
|
with patch("scribe.services.dedup.async_session", return_value=sess):
|
|
dup = await find_duplicate_rule("anything")
|
|
assert dup is None
|
|
sess.__aenter__.assert_not_called()
|
|
|
|
|
|
def test_duplicate_response_shape():
|
|
dm = DuplicateMatch(id=5, title="Foo", similarity=1.0, reason="title")
|
|
r = duplicate_response(dm, "task")
|
|
assert r["duplicate"] is True
|
|
assert r["existing_id"] == 5
|
|
assert r["match"] == "title"
|
|
assert "force=true" in r["message"]
|
|
assert "update_task" in r["message"]
|