feat: add local image enhancement and lossless mod assets

backend/tests/test_image_enhance.py

@@ -0,0 +1,145 @@
+import asyncio
+import base64
+import copy
+import io
+import json
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+from PIL import Image
+from starlette.requests import Request
+
+from app import mods
+from app.blobs import blob_path
+from app.converter.assets import WEBP_MAX, lossless_image
+from app.routers import mods as routes
+from app.translation import file_json
+from tests.test_mods import pg_manifest  # noqa: F401
+
+
+def image_bytes(size=(40, 20), format="PNG"):
+    out = io.BytesIO()
+    Image.new("RGBA", size, (40, 80, 160, 90)).save(out, format)
+    return out.getvalue()
+
+
+def test_lossless_preserves_bytes_and_alpha(tmp_path):
+    for format, suffix in [("PNG", ".png"), ("WEBP", ".webp")]:
+        data = image_bytes(format=format)
+        source = tmp_path / ("source" + suffix)
+        source.write_bytes(data)
+        result, ext = lossless_image(source, tmp_path / "output")
+        assert ext == suffix and result.read_bytes() == data
+        with Image.open(result) as image:
+            assert image.convert("RGBA").getpixel((0, 0))[3] == 90
+
+
+def test_large_images_stay_png_and_animation_is_rejected(tmp_path):
+    source = tmp_path / "large.png"
+    source.write_bytes(image_bytes((WEBP_MAX + 1, 1)))
+    out, ext = lossless_image(source, tmp_path / "output")
+    assert ext == ".png"
+    with Image.open(out) as image:
+        assert image.size == (WEBP_MAX + 1, 1)
+    source = tmp_path / "animated.webp"
+    Image.new("RGBA", (4, 4), "red").save(source, "WEBP", save_all=True,
+        append_images=[Image.new("RGBA", (4, 4), "blue")], duration=100)
+    with pytest.raises(ValueError, match="PNG"):
+        lossless_image(source, tmp_path / "animation")
+    source.write_bytes(b"not an image")
+    with pytest.raises(OSError):
+        lossless_image(source, tmp_path / "broken")
+
+
+def test_asset_scale_validation(pg_manifest):
+    doc = mods.prepare(mods.document(pg_manifest))
+    path = next(p for p in pg_manifest["files"] if p.endswith(".webp"))
+    doc["game"]["assetScale"] = {path: 2}
+    mods.validate(doc, pg_manifest)
+    for bad in [0, -1, True, "2", float("nan"), float("inf"), 1025]:
+        doc["game"]["assetScale"] = {path: bad}
+        with pytest.raises(ValueError, match="assetScale"):
+            mods.validate(doc, pg_manifest)
+    for bad in [{"missing.png": 2}, {"game.json": 2}, [2]]:
+        doc["game"]["assetScale"] = bad
+        with pytest.raises(ValueError, match="assetScale"):
+            mods.validate(doc, pg_manifest)
+
+
+def request(body):
+    async def receive():
+        return {"type": "http.request", "body": body, "more_body": False}
+    return Request({"type": "http", "method": "POST", "headers": []}, receive)
+
+
+@pytest.fixture
+def editor(pg_manifest, monkeypatch):
+    v = SimpleNamespace(status="ready", manifest=copy.deepcopy(pg_manifest), kind="mod", visibility="draft", owner_id=1, revision_id="r1")
+    user = SimpleNamespace(id=1, permissions=["games.play", "versions.fork"])
+    async def version(*_):
+        return v
+    monkeypatch.setattr(routes, "get_version", version)
+    monkeypatch.setattr(routes, "lock_version", version)
+    class DB:
+        commits = 0
+        async def commit(self):
+            self.commits += 1
+            v.revision_id = f"r{self.commits + 1}"
+    return v, user, DB()
+
+
+def test_single_lossless_upload_stores_scale_in_same_commit(editor):
+    v, user, db = editor
+    path = next(p for p in v.manifest["files"] if p.endswith(".webp"))
+    data = image_bytes()
+    reply = asyncio.run(routes.replace_asset(1, path, "r1", "result.png", request(data), user, db, True, 2))
+    assert reply["revision"] == "r2" and db.commits == 1
+    assert blob_path(reply["files"][path]["hash"]).read_bytes() == data
+    assert file_json(v.manifest, "game.json")["assetScale"][path] == 2
+    # Ordinary manual replacement resets the previous scale.
+    asyncio.run(routes.replace_asset(1, path, "r2", "manual.png", request(data), user, db))
+    assert path not in file_json(v.manifest, "game.json")["assetScale"]
+
+
+def test_batch_is_atomic_and_rejects_stale_duplicate_and_non_owner(editor):
+    v, user, db = editor
+    paths = [p for p in v.manifest["files"] if p.endswith(".webp")][:2]
+    assert len(paths) == 2
+    def asset(p):
+        return {"path": p, "filename": "result.png", "scale": 2, "data": base64.b64encode(image_bytes()).decode()}
+    before = copy.deepcopy(v.manifest)
+    bad = [asset(paths[0]), {**asset(paths[1]), "data": "bad base64!"}]
+    with pytest.raises(HTTPException) as error:
+        asyncio.run(routes.replace_assets(1, request(json.dumps({"revision": "r1", "assets": bad}).encode()), user, db))
+    assert error.value.status_code == 422 and v.manifest == before and db.commits == 0
+    assets = [asset(p) for p in paths]
+    reply = asyncio.run(routes.replace_assets(1, request(json.dumps({"revision": "r1", "assets": assets}).encode()), user, db))
+    assert db.commits == 1 and reply["revision"] == "r2"
+    assert file_json(v.manifest, "game.json")["assetScale"] == dict.fromkeys(paths, 2)
+    for revision, items, actor, code in [("r1", assets, user, 409), ("r2", [assets[0], assets[0]], user, 422),
+        ("r2", assets, SimpleNamespace(id=2, permissions=[]), 403)]:
+        with pytest.raises(HTTPException) as error:
+            asyncio.run(routes.replace_assets(1, request(json.dumps({"revision": revision, "assets": items}).encode()), actor, db))
+        assert error.value.status_code == code
+    assert db.commits == 1
+
+
+def test_restore_asset_uses_frozen_base_with_its_scale(editor):
+    v, user, db = editor
+    v.base_revision_id = "base"
+    v.game_id = 17
+    base_manifest = copy.deepcopy(v.manifest)
+    path = next(p for p in v.manifest["files"] if p.endswith(".webp"))
+    routes.set_asset_scale(base_manifest, {path: 2})
+    base = SimpleNamespace(game_id=17, manifest=base_manifest)
+    async def get(_type, _id):
+        assert _id == "base"
+        return base
+    db.get = get
+    asyncio.run(routes.replace_asset(1, path, "r1", "result.png", request(image_bytes()), user, db, True, 4))
+    assert file_json(v.manifest, "game.json")["assetScale"][path] == 4
+    reply = asyncio.run(routes.restore_asset(1, routes.RestoreAssetIn(revision="r2", path=path), user, db))
+    assert reply["files"][path] == base_manifest["files"][path]
+    assert file_json(v.manifest, "game.json")["assetScale"][path] == 2
+    assert db.commits == 2