82 lines
2.2 KiB
Python
82 lines
2.2 KiB
Python
"""FC-2d-vi: library-wide phash backfill (keyset, idempotent)."""
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
from backend.app.models import ImageRecord
|
|
from backend.app.tasks.maintenance import backfill_phash
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
class _Ctx:
|
|
def __init__(self, s):
|
|
self.s = s
|
|
|
|
def __enter__(self):
|
|
return self.s
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
|
|
def _sf(db_sync):
|
|
"""sessionmaker-like returning the test's bound session, so the task
|
|
runs against the same transaction / sees the seeded rows."""
|
|
class _SM:
|
|
def __call__(self):
|
|
return _Ctx(db_sync)
|
|
|
|
return _SM()
|
|
|
|
|
|
def _img(path, color):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
Image.new("RGB", (64, 64), color).save(path, "JPEG")
|
|
|
|
|
|
def _rec(db_sync, path, *, sha, mime="image/jpeg", phash=None):
|
|
rec = ImageRecord(
|
|
path=str(path), sha256=sha, phash=phash, size_bytes=1,
|
|
mime=mime, width=64, height=64, origin="imported_filesystem",
|
|
integrity_status="unknown",
|
|
)
|
|
db_sync.add(rec)
|
|
db_sync.flush()
|
|
return rec
|
|
|
|
|
|
def test_backfill_fills_null_image_phash(db_sync, tmp_path, monkeypatch):
|
|
from backend.app.tasks import maintenance as m
|
|
|
|
p = tmp_path / "a.jpg"
|
|
_img(p, (10, 120, 200))
|
|
pre = tmp_path / "preset.jpg"
|
|
_img(pre, (5, 5, 5))
|
|
|
|
rec = _rec(db_sync, p, sha="a" + "0" * 63, phash=None)
|
|
vid = _rec(
|
|
db_sync, tmp_path / "v.mp4", sha="v" + "0" * 63,
|
|
mime="video/mp4", phash=None,
|
|
)
|
|
missing = _rec(
|
|
db_sync, tmp_path / "gone.jpg", sha="g" + "0" * 63, phash=None,
|
|
)
|
|
preset = _rec(
|
|
db_sync, pre, sha="p" + "0" * 63, phash="deadbeefdeadbeef",
|
|
)
|
|
db_sync.commit()
|
|
|
|
monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
|
|
|
|
updated = backfill_phash()
|
|
db_sync.expire_all()
|
|
assert updated == 1
|
|
assert db_sync.get(ImageRecord, rec.id).phash is not None
|
|
assert db_sync.get(ImageRecord, vid.id).phash is None
|
|
assert db_sync.get(ImageRecord, missing.id).phash is None
|
|
assert db_sync.get(ImageRecord, preset.id).phash == "deadbeefdeadbeef"
|
|
|
|
# idempotent: a second run touches nothing
|
|
assert backfill_phash() == 0
|