feat: M1 foundation packages + trip-cluster app #1
@@ -0,0 +1,60 @@
|
|||||||
|
import io
|
||||||
|
import os
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from PIL import Image
|
||||||
|
from werkzeug.serving import make_server
|
||||||
|
|
||||||
|
from app import create_app
|
||||||
|
from app.config import Config
|
||||||
|
from photoflow.core import Store
|
||||||
|
from photoflow.core.models import Asset, Cluster, ClusterMember
|
||||||
|
|
||||||
|
|
||||||
|
def _jpeg(color):
|
||||||
|
buf = io.BytesIO()
|
||||||
|
Image.new("RGB", (48, 48), color).save(buf, format="JPEG")
|
||||||
|
return buf.getvalue()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="session")
|
||||||
|
def data_dir(tmp_path_factory):
|
||||||
|
d = str(tmp_path_factory.mktemp("data"))
|
||||||
|
thumbs = os.path.join(d, "thumbs")
|
||||||
|
os.makedirs(thumbs, exist_ok=True)
|
||||||
|
s = Store(os.path.join(d, "trip-cluster.db")).connect()
|
||||||
|
palette = {"a": (10, 20, 30), "b": (40, 80, 120), "c": (200, 50, 90)}
|
||||||
|
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))
|
||||||
|
with open(os.path.join(thumbs, f"{i}.jpg"), "wb") as f:
|
||||||
|
f.write(_jpeg(palette[i]))
|
||||||
|
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")])
|
||||||
|
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 d
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="session")
|
||||||
|
def flask_app(data_dir):
|
||||||
|
cfg = Config(immich_url="http://127.0.0.1:1", immich_api_key="k",
|
||||||
|
anthropic_api_key="", data_dir=data_dir)
|
||||||
|
return create_app(cfg)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="session")
|
||||||
|
def base_url(flask_app):
|
||||||
|
server = make_server("127.0.0.1", 8095, flask_app)
|
||||||
|
t = threading.Thread(target=server.serve_forever, daemon=True)
|
||||||
|
t.start()
|
||||||
|
time.sleep(0.2)
|
||||||
|
yield "http://127.0.0.1:8095"
|
||||||
|
server.shutdown()
|
||||||
@@ -0,0 +1,45 @@
|
|||||||
|
def _open(page, base_url):
|
||||||
|
page.goto(base_url)
|
||||||
|
page.wait_for_selector(".photo-card")
|
||||||
|
|
||||||
|
|
||||||
|
def test_grid_arrow_navigation_moves_focus_ring(base_url, page):
|
||||||
|
# Venice (2 photos) is high-confidence; click it to load its detail.
|
||||||
|
_open(page, base_url)
|
||||||
|
page.click("text=Venice")
|
||||||
|
page.wait_for_selector(".photo-card")
|
||||||
|
cards = page.query_selector_all(".photo-card")
|
||||||
|
assert len(cards) == 2
|
||||||
|
page.keyboard.press("ArrowRight")
|
||||||
|
# second card gains the focus ring class
|
||||||
|
assert "ring-4" in (page.query_selector_all(".photo-card")[1].get_attribute("class"))
|
||||||
|
|
||||||
|
|
||||||
|
def test_lightbox_opens_and_closes(base_url, page):
|
||||||
|
_open(page, base_url)
|
||||||
|
page.click("text=Venice")
|
||||||
|
page.wait_for_selector(".photo-card")
|
||||||
|
page.keyboard.press("ArrowRight")
|
||||||
|
page.keyboard.press("Enter")
|
||||||
|
assert page.is_visible("#lb")
|
||||||
|
page.keyboard.press("Escape")
|
||||||
|
page.wait_for_selector("#lb", state="hidden")
|
||||||
|
|
||||||
|
|
||||||
|
def test_cluster_switching_with_brackets(base_url, page):
|
||||||
|
_open(page, base_url)
|
||||||
|
page.click("text=Venice")
|
||||||
|
page.wait_for_selector(".photo-card")
|
||||||
|
page.keyboard.press("]") # move to the next cluster
|
||||||
|
page.wait_for_function(
|
||||||
|
"document.querySelectorAll('.photo-card').length === 1") # Rome has 1 photo
|
||||||
|
|
||||||
|
|
||||||
|
def test_approve_updates_status(base_url, page):
|
||||||
|
_open(page, base_url)
|
||||||
|
page.click("text=Venice")
|
||||||
|
page.wait_for_selector("#cluster-name")
|
||||||
|
page.once("dialog", lambda d: d.accept()) # no dialog expected, but be safe
|
||||||
|
page.click("text=Approve (A)")
|
||||||
|
page.wait_for_selector("text=approved")
|
||||||
|
assert "approved" in page.inner_text("body")
|
||||||
@@ -0,0 +1,9 @@
|
|||||||
|
def test_health(base_url, page):
|
||||||
|
page.goto(f"{base_url}/health")
|
||||||
|
assert "ok" in page.content()
|
||||||
|
|
||||||
|
|
||||||
|
def test_index_lists_clusters(base_url, page):
|
||||||
|
page.goto(base_url)
|
||||||
|
assert "Venice" in page.inner_text("body")
|
||||||
|
assert "Rome" in page.inner_text("body")
|
||||||
Reference in New Issue
Block a user