77 lines
3.0 KiB
Python
77 lines
3.0 KiB
Python
"""Tests for cleanup_service's library-audit additions.
|
|
|
|
Covers:
|
|
- project_min_dimension_violations: SQL-only query against width/height
|
|
- delete_min_dimension_violations: routes through existing delete_images
|
|
- audit lifecycle (start_audit_run / apply_audit_run / cancel_audit_run)
|
|
|
|
Tests assert via column selects per reference-async-coredml-test-assertions
|
|
(post-DML ORM entity access via session.get() raises MissingGreenlet on
|
|
async sessions; we use db_sync here but the convention is preserved for
|
|
consistency).
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
from sqlalchemy import func, select
|
|
|
|
from backend.app.models import ImageRecord, LibraryAuditRun
|
|
from backend.app.services import cleanup_service
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _make_image_record(db_sync, tmp_path, *, width, height, color):
|
|
"""Helper: write a real PIL file to a UNIQUE path (reference-image-record-path-unique)
|
|
and insert an ImageRecord row with all required NOT-NULL columns set."""
|
|
path = tmp_path / f"img_{width}x{height}_{color[0]}.png"
|
|
Image.new("RGB", (width, height), color).save(path)
|
|
rec = ImageRecord(
|
|
path=str(path),
|
|
sha256=f"fake-{width}-{height}-{color[0]:03d}" + "0" * 50,
|
|
size_bytes=path.stat().st_size,
|
|
mime="image/png", # reference-image-record-required-columns
|
|
width=width,
|
|
height=height,
|
|
origin="imported_filesystem", # feedback-check-existing-enums
|
|
integrity_status="ok",
|
|
)
|
|
db_sync.add(rec)
|
|
db_sync.flush()
|
|
return rec
|
|
|
|
|
|
def test_project_min_dimension_violations_returns_count_and_samples(db_sync, tmp_path):
|
|
_make_image_record(db_sync, tmp_path, width=100, height=100, color=(10, 0, 0))
|
|
_make_image_record(db_sync, tmp_path, width=50, height=50, color=(20, 0, 0))
|
|
_make_image_record(db_sync, tmp_path, width=400, height=400, color=(30, 0, 0))
|
|
db_sync.commit()
|
|
|
|
result = cleanup_service.project_min_dimension_violations(
|
|
db_sync, min_width=200, min_height=200,
|
|
)
|
|
assert result["count"] == 2 # 100x100 and 50x50 violate
|
|
assert len(result["sample_ids"]) == 2
|
|
|
|
|
|
def test_delete_min_dimension_violations_unlinks_and_cascades(db_sync, tmp_path):
|
|
rec_small = _make_image_record(db_sync, tmp_path, width=50, height=50, color=(40, 0, 0))
|
|
rec_big = _make_image_record(db_sync, tmp_path, width=500, height=500, color=(50, 0, 0))
|
|
small_path = Path(rec_small.path)
|
|
db_sync.commit()
|
|
|
|
# images_root is tmp_path here because the test fixtures stored files there;
|
|
# delete_images() uses it to unlink originals + thumbs from the right tree.
|
|
deleted = cleanup_service.delete_min_dimension_violations(
|
|
db_sync, min_width=200, min_height=200, images_root=tmp_path,
|
|
)
|
|
assert deleted == 1
|
|
# Verify via column selects per banked rule.
|
|
remaining_ids = db_sync.execute(
|
|
select(ImageRecord.id).order_by(ImageRecord.id)
|
|
).scalars().all()
|
|
assert remaining_ids == [rec_big.id]
|
|
assert not small_path.exists() # file unlinked
|