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

from photoflow.core import Store
from photoflow.core.models import Asset, Cluster, ClusterMember
from photoflow.immich import pipeline
from app.writeback import apply_cluster, apply_all


class FakeImmich:
    def __init__(self):
        self.tagged = []
        self.fail_tag_id = None
        self.fail_upsert = None          # tag name whose upsert_tag should raise
        self._ids = {}

    def upsert_tag(self, name):
        if self.fail_upsert is not None and name == self.fail_upsert:
            raise RuntimeError("upsert boom")
        self._ids.setdefault(name, f"id:{name}")
        return self._ids[name]

    def tag_assets(self, tag_id, ids):
        if self.fail_tag_id is not None and tag_id == self.fail_tag_id:
            raise RuntimeError("boom")
        self.tagged.append((tag_id, list(ids)))


def _store(tmp_path):
    s = Store(str(tmp_path / "t.db")).connect()
    for i in ("a", "b", "x"):
        s.upsert_asset(Asset(immich_id=i, taken_at="2019-06-01"))
    return s


def _approved(s, name="Venice"):
    cid = s.insert_cluster(
        Cluster(start_at="2019-06-01", end_at="2019-06-02", suggested_name=name,
                status="approved", decided_name=name),
        [ClusterMember(cluster_id=0, immich_id="a"),
         ClusterMember(cluster_id=0, immich_id="b"),
         ClusterMember(cluster_id=0, immich_id="x", included=False, flagged_coverage=True)])
    return cid


def test_apply_approved_tags_included_then_processed(tmp_path):
    s = _store(tmp_path)
    cid = _approved(s)
    client = FakeImmich()
    res = apply_cluster(client, s, cid)
    assert sorted(res["succeeded"]) == ["a", "b"] and res["failed"] == []
    # trip tag on a,b ; processed on a,b ; x (excluded) never tagged
    assert ("id:Venice", ["a", "b"]) in client.tagged
    assert ("id:_pipeline/processed", ["a", "b"]) in client.tagged
    assert all("x" not in ids for _, ids in client.tagged)
    assert s.get_asset("a").processed is True
    s.close()


def test_apply_is_idempotent(tmp_path):
    s = _store(tmp_path)
    cid = _approved(s)
    client = FakeImmich()
    apply_cluster(client, s, cid)
    before = len(client.tagged)
    apply_cluster(client, s, cid)             # second run writes nothing new
    assert len(client.tagged) == before
    s.close()


def test_partial_failure_leaves_retryable(tmp_path):
    s = _store(tmp_path)
    cid = _approved(s)
    client = FakeImmich()
    client.fail_tag_id = "id:Venice"          # trip tag write fails
    res = apply_cluster(client, s, cid)
    assert res["succeeded"] == [] and sorted(i for i, _ in res["failed"]) == ["a", "b"]
    assert s.get_asset("a").processed is False         # not marked processed on failure
    assert s.already_applied("a", "trip", "Venice") is False  # retryable
    s.close()


def test_apply_non_trip(tmp_path):
    s = _store(tmp_path)
    cid = s.insert_cluster(
        Cluster(start_at="2019-06-01", end_at="2019-06-02", status="non_trip"),
        [ClusterMember(cluster_id=0, immich_id="a")])
    client = FakeImmich()
    apply_cluster(client, s, cid)
    assert ("id:_pipeline/non-trip", ["a"]) in client.tagged
    assert ("id:_pipeline/processed", ["a"]) in client.tagged
    s.close()


def test_apply_all_reports_per_cluster(tmp_path):
    s = _store(tmp_path)
    _approved(s, name="Venice")
    client = FakeImmich()
    results = apply_all(client, s)
    assert len(results) == 1 and results[0]["status"] == "approved"
    s.close()


def test_apply_skipped(tmp_path):
    s = _store(tmp_path)
    cid = s.insert_cluster(
        Cluster(start_at="2019-06-01", end_at="2019-06-02", status="skipped"),
        [ClusterMember(cluster_id=0, immich_id="a"),
         ClusterMember(cluster_id=0, immich_id="b")])
    client = FakeImmich()
    res = apply_cluster(client, s, cid)
    # skipped: only _pipeline/processed is written, no content/non-trip tag
    assert res["status"] == "skipped"
    assert res["succeeded"] == [] and res["failed"] == []
    assert ("id:_pipeline/processed", ["a", "b"]) in client.tagged
    assert len(client.tagged) == 1            # nothing but the processed tag
    assert s.get_asset("a").processed is True
    assert s.get_asset("b").processed is True
    s.close()


def test_upsert_tag_failure_is_caught_and_does_not_abort_batch(tmp_path):
    # A failing upsert_tag must be recorded as a per-asset failure (retryable),
    # not propagate out of apply_cluster and abort apply_all (review finding R1).
    s = _store(tmp_path)
    _approved(s, name="Venice")
    _approved(s, name="Rome")           # second cluster must still be applied
    client = FakeImmich()
    client.fail_upsert = "Venice"       # first cluster's trip-tag upsert fails
    results = apply_all(client, s)
    assert len(results) == 2            # batch was not aborted by the first failure
    venice = next(r for r in results if r["cluster_id"] == 1)
    assert venice["succeeded"] == [] and sorted(i for i, _ in venice["failed"]) == ["a", "b"]
    assert s.already_applied("a", "trip", "Venice") is False   # retryable
    rome = next(r for r in results if r["cluster_id"] == 2)
    assert sorted(rome["succeeded"]) == ["a", "b"]
    s.close()


def test_processed_write_failure_is_surfaced_in_result(tmp_path):
    # Trip tag succeeds but the _pipeline/processed write fails: the failure must
    # appear in result["failed"] and the asset must NOT be marked processed.
    s = _store(tmp_path)
    cid = _approved(s, name="Venice")
    client = FakeImmich()
    client.fail_upsert = pipeline.PROCESSED
    res = apply_cluster(client, s, cid)
    assert sorted(res["succeeded"]) == ["a", "b"]              # trip tag still applied
    assert sorted(i for i, _ in res["failed"]) == ["a", "b"]  # processed surfaced
    assert s.get_asset("a").processed is False
    s.close()