114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
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._ids = {}
|
|
|
|
def upsert_tag(self, name):
|
|
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()
|