121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
"""FC-2d-iii: archive → import media members + store the archive."""
|
|
|
|
import io
|
|
import json
|
|
import zipfile
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
from sqlalchemy import func, select
|
|
|
|
from backend.app.models import (
|
|
ImageProvenance,
|
|
ImageRecord,
|
|
ImportSettings,
|
|
Post,
|
|
PostAttachment,
|
|
)
|
|
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
|
|
|
|
|
|
@pytest.fixture
|
|
def importer(db_sync, import_layout):
|
|
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,
|
|
)
|
|
|
|
|
|
def _jpeg_bytes(color):
|
|
buf = io.BytesIO()
|
|
Image.new("RGB", (40, 40), color).save(buf, "JPEG")
|
|
return buf.getvalue()
|
|
|
|
|
|
def _split_bytes(orient):
|
|
"""Structured half/half image — solid colors phash-collapse (distance
|
|
0); orthogonal splits are distinct only at phash_threshold=0 (see
|
|
reference-phash-test-images)."""
|
|
im = Image.new("L", (256, 256), 0)
|
|
px = im.load()
|
|
for y in range(256):
|
|
for x in range(256):
|
|
if (x / 256 if orient == "v" else y / 256) >= 0.5:
|
|
px[x, y] = 255
|
|
buf = io.BytesIO()
|
|
im.convert("RGB").save(buf, "JPEG")
|
|
return buf.getvalue()
|
|
|
|
|
|
def test_archive_imports_members_and_stores_archive(importer, import_layout):
|
|
import_root, _ = import_layout
|
|
# Distinct structure + threshold 0 so BOTH members import (solid
|
|
# colors would phash-collapse; reference-phash-test-images).
|
|
importer.settings.phash_threshold = 0
|
|
arc = import_root / "Bob" / "set.cbz"
|
|
arc.parent.mkdir(parents=True, exist_ok=True)
|
|
with zipfile.ZipFile(arc, "w") as zf:
|
|
zf.writestr("a.jpg", _split_bytes("v"))
|
|
zf.writestr("b.jpg", _split_bytes("h"))
|
|
zf.writestr("readme.txt", b"hello")
|
|
arc.with_suffix(".cbz.json").write_text(json.dumps(
|
|
{"category": "patreon", "id": "777", "title": "Set"}))
|
|
|
|
r = importer.import_one(arc)
|
|
assert r.status == "imported"
|
|
assert len(r.member_image_ids) == 2
|
|
|
|
imgs = importer.session.execute(select(ImageRecord)).scalars().all()
|
|
assert len(imgs) == 2
|
|
post = importer.session.execute(select(Post)).scalar_one()
|
|
provs = importer.session.execute(
|
|
select(func.count()).select_from(ImageProvenance)
|
|
).scalar_one()
|
|
assert provs == 2 # one Post, both members
|
|
att = importer.session.execute(select(PostAttachment)).scalar_one()
|
|
assert att.original_filename == "set.cbz"
|
|
assert att.post_id == post.id
|
|
|
|
|
|
def test_corrupt_archive_still_stored(importer, import_layout):
|
|
import_root, _ = import_layout
|
|
arc = import_root / "Bob" / "broken.zip"
|
|
arc.parent.mkdir(parents=True, exist_ok=True)
|
|
arc.write_bytes(b"definitely not a zip")
|
|
r = importer.import_one(arc)
|
|
assert r.status == "attached"
|
|
att = importer.session.execute(select(PostAttachment)).scalar_one()
|
|
assert att.original_filename == "broken.zip"
|
|
assert importer.session.execute(
|
|
select(func.count()).select_from(ImageRecord)
|
|
).scalar_one() == 0
|
|
|
|
|
|
def test_reimport_archive_is_idempotent(importer, import_layout):
|
|
import_root, _ = import_layout
|
|
arc = import_root / "Bob" / "set.zip"
|
|
arc.parent.mkdir(parents=True, exist_ok=True)
|
|
with zipfile.ZipFile(arc, "w") as zf:
|
|
zf.writestr("a.jpg", _jpeg_bytes((1, 2, 3)))
|
|
importer.import_one(arc)
|
|
importer.import_one(arc)
|
|
assert importer.session.execute(
|
|
select(func.count()).select_from(PostAttachment)
|
|
).scalar_one() == 1
|