64 lines
2.9 KiB
Python
64 lines
2.9 KiB
Python
from werkzeug.wrappers import Response
|
|
from photoflow.immich import ImmichClient
|
|
|
|
|
|
def test_resolve_tag_id_matches_name_or_value(httpserver):
|
|
httpserver.expect_request("/api/tags").respond_with_json([
|
|
{"id": "t1", "name": "Italy 2019", "value": "Italy 2019"},
|
|
{"id": "t2", "name": "processed", "value": "_pipeline/processed"},
|
|
])
|
|
c = ImmichClient(httpserver.url_for(""), "k")
|
|
assert c.resolve_tag_id("Italy 2019") == "t1"
|
|
assert c.resolve_tag_id("_pipeline/processed") == "t2"
|
|
assert c.resolve_tag_id("nope") is None
|
|
|
|
|
|
def test_search_assets_normalizes_and_paginates(httpserver):
|
|
def handler(request):
|
|
page = request.json.get("page", 1)
|
|
assert request.json.get("withExif") is True
|
|
if page == 1:
|
|
return Response(
|
|
'{"assets": {"items": [{"id": "a", "originalFileName": "a.jpg",'
|
|
' "localDateTime": "2019-06-01T10:00:00.000Z", "type": "IMAGE",'
|
|
' "updatedAt": "2026-01-01T00:00:00Z",'
|
|
' "exifInfo": {"latitude": 45.4, "longitude": 12.3, "city": "Venezia",'
|
|
' "country": "Italy", "rating": 4},'
|
|
' "tags": [{"name": "Italy 2019"}]}], "nextPage": 2}}',
|
|
content_type="application/json")
|
|
return Response(
|
|
'{"assets": {"items": [{"id": "b", "originalFileName": "b.jpg",'
|
|
' "localDateTime": "2019-06-02T11:00:00.000Z", "type": "IMAGE",'
|
|
' "updatedAt": "2026-01-02T00:00:00Z", "exifInfo": {}, "tags": []}],'
|
|
' "nextPage": null}}', content_type="application/json")
|
|
|
|
httpserver.expect_request("/api/search/metadata", method="POST").respond_with_handler(handler)
|
|
c = ImmichClient(httpserver.url_for(""), "k")
|
|
assets = c.search_assets(taken_after="2019-01-01", taken_before="2020-01-01")
|
|
assert [a["id"] for a in assets] == ["a", "b"]
|
|
a = assets[0]
|
|
assert a["gps_lat"] == 45.4 and a["place_city"] == "Venezia"
|
|
assert a["tags"] == ["Italy 2019"] and a["rating"] == 4
|
|
assert a["taken_at"] == "2019-06-01T10:00:00.000Z"
|
|
assert assets[1]["gps_lat"] is None and assets[1]["tags"] == []
|
|
|
|
|
|
def test_download_thumbnail(httpserver):
|
|
httpserver.expect_request("/api/assets/a/thumbnail").respond_with_data(
|
|
b"\xff\xd8\xffjpegbytes", content_type="image/jpeg")
|
|
c = ImmichClient(httpserver.url_for(""), "k")
|
|
assert c.download_thumbnail("a").startswith(b"\xff\xd8\xff")
|
|
|
|
|
|
def test_upsert_tag_returns_id_by_value(httpserver):
|
|
httpserver.expect_request("/api/tags", method="PUT").respond_with_json(
|
|
[{"id": "p1", "name": "non-trip", "value": "_pipeline/non-trip"}])
|
|
c = ImmichClient(httpserver.url_for(""), "k")
|
|
assert c.upsert_tag("_pipeline/non-trip") == "p1"
|
|
|
|
|
|
def test_tag_assets_posts_ids(httpserver):
|
|
httpserver.expect_request("/api/tags/p1/assets", method="PUT").respond_with_json({"ok": True})
|
|
c = ImmichClient(httpserver.url_for(""), "k")
|
|
c.tag_assets("p1", ["a", "b"]) # should not raise
|