292 lines
9.7 KiB
Python
292 lines
9.7 KiB
Python
"""Duplicate engine: banded clustering, reversible decisions, negative links,
|
|
reopening on contradiction, no canonical cycles, and durability across restart.
|
|
|
|
Fixtures are inlined (not in a sibling conftest.py) to avoid shadowing the
|
|
characterization suite's bare ``conftest`` import under pytest's prepend mode.
|
|
"""
|
|
|
|
import shutil
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from PIL import Image
|
|
from sqlalchemy import select
|
|
|
|
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
|
from photo_pipeline.models import Asset, DuplicateNegativeLink
|
|
from photo_pipeline.services.duplicates import (
|
|
ClusterState,
|
|
ConflictError,
|
|
Decision,
|
|
DuplicateError,
|
|
DuplicateService,
|
|
Method,
|
|
)
|
|
from photo_pipeline.services.inventory import InventoryService
|
|
|
|
|
|
@pytest.fixture
|
|
def db_url(tmp_path):
|
|
url = f"sqlite:///{tmp_path / 'dup.db'}"
|
|
run_migrations(url)
|
|
return url
|
|
|
|
|
|
@pytest.fixture
|
|
def make_factory():
|
|
engines = []
|
|
|
|
def _make(url):
|
|
engine = create_db_engine(url)
|
|
engines.append(engine)
|
|
return create_session_factory(engine)
|
|
|
|
yield _make
|
|
for engine in engines:
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
def factory(db_url, make_factory):
|
|
return make_factory(db_url)
|
|
|
|
|
|
@pytest.fixture
|
|
def inventory(factory):
|
|
return InventoryService(factory)
|
|
|
|
|
|
@pytest.fixture
|
|
def duplicates(factory):
|
|
return DuplicateService(factory)
|
|
|
|
|
|
def structured(path, seed, size=(256, 192)):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
rng = np.random.default_rng(seed)
|
|
w, h = size
|
|
base = np.zeros((h, w, 3), dtype=np.uint8)
|
|
for _ in range(6):
|
|
x0 = int(rng.integers(0, w - 60))
|
|
y0 = int(rng.integers(0, h - 60))
|
|
base[y0 : y0 + 60, x0 : x0 + 60] = rng.integers(0, 256, 3)
|
|
grad = np.linspace(0, 120, w, dtype=np.uint8)
|
|
base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255)
|
|
Image.fromarray(base).save(path, quality=95)
|
|
return path
|
|
|
|
|
|
def resized_copy(src, dst, scale=0.5):
|
|
with Image.open(src) as image:
|
|
image.resize(
|
|
(int(image.width * scale), int(image.height * scale)), Image.LANCZOS
|
|
).save(dst, quality=95)
|
|
return dst
|
|
|
|
|
|
def only(clusters, method):
|
|
return [c for c in clusters if c["method"] == method]
|
|
|
|
|
|
def test_exact_copies_form_auto_decided_cluster(tmp_path, inventory, duplicates, factory):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
shutil.copy2(a, lib / "a_copy.jpg")
|
|
inventory.scan(lib)
|
|
|
|
clusters = duplicates.detect().clusters
|
|
exact = only(clusters, Method.EXACT.value)
|
|
assert len(exact) == 1
|
|
cluster = exact[0]
|
|
assert cluster["state"] == ClusterState.DECIDED.value
|
|
assert cluster["decision"] == Decision.CANONICAL.value
|
|
roles = {m["role"] for m in cluster["members"]}
|
|
assert roles == {"canonical", "variant"}
|
|
|
|
# The variant asset points at the canonical; the canonical points at nothing.
|
|
assets = {a.id: a for a in factory().execute(select(Asset)).scalars()}
|
|
canonical = cluster["canonical_asset_id"]
|
|
for m in cluster["members"]:
|
|
if m["role"] == "variant":
|
|
assert assets[m["asset_id"]].canonical_asset_id == canonical
|
|
else:
|
|
assert assets[m["asset_id"]].canonical_asset_id is None
|
|
|
|
|
|
def test_same_pixels_different_bytes_form_pixel_cluster(tmp_path, inventory, duplicates):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
twin = lib / "a_meta.jpg"
|
|
shutil.copy2(a, twin)
|
|
with open(twin, "ab") as handle: # trailing bytes: same pixels, different bytes
|
|
handle.write(b"\xff\xfe metadata")
|
|
inventory.scan(lib)
|
|
|
|
clusters = duplicates.detect().clusters
|
|
pixel = only(clusters, Method.PIXEL.value)
|
|
assert len(pixel) == 1
|
|
assert pixel[0]["confidence"] == "pixel"
|
|
assert pixel[0]["state"] == ClusterState.DECIDED.value
|
|
|
|
|
|
def test_perceptual_variant_is_review_only(tmp_path, inventory, duplicates):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
resized_copy(a, lib / "a_small.jpg")
|
|
inventory.scan(lib)
|
|
|
|
clusters = duplicates.detect().clusters
|
|
perceptual = only(clusters, Method.PERCEPTUAL.value)
|
|
assert len(perceptual) == 1
|
|
assert perceptual[0]["state"] == ClusterState.OPEN.value # never auto-decided
|
|
assert perceptual[0]["decision"] is None
|
|
assert perceptual[0]["confidence"] in ("near", "similar")
|
|
|
|
|
|
def test_distinct_images_do_not_cluster(tmp_path, inventory, duplicates):
|
|
lib = tmp_path / "lib"
|
|
structured(lib / "a.jpg", 1)
|
|
structured(lib / "b.jpg", 2)
|
|
inventory.scan(lib)
|
|
assert duplicates.detect().clusters == []
|
|
|
|
|
|
def test_not_duplicate_decision_persists_negative_link_and_suppresses(
|
|
tmp_path, inventory, duplicates, factory
|
|
):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
resized_copy(a, lib / "a_small.jpg")
|
|
inventory.scan(lib)
|
|
cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0]
|
|
|
|
duplicates.decide(cluster["id"], Decision.NOT_DUPLICATE)
|
|
|
|
links = list(factory().execute(select(DuplicateNegativeLink)).scalars())
|
|
assert len(links) == 1
|
|
|
|
# Re-detecting must not re-suggest the rejected pair.
|
|
again = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)
|
|
assert again == []
|
|
|
|
|
|
def test_decision_is_reversible(tmp_path, inventory, duplicates, factory):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
resized_copy(a, lib / "a_small.jpg")
|
|
inventory.scan(lib)
|
|
cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0]
|
|
members = [m["asset_id"] for m in cluster["members"]]
|
|
|
|
duplicates.decide(cluster["id"], Decision.NOT_DUPLICATE)
|
|
assert factory().execute(select(DuplicateNegativeLink)).scalars().all()
|
|
|
|
# Reverse: choose a canonical instead.
|
|
reversed_snapshot = duplicates.decide(
|
|
cluster["id"], Decision.CANONICAL, canonical_asset_id=members[0]
|
|
)
|
|
assert reversed_snapshot["state"] == ClusterState.DECIDED.value
|
|
assert not factory().execute(select(DuplicateNegativeLink)).scalars().all()
|
|
|
|
|
|
def test_version_conflict_is_rejected(tmp_path, inventory, duplicates):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
resized_copy(a, lib / "a_small.jpg")
|
|
inventory.scan(lib)
|
|
cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0]
|
|
|
|
with pytest.raises(ConflictError):
|
|
duplicates.decide(
|
|
cluster["id"], Decision.DEFERRED, expected_version=cluster["version"] + 5
|
|
)
|
|
|
|
|
|
def test_canonical_must_be_a_member(tmp_path, inventory, duplicates):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
resized_copy(a, lib / "a_small.jpg")
|
|
inventory.scan(lib)
|
|
cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0]
|
|
|
|
with pytest.raises(DuplicateError):
|
|
duplicates.decide(cluster["id"], Decision.CANONICAL, canonical_asset_id="ghost")
|
|
|
|
|
|
def test_no_canonical_cycles(tmp_path, inventory, duplicates, factory):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
shutil.copy2(a, lib / "a_copy.jpg")
|
|
inventory.scan(lib)
|
|
duplicates.detect()
|
|
|
|
# Every asset's canonical chain terminates (no loop).
|
|
assets = {a.id: a for a in factory().execute(select(Asset)).scalars()}
|
|
for start in assets:
|
|
seen, cur = set(), start
|
|
while cur is not None:
|
|
assert cur not in seen
|
|
seen.add(cur)
|
|
cur = assets[cur].canonical_asset_id
|
|
|
|
|
|
def test_new_exact_member_inherits_decision(tmp_path, inventory, duplicates, factory):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
shutil.copy2(a, lib / "a2.jpg")
|
|
inventory.scan(lib)
|
|
cluster = only(duplicates.detect().clusters, Method.EXACT.value)[0]
|
|
canonical = cluster["canonical_asset_id"]
|
|
|
|
# A third exact copy appears later.
|
|
shutil.copy2(a, lib / "a3.jpg")
|
|
inventory.scan(lib)
|
|
cluster2 = only(duplicates.detect().clusters, Method.EXACT.value)[0]
|
|
|
|
assert cluster2["id"] == cluster["id"]
|
|
assert cluster2["state"] == ClusterState.DECIDED.value
|
|
assert len(cluster2["members"]) == 3
|
|
assets = {a.id: a for a in factory().execute(select(Asset)).scalars()}
|
|
variants = [m for m in cluster2["members"] if m["role"] == "variant"]
|
|
assert len(variants) == 2
|
|
for m in variants:
|
|
assert assets[m["asset_id"]].canonical_asset_id == canonical
|
|
|
|
|
|
def test_contradictory_member_reopens_reviewed_cluster(
|
|
tmp_path, inventory, duplicates, db_url, make_factory
|
|
):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
b = resized_copy(a, lib / "b.jpg")
|
|
inventory.scan(lib)
|
|
cluster = only(duplicates.detect().clusters, Method.PERCEPTUAL.value)[0]
|
|
duplicates.decide(cluster["id"], Decision.NOT_DUPLICATE) # dismissed + negative link
|
|
|
|
# b is replaced by an exact copy of a — now byte-identical to a "not duplicate".
|
|
shutil.copy2(a, b)
|
|
inventory.scan(lib)
|
|
clusters = duplicates.detect().clusters
|
|
|
|
reopened = [c for c in clusters if c["state"] == ClusterState.REOPENED.value]
|
|
assert reopened, "byte-identical contradiction of a not_duplicate must reopen review"
|
|
|
|
|
|
def test_decisions_durable_across_restart(tmp_path, db_url, make_factory):
|
|
lib = tmp_path / "lib"
|
|
a = structured(lib / "a.jpg", 1)
|
|
shutil.copy2(a, lib / "a_copy.jpg")
|
|
|
|
InventoryService(make_factory(db_url)).scan(lib)
|
|
cluster = only(
|
|
DuplicateService(make_factory(db_url)).detect().clusters, Method.EXACT.value
|
|
)[0]
|
|
|
|
# Fresh factory against the same database: the decision persisted.
|
|
from photo_pipeline.models import DuplicateCluster
|
|
|
|
with make_factory(db_url)() as session:
|
|
persisted = session.get(DuplicateCluster, cluster["id"])
|
|
assert persisted.state == ClusterState.DECIDED.value
|
|
assert persisted.decision == Decision.CANONICAL.value
|