176 lines
6.2 KiB
Python
176 lines
6.2 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)
|
|
# sha256 column is varchar(64) — use a deterministic 64-char pseudo
|
|
# hash built from the unique inputs.
|
|
sha = f"{width:04d}{height:04d}{color[0]:03d}".ljust(64, "0")[:64]
|
|
rec = ImageRecord(
|
|
path=str(path),
|
|
sha256=sha,
|
|
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
|
|
|
|
|
|
# --- Audit lifecycle tests (Task 5) ---
|
|
|
|
import backend.app.tasks.library_audit # noqa: F401, E402 — celery registration
|
|
|
|
|
|
def test_start_audit_run_creates_row_and_dispatches(db_sync, monkeypatch):
|
|
dispatched = []
|
|
from backend.app.tasks import library_audit as la_mod
|
|
monkeypatch.setattr(
|
|
la_mod.scan_library_for_rule, "delay",
|
|
lambda audit_id: dispatched.append(audit_id),
|
|
)
|
|
audit_id = cleanup_service.start_audit_run(
|
|
db_sync, rule="transparency", params={"threshold": 0.9},
|
|
)
|
|
db_sync.commit()
|
|
row_status = db_sync.execute(
|
|
select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit_id)
|
|
).scalar_one()
|
|
assert row_status == "running"
|
|
assert dispatched == [audit_id]
|
|
|
|
|
|
def test_start_audit_run_rejects_when_another_is_running(db_sync, monkeypatch):
|
|
from backend.app.tasks import library_audit as la_mod
|
|
monkeypatch.setattr(
|
|
la_mod.scan_library_for_rule, "delay", lambda audit_id: None,
|
|
)
|
|
cleanup_service.start_audit_run(
|
|
db_sync, rule="transparency", params={"threshold": 0.9},
|
|
)
|
|
db_sync.commit()
|
|
with pytest.raises(cleanup_service.AuditAlreadyRunning):
|
|
cleanup_service.start_audit_run(
|
|
db_sync, rule="single_color",
|
|
params={"threshold": 0.95, "tolerance": 30},
|
|
)
|
|
|
|
|
|
def test_apply_audit_run_with_correct_token_deletes_matched(db_sync, tmp_path):
|
|
rec = _make_image_record(
|
|
db_sync, tmp_path, width=100, height=100, color=(60, 0, 0),
|
|
)
|
|
db_sync.flush()
|
|
audit = LibraryAuditRun(
|
|
rule="transparency", params={"threshold": 0.9},
|
|
status="ready", scanned_count=1, matched_count=1,
|
|
matched_ids=[rec.id],
|
|
)
|
|
db_sync.add(audit)
|
|
db_sync.commit()
|
|
|
|
deleted = cleanup_service.apply_audit_run(
|
|
db_sync, audit_id=audit.id,
|
|
confirm_token=f"delete-audit-{audit.id}",
|
|
images_root=tmp_path,
|
|
)
|
|
assert deleted == 1
|
|
remaining_count = db_sync.execute(
|
|
select(func.count()).select_from(ImageRecord)
|
|
).scalar_one()
|
|
assert remaining_count == 0
|
|
new_status = db_sync.execute(
|
|
select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit.id)
|
|
).scalar_one()
|
|
assert new_status == "applied"
|
|
|
|
|
|
def test_apply_audit_run_with_wrong_token_raises(db_sync, tmp_path):
|
|
audit = LibraryAuditRun(
|
|
rule="transparency", params={"threshold": 0.9},
|
|
status="ready", matched_ids=[],
|
|
)
|
|
db_sync.add(audit)
|
|
db_sync.commit()
|
|
with pytest.raises(cleanup_service.ConfirmTokenMismatch):
|
|
cleanup_service.apply_audit_run(
|
|
db_sync, audit_id=audit.id,
|
|
confirm_token="wrong-token", images_root=tmp_path,
|
|
)
|
|
|
|
|
|
def test_cancel_audit_run_flips_status(db_sync):
|
|
audit = LibraryAuditRun(
|
|
rule="transparency", params={"threshold": 0.9},
|
|
status="running", matched_ids=[],
|
|
)
|
|
db_sync.add(audit)
|
|
db_sync.commit()
|
|
cleanup_service.cancel_audit_run(db_sync, audit_id=audit.id)
|
|
new_status = db_sync.execute(
|
|
select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit.id)
|
|
).scalar_one()
|
|
assert new_status == "cancelled"
|