cc-ci/tests/immich/functional/test_asset_processing.py

"""immich — 2nd recipe-specific functional test (Phase 2 P3): asset PROCESSING pipeline.

Distinct from `test_asset_upload.py` (which proves storage round-trip + thumbnail generation):
this exercises immich's **metadata-extraction microservice** and **library cataloging** — after an
upload, immich's background jobs read the image, populate `exifInfo` (image dimensions / file size),
and count the asset into the owner's library statistics. Characteristic immich behavior (it is a media
*management* server, not just a blob store), asserted on real stored app state — not a 200/health
stand-in.

  1. admin-sign-up / login (first-run local admin; no SSO needed)
  2. POST /api/assets (upload a uniquely-keyed valid PNG) → 201 {id}
  3. poll GET /api/assets/{id} until immich's metadata-extraction job populated exifInfo with the
     image dimensions (exifImageWidth/Height) — proves the extraction pipeline ran
  4. GET /api/assets/statistics → the asset is counted in the owner's library (images/total >= 1)

Verified against immich app version 2.7.5.
"""

from __future__ import annotations

import os
import ssl
import sys
import time
import uuid

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
from harness import http as harness_http  # noqa: E402

# A valid 1x1 PNG (immich's decoder reads its dimensions → exifImageWidth/Height = 1).
_PNG = bytes.fromhex(
    "89504e470d0a1a0a0000000d4948445200000001000000010806000000"
    "1f15c4890000000a49444154789c6300010000050001"
    "0d0a2db40000000049454e44ae426082"
)

_ADMIN = {"email": "ci@ccci.local", "password": "ccciPass12345", "name": "ci"}


def _ctx():
    c = ssl.create_default_context()
    c.check_hostname = False
    c.verify_mode = ssl.CERT_NONE
    return c


def _upload_asset(base: str, token: str, device_asset_id: str):
    import json
    import urllib.error
    import urllib.request

    boundary = "----ccci" + uuid.uuid4().hex
    parts: list[bytes] = []

    def field(name: str, value: str) -> None:
        parts.append(
            f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n{value}\r\n'.encode()
        )

    field("deviceAssetId", device_asset_id)
    field("deviceId", "ccci")
    field("fileCreatedAt", "2026-05-29T00:00:00.000Z")
    field("fileModifiedAt", "2026-05-29T00:00:00.000Z")
    # unique trailing bytes after IEND → unique content checksum so immich does not dedup (returns 201)
    png = _PNG + device_asset_id.encode()
    parts.append(
        f'--{boundary}\r\nContent-Disposition: form-data; name="assetData"; filename="ccci.png"\r\n'
        f"Content-Type: image/png\r\n\r\n".encode()
        + png
        + b"\r\n"
    )
    parts.append(f"--{boundary}--\r\n".encode())
    body = b"".join(parts)
    req = urllib.request.Request(f"{base}/api/assets", data=body, method="POST")
    req.add_header("Content-Type", f"multipart/form-data; boundary={boundary}")
    req.add_header("Authorization", f"Bearer {token}")
    try:
        with urllib.request.urlopen(req, timeout=60, context=_ctx()) as r:
            return r.status, json.loads(r.read() or b"null")
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read() or b"null")
        except Exception:  # noqa: BLE001
            return e.code, None


def test_immich_processes_uploaded_asset_metadata_and_statistics(live_app):
    base = f"https://{live_app}"

    # admin + login
    st, _ = harness_http.http_post(f"{base}/api/auth/admin-sign-up", data=_ADMIN)
    assert st in (201, 400, 409), f"admin-sign-up unexpected HTTP {st}"
    st, body = harness_http.http_post(
        f"{base}/api/auth/login", data={"email": _ADMIN["email"], "password": _ADMIN["password"]}
    )
    assert st in (200, 201) and isinstance(body, dict), f"login HTTP {st}: {body!r}"
    token = body["accessToken"]
    auth = {"Authorization": f"Bearer {token}"}

    # upload a uniquely-keyed asset
    st, up = _upload_asset(base, token, f"ccci-meta-{uuid.uuid4().hex}")
    assert st == 201 and isinstance(up, dict), f"asset upload HTTP {st}: {up!r}"
    asset_id = up["id"]

    # immich's metadata-extraction job populates exifInfo (image dimensions) — poll (async job)
    exif = None
    for _ in range(24):
        gst, got = harness_http.http_request("GET", f"{base}/api/assets/{asset_id}", headers=auth)
        assert gst == 200 and isinstance(got, dict), f"asset read HTTP {gst}: {got!r}"
        exif = got.get("exifInfo")
        if exif and exif.get("exifImageWidth"):
            break
        time.sleep(5)
    assert (
        exif and exif.get("exifImageWidth") == 1 and exif.get("exifImageHeight") == 1
    ), f"immich metadata-extraction did not populate the 1x1 PNG dimensions in exifInfo: {exif!r}"

    # the asset is catalogued into the owner's library statistics (list-back in aggregate)
    sst, stats = harness_http.http_request("GET", f"{base}/api/assets/statistics", headers=auth)
    assert sst == 200 and isinstance(stats, dict), f"statistics HTTP {sst}: {stats!r}"
    assert (
        stats.get("images", 0) >= 1 and stats.get("total", 0) >= 1
    ), f"uploaded asset not reflected in library statistics: {stats!r}"