146 lines
6.3 KiB
Python
146 lines
6.3 KiB
Python
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
|