feat(core): asset & tag data-access

shared/tests/test_store_assets.py

@@ -22,3 +22,60 @@ def test_meta_roundtrip_and_default(tmp_path):
     s.set_meta("last_ingest_at", "newer")  # upsert
     assert s.get_meta("last_ingest_at") == "newer"
     s.close()
+
+
+from photoflow.core.models import Asset
+
+
+def _store(tmp_path):
+    return Store(str(tmp_path / "t.db")).connect()
+
+
+def test_upsert_asset_roundtrip_and_has_gps(tmp_path):
+    s = _store(tmp_path)
+    s.upsert_asset(Asset(immich_id="a", taken_at="2019-06-01T10:00:00",
+                         gps_lat=45.4, gps_lon=12.3, place_city="Venezia", type="IMAGE"))
+    got = s.get_asset("a")
+    assert got.place_city == "Venezia" and got.has_gps is True
+    s.upsert_asset(Asset(immich_id="b", taken_at="2019-06-02T10:00:00"))
+    assert s.get_asset("b").has_gps is False
+    s.close()
+
+
+def test_upsert_preserves_processed(tmp_path):
+    s = _store(tmp_path)
+    s.upsert_asset(Asset(immich_id="a", taken_at="t"))
+    s.mark_processed("a")
+    s.upsert_asset(Asset(immich_id="a", taken_at="t2"))  # re-ingest
+    got = s.get_asset("a")
+    assert got.processed is True and got.taken_at == "t2"
+    s.close()
+
+
+def test_all_assets_ordered_and_range(tmp_path):
+    s = _store(tmp_path)
+    for i, t in [("c", "2019-06-03"), ("a", "2019-06-01"), ("b", "2019-06-02")]:
+        s.upsert_asset(Asset(immich_id=i, taken_at=t))
+    assert [a.immich_id for a in s.all_assets()] == ["a", "b", "c"]
+    rng = s.assets_in_range("2019-06-02", "2019-06-03")
+    assert [a.immich_id for a in rng] == ["b", "c"]
+    s.close()
+
+
+def test_asset_tags_replace(tmp_path):
+    s = _store(tmp_path)
+    s.upsert_asset(Asset(immich_id="a", taken_at="t"))
+    s.set_asset_tags("a", ["Italy 2019", "Kiev"])
+    assert sorted(s.asset_tags("a")) == ["Italy 2019", "Kiev"]
+    s.set_asset_tags("a", ["Italy 2019"])  # replace
+    assert s.asset_tags("a") == ["Italy 2019"]
+    s.close()
+
+
+def test_tags_inventory(tmp_path):
+    s = _store(tmp_path)
+    s.upsert_tag("Italy 2019", immich_tag_id="t1", count=42)
+    s.upsert_tag("Italy 2019", immich_tag_id="t1", count=43)  # upsert
+    names = {t.name: t for t in s.all_tags()}
+    assert names["Italy 2019"].count == 43 and names["Italy 2019"].immich_tag_id == "t1"
+    s.close()