m038
ae093881fa
- write-back: call upsert_tag inside _apply_tag's try so a tag-create failure is caught per-asset and apply_all no longer aborts mid-batch (was P1) - write-back: surface _pipeline/processed write failures in the result instead of discarding them (was reported as success) - ui: add title to the high confidence badge so approve-high-confidence's pre-action count is non-zero; confirm() before single-cluster apply - core: Store context manager; close DB connection even if a route raises; guard split_cluster against a first-member boundary (empty cluster); add writeback_log lookup index - ingest: split thumbnail download/write error handling and clean up the .tmp file on a write failure - tests: upsert/processed write-failure regression tests + split-guard test
111 lines
4.7 KiB
Python
111 lines
4.7 KiB
Python
from photoflow.core import Store
|
|
from photoflow.core.models import Asset, Cluster, ClusterMember
|
|
|
|
|
|
def _seed(tmp_path):
|
|
s = Store(str(tmp_path / "t.db")).connect()
|
|
for i, t in [("a", "2019-06-01"), ("b", "2019-06-02"), ("c", "2019-06-03"),
|
|
("d", "2019-07-01"), ("e", "2019-07-02")]:
|
|
s.upsert_asset(Asset(immich_id=i, taken_at=t))
|
|
return s
|
|
|
|
|
|
def _members(ids):
|
|
return [ClusterMember(cluster_id=0, immich_id=i) for i in ids]
|
|
|
|
|
|
def test_insert_and_members_ordered(tmp_path):
|
|
s = _seed(tmp_path)
|
|
cid = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03",
|
|
count=3, suggested_name="Trip 1", confidence=0.9),
|
|
_members(["c", "a", "b"]))
|
|
pairs = s.cluster_members(cid)
|
|
assert [a.immich_id for a, m in pairs] == ["a", "b", "c"]
|
|
assert s.get_cluster(cid).suggested_name == "Trip 1"
|
|
s.close()
|
|
|
|
|
|
def test_attention_sort(tmp_path):
|
|
s = _seed(tmp_path)
|
|
s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-10",
|
|
confidence=0.9, status="pending"), _members(["a"]))
|
|
low = s.insert_cluster(Cluster(start_at="2019-07-01", end_at="2019-07-02",
|
|
confidence=0.2, status="pending"), _members(["d"]))
|
|
s.insert_cluster(Cluster(start_at="2019-06-02", end_at="2019-06-03",
|
|
confidence=0.1, status="approved"), _members(["b"]))
|
|
order = [c.id for c in s.clusters_by_attention()]
|
|
assert order[0] == low # lowest-confidence pending first
|
|
assert order[-1] != low # approved sinks to the bottom
|
|
s.close()
|
|
|
|
|
|
def test_chronological_neighbors(tmp_path):
|
|
s = _seed(tmp_path)
|
|
c1 = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03"), _members(["a"]))
|
|
c2 = s.insert_cluster(Cluster(start_at="2019-07-01", end_at="2019-07-02"), _members(["d"]))
|
|
assert s.chronological_neighbors(c1) == (None, c2)
|
|
assert s.chronological_neighbors(c2) == (c1, None)
|
|
s.close()
|
|
|
|
|
|
def test_update_and_member_inclusion(tmp_path):
|
|
s = _seed(tmp_path)
|
|
cid = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03"),
|
|
_members(["a", "b"]))
|
|
s.update_cluster(cid, status="approved", decided_name="Venice", reviewed_at="now")
|
|
c = s.get_cluster(cid)
|
|
assert c.status == "approved" and c.decided_name == "Venice"
|
|
s.set_member_inclusion(cid, "b", False)
|
|
inc = {m.immich_id: m.included for _, m in s.cluster_members(cid)}
|
|
assert inc == {"a": True, "b": False}
|
|
s.close()
|
|
|
|
|
|
def test_split_cluster(tmp_path):
|
|
s = _seed(tmp_path)
|
|
cid = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03",
|
|
suggested_name="Trip"), _members(["a", "b", "c"]))
|
|
id1, id2 = s.split_cluster(cid, "c") # boundary "c" begins the second cluster
|
|
assert s.get_cluster(cid).status == "split"
|
|
left = [a.immich_id for a, _ in s.cluster_members(id1)]
|
|
right = [a.immich_id for a, _ in s.cluster_members(id2)]
|
|
assert left == ["a", "b"] and right == ["c"]
|
|
assert s.get_cluster(id1).status == "pending"
|
|
s.close()
|
|
|
|
|
|
def test_merge_clusters(tmp_path):
|
|
s = _seed(tmp_path)
|
|
a = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03",
|
|
suggested_name="A"), _members(["a", "b"]))
|
|
b = s.insert_cluster(Cluster(start_at="2019-07-01", end_at="2019-07-02",
|
|
suggested_name="B"), _members(["d", "e"]))
|
|
new = s.merge_clusters(a, b)
|
|
assert s.get_cluster(a).status == "merged" and s.get_cluster(b).status == "merged"
|
|
ids = [x.immich_id for x, _ in s.cluster_members(new)]
|
|
assert ids == ["a", "b", "d", "e"]
|
|
c = s.get_cluster(new)
|
|
assert c.start_at == "2019-06-01" and c.end_at == "2019-07-02" and c.status == "pending"
|
|
s.close()
|
|
|
|
|
|
def test_writeback_log_idempotency(tmp_path):
|
|
s = _seed(tmp_path)
|
|
assert s.already_applied("a", "trip", "Venice") is False
|
|
s.log_writeback("a", "trip", "Venice", "ok")
|
|
assert s.already_applied("a", "trip", "Venice") is True
|
|
s.log_writeback("b", "trip", "Venice", "error:boom")
|
|
assert s.already_applied("b", "trip", "Venice") is False # only ok counts
|
|
s.close()
|
|
|
|
|
|
def test_split_cluster_rejects_first_member_boundary(tmp_path):
|
|
import pytest
|
|
s = _seed(tmp_path)
|
|
cid = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03",
|
|
suggested_name="Trip"), _members(["a", "b", "c"]))
|
|
with pytest.raises(ValueError):
|
|
s.split_cluster(cid, "a") # boundary == first member -> empty left
|
|
assert s.get_cluster(cid).status != "split" # original untouched on rejection
|
|
s.close()
|