82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
from photoflow.core import Store
|
|
|
|
|
|
def test_connect_creates_schema_and_version(tmp_path):
|
|
db = str(tmp_path / "t.db")
|
|
s = Store(db).connect()
|
|
assert s.get_meta("schema_version") == "1"
|
|
# tables exist
|
|
names = {r["name"] for r in s.conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'")}
|
|
assert {"assets", "tags", "asset_tags", "clusters",
|
|
"cluster_members", "writeback_log", "meta"} <= names
|
|
s.close()
|
|
|
|
|
|
def test_meta_roundtrip_and_default(tmp_path):
|
|
s = Store(str(tmp_path / "t.db")).connect()
|
|
assert s.get_meta("missing") is None
|
|
assert s.get_meta("missing", "x") == "x"
|
|
s.set_meta("last_ingest_at", "2026-06-27T00:00:00Z")
|
|
assert s.get_meta("last_ingest_at") == "2026-06-27T00:00:00Z"
|
|
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()
|