immich-photo-flow/apps/trip-cluster/tests/test_ingest.py

import os
from photoflow.core import Store
from photoflow.immich import pipeline
from app.ingest import run_ingest


class FakeImmich:
    def __init__(self, assets, tag_map=None, thumb=b"\xff\xd8\xffjpeg"):
        self._assets = assets
        self._tag_map = tag_map or {}
        self._thumb = thumb
        self.searches = []

    def resolve_tag_id(self, name):
        return self._tag_map.get(name)

    def search_assets(self, **kwargs):
        self.searches.append(kwargs)
        return list(self._assets)

    def download_thumbnail(self, asset_id):
        return self._thumb


def _asset(i, taken, tags=None, updated="2026-01-01T00:00:00Z"):
    return {"id": i, "original_filename": f"{i}.jpg", "taken_at": taken,
            "gps_lat": None, "gps_lon": None, "place_city": None,
            "place_country": None, "type": "IMAGE", "tags": tags or [],
            "rating": 0, "updated_at": updated}


def test_ingest_upserts_assets_tags_and_thumbs(tmp_path):
    store = Store(str(tmp_path / "t.db")).connect()
    thumbs = str(tmp_path / "thumbs")
    os.makedirs(thumbs, exist_ok=True)
    client = FakeImmich([_asset("a", "2019-06-01", tags=["Italy 2019"]),
                         _asset("b", "2019-06-02")])
    res = run_ingest(client, store, thumbs, date_from="2019-01-01", date_to="2020-01-01")
    assert res["fetched"] == 2
    assert store.get_asset("a").taken_at == "2019-06-01"
    assert store.asset_tags("a") == ["Italy 2019"]
    assert os.path.exists(os.path.join(thumbs, "a.jpg"))
    assert store.get_meta("last_ingest_at") == "2026-01-01T00:00:00Z"
    store.close()


def test_ingest_marks_processed_from_pipeline_tag(tmp_path):
    store = Store(str(tmp_path / "t.db")).connect()
    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
    client = FakeImmich([_asset("a", "2019-06-01", tags=[pipeline.PROCESSED])])
    res = run_ingest(client, store, thumbs)
    assert res["processed_marked"] == 1
    assert store.get_asset("a").processed is True
    store.close()


def test_ingest_incremental_passes_updated_after(tmp_path):
    store = Store(str(tmp_path / "t.db")).connect()
    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
    store.set_meta("last_ingest_at", "2026-05-01T00:00:00Z")
    client = FakeImmich([_asset("a", "2019-06-01")])
    run_ingest(client, store, thumbs)
    assert client.searches[0].get("updated_after") == "2026-05-01T00:00:00Z"
    # --full ignores it
    client2 = FakeImmich([_asset("a", "2019-06-01")])
    run_ingest(client2, store, thumbs, full=True)
    assert client2.searches[0].get("updated_after") is None
    store.close()


def test_ingest_subset_and_tag_resolution(tmp_path):
    store = Store(str(tmp_path / "t.db")).connect()
    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
    client = FakeImmich([_asset("a", "2019-06-01"), _asset("b", "2019-06-02")],
                        tag_map={"Italy 2019": "t1"})
    res = run_ingest(client, store, thumbs, tag="Italy 2019", subset=1)
    assert res["fetched"] == 1
    assert client.searches[0].get("tag_ids") == ["t1"]
    store.close()


def test_ingest_resilient_to_thumb_failure_and_zero_byte(tmp_path):
    store = Store(str(tmp_path / "t.db")).connect()
    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)

    class FailThenOk(FakeImmich):
        def download_thumbnail(self, asset_id):
            if asset_id == "a":
                raise RuntimeError("boom")
            return self._thumb

    # Pre-existing 0-byte file for "b" must be treated as missing and refetched.
    with open(os.path.join(thumbs, "b.jpg"), "wb"):
        pass

    client = FailThenOk([_asset("a", "2019-06-01"), _asset("b", "2019-06-02")])
    res = run_ingest(client, store, thumbs)
    # The run completes despite "a" failing.
    assert res["fetched"] == 2
    # No 0-byte file left for the failed asset.
    assert not os.path.exists(os.path.join(thumbs, "a.jpg"))
    # The 0-byte file for "b" was replaced with real bytes.
    assert os.path.getsize(os.path.join(thumbs, "b.jpg")) > 0
    store.close()


def test_ingest_watermark_not_advanced_past_failed_asset(tmp_path):
    # A later successful asset must NOT advance last_ingest_at past an earlier
    # asset whose thumbnail failed, or that asset is stranded on the next
    # incremental run (review finding #9).
    store = Store(str(tmp_path / "t.db")).connect()
    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)

    class FailOld(FakeImmich):
        def download_thumbnail(self, asset_id):
            if asset_id == "old":
                raise RuntimeError("boom")
            return self._thumb

    client = FailOld([_asset("old", "2008-06-01", updated="2026-01-01T00:00:00Z"),
                      _asset("new", "2024-03-01", updated="2026-03-01T00:00:00Z")])
    run_ingest(client, store, thumbs)
    assert store.get_asset("old") is None and store.get_asset("new") is not None
    wm = store.get_meta("last_ingest_at")
    # cursor must stay below the failed asset (here: not advanced at all)
    assert wm is None or wm < "2026-01-01T00:00:00Z"
    assert wm != "2026-03-01T00:00:00Z"
    store.close()


def test_ingest_subset_run_does_not_advance_watermark(tmp_path):
    # --subset is a sampling run; it must not move the incremental cursor or it
    # would strand the un-fetched remainder.
    store = Store(str(tmp_path / "t.db")).connect()
    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
    client = FakeImmich([_asset("a", "2019-06-01", updated="2026-02-01T00:00:00Z"),
                         _asset("b", "2019-06-02", updated="2026-02-02T00:00:00Z")])
    run_ingest(client, store, thumbs, subset=1)
    assert store.get_meta("last_ingest_at") is None
    store.close()