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

"""immich — §4.3 create-an-object + read-it-back + characteristic feature (Phase 2 P3).

immich is a photo/video manager; its characteristic behavior is ingesting an asset and generating
derivatives (thumbnails) — exercised here end-to-end against the live deployment (no SSO needed;
immich's first-run creates a local admin):
  1. POST /api/auth/admin-sign-up {email,password,name}  (first-run admin; 201, or already-exists)
  2. POST /api/auth/login → accessToken
  3. POST /api/assets (multipart: assetData PNG + deviceAssetId/deviceId/fileCreatedAt/fileModifiedAt)
     → 201 {id}  — CREATE the object
  4. GET /api/assets/{id} → 200, asset id matches + type IMAGE  — READ IT BACK
  5. GET /api/assets/{id}/thumbnail → 200  — immich GENERATED a derivative (poll; async)

Verified against the deployed immich app version 2.7.5 (GET /api/server/version → {2,7,5}). Real
assertions on app state (the stored asset + its generated thumbnail), not a health/200 stand-in.
"""

from __future__ import annotations

import json
import os
import ssl
import sys
import time
import urllib.request
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 (so immich's decoder + thumbnailer accept it and generate a derivative).
_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) -> tuple[int, dict | None]:
    """Multipart POST /api/assets (immich's upload endpoint). Returns (status, json)."""
    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")
    # Append unique trailing bytes after IEND so each run's file has a UNIQUE checksum — immich dedups
    # by content hash, so without this a re-run (or a probe that already has this exact PNG) returns
    # 200 "duplicate" instead of 201 "created". PNG decoders ignore post-IEND bytes, so it's still a
    # valid 1x1 image the thumbnailer can process.
    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:  # noqa: PERF203
        try:
            return e.code, json.loads(e.read() or b"null")
        except Exception:  # noqa: BLE001
            return e.code, None


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

    # 1) first-run admin (201 created; tolerate 400 if one already exists from a prior tier)
    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}"

    # 2) login → accessToken
    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.get("accessToken")
    assert token and isinstance(token, str), f"login returned no accessToken: {body!r}"
    auth = {"Authorization": f"Bearer {token}"}

    # 3) upload an asset (CREATE the object) — unique deviceAssetId so immich doesn't dedup
    st, up = _upload_asset(base, token, f"ccci-{uuid.uuid4().hex}")
    assert st == 201 and isinstance(up, dict), f"asset upload HTTP {st}: {up!r}"
    asset_id = up.get("id")
    assert asset_id, f"upload returned no asset id: {up!r}"

    # 4) READ IT BACK — GET the asset; assert it's the one we uploaded, an IMAGE
    st, got = harness_http.http_request("GET", f"{base}/api/assets/{asset_id}", headers=auth)
    assert st == 200 and isinstance(got, dict), f"asset read-back HTTP {st}: {got!r}"
    assert got.get("id") == asset_id, f"read-back asset id {got.get('id')!r} != {asset_id!r}"
    assert got.get("type") == "IMAGE", f"read-back asset type {got.get('type')!r} != IMAGE"

    # 5) immich GENERATED a derivative — the thumbnail (poll; generation is async)
    thumb = 0
    for _ in range(12):
        thumb, _ = harness_http.http_request(
            "GET", f"{base}/api/assets/{asset_id}/thumbnail", headers=auth
        )
        if thumb == 200:
            break
        time.sleep(5)
    assert (
        thumb == 200
    ), f"immich did not generate a thumbnail/derivative for the uploaded asset (last HTTP {thumb})"