Forge-Engine/tests/test_user_assets.py

""""Mes assets" (voir plan Phase 1, §2) — bibliothèque d'images PAR
COMPTE, distincte du dossier d'un jeu (routes/uploads/upload_file.py) :
réutilisable d'un projet à l'autre du MÊME compte, jamais visible par un
autre."""
import io
import struct
import wave

import auth


def _png_bytes(width, height):
    """Un PNG minimal — seule la signature + l'en-tête IHDR (largeur/
    hauteur) comptent pour auth.image_dimensions, le reste du fichier
    n'a pas besoin d'être une image RÉELLEMENT décodable (voir
    add_scene_object.py, qui ne fait que lire ces dimensions, jamais le
    pixel lui-même)."""
    ihdr = struct.pack(">II", width, height) + b"\x08\x02\x00\x00\x00"
    return b"\x89PNG\r\n\x1a\n" + b"\x00\x00\x00\x0d" + b"IHDR" + ihdr + b"\x00\x00\x00\x00"


def _mp4_box(box_type, payload):
    return struct.pack(">I4s", len(payload) + 8, box_type) + payload


def _mp4_bytes(seconds, timescale=1000):
    """Un MP4 minimal — juste "ftyp" + "moov/mvhd" (durée/timescale),
    suffisant pour que mutagen (auth/validate_video_duration.py) lise sa
    durée, sans piste vidéo/audio réelle ni "mdat" (jamais décodé, seule
    la durée annoncée compte ici)."""
    ftyp = _mp4_box(b"ftyp", b"isom" + struct.pack(">I", 0) + b"isom" + b"mp42")
    duration = int(seconds * timescale)
    mvhd_payload = (
        struct.pack(">I", 0) + struct.pack(">II", 0, 0)
        + struct.pack(">II", timescale, duration) + b"\x00" * 80
    )
    moov = _mp4_box(b"moov", _mp4_box(b"mvhd", mvhd_payload))
    return ftyp + moov


def _wav_bytes(seconds, framerate=8000):
    """Un WAV de silence, durée EXACTE (au sample près) — 8kHz mono 8-bit
    pour rester petit même sur 3+ minutes (voir test de la limite de
    durée d'un son, auth/validate_audio_duration.py)."""
    buf = io.BytesIO()
    with wave.open(buf, "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(1)
        w.setframerate(framerate)
        w.writeframes(b"\x80" * int(seconds * framerate))
    return buf.getvalue()


def test_create_and_list_user_asset(client):
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    asset_id = auth.create_user_asset(user_id, "abc123.png", original_name="photo.png", source="upload")
    assets = auth.list_user_assets(user_id)
    assert any(a["id"] == asset_id and a["filename"] == "abc123.png" for a in assets)


def test_list_user_assets_is_isolated_per_account(client, user_client):
    with client.session_transaction() as sess:
        admin_id = sess["user_id"]
    with user_client.session_transaction() as sess:
        user_id = sess["user_id"]
    auth.create_user_asset(admin_id, "admin_only.png")
    assets = auth.list_user_assets(user_id)
    assert all(a["filename"] != "admin_only.png" for a in assets)


def test_get_user_asset_returns_none_for_unknown_id():
    assert auth.get_user_asset(999999) is None


def test_delete_user_asset_refuses_when_not_the_owner(client, user_client):
    with client.session_transaction() as sess:
        admin_id = sess["user_id"]
    with user_client.session_transaction() as sess:
        user_id = sess["user_id"]
    asset_id = auth.create_user_asset(admin_id, "admin_only.png")
    assert auth.delete_user_asset(asset_id, user_id) is False
    assert auth.get_user_asset(asset_id) is not None


def test_upload_user_asset_route_stores_the_file_and_returns_a_url(client):
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(b"fake-image-bytes"), "photo.png")},
        content_type="multipart/form-data",
    )
    assert resp.status_code == 200
    body = resp.get_json()
    assert body["id"]
    assert "/assets/" in body["url"]

    dl = client.get(body["url"])
    assert dl.status_code == 200
    assert dl.data == b"fake-image-bytes"


def test_upload_user_asset_route_requires_a_file(client):
    resp = client.post("/assets/upload", data={}, content_type="multipart/form-data")
    assert resp.status_code == 400


# ---------- scene_kind ("fond"/"decor" — demande explicite : "un bouton
# d'import séparé pour les images de fond et les objets") ----------

def test_create_user_asset_stores_the_scene_kind(client):
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    asset_id = auth.create_user_asset(user_id, "ciel.png", source="upload", scene_kind="fond")
    assets = auth.list_user_assets(user_id)
    asset = next(a for a in assets if a["id"] == asset_id)
    assert asset["scene_kind"] == "fond"


def test_create_user_asset_scene_kind_defaults_to_none(client):
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    asset_id = auth.create_user_asset(user_id, "ancienne.png", source="upload")
    assets = auth.list_user_assets(user_id)
    asset = next(a for a in assets if a["id"] == asset_id)
    assert asset["scene_kind"] is None


def test_upload_route_stores_the_posted_scene_kind_for_an_image(client):
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(b"fake-image-bytes"), "ciel.png"), "scene_kind": "fond"},
        content_type="multipart/form-data",
    )
    assert resp.status_code == 200
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    assets = auth.list_user_assets(user_id)
    asset = next(a for a in assets if a["id"] == resp.get_json()["id"])
    assert asset["scene_kind"] == "fond"


def test_upload_route_defaults_an_image_without_scene_kind_to_decor(client):
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(b"fake-image-bytes"), "photo.png")},
        content_type="multipart/form-data",
    )
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    assets = auth.list_user_assets(user_id)
    asset = next(a for a in assets if a["id"] == resp.get_json()["id"])
    assert asset["scene_kind"] == "decor"


def test_upload_route_ignores_scene_kind_for_a_non_image_file(client):
    """scene_kind n'a de sens que pour une IMAGE (voir
    auth/user_asset_kind.py) — un son/une vidéo ne "posent" rien sur la
    scène, peu importe ce que le client aurait posté."""
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(_wav_bytes(2)), "bip.wav"), "scene_kind": "fond"},
        content_type="multipart/form-data",
    )
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    assets = auth.list_user_assets(user_id)
    asset = next(a for a in assets if a["id"] == resp.get_json()["id"])
    assert asset["scene_kind"] is None


def test_serve_user_asset_route_403s_for_another_account(client, user_client):
    # Un compte "user" fraîchement inscrit (sans projet) est redirigé vers
    # l'onboarding pour toute route hors _REACHABLE_WITHOUT_PROJECT (voir
    # core/auth_guard.py) — lui assigner un projet ici isole ce test sur
    # le SEUL comportement qu'on veut vérifier : l'isolation par
    # propriétaire de "Mes assets", pas le parcours d'onboarding.
    with user_client.session_transaction() as sess:
        other_user_id = sess["user_id"]
    auth.set_project_slug(other_user_id, "pytest_dummy_project")

    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(b"fake-image-bytes"), "photo.png")},
        content_type="multipart/form-data",
    )
    url = resp.get_json()["url"]
    resp2 = user_client.get(url)
    assert resp2.status_code == 403


# ---------- "Mes assets" — upload d'un son, limité à 3 minutes (demande
# explicite) ----------

def test_user_asset_kind_classifies_by_extension():
    assert auth.user_asset_kind("photo.PNG") == "image"
    assert auth.user_asset_kind("bip.mp3") == "audio"
    assert auth.user_asset_kind("bip.WAV") == "audio"
    assert auth.user_asset_kind("notes.txt") == "other"


def test_validate_audio_duration_accepts_a_short_wav():
    content = _wav_bytes(2)
    assert auth.validate_audio_duration(content) is None


def test_validate_audio_duration_rejects_a_wav_over_the_limit():
    content = _wav_bytes(200)  # > 180s (auth.MAX_AUDIO_SECONDS)
    error = auth.validate_audio_duration(content)
    assert error is not None
    assert "3 minutes" in error


def test_validate_audio_duration_rejects_unrecognized_content():
    assert auth.validate_audio_duration(b"pas-un-vrai-fichier-audio") is not None


def test_upload_route_accepts_a_short_audio_file(client):
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(_wav_bytes(2)), "bip.wav")},
        content_type="multipart/form-data",
    )
    assert resp.status_code == 200
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    assets = auth.list_user_assets(user_id)
    assert any(a["filename"].endswith(".wav") for a in assets)


def test_upload_route_rejects_an_audio_file_over_three_minutes(client):
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(_wav_bytes(200)), "trop-long.wav")},
        content_type="multipart/form-data",
    )
    assert resp.status_code == 400
    assert "3 minutes" in resp.get_json()["error"]
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    assets = auth.list_user_assets(user_id)
    assert not any(a["original_name"] == "trop-long.wav" for a in assets)


# ---------- "Mes assets" — upload d'une vidéo, format MP4 uniquement,
# limité à 5 minutes (demande explicite) ----------

def test_validate_video_duration_accepts_a_short_mp4():
    assert auth.validate_video_duration(_mp4_bytes(30)) is None


def test_validate_video_duration_rejects_an_mp4_over_the_limit():
    error = auth.validate_video_duration(_mp4_bytes(400))
    assert error is not None
    assert "5 minutes" in error


def test_validate_video_duration_rejects_unrecognized_content():
    assert auth.validate_video_duration(b"pas-une-vraie-video") is not None


def test_upload_route_accepts_a_short_mp4_video(client):
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(_mp4_bytes(30)), "clip.mp4")},
        content_type="multipart/form-data",
    )
    assert resp.status_code == 200
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    assets = auth.list_user_assets(user_id)
    assert any(a["filename"].endswith(".mp4") for a in assets)


def test_upload_route_rejects_an_mp4_video_over_five_minutes(client):
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(_mp4_bytes(400)), "trop-long.mp4")},
        content_type="multipart/form-data",
    )
    assert resp.status_code == 400
    assert "5 minutes" in resp.get_json()["error"]
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    assets = auth.list_user_assets(user_id)
    assert not any(a["original_name"] == "trop-long.mp4" for a in assets)


def test_upload_route_rejects_a_non_mp4_video_format(client):
    """Demande explicite : "une vidéo au format MP4" — webm/mov/avi sont
    reconnus comme "vidéo" par auth/user_asset_kind.py (rétrocompatibilité
    de l'organisation "Mes assets"), mais plus ACCEPTÉS à l'upload."""
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(b"peu-importe-le-contenu"), "clip.webm")},
        content_type="multipart/form-data",
    )
    assert resp.status_code == 400
    assert "MP4" in resp.get_json()["error"]


def test_upload_route_rejects_unrecognized_mp4_content(client):
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(b"pas-une-vraie-video"), "faux.mp4")},
        content_type="multipart/form-data",
    )
    assert resp.status_code == 400


# ---------- Suppression d'un asset (demande explicite : "voir, utiliser
# ou supprimer des son, image et vidéo") ----------

def test_delete_user_asset_route_removes_it(client):
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    asset_id = auth.create_user_asset(user_id, "abc123.png", original_name="photo.png", source="upload")
    resp = client.delete(f"/assets/{asset_id}")
    assert resp.status_code == 200
    assert auth.get_user_asset(asset_id) is None


def test_delete_user_asset_route_404s_for_an_unknown_id(client):
    resp = client.delete("/assets/999999")
    assert resp.status_code == 404


def test_delete_user_asset_route_refuses_another_accounts_asset(client, user_client):
    # Un compte "user" fraîchement inscrit (sans projet) est redirigé vers
    # l'onboarding pour toute route hors _REACHABLE_WITHOUT_PROJECT (voir
    # core/auth_guard.py) — lui assigner un projet isole ce test sur le
    # SEUL comportement voulu : l'isolation par propriétaire.
    with client.session_transaction() as sess:
        admin_id = sess["user_id"]
    with user_client.session_transaction() as sess:
        other_user_id = sess["user_id"]
    auth.set_project_slug(other_user_id, "pytest_dummy_project_delete")
    asset_id = auth.create_user_asset(admin_id, "admin_only.png")
    resp = user_client.delete(f"/assets/{asset_id}")
    assert resp.status_code == 404
    assert auth.get_user_asset(asset_id) is not None


# ---------- Reclasser un asset DÉJÀ importé entre "Fonds" et
# "Décors/objets" sans ré-upload (diagnostiqué depuis un "décalage" : une
# image importée par le mauvais bouton, ou avant la séparation fond/décor,
# restait bloquée dans la mauvaise section) ----------

def test_update_user_asset_scene_kind_reclassifies_the_asset(client):
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    asset_id = auth.create_user_asset(user_id, "ciel.png", source="upload", scene_kind="decor")
    assert auth.update_user_asset_scene_kind(asset_id, user_id, "fond") is True
    asset = auth.get_user_asset(asset_id)
    assets = auth.list_user_assets(user_id)
    assert next(a for a in assets if a["id"] == asset_id)["scene_kind"] == "fond"


def test_update_user_asset_scene_kind_refuses_when_not_the_owner(client, user_client):
    with client.session_transaction() as sess:
        admin_id = sess["user_id"]
    with user_client.session_transaction() as sess:
        other_user_id = sess["user_id"]
    asset_id = auth.create_user_asset(admin_id, "ciel.png", source="upload", scene_kind="decor")
    assert auth.update_user_asset_scene_kind(asset_id, other_user_id, "fond") is False


def test_update_user_asset_scene_kind_refuses_a_non_image_asset(client):
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    asset_id = auth.create_user_asset(user_id, "bip.mp3", source="upload")
    assert auth.update_user_asset_scene_kind(asset_id, user_id, "fond") is False


def test_set_scene_kind_route_reclassifies_the_asset(client):
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    asset_id = auth.create_user_asset(user_id, "ciel.png", source="upload", scene_kind="decor")
    resp = client.post(f"/assets/{asset_id}/scene-kind", json={"scene_kind": "fond"})
    assert resp.status_code == 200
    assets = auth.list_user_assets(user_id)
    assert next(a for a in assets if a["id"] == asset_id)["scene_kind"] == "fond"


def test_set_scene_kind_route_rejects_an_invalid_value(client):
    with client.session_transaction() as sess:
        user_id = sess["user_id"]
    asset_id = auth.create_user_asset(user_id, "ciel.png", source="upload", scene_kind="decor")
    resp = client.post(f"/assets/{asset_id}/scene-kind", json={"scene_kind": "autre"})
    assert resp.status_code == 400


def test_set_scene_kind_route_404s_for_another_accounts_asset(client, user_client):
    with client.session_transaction() as sess:
        admin_id = sess["user_id"]
    with user_client.session_transaction() as sess:
        other_user_id = sess["user_id"]
    auth.set_project_slug(other_user_id, "pytest_dummy_project_scene_kind")
    asset_id = auth.create_user_asset(admin_id, "ciel.png", source="upload", scene_kind="decor")
    resp = user_client.post(f"/assets/{asset_id}/scene-kind", json={"scene_kind": "fond"})
    assert resp.status_code == 404


# ---------- Dimensions réelles d'une image importée (bug : "je ne peux
# pas poser un objet hors du champ de la caméra quand j'utilise un fond
# de scène importé" — voir auth/image_dimensions.py) ----------

def test_image_dimensions_reads_a_png_header():
    assert auth.image_dimensions(_png_bytes(1920, 1080)) == (1920, 1080)


def test_image_dimensions_returns_none_for_unrecognized_content():
    assert auth.image_dimensions(b"pas-une-image") is None


def test_upload_route_rejects_unrecognized_audio_content(client):
    resp = client.post(
        "/assets/upload",
        data={"file": (io.BytesIO(b"pas-un-vrai-son"), "faux.mp3")},
        content_type="multipart/form-data",
    )
    assert resp.status_code == 400