"""Unit tests for process.py media preprocessing."""
import json
import os
import subprocess
import tempfile
import time
import unittest
from PIL import Image
from process import (
process_collection,
process_video_file,
process_video_collection,
get_video_duration,
generate_manifest,
slugify,
load_config,
)
class TestSlugify(unittest.TestCase):
"""Tests for the slugify helper."""
def test_basic_slugify(self):
self.assertEqual(slugify("Paint and Pixels"), "paint-and-pixels")
def test_special_characters(self):
self.assertEqual(slugify("Hello, World!"), "hello-world")
def test_multiple_spaces(self):
self.assertEqual(slugify("too many spaces"), "too-many-spaces")
def test_already_slug(self):
self.assertEqual(slugify("already-a-slug"), "already-a-slug")
class TestLoadConfig(unittest.TestCase):
"""Tests for config loading."""
def test_load_config_defaults(self):
"""Missing config file should return defaults."""
config = load_config("/nonexistent/path.json")
self.assertEqual(config["fullsize"], 1500)
self.assertEqual(config["thumbnail"], 400)
self.assertEqual(config["quality"], 85)
self.assertEqual(config["video_thumbnail_time"], 2)
def test_load_config_from_file(self):
"""Config file should override defaults."""
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump({"processing": {"fullsize": 2000, "quality": 90}}, f)
f.flush()
config = load_config(f.name)
os.unlink(f.name)
self.assertEqual(config["fullsize"], 2000)
self.assertEqual(config["quality"], 90)
# Non-overridden values keep defaults
self.assertEqual(config["thumbnail"], 400)
class TestProcessCollection(unittest.TestCase):
"""Tests for the image preprocessing pipeline."""
def test_process_single_image(self):
"""A 4000x3000 image should be resized to 1500x1125 full-res and 400x300 thumbnail."""
with tempfile.TemporaryDirectory() as tmpdir:
src_dir = os.path.join(tmpdir, "src")
dst_dir = os.path.join(tmpdir, "dst")
os.makedirs(src_dir)
# Create a 4000x3000 test image
img = Image.new("RGB", (4000, 3000), color=(255, 0, 0))
img.save(os.path.join(src_dir, "big.jpg"))
process_collection(src_dir, dst_dir, fullsize=1500, thumbnail=400, quality=85)
# Check full-res output exists and has correct dimensions
fullres_path = os.path.join(dst_dir, "big.jpg")
self.assertTrue(os.path.exists(fullres_path), "Full-res output should exist")
with Image.open(fullres_path) as out:
self.assertEqual(out.size, (1500, 1125))
# Check thumbnail exists and has correct dimensions
thumb_path = os.path.join(dst_dir, "thumbs", "big.jpg")
self.assertTrue(os.path.exists(thumb_path), "Thumbnail should exist")
with Image.open(thumb_path) as out:
self.assertEqual(out.size, (400, 300))
def test_skip_already_processed(self):
"""Processing again should not overwrite existing outputs (same mtime)."""
with tempfile.TemporaryDirectory() as tmpdir:
src_dir = os.path.join(tmpdir, "src")
dst_dir = os.path.join(tmpdir, "dst")
os.makedirs(src_dir)
# Create and process a test image
img = Image.new("RGB", (2000, 1000), color=(0, 255, 0))
img.save(os.path.join(src_dir, "test.jpg"))
process_collection(src_dir, dst_dir)
fullres_path = os.path.join(dst_dir, "test.jpg")
thumb_path = os.path.join(dst_dir, "thumbs", "test.jpg")
# Record mtimes after first processing
mtime_full_1 = os.path.getmtime(fullres_path)
mtime_thumb_1 = os.path.getmtime(thumb_path)
# Small delay to ensure any rewrite would have a different mtime
time.sleep(0.1)
# Process again
process_collection(src_dir, dst_dir)
# mtimes should be unchanged (files were skipped)
mtime_full_2 = os.path.getmtime(fullres_path)
mtime_thumb_2 = os.path.getmtime(thumb_path)
self.assertEqual(mtime_full_1, mtime_full_2,
"Full-res mtime should not change on re-process")
self.assertEqual(mtime_thumb_1, mtime_thumb_2,
"Thumbnail mtime should not change on re-process")
def test_png_to_jpeg(self):
"""PNG input should produce .jpg output."""
with tempfile.TemporaryDirectory() as tmpdir:
src_dir = os.path.join(tmpdir, "src")
dst_dir = os.path.join(tmpdir, "dst")
os.makedirs(src_dir)
# Create a PNG with alpha channel (RGBA)
img = Image.new("RGBA", (2000, 1500), color=(0, 0, 255, 128))
img.save(os.path.join(src_dir, "alpha.png"))
process_collection(src_dir, dst_dir)
# Output should be .jpg, not .png
jpg_path = os.path.join(dst_dir, "alpha.jpg")
png_path = os.path.join(dst_dir, "alpha.png")
self.assertTrue(os.path.exists(jpg_path),
"PNG should be converted to .jpg output")
self.assertFalse(os.path.exists(png_path),
"Original .png extension should not appear in output")
# Verify it is actually a JPEG
with Image.open(jpg_path) as out:
self.assertEqual(out.format, "JPEG")
def test_small_image_not_upscaled(self):
"""A 300x200 image should not be upscaled."""
with tempfile.TemporaryDirectory() as tmpdir:
src_dir = os.path.join(tmpdir, "src")
dst_dir = os.path.join(tmpdir, "dst")
os.makedirs(src_dir)
# Create a small image (smaller than both fullsize and thumbnail targets)
img = Image.new("RGB", (300, 200), color=(128, 128, 0))
img.save(os.path.join(src_dir, "small.jpg"))
process_collection(src_dir, dst_dir, fullsize=1500, thumbnail=400, quality=85)
# Full-res should keep original size (no upscaling)
fullres_path = os.path.join(dst_dir, "small.jpg")
self.assertTrue(os.path.exists(fullres_path))
with Image.open(fullres_path) as out:
self.assertEqual(out.size, (300, 200),
"Small image should not be upscaled for full-res")
# Thumbnail should also keep original size (no upscaling)
thumb_path = os.path.join(dst_dir, "thumbs", "small.jpg")
self.assertTrue(os.path.exists(thumb_path))
with Image.open(thumb_path) as out:
self.assertEqual(out.size, (300, 200),
"Small image should not be upscaled for thumbnail")
class TestVideoProcessing(unittest.TestCase):
"""Tests for the video preprocessing pipeline."""
def _create_test_video(self, path, duration=3):
"""Create minimal test video with ffmpeg."""
os.makedirs(os.path.dirname(path), exist_ok=True)
subprocess.run([
"ffmpeg", "-y", "-f", "lavfi", "-i",
f"color=c=red:s=320x240:d={duration}",
"-f", "lavfi", "-i", f"anullsrc=r=44100:cl=mono",
"-t", str(duration), "-c:v", "libx264", "-c:a", "aac",
"-shortest", path
], capture_output=True, check=True)
def test_process_video(self):
"""Video should produce faststart MP4 + poster thumbnail."""
with tempfile.TemporaryDirectory() as tmpdir:
src_dir = os.path.join(tmpdir, "src")
dst_dir = os.path.join(tmpdir, "dst")
thumbs_dir = os.path.join(dst_dir, "thumbs")
src_path = os.path.join(src_dir, "clip.mp4")
self._create_test_video(src_path, duration=3)
result = process_video_file(src_path, dst_dir, thumbs_dir, thumbnail_time=1)
self.assertTrue(result, "Video should be processed (not skipped)")
# Output video should exist
out_path = os.path.join(dst_dir, "clip.mp4")
self.assertTrue(os.path.exists(out_path), "Output MP4 should exist")
self.assertGreater(os.path.getsize(out_path), 0, "Output MP4 should not be empty")
# Poster thumbnail should exist
thumb_path = os.path.join(thumbs_dir, "clip.jpg")
self.assertTrue(os.path.exists(thumb_path), "Poster thumbnail should exist")
with Image.open(thumb_path) as img:
self.assertEqual(img.format, "JPEG")
def test_video_skip_already_processed(self):
"""Already-processed videos should be skipped."""
with tempfile.TemporaryDirectory() as tmpdir:
src_dir = os.path.join(tmpdir, "src")
dst_dir = os.path.join(tmpdir, "dst")
thumbs_dir = os.path.join(dst_dir, "thumbs")
src_path = os.path.join(src_dir, "clip.mp4")
self._create_test_video(src_path, duration=2)
# First run should process
result1 = process_video_file(src_path, dst_dir, thumbs_dir)
self.assertTrue(result1, "First run should process")
out_path = os.path.join(dst_dir, "clip.mp4")
thumb_path = os.path.join(thumbs_dir, "clip.jpg")
mtime_vid = os.path.getmtime(out_path)
mtime_thumb = os.path.getmtime(thumb_path)
time.sleep(0.1)
# Second run should skip
result2 = process_video_file(src_path, dst_dir, thumbs_dir)
self.assertFalse(result2, "Second run should skip")
self.assertEqual(os.path.getmtime(out_path), mtime_vid,
"Video mtime should not change on re-process")
self.assertEqual(os.path.getmtime(thumb_path), mtime_thumb,
"Thumbnail mtime should not change on re-process")
def test_get_video_duration(self):
"""Should return correct duration for a test video."""
with tempfile.TemporaryDirectory() as tmpdir:
path = os.path.join(tmpdir, "test.mp4")
self._create_test_video(path, duration=4)
duration = get_video_duration(path)
# Allow some tolerance for encoding
self.assertGreater(duration, 3.0)
self.assertLess(duration, 5.0)
def test_get_video_duration_invalid(self):
"""Should return 0.0 for non-video files."""
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
f.write(b"not a video")
f.flush()
duration = get_video_duration(f.name)
os.unlink(f.name)
self.assertEqual(duration, 0.0)
def test_process_video_collection(self):
"""Should process all videos in a directory."""
with tempfile.TemporaryDirectory() as tmpdir:
src_dir = os.path.join(tmpdir, "src")
dst_dir = os.path.join(tmpdir, "dst")
os.makedirs(src_dir)
# Create two test videos
self._create_test_video(os.path.join(src_dir, "a.mp4"), duration=2)
self._create_test_video(os.path.join(src_dir, "b.mp4"), duration=2)
# Also create a non-video file (should be ignored)
with open(os.path.join(src_dir, "readme.txt"), "w") as f:
f.write("not a video")
processed, skipped = process_video_collection(src_dir, dst_dir)
self.assertEqual(processed, 2)
self.assertEqual(skipped, 0)
self.assertTrue(os.path.exists(os.path.join(dst_dir, "a.mp4")))
self.assertTrue(os.path.exists(os.path.join(dst_dir, "b.mp4")))
self.assertTrue(os.path.exists(os.path.join(dst_dir, "thumbs", "a.jpg")))
self.assertTrue(os.path.exists(os.path.join(dst_dir, "thumbs", "b.jpg")))
class TestManifestGeneration(unittest.TestCase):
"""Tests for manifest generation."""
def _setup_images(self, base_dir, collection, filenames):
"""Create dummy image files in images/<collection>/."""
coll_dir = os.path.join(base_dir, "images", collection)
thumbs_dir = os.path.join(coll_dir, "thumbs")
os.makedirs(coll_dir, exist_ok=True)
os.makedirs(thumbs_dir, exist_ok=True)
for fname in filenames:
img = Image.new("RGB", (100, 100), color=(255, 0, 0))
img.save(os.path.join(coll_dir, fname))
img.save(os.path.join(thumbs_dir, fname))
def _setup_videos(self, base_dir, collection, filenames):
"""Create dummy video files in videos/<collection>/."""
coll_dir = os.path.join(base_dir, "videos", collection)
thumbs_dir = os.path.join(coll_dir, "thumbs")
os.makedirs(coll_dir, exist_ok=True)
os.makedirs(thumbs_dir, exist_ok=True)
for fname in filenames:
# Create a minimal valid MP4 for duration detection
# (just an empty file -- duration will be 0.0)
with open(os.path.join(coll_dir, fname), "wb") as f:
f.write(b"\x00" * 100)
# Create thumbnail
basename = os.path.splitext(fname)[0]
img = Image.new("RGB", (100, 100), color=(0, 0, 255))
img.save(os.path.join(thumbs_dir, basename + ".jpg"))
def test_generate_manifest_images(self):
"""Manifest should include all images with correct structure."""
with tempfile.TemporaryDirectory() as tmpdir:
self._setup_images(tmpdir, "nature", ["sunset.jpg", "forest.jpg"])
config = {"collections": {"order": ["nature"], "ignore": []}}
manifest = generate_manifest(tmpdir, config)
self.assertEqual(manifest["collections"], ["nature"])
self.assertEqual(len(manifest["items"]), 2)
item = manifest["items"][0]
self.assertEqual(item["type"], "image")
self.assertEqual(item["collection"], "nature")
self.assertIn("name", item)
self.assertIn("full", item)
self.assertIn("thumb", item)
self.assertIn("hasInfo", item)
self.assertIn("mdUrl", item)
def test_generate_manifest_with_sidecar(self):
"""Images with .md sidecars should include sidecar content."""
with tempfile.TemporaryDirectory() as tmpdir:
self._setup_images(tmpdir, "art", ["piece.jpg"])
# Write a sidecar
sidecar_path = os.path.join(tmpdir, "images", "art", "piece.md")
with open(sidecar_path, "w") as f:
f.write("Oil on canvas, 2024")
config = {"collections": {"order": ["art"], "ignore": []}}
manifest = generate_manifest(tmpdir, config)
self.assertEqual(len(manifest["items"]), 1)
self.assertTrue(manifest["items"][0]["hasInfo"])
self.assertEqual(manifest["items"][0]["mdUrl"], "images/art/piece.md")
def test_generate_manifest_respects_order(self):
"""Collections should follow site.json order."""
with tempfile.TemporaryDirectory() as tmpdir:
self._setup_images(tmpdir, "alpha", ["a.jpg"])
self._setup_images(tmpdir, "beta", ["b.jpg"])
self._setup_images(tmpdir, "gamma", ["g.jpg"])
config = {"collections": {"order": ["gamma", "alpha", "beta"], "ignore": []}}
manifest = generate_manifest(tmpdir, config)
self.assertEqual(manifest["collections"], ["gamma", "alpha", "beta"])
def test_generate_manifest_unordered_collections(self):
"""Collections not in order list should appear alphabetically after ordered ones."""
with tempfile.TemporaryDirectory() as tmpdir:
self._setup_images(tmpdir, "alpha", ["a.jpg"])
self._setup_images(tmpdir, "beta", ["b.jpg"])
self._setup_images(tmpdir, "gamma", ["g.jpg"])
config = {"collections": {"order": ["gamma"], "ignore": []}}
manifest = generate_manifest(tmpdir, config)
self.assertEqual(manifest["collections"], ["gamma", "alpha", "beta"])
def test_generate_manifest_ignores(self):
"""Ignored collections should not appear."""
with tempfile.TemporaryDirectory() as tmpdir:
self._setup_images(tmpdir, "keep", ["a.jpg"])
self._setup_images(tmpdir, "old", ["b.jpg"])
self._setup_images(tmpdir, "_archive", ["c.jpg"])
config = {"collections": {"order": [], "ignore": ["old", "_archive"]}}
manifest = generate_manifest(tmpdir, config)
self.assertEqual(manifest["collections"], ["keep"])
self.assertEqual(len(manifest["items"]), 1)
def test_generate_manifest_mixed_media(self):
"""Manifest should include both images and videos from same collection."""
with tempfile.TemporaryDirectory() as tmpdir:
self._setup_images(tmpdir, "portfolio", ["photo.jpg"])
self._setup_videos(tmpdir, "portfolio", ["clip.mp4"])
config = {"collections": {"order": ["portfolio"], "ignore": []}}
manifest = generate_manifest(tmpdir, config)
self.assertEqual(manifest["collections"], ["portfolio"])
self.assertEqual(len(manifest["items"]), 2)
types = {item["type"] for item in manifest["items"]}
self.assertEqual(types, {"image", "video"})
def test_generate_manifest_empty(self):
"""Empty directories should produce empty manifest."""
with tempfile.TemporaryDirectory() as tmpdir:
config = {"collections": {"order": [], "ignore": []}}
manifest = generate_manifest(tmpdir, config)
self.assertEqual(manifest["collections"], [])
self.assertEqual(manifest["items"], [])
if __name__ == "__main__":
unittest.main()