104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
"""FC-2d-vi: Importer(deep=True) re-derives on an already-imported row."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
from sqlalchemy import func, select
|
|
|
|
from backend.app.models import (
|
|
ImageProvenance,
|
|
ImageRecord,
|
|
ImportSettings,
|
|
Post,
|
|
)
|
|
from backend.app.services.importer import Importer
|
|
from backend.app.services.thumbnailer import Thumbnailer
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
@pytest.fixture
|
|
def import_layout(tmp_path):
|
|
import_root = tmp_path / "import"
|
|
images_root = tmp_path / "images"
|
|
import_root.mkdir()
|
|
images_root.mkdir()
|
|
return import_root, images_root
|
|
|
|
|
|
def _mk(db_sync, import_layout, *, deep):
|
|
import_root, images_root = import_layout
|
|
settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
return Importer(
|
|
session=db_sync, images_root=images_root, import_root=import_root,
|
|
thumbnailer=Thumbnailer(images_root=images_root), settings=settings,
|
|
deep=deep,
|
|
)
|
|
|
|
|
|
def _img(path):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
Image.new("RGB", (50, 50), (90, 30, 200)).save(path, "JPEG")
|
|
|
|
|
|
def test_deep_rederives_phash_and_provenance(db_sync, import_layout):
|
|
import_root, _ = import_layout
|
|
src = import_root / "Mae" / "p.jpg"
|
|
_img(src)
|
|
|
|
# First import (quick). Images get a phash on import; null it +
|
|
# primary_post_id to simulate a pre-FC-2d-i+ii/v record, then add a
|
|
# sidecar to simulate provenance that didn't exist at first import.
|
|
quick = _mk(db_sync, import_layout, deep=False)
|
|
r1 = quick.import_one(src)
|
|
assert r1.status == "imported"
|
|
rec = db_sync.get(ImageRecord, r1.image_id)
|
|
rec.phash = None
|
|
rec.primary_post_id = None
|
|
db_sync.commit()
|
|
src.with_suffix(".jpg.json").write_text(json.dumps(
|
|
{"category": "patreon", "id": "9", "title": "P"}))
|
|
|
|
deep = _mk(db_sync, import_layout, deep=True)
|
|
r2 = deep.import_one(src)
|
|
# Outcome flipped from "skipped+duplicate_hash" to "refreshed" 2026-05-25
|
|
# so the UI can surface deep scan's actual work instead of showing it as
|
|
# a no-op. See ImportResult.status comment + _deep_rederive docstring.
|
|
assert r2.status == "refreshed"
|
|
assert r2.image_id == rec.id
|
|
assert r2.skip_reason is None
|
|
assert r2.error is None
|
|
|
|
db_sync.expire_all()
|
|
rec2 = db_sync.get(ImageRecord, rec.id)
|
|
assert rec2.phash is not None
|
|
assert db_sync.execute(
|
|
select(func.count()).select_from(ImageRecord)
|
|
).scalar_one() == 1 # no new record
|
|
post = db_sync.execute(select(Post)).scalar_one()
|
|
assert db_sync.execute(
|
|
select(func.count()).select_from(ImageProvenance)
|
|
.where(ImageProvenance.image_record_id == rec.id)
|
|
).scalar_one() == 1
|
|
assert rec2.primary_post_id == post.id
|
|
|
|
# Idempotent: a second deep pass adds no duplicate provenance.
|
|
_mk(db_sync, import_layout, deep=True).import_one(src)
|
|
db_sync.expire_all()
|
|
assert db_sync.execute(
|
|
select(func.count()).select_from(ImageProvenance)
|
|
).scalar_one() == 1
|
|
|
|
|
|
def test_deep_false_is_plain_skip(db_sync, import_layout):
|
|
import_root, _ = import_layout
|
|
src = import_root / "Mae" / "q.jpg"
|
|
_img(src)
|
|
_mk(db_sync, import_layout, deep=False).import_one(src)
|
|
r = _mk(db_sync, import_layout, deep=False).import_one(src)
|
|
assert r.status == "skipped"
|
|
assert r.error == "sha256 already present"
|