feat(validator): add magic-byte file validator for partial-download detection
Detects truncated/incomplete downloads by checking format-specific head and tail bytes (JPEG SOI/EOI, PNG signature/IEND, GIF header/3B, WEBP RIFF size). Catches the production failure mode where a file landed missing its 2-byte JPEG EOI marker and PIL.ImageFile.load raised "image file is truncated (6 bytes not processed)" downstream. Reads only 16 bytes from each end — O(1) per file. Unknown extensions (JSON sidecars, ugoira zips, etc.) pass through. Hooks into the download flow in a follow-up commit. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,178 @@
|
||||
"""Unit tests for file_validator.
|
||||
|
||||
Covers the actual production failure mode (JPEG missing FF D9 EOI) and
|
||||
the symmetric checks for the other formats we validate. Unknown formats
|
||||
must pass through — we don't want to police gallery-dl's own JSON
|
||||
sidecars or zip ugoira archives.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.services.file_validator import (
|
||||
GIF_HEAD_89A,
|
||||
JPEG_HEAD,
|
||||
JPEG_TAIL,
|
||||
PNG_HEAD,
|
||||
PNG_TAIL,
|
||||
WEBP_FORMAT,
|
||||
WEBP_RIFF,
|
||||
is_validatable,
|
||||
validate_file,
|
||||
)
|
||||
|
||||
|
||||
def _write(tmp_path: Path, name: str, data: bytes) -> Path:
|
||||
p = tmp_path / name
|
||||
p.write_bytes(data)
|
||||
return p
|
||||
|
||||
|
||||
# --- JPEG ---
|
||||
|
||||
|
||||
def test_jpeg_well_formed_passes(tmp_path):
|
||||
body = JPEG_HEAD + b"\x00" * 64 + JPEG_TAIL
|
||||
p = _write(tmp_path, "ok.jpg", body)
|
||||
result = validate_file(p)
|
||||
assert result.ok
|
||||
assert result.format == "jpeg"
|
||||
|
||||
|
||||
def test_jpeg_missing_eoi_fails(tmp_path):
|
||||
"""The exact production failure: file ends 6 bytes short, no FF D9.
|
||||
|
||||
Downstream PIL.ImageFile.load raised "image file is truncated (6 bytes
|
||||
not processed)" — we must catch this at write time.
|
||||
"""
|
||||
body = JPEG_HEAD + b"\x00" * 64 # No JPEG_TAIL
|
||||
p = _write(tmp_path, "truncated.jpg", body)
|
||||
result = validate_file(p)
|
||||
assert not result.ok
|
||||
assert result.format == "jpeg"
|
||||
assert "EOI" in result.reason
|
||||
|
||||
|
||||
def test_jpeg_missing_soi_fails(tmp_path):
|
||||
p = _write(tmp_path, "bogus.jpg", b"NOTAJPEG" * 8 + JPEG_TAIL)
|
||||
result = validate_file(p)
|
||||
assert not result.ok
|
||||
assert "SOI" in result.reason
|
||||
|
||||
|
||||
def test_jpeg_extension_jpeg_also_works(tmp_path):
|
||||
body = JPEG_HEAD + b"\x00" * 32 + JPEG_TAIL
|
||||
p = _write(tmp_path, "ok.jpeg", body)
|
||||
assert validate_file(p).ok
|
||||
|
||||
|
||||
# --- PNG ---
|
||||
|
||||
|
||||
def test_png_well_formed_passes(tmp_path):
|
||||
body = PNG_HEAD + b"\x00" * 32 + PNG_TAIL
|
||||
p = _write(tmp_path, "ok.png", body)
|
||||
result = validate_file(p)
|
||||
assert result.ok
|
||||
assert result.format == "png"
|
||||
|
||||
|
||||
def test_png_missing_iend_fails(tmp_path):
|
||||
body = PNG_HEAD + b"\x00" * 32 # No IEND chunk
|
||||
p = _write(tmp_path, "truncated.png", body)
|
||||
result = validate_file(p)
|
||||
assert not result.ok
|
||||
assert "IEND" in result.reason
|
||||
|
||||
|
||||
def test_png_missing_signature_fails(tmp_path):
|
||||
p = _write(tmp_path, "bogus.png", b"\x00" * 32 + PNG_TAIL)
|
||||
result = validate_file(p)
|
||||
assert not result.ok
|
||||
assert "signature" in result.reason
|
||||
|
||||
|
||||
# --- GIF ---
|
||||
|
||||
|
||||
def test_gif_well_formed_passes(tmp_path):
|
||||
body = GIF_HEAD_89A + b"\x00" * 32 + b"\x3b"
|
||||
p = _write(tmp_path, "ok.gif", body)
|
||||
result = validate_file(p)
|
||||
assert result.ok
|
||||
assert result.format == "gif"
|
||||
|
||||
|
||||
def test_gif_missing_trailer_fails(tmp_path):
|
||||
body = GIF_HEAD_89A + b"\x00" * 32 # No 0x3B
|
||||
p = _write(tmp_path, "truncated.gif", body)
|
||||
result = validate_file(p)
|
||||
assert not result.ok
|
||||
assert "trailer" in result.reason
|
||||
|
||||
|
||||
# --- WEBP ---
|
||||
|
||||
|
||||
def test_webp_well_formed_passes(tmp_path):
|
||||
payload = b"\x00" * 32
|
||||
declared = (4 + len(payload)).to_bytes(4, "little") # WEBP marker + payload
|
||||
body = WEBP_RIFF + declared + WEBP_FORMAT + payload
|
||||
p = _write(tmp_path, "ok.webp", body)
|
||||
result = validate_file(p)
|
||||
assert result.ok, f"unexpected failure: {result.reason}"
|
||||
assert result.format == "webp"
|
||||
|
||||
|
||||
def test_webp_size_mismatch_fails(tmp_path):
|
||||
"""RIFF declares more bytes than the file actually has — classic truncation."""
|
||||
payload = b"\x00" * 32
|
||||
lying_size = (4 + len(payload) + 100).to_bytes(4, "little")
|
||||
body = WEBP_RIFF + lying_size + WEBP_FORMAT + payload
|
||||
p = _write(tmp_path, "truncated.webp", body)
|
||||
result = validate_file(p)
|
||||
assert not result.ok
|
||||
assert "size mismatch" in result.reason
|
||||
|
||||
|
||||
# --- Unknown formats / edge cases ---
|
||||
|
||||
|
||||
def test_zero_byte_file_fails(tmp_path):
|
||||
p = _write(tmp_path, "empty.jpg", b"")
|
||||
result = validate_file(p)
|
||||
assert not result.ok
|
||||
assert "zero-byte" in result.reason
|
||||
|
||||
|
||||
def test_missing_file_fails(tmp_path):
|
||||
p = tmp_path / "ghost.jpg"
|
||||
result = validate_file(p)
|
||||
assert not result.ok
|
||||
assert "not found" in result.reason
|
||||
|
||||
|
||||
def test_unknown_extension_passes_through(tmp_path):
|
||||
"""JSON sidecars from gallery-dl's metadata postprocessor and other
|
||||
non-image artifacts must not be quarantined."""
|
||||
p = _write(tmp_path, "metadata.json", b'{"ok": true}')
|
||||
result = validate_file(p)
|
||||
assert result.ok
|
||||
assert result.format is None
|
||||
|
||||
|
||||
def test_zip_extension_passes_through(tmp_path):
|
||||
"""Ugoira archives (Pixiv) — gallery-dl writes .zip, not our concern."""
|
||||
p = _write(tmp_path, "anim.zip", b"PK\x03\x04random")
|
||||
result = validate_file(p)
|
||||
assert result.ok
|
||||
|
||||
|
||||
def test_is_validatable_flags_known_formats():
|
||||
assert is_validatable(Path("a.jpg"))
|
||||
assert is_validatable(Path("a.JPG")) # case-insensitive
|
||||
assert is_validatable(Path("a.png"))
|
||||
assert is_validatable(Path("a.gif"))
|
||||
assert is_validatable(Path("a.webp"))
|
||||
assert not is_validatable(Path("a.json"))
|
||||
assert not is_validatable(Path("a.mp4"))
|
||||
Reference in New Issue
Block a user