142 lines
4.4 KiB
Python
142 lines
4.4 KiB
Python
"""FC-2e: verify_integrity — sha + decode/probe per ImageRecord."""
|
|
|
|
import subprocess
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
from backend.app.models import ImageRecord
|
|
from backend.app.tasks.maintenance import verify_integrity
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
class _Ctx:
|
|
def __init__(self, s):
|
|
self.s = s
|
|
|
|
def __enter__(self):
|
|
return self.s
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
|
|
def _sf(db_sync):
|
|
class _SM:
|
|
def __call__(self):
|
|
return _Ctx(db_sync)
|
|
|
|
return _SM()
|
|
|
|
|
|
def _good_jpeg(path):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
Image.new("RGB", (32, 32), (40, 80, 160)).save(path, "JPEG")
|
|
|
|
|
|
def _rec(db_sync, path, *, sha, mime="image/jpeg"):
|
|
rec = ImageRecord(
|
|
path=str(path), sha256=sha, phash=None, size_bytes=1,
|
|
mime=mime, width=32, height=32, origin="imported_filesystem",
|
|
integrity_status="unknown",
|
|
)
|
|
db_sync.add(rec)
|
|
db_sync.flush()
|
|
return rec
|
|
|
|
|
|
def _sha_of(path):
|
|
import hashlib
|
|
|
|
h = hashlib.sha256()
|
|
h.update(path.read_bytes())
|
|
return h.hexdigest()
|
|
|
|
|
|
def test_ok_corrupt_failed_paths(db_sync, tmp_path, monkeypatch):
|
|
from backend.app.tasks import maintenance as m
|
|
|
|
good = tmp_path / "good.jpg"
|
|
_good_jpeg(good)
|
|
good2 = tmp_path / "good2.jpg" # distinct path — image_record.path is unique
|
|
_good_jpeg(good2)
|
|
bad = tmp_path / "bad.jpg"
|
|
bad.write_bytes(b"not a real jpeg")
|
|
truncated = tmp_path / "trunc.jpg"
|
|
_good_jpeg(truncated)
|
|
truncated.write_bytes(truncated.read_bytes()[:30]) # break the bitstream
|
|
|
|
rec_ok = _rec(db_sync, good, sha=_sha_of(good))
|
|
rec_sha_mismatch = _rec(db_sync, good2, sha="0" * 64) # bytes intact, sha wrong → corrupt
|
|
rec_decode_fail = _rec(db_sync, truncated, sha=_sha_of(truncated))
|
|
# 'bad' has not-jpeg bytes; sha matches the bad bytes → decode fails
|
|
rec_bad_decode = _rec(db_sync, bad, sha=_sha_of(bad))
|
|
rec_missing = _rec(
|
|
db_sync, tmp_path / "gone.jpg", sha="g" + "0" * 63,
|
|
)
|
|
db_sync.commit()
|
|
|
|
monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
|
|
|
|
total = verify_integrity()
|
|
db_sync.expire_all()
|
|
assert total == 5
|
|
assert db_sync.get(ImageRecord, rec_ok.id).integrity_status == "ok"
|
|
assert db_sync.get(ImageRecord, rec_sha_mismatch.id).integrity_status == "corrupt"
|
|
assert db_sync.get(ImageRecord, rec_decode_fail.id).integrity_status == "corrupt"
|
|
assert db_sync.get(ImageRecord, rec_bad_decode.id).integrity_status == "corrupt"
|
|
assert db_sync.get(ImageRecord, rec_missing.id).integrity_status == "failed_verification"
|
|
|
|
# Idempotent — second run re-derives the same verdict.
|
|
verify_integrity()
|
|
db_sync.expire_all()
|
|
assert db_sync.get(ImageRecord, rec_ok.id).integrity_status == "ok"
|
|
|
|
|
|
def test_video_ffprobe_paths(db_sync, tmp_path, monkeypatch):
|
|
from backend.app.tasks import maintenance as m
|
|
|
|
vid = tmp_path / "v.mp4"
|
|
vid.write_bytes(b"fake but consistent video bytes")
|
|
rec = _rec(db_sync, vid, sha=_sha_of(vid), mime="video/mp4")
|
|
db_sync.commit()
|
|
|
|
monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
|
|
|
|
# Case A: ffprobe succeeds → ok
|
|
def _run_ok(args, **kw):
|
|
return subprocess.CompletedProcess(args, 0, b"", b"")
|
|
|
|
monkeypatch.setattr(m.subprocess, "run", _run_ok)
|
|
verify_integrity()
|
|
db_sync.expire_all()
|
|
assert db_sync.get(ImageRecord, rec.id).integrity_status == "ok"
|
|
|
|
# Case B: ffprobe non-zero → corrupt
|
|
def _run_bad(args, **kw):
|
|
return subprocess.CompletedProcess(args, 1, b"", b"err")
|
|
|
|
monkeypatch.setattr(m.subprocess, "run", _run_bad)
|
|
verify_integrity()
|
|
db_sync.expire_all()
|
|
assert db_sync.get(ImageRecord, rec.id).integrity_status == "corrupt"
|
|
|
|
# Case C: ffprobe times out → corrupt
|
|
def _run_timeout(args, **kw):
|
|
raise subprocess.TimeoutExpired(cmd=args, timeout=10)
|
|
|
|
monkeypatch.setattr(m.subprocess, "run", _run_timeout)
|
|
verify_integrity()
|
|
db_sync.expire_all()
|
|
assert db_sync.get(ImageRecord, rec.id).integrity_status == "corrupt"
|
|
|
|
# Case D: ffprobe binary missing → failed_verification
|
|
def _run_missing(args, **kw):
|
|
raise FileNotFoundError("ffprobe")
|
|
|
|
monkeypatch.setattr(m.subprocess, "run", _run_missing)
|
|
verify_integrity()
|
|
db_sync.expire_all()
|
|
assert db_sync.get(ImageRecord, rec.id).integrity_status == "failed_verification"
|