216 lines
6.5 KiB
Python
216 lines
6.5 KiB
Python
"""API tests for the /api/cleanup/* blueprint.
|
|
|
|
Per reference-async-coredml-test-assertions, post-DML state checks go
|
|
via column selects, not ORM entity access.
|
|
"""
|
|
|
|
import hashlib
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
from sqlalchemy import func, select
|
|
|
|
from backend.app import create_app
|
|
from backend.app.celery_app import celery
|
|
from backend.app.models import ImageRecord, LibraryAuditRun
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def disable_celery_eager(monkeypatch):
|
|
monkeypatch.setattr(celery.conf, "task_always_eager", False)
|
|
|
|
|
|
@pytest.fixture
|
|
async def app():
|
|
return create_app()
|
|
|
|
|
|
@pytest.fixture
|
|
async def client(app):
|
|
async with app.test_client() as c:
|
|
yield c
|
|
|
|
|
|
def _sha256_min_dim_token(min_w: int, min_h: int) -> str:
|
|
canon = f"{min_w}x{min_h}"
|
|
return f"delete-min-dim-{hashlib.sha256(canon.encode()).hexdigest()[:8]}"
|
|
|
|
|
|
async def _seed_image(db, tmp_path, *, w, h, name):
|
|
path = tmp_path / name
|
|
Image.new("RGB", (w, h), (w % 256, h % 256, 0)).save(path)
|
|
sha = f"api-cleanup-{name}".ljust(64, "x")[:64]
|
|
rec = ImageRecord(
|
|
path=str(path), sha256=sha,
|
|
size_bytes=path.stat().st_size, mime="image/png",
|
|
width=w, height=h, origin="imported_filesystem",
|
|
integrity_status="ok",
|
|
)
|
|
db.add(rec)
|
|
await db.flush()
|
|
return rec
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_min_dimension_preview_returns_count(client, db, tmp_path):
|
|
await _seed_image(db, tmp_path, w=50, h=50, name="small.png")
|
|
await _seed_image(db, tmp_path, w=500, h=500, name="big.png")
|
|
await db.commit()
|
|
resp = await client.post(
|
|
"/api/cleanup/min-dimension/preview",
|
|
json={"min_width": 200, "min_height": 200},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["count"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_min_dimension_delete_with_token_removes_rows(client, db, tmp_path):
|
|
await _seed_image(db, tmp_path, w=50, h=50, name="s2.png")
|
|
await _seed_image(db, tmp_path, w=500, h=500, name="b2.png")
|
|
await db.commit()
|
|
token = _sha256_min_dim_token(200, 200)
|
|
resp = await client.post(
|
|
"/api/cleanup/min-dimension/delete",
|
|
json={"min_width": 200, "min_height": 200, "confirm": token},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["deleted"] == 1
|
|
remaining = await db.execute(select(func.count()).select_from(ImageRecord))
|
|
assert remaining.scalar_one() == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_min_dimension_delete_with_bad_token_returns_400(client, db, tmp_path):
|
|
await _seed_image(db, tmp_path, w=50, h=50, name="s3.png")
|
|
await db.commit()
|
|
resp = await client.post(
|
|
"/api/cleanup/min-dimension/delete",
|
|
json={"min_width": 200, "min_height": 200, "confirm": "nope"},
|
|
)
|
|
assert resp.status_code == 400
|
|
body = await resp.get_json()
|
|
assert body["error"] == "confirm_mismatch"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_create_returns_id_and_running_status(
|
|
client, db, monkeypatch,
|
|
):
|
|
from backend.app.tasks import library_audit
|
|
monkeypatch.setattr(
|
|
library_audit.scan_library_for_rule, "delay", lambda audit_id: None,
|
|
)
|
|
resp = await client.post(
|
|
"/api/cleanup/audit",
|
|
json={"rule": "transparency", "params": {"threshold": 0.9}},
|
|
)
|
|
assert resp.status_code == 202
|
|
body = await resp.get_json()
|
|
assert body["status"] == "running"
|
|
assert isinstance(body["audit_id"], int)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_create_returns_409_when_another_is_running(client, db):
|
|
audit = LibraryAuditRun(
|
|
rule="transparency", params={"threshold": 0.9},
|
|
status="running", matched_ids=[],
|
|
)
|
|
db.add(audit)
|
|
await db.commit()
|
|
resp = await client.post(
|
|
"/api/cleanup/audit",
|
|
json={"rule": "single_color", "params": {"threshold": 0.95, "tolerance": 30}},
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_get_by_id_returns_full_row(client, db):
|
|
audit = LibraryAuditRun(
|
|
rule="transparency", params={"threshold": 0.85},
|
|
status="ready", scanned_count=100, matched_count=3,
|
|
matched_ids=[1, 2, 3],
|
|
)
|
|
db.add(audit)
|
|
await db.commit()
|
|
resp = await client.get(f"/api/cleanup/audit/{audit.id}")
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["rule"] == "transparency"
|
|
assert body["matched_count"] == 3
|
|
assert body["matched_ids"] == [1, 2, 3]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_history_returns_recent_runs(client, db):
|
|
for _ in range(3):
|
|
db.add(LibraryAuditRun(
|
|
rule="transparency", params={"threshold": 0.9},
|
|
status="applied", matched_ids=[],
|
|
finished_at=datetime.now(UTC),
|
|
))
|
|
await db.commit()
|
|
resp = await client.get("/api/cleanup/audit?limit=5")
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert len(body["runs"]) >= 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_apply_with_token_deletes(client, db, tmp_path):
|
|
rec = await _seed_image(db, tmp_path, w=100, h=100, name="apply.png")
|
|
audit = LibraryAuditRun(
|
|
rule="transparency", params={"threshold": 0.9},
|
|
status="ready", scanned_count=1, matched_count=1,
|
|
matched_ids=[rec.id],
|
|
)
|
|
db.add(audit)
|
|
await db.commit()
|
|
resp = await client.post(
|
|
f"/api/cleanup/audit/{audit.id}/apply",
|
|
json={"confirm": f"delete-audit-{audit.id}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["deleted"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_apply_with_bad_token_returns_400(client, db):
|
|
audit = LibraryAuditRun(
|
|
rule="transparency", params={"threshold": 0.9},
|
|
status="ready", matched_ids=[],
|
|
)
|
|
db.add(audit)
|
|
await db.commit()
|
|
resp = await client.post(
|
|
f"/api/cleanup/audit/{audit.id}/apply",
|
|
json={"confirm": "wrong-token"},
|
|
)
|
|
assert resp.status_code == 400
|
|
body = await resp.get_json()
|
|
assert body["error"] == "confirm_mismatch"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_cancel_flips_status(client, db):
|
|
audit = LibraryAuditRun(
|
|
rule="transparency", params={"threshold": 0.9},
|
|
status="running", matched_ids=[],
|
|
)
|
|
db.add(audit)
|
|
await db.commit()
|
|
resp = await client.post(f"/api/cleanup/audit/{audit.id}/cancel")
|
|
assert resp.status_code == 200
|
|
new_status = await db.execute(
|
|
select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit.id)
|
|
)
|
|
assert new_status.scalar_one() == "cancelled"
|