1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
|
"""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 meshbay_common.protocol import IndexEntry
from meshbay_node.indexer.enrich_photo import PhotoEnricher
from meshbay_node.media_cache import MediaCache
from PIL import Image
@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/MESHBAY_DESIGN.md §9.9: 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(encoding="utf-8")
for needle in ("GPSInfo", "GPSTAGS", "0x8825", "34853"):
assert needle not in src, f"found {needle!r} — GPS extraction must never be added here"
|