167 lines
7.5 KiB
Python
167 lines
7.5 KiB
Python
# tests/test_duplicates.py — Duplicate detection & merge (#166)
|
|
"""Tests for backend.services.duplicates, its REST routes and AI tools.
|
|
|
|
Fusion is destructive: the service takes backups first and the tool/route
|
|
require an explicit confirmation (DANGEROUS + ``confirm: true``).
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from backend.services import duplicates as _duplicates
|
|
from backend.services.errors import ServiceError
|
|
from backend.tools.api import ToolContext, call_tool
|
|
from backend.tools.context import ToolConfirmationRequired
|
|
|
|
|
|
def _ctx(**kwargs) -> ToolContext:
|
|
user = {"username": "tester", "role": "admin", "vaults": ["*"]}
|
|
kwargs.setdefault("audit_enabled", False)
|
|
return ToolContext(user=user, **kwargs)
|
|
|
|
|
|
class TestSimilarity:
|
|
def test_identical_texts_score_one(self):
|
|
text = "# Titre\nContenu identique avec plusieurs mots significatifs."
|
|
assert _duplicates.similarity_score(text, text) == 1.0
|
|
|
|
def test_different_texts_score_low(self):
|
|
a = "# Recette pizza\nFarine, tomate, mozzarella, four à bois."
|
|
b = "# Config réseau\nAdresse IP, masque, passerelle, DNS du datacenter."
|
|
assert _duplicates.similarity_score(a, b) < 0.5
|
|
|
|
def test_empty_text_scores_zero(self):
|
|
assert _duplicates.similarity_score("", "# Titre\ncontenu") == 0.0
|
|
|
|
def test_frontmatter_ignored(self):
|
|
a = "---\ntitle: A\ntags: [x]\n---\n# Note\nContenu commun significatif ici."
|
|
b = "---\ntitle: B\ntags: [y]\n---\n# Note\nContenu commun significatif ici."
|
|
assert _duplicates.similarity_score(a, b) > 0.8
|
|
|
|
|
|
class TestFindPairs:
|
|
def test_finds_planted_duplicates(self, client, test_vault_dir):
|
|
vault = Path(test_vault_dir)
|
|
body = "# Rapport mensuel\nVentes en hausse de vingt pour cent ce trimestre."
|
|
(vault / "rapport-a.md").write_text(body, encoding="utf-8")
|
|
(vault / "rapport-b.md").write_text(body + "\nLigne complémentaire mineure.", encoding="utf-8")
|
|
result = _duplicates.find_duplicate_pairs("TestVault", threshold=0.5, limit=20)
|
|
assert result["vault"] == "TestVault"
|
|
assert result["files_scanned"] >= 2
|
|
pair_files = {(p["file_a"], p["file_b"]) for p in result["pairs"]}
|
|
assert ("rapport-a.md", "rapport-b.md") in pair_files
|
|
|
|
def test_threshold_filters(self, client, test_vault_dir):
|
|
result = _duplicates.find_duplicate_pairs("TestVault", threshold=1.0, limit=20)
|
|
assert result["pairs"] == []
|
|
|
|
def test_invalid_threshold_rejected(self, client):
|
|
with pytest.raises(ServiceError):
|
|
_duplicates.find_duplicate_pairs("TestVault", threshold=0.1)
|
|
|
|
def test_unknown_vault_rejected(self, client):
|
|
with pytest.raises(ServiceError):
|
|
_duplicates.find_duplicate_pairs("NoSuchVault")
|
|
|
|
|
|
class TestMerge:
|
|
def _plant(self, test_vault_dir: str, name: str, content: str) -> None:
|
|
Path(test_vault_dir, name).write_text(content, encoding="utf-8")
|
|
|
|
def test_merge_append(self, client, test_vault_dir):
|
|
self._plant(test_vault_dir, "src.md", "# Source\nContenu source unique.")
|
|
self._plant(test_vault_dir, "dst.md", "# Cible\nContenu cible unique.")
|
|
result = _duplicates.merge_duplicates("TestVault", "src.md", "dst.md", strategy="append")
|
|
assert result["strategy"] == "append"
|
|
assert result["deleted"] == "src.md"
|
|
assert not Path(test_vault_dir, "src.md").exists()
|
|
merged = Path(test_vault_dir, "dst.md").read_text(encoding="utf-8")
|
|
assert "Contenu cible unique" in merged
|
|
assert "Contenu source unique" in merged
|
|
|
|
def test_merge_prefer_target(self, client, test_vault_dir):
|
|
self._plant(test_vault_dir, "old.md", "# Vieux\nAncien contenu.")
|
|
self._plant(test_vault_dir, "new.md", "# Neuf\nNouveau contenu.")
|
|
result = _duplicates.merge_duplicates("TestVault", "old.md", "new.md", strategy="prefer_target")
|
|
assert result["strategy"] == "prefer_target"
|
|
assert Path(test_vault_dir, "new.md").read_text(encoding="utf-8") == "# Neuf\nNouveau contenu."
|
|
assert not Path(test_vault_dir, "old.md").exists()
|
|
|
|
def test_merge_same_path_rejected(self, client, test_vault_dir):
|
|
self._plant(test_vault_dir, "same.md", "# Same\nContenu.")
|
|
with pytest.raises(ServiceError):
|
|
_duplicates.merge_duplicates("TestVault", "same.md", "same.md")
|
|
|
|
def test_merge_missing_source_rejected(self, client, test_vault_dir):
|
|
self._plant(test_vault_dir, "exists.md", "# Exists\nContenu.")
|
|
with pytest.raises(ServiceError):
|
|
_duplicates.merge_duplicates("TestVault", "missing.md", "exists.md")
|
|
|
|
|
|
class TestDuplicatesApi:
|
|
def test_list_endpoint(self, client, test_vault_dir):
|
|
(Path(test_vault_dir) / "dup1.md").write_text("# Doublon\nTexte commun significatif.", encoding="utf-8")
|
|
(Path(test_vault_dir) / "dup2.md").write_text("# Doublon\nTexte commun significatif.", encoding="utf-8")
|
|
resp = client.get("/api/duplicates", params={"vault": "TestVault", "threshold": 0.5})
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["vault"] == "TestVault"
|
|
|
|
def test_merge_requires_confirm(self, client, test_vault_dir):
|
|
(Path(test_vault_dir) / "m1.md").write_text("# M1\nContenu.", encoding="utf-8")
|
|
(Path(test_vault_dir) / "m2.md").write_text("# M2\nContenu.", encoding="utf-8")
|
|
resp = client.post(
|
|
"/api/duplicates/merge",
|
|
json={"vault": "TestVault", "source_path": "m1.md", "target_path": "m2.md"},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_merge_with_confirm(self, client, test_vault_dir):
|
|
(Path(test_vault_dir) / "c1.md").write_text("# C1\nContenu un.", encoding="utf-8")
|
|
(Path(test_vault_dir) / "c2.md").write_text("# C2\nContenu deux.", encoding="utf-8")
|
|
resp = client.post(
|
|
"/api/duplicates/merge",
|
|
json={
|
|
"vault": "TestVault",
|
|
"source_path": "c1.md",
|
|
"target_path": "c2.md",
|
|
"strategy": "append",
|
|
"confirm": True,
|
|
},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["deleted"] == "c1.md"
|
|
|
|
|
|
class TestDuplicateTools:
|
|
def test_find_tool(self, client, test_vault_dir):
|
|
(Path(test_vault_dir) / "t1.md").write_text("# Outil\nRecherche de doublons.", encoding="utf-8")
|
|
result = call_tool(
|
|
"find_duplicates",
|
|
_ctx(),
|
|
{"vault": "TestVault", "threshold": 0.5, "limit": 10},
|
|
)
|
|
assert result.ok
|
|
assert result.data["vault"] == "TestVault"
|
|
|
|
def test_merge_tool_requires_confirmation(self, client, test_vault_dir):
|
|
(Path(test_vault_dir) / "s1.md").write_text("# S1\nContenu.", encoding="utf-8")
|
|
(Path(test_vault_dir) / "s2.md").write_text("# S2\nContenu.", encoding="utf-8")
|
|
with pytest.raises(ToolConfirmationRequired):
|
|
call_tool(
|
|
"merge_duplicate_notes",
|
|
_ctx(),
|
|
{"vault": "TestVault", "source_path": "s1.md", "target_path": "s2.md"},
|
|
)
|
|
|
|
def test_merge_tool_confirmed(self, client, test_vault_dir):
|
|
(Path(test_vault_dir) / "k1.md").write_text("# K1\nContenu k.", encoding="utf-8")
|
|
(Path(test_vault_dir) / "k2.md").write_text("# K2\nContenu l.", encoding="utf-8")
|
|
result = call_tool(
|
|
"merge_duplicate_notes",
|
|
_ctx(confirmed=True),
|
|
{"vault": "TestVault", "source_path": "k1.md", "target_path": "k2.md", "strategy": "append"},
|
|
)
|
|
assert result.ok
|
|
assert result.data["deleted"] == "k1.md"
|