"""Tests for indexer/enrich_photo.py — the Photos app's thumbnail/EXIF pass.""" import asyncio import io from pathlib import Path import piexif import pytest from PIL import Image from meshbay_common.protocol import IndexEntry from meshbay_node.indexer.enrich_photo import PhotoEnricher from meshbay_node.media_cache import MediaCache @pytest.fixture async def media_cache(tmp_path): c = MediaCache(db_path=tmp_path / "media_cache.db") await c.open() yield c await c.close() def _save_jpeg(path: Path, size=(300, 200), color="red", exif_bytes: bytes | None = None): img = Image.new("RGB", size, color) kwargs = {"format": "JPEG"} if exif_bytes is not None: kwargs["exif"] = exif_bytes img.save(path, **kwargs) async def _run(enricher: PhotoEnricher, entry: IndexEntry, path: Path): done = asyncio.get_event_loop().create_future() async def on_done(file_id, fields): done.set_result((file_id, fields)) enricher.spawn(entry, path, on_done) return await asyncio.wait_for(done, timeout=15) @pytest.mark.asyncio async def test_enricher_populates_dimensions_and_stores_thumbnail(tmp_path, media_cache): img = tmp_path / "plain.jpg" _save_jpeg(img, size=(300, 200)) entry = IndexEntry(id="fileid1", name=img.name, path=img.name, size=img.stat().st_size, type="image", added_at=0) enricher = PhotoEnricher(media_cache) file_id, fields = await _run(enricher, entry, img) assert file_id == "fileid1" assert fields["width"] == 300 assert fields["height"] == 200 assert fields.get("thumb_hash") stored = await media_cache.get_thumb(fields["thumb_hash"]) assert stored is not None and len(stored) > 0 # Actually decodes as a downsized JPEG, not just non-empty bytes. thumb = Image.open(io.BytesIO(stored)) assert thumb.format == "JPEG" assert max(thumb.size) <= 480 @pytest.mark.asyncio async def test_enricher_reuses_cached_meta_without_touching_the_file_again(tmp_path, media_cache): """ The gap this closes: enrichment fields only ever lived in the in-memory GroupIndex, so every daemon restart re-ran Pillow over every photo in every configured root from scratch, even though media_cache.db (the thumbnail bytes) already had the answer. A second PhotoEnricher sharing the same media_cache — standing in for "the daemon restarted" — must reuse it instead. Proven strongly: the source file is deleted between the two runs, so a second real decode attempt would fail outright rather than merely being redundant. """ img = tmp_path / "reused.jpg" _save_jpeg(img, size=(300, 200)) entry = IndexEntry(id="fileid_reuse", name=img.name, path=img.name, size=img.stat().st_size, type="image", added_at=0) first_enricher = PhotoEnricher(media_cache) _, first_fields = await _run(first_enricher, entry, img) assert first_fields.get("thumb_hash") img.unlink() # a real second decode would now raise, not just be wasteful second_enricher = PhotoEnricher(media_cache) _, second_fields = await _run(second_enricher, entry, img) assert second_fields == first_fields @pytest.mark.asyncio async def test_enricher_no_exif_degrades_gracefully(tmp_path, media_cache): """A screenshot or a re-saved image with no EXIF block at all is the ordinary case, not an error — must not raise and must leave taken_at/ camera unset rather than guessing.""" img = tmp_path / "no_exif.jpg" _save_jpeg(img) entry = IndexEntry(id="fileid2", name=img.name, path=img.name, size=img.stat().st_size, type="image", added_at=0) enricher = PhotoEnricher(media_cache) _, fields = await _run(enricher, entry, img) assert fields.get("taken_at") is None assert fields.get("camera") is None assert fields.get("thumb_hash") @pytest.mark.asyncio async def test_enricher_reads_taken_at_from_realistic_camera_exif(tmp_path, media_cache): """ Built with piexif rather than a flat Image.Exif dict: a real camera stores DateTimeOriginal in the Exif sub-IFD (tag 0x8769), not the 0th IFD — Pillow's own getexif() only sees the 0th IFD directly. A flat dict (`img.getexif()[36867] = ...`) round-trips inside Pillow without ever exercising that distinction, which is exactly what let an earlier version of enrich_photo.py read `img.getexif().get(36867)` and always get None against a real photo while Make/Model (genuine 0th-IFD tags) kept working — found only once this test built EXIF the way real hardware does. """ img = tmp_path / "camera.jpg" exif_dict = { "0th": {piexif.ImageIFD.Make: b"Acme", piexif.ImageIFD.Model: b"Camera X"}, "Exif": {piexif.ExifIFD.DateTimeOriginal: b"2024:01:02 03:04:05"}, "GPS": {}, "1st": {}, "thumbnail": None, } _save_jpeg(img, exif_bytes=piexif.dump(exif_dict)) entry = IndexEntry(id="fileid3", name=img.name, path=img.name, size=img.stat().st_size, type="image", added_at=0) enricher = PhotoEnricher(media_cache) _, fields = await _run(enricher, entry, img) assert fields["camera"] == "Acme Camera X" assert fields["taken_at"] is not None # 2024-01-02 03:04:05 UTC-ish, tolerant of local-time parsing: same day. import datetime dt = datetime.datetime.fromtimestamp(fields["taken_at"]) assert (dt.year, dt.month, dt.day) == (2024, 1, 2) @pytest.mark.asyncio async def test_enricher_corrects_orientation(tmp_path, media_cache): """ A phone photo is routinely stored "sideways" with an EXIF Orientation tag telling viewers how to rotate it — width/height, and the thumbnail itself, must describe the *displayed* image, not the raw stored one. Orientation 6 stores a 300x200 frame that displays as 200x300. """ img = tmp_path / "rotated.jpg" raw = Image.new("RGB", (300, 200), "blue") exif = raw.getexif() exif[274] = 6 # Orientation buf = io.BytesIO() raw.save(buf, format="JPEG", exif=exif) img.write_bytes(buf.getvalue()) entry = IndexEntry(id="fileid4", name=img.name, path=img.name, size=img.stat().st_size, type="image", added_at=0) enricher = PhotoEnricher(media_cache) _, fields = await _run(enricher, entry, img) assert (fields["width"], fields["height"]) == (200, 300), ( "width/height must reflect the EXIF-corrected orientation, not the raw stored frame") thumb_bytes = await media_cache.get_thumb(fields["thumb_hash"]) thumb = Image.open(io.BytesIO(thumb_bytes)) assert thumb.size[0] < thumb.size[1], "the stored thumbnail itself must be portrait, not sideways" def test_gps_is_never_read_by_this_module(): """ docs/photos.md §2.4/§11: GPS must never be extracted, cached, or handed to a caller — a location disclosure the instant it is surfaced to every group member. Grep-based, the same discipline test_hub_address_seam.py/ test_task_lifetime.py already apply elsewhere in this codebase to a property that must never silently regress. Checks for the actual extraction API (the GPS sub-IFD constant, its numeric tag ids in either byte order, and the GPS tag-name table) rather than the bare word "GPS" — this module's own docstrings and comments say GPS *on purpose*, explaining why it is never read; that prose is not what this test is guarding against. """ src = (Path(__file__).resolve().parents[1] / "src" / "meshbay_node" / "indexer" / "enrich_photo.py").read_text() for needle in ("GPSInfo", "GPSTAGS", "0x8825", "34853"): assert needle not in src, f"found {needle!r} — GPS extraction must never be added here"