101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
import os
|
|
from app import create_app
|
|
from app.config import Config
|
|
from photoflow.core import Store
|
|
from photoflow.core.models import Asset, Cluster, ClusterMember
|
|
|
|
|
|
def _seed_store(cfg):
|
|
s = Store(cfg.db_path).connect()
|
|
for i, t in [("a", "2019-06-01"), ("b", "2019-06-02"), ("c", "2019-07-10")]:
|
|
s.upsert_asset(Asset(immich_id=i, taken_at=t))
|
|
cid = s.insert_cluster(
|
|
Cluster(start_at="2019-06-01", end_at="2019-06-02", suggested_name="Venice",
|
|
confidence=0.9, kind_guess="trip", status="pending"),
|
|
[ClusterMember(cluster_id=0, immich_id="a"),
|
|
ClusterMember(cluster_id=0, immich_id="b")])
|
|
other = s.insert_cluster(
|
|
Cluster(start_at="2019-07-10", end_at="2019-07-10", suggested_name="Rome",
|
|
confidence=0.3, kind_guess="trip", status="pending"),
|
|
[ClusterMember(cluster_id=0, immich_id="c")])
|
|
s.close()
|
|
return cid, other
|
|
|
|
|
|
def _app(tmp_path):
|
|
cfg = Config(immich_url="http://x", immich_api_key="k",
|
|
anthropic_api_key="", data_dir=str(tmp_path))
|
|
app = create_app(cfg)
|
|
app.config.update(TESTING=True)
|
|
return app
|
|
|
|
|
|
def test_health(tmp_path):
|
|
assert _app(tmp_path).test_client().get("/health").data == b"ok"
|
|
|
|
|
|
def test_thumb_served(tmp_path):
|
|
thumbs = os.path.join(str(tmp_path), "thumbs")
|
|
os.makedirs(thumbs, exist_ok=True)
|
|
with open(os.path.join(thumbs, "a.jpg"), "wb") as f:
|
|
f.write(b"\xff\xd8\xffjpeg")
|
|
client = _app(tmp_path).test_client()
|
|
r = client.get("/thumb/a")
|
|
assert r.status_code == 200 and r.mimetype == "image/jpeg"
|
|
assert _app(tmp_path).test_client().get("/thumb/missing").status_code == 404
|
|
|
|
|
|
def test_index_lists_clusters(tmp_path):
|
|
app = _app(tmp_path)
|
|
_seed_store(app.config["APP_CONFIG"])
|
|
r = app.test_client().get("/")
|
|
assert r.status_code == 200 and b"Venice" in r.data and b"Rome" in r.data
|
|
|
|
|
|
def test_index_empty_state(tmp_path):
|
|
r = _app(tmp_path).test_client().get("/")
|
|
assert r.status_code == 200 and b"No clusters yet" in r.data
|
|
|
|
|
|
def test_detail_and_approve(tmp_path):
|
|
app = _app(tmp_path)
|
|
cid, _ = _seed_store(app.config["APP_CONFIG"])
|
|
client = app.test_client()
|
|
d = client.get(f"/cluster/{cid}")
|
|
assert d.status_code == 200 and b"/thumb/a" in d.data
|
|
r = client.post(f"/cluster/{cid}/approve", json={"name": "Venezia"})
|
|
assert r.get_json()["status"] == "approved"
|
|
s = Store(app.config["APP_CONFIG"].db_path).connect()
|
|
assert s.get_cluster(cid).decided_name == "Venezia"
|
|
s.close()
|
|
|
|
|
|
def test_member_toggle_and_high_confidence(tmp_path):
|
|
app = _app(tmp_path)
|
|
cid, _ = _seed_store(app.config["APP_CONFIG"])
|
|
client = app.test_client()
|
|
client.post(f"/cluster/{cid}/member", json={"asset_id": "b", "included": False})
|
|
s = Store(app.config["APP_CONFIG"].db_path).connect()
|
|
assert {m.immich_id: m.included for _, m in s.cluster_members(cid)}["b"] is False
|
|
s.close()
|
|
r = client.post("/approve-high-confidence", json={"threshold": 0.75})
|
|
assert r.get_json()["approved"] == 1 # only the 0.9 cluster
|
|
|
|
|
|
def test_apply_all_with_injected_client(tmp_path):
|
|
app = _app(tmp_path)
|
|
cid, _ = _seed_store(app.config["APP_CONFIG"])
|
|
|
|
class FakeImmich:
|
|
def __init__(self): self.tagged = []
|
|
def upsert_tag(self, name): return f"id:{name}"
|
|
def tag_assets(self, tid, ids): self.tagged.append((tid, list(ids)))
|
|
|
|
fake = FakeImmich()
|
|
app.config["IMMICH_FACTORY"] = lambda: fake
|
|
client = app.test_client()
|
|
client.post(f"/cluster/{cid}/approve", json={"name": "Venice"})
|
|
r = client.post("/apply-all", json={})
|
|
assert r.status_code == 200
|
|
assert any(tid == "id:Venice" for tid, _ in fake.tagged)
|