diff --git a/backend/tests/endpoints/roms/test_patch.py b/backend/tests/endpoints/roms/test_patch.py index f91e9c3b8..60b070088 100644 --- a/backend/tests/endpoints/roms/test_patch.py +++ b/backend/tests/endpoints/roms/test_patch.py @@ -1,15 +1,29 @@ +import io +import tempfile +import zipfile +from datetime import timedelta from pathlib import Path +from typing import Any +import httpx2 import pytest from anyio import Path as AnyioPath from fastapi import status from fastapi.testclient import TestClient +from tests.rom_patcher_stubs import APPEND_PATCH, install_fake_node +from config import OAUTH_ACCESS_TOKEN_EXPIRE_SECONDS from endpoints.roms import patch as patch_endpoint +from handler.auth.base_handler import oauth_handler from handler.database import db_rom_handler +from handler.database.base_handler import sync_session from handler.filesystem import fs_rom_handler +from models.permission import HiddenEntity, PermEntity from models.rom import Rom, RomFile, RomFileCategory +from models.user import User +from utils.nginx import content_disposition from utils.rom_patcher import PatcherInputError +from utils.zip_cache import ensure_zipfile_writable def _auth(token: str) -> dict[str, str]: @@ -92,3 +106,487 @@ def test_patch_rom_returns_bad_request_for_invalid_archive( assert response.json() == { "detail": "Select which file inside the ROM archive to patch" } + + +@pytest.fixture +def library(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + root = tmp_path / "library" + root.mkdir() + monkeypatch.setattr(fs_rom_handler, "base_path", root) + return root + + +class Scratch: + """The temp dirs the endpoint makes, so the tests can see them go.""" + + def __init__(self) -> None: + self.made: list[Path] = [] + + def all_removed(self) -> bool: + return not any(path.exists() for path in self.made) + + +@pytest.fixture +def scratch(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Scratch: + root = tmp_path / "scratch" + root.mkdir() + made = Scratch() + mkdtemp = tempfile.mkdtemp + + def recording_mkdtemp(prefix: str | None = None) -> str: + path = mkdtemp(prefix=prefix, dir=root) + made.made.append(Path(path)) + return path + + monkeypatch.setattr(tempfile, "mkdtemp", recording_mkdtemp) + return made + + +@pytest.fixture +def node(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + install_fake_node(tmp_path, monkeypatch, APPEND_PATCH) + + +def _file( + library: Path, + rom: Rom, + name: str, + content: bytes, + category: RomFileCategory | None = None, + **overrides: Any, +) -> RomFile: + path = library / rom.fs_path / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + return db_rom_handler.add_rom_file( + RomFile( + rom_id=rom.id, + file_name=name, + file_path=rom.fs_path, + file_size_bytes=len(content), + category=category, + **overrides, + ) + ) + + +def _hide(rom: Rom, user: User) -> None: + with sync_session.begin() as session: + session.add( + HiddenEntity(entity=PermEntity.ROMS, entity_id=rom.id, user_id=user.id) + ) + + +def _user_auth(user: User) -> dict[str, str]: + token = oauth_handler.create_access_token( + data={ + "sub": user.username, + "iss": "romm:oauth", + "scopes": " ".join(user.oauth_scopes), + }, + expires_delta=timedelta(seconds=OAUTH_ACCESS_TOKEN_EXPIRE_SECONDS), + ) + return _auth(token) + + +class TestPatchingALibraryFile: + def test_returns_the_patched_rom( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + scratch: Scratch, + node: None, + ): + game = _file(library, rom, "game.sfc", b"rom") + patch = _file(library, rom, "translation.bps", b"+patch", RomFileCategory.PATCH) + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(access_token), + data={"patch_file_id": str(patch.id)}, + ) + + assert response.status_code == status.HTTP_200_OK + assert response.content == b"rom+patch" + assert response.headers["x-patch-validated"] == "true" + assert response.headers["content-disposition"] == content_disposition( + "game (patched-translation).sfc" + ) + assert len(scratch.made) == 1 + assert scratch.all_removed() + + def test_names_the_output_as_asked_keeping_the_roms_extension( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + scratch: Scratch, + node: None, + ): + game = _file(library, rom, "game.sfc", b"rom") + patch = _file(library, rom, "translation.bps", b"+patch", RomFileCategory.PATCH) + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(access_token), + data={"patch_file_id": str(patch.id), "output_file_name": "My Hack.bin"}, + ) + + assert response.headers["content-disposition"] == content_disposition( + "My Hack.sfc" + ) + + def test_patches_a_member_of_a_zipped_rom( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + scratch: Scratch, + node: None, + ): + archive = io.BytesIO() + ensure_zipfile_writable() + with zipfile.ZipFile(archive, "w") as zipped: + zipped.writestr("game.sfc", b"rom") + zipped.writestr("readme.txt", b"hi") + game = _file(library, rom, "game.zip", archive.getvalue()) + patch = _file(library, rom, "translation.bps", b"+patch", RomFileCategory.PATCH) + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(access_token), + data={"patch_file_id": str(patch.id), "archive_member_name": "game.sfc"}, + ) + + assert response.status_code == status.HTTP_200_OK + with zipfile.ZipFile(io.BytesIO(response.content)) as zipped: + assert zipped.read("game.sfc") == b"rom+patch" + assert zipped.read("readme.txt") == b"hi" + + +class TestPatchingWithAnUpload: + def test_returns_the_patched_rom( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + scratch: Scratch, + node: None, + ): + game = _file(library, rom, "game.sfc", b"rom") + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(access_token), + files={"patch_file": ("fix.ips", b"+upload")}, + ) + + assert response.status_code == status.HTTP_200_OK + assert response.content == b"rom+upload" + assert response.headers["content-disposition"] == content_disposition( + "game (patched-fix).sfc" + ) + assert len(scratch.made) == 1 + assert scratch.all_removed() + + @pytest.mark.parametrize( + ("name", "content", "detail"), + [ + ("fix.txt", b"+upload", "Unsupported patch format '.txt'"), + ("fix", b"+upload", "Unsupported patch format ''"), + ("fix.ips", b"", "Uploaded patch file is empty"), + ("fix.ips", b"x" * 65, "Patch file is too large to patch (max 64 bytes)"), + ], + ids=["unsupported", "no_extension", "empty", "too_large"], + ) + def test_a_bad_upload_is_rejected( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + scratch: Scratch, + node: None, + monkeypatch: pytest.MonkeyPatch, + name: str, + content: bytes, + detail: str, + ): + monkeypatch.setattr(patch_endpoint, "ROM_PATCHER_MAX_FILE_SIZE_BYTES", 64) + game = _file(library, rom, "game.sfc", b"rom") + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(access_token), + files={"patch_file": (name, content)}, + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"].startswith(detail) + assert len(scratch.made) == 1 + assert scratch.all_removed() + + +class TestRejections: + @pytest.mark.parametrize("both", [False, True], ids=["neither", "both"]) + def test_needs_exactly_one_patch_source( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + both: bool, + ): + game = _file(library, rom, "game.sfc", b"rom") + patch = _file(library, rom, "translation.bps", b"+patch", RomFileCategory.PATCH) + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(access_token), + data={"patch_file_id": str(patch.id)} if both else {}, + files={"patch_file": ("fix.ips", b"+upload")} if both else None, + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"] == ( + "Provide exactly one of a library patch file or an uploaded patch file" + ) + + def test_an_unknown_rom_file_is_not_found( + self, client: TestClient, access_token: str + ): + response = client.post( + "/api/roms/999999/patch", + headers=_auth(access_token), + files={"patch_file": ("fix.ips", b"+upload")}, + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.json()["detail"] == "ROM file with id 999999 not found" + + def test_a_rom_hidden_from_the_caller_is_not_found( + self, client: TestClient, viewer_user: User, rom: Rom, library: Path + ): + game = _file(library, rom, "game.sfc", b"rom") + _hide(rom, viewer_user) + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_user_auth(viewer_user), + files={"patch_file": ("fix.ips", b"+upload")}, + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.json()["detail"] == f"ROM file with id {game.id} not found" + + def test_a_rom_missing_from_disk_is_not_found( + self, client: TestClient, access_token: str, rom: Rom, library: Path + ): + game = _file(library, rom, "game.sfc", b"rom", missing_from_fs=True) + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(access_token), + files={"patch_file": ("fix.ips", b"+upload")}, + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.json()["detail"] == ( + "ROM file 'game.sfc' is missing from filesystem" + ) + + def test_a_rom_over_the_size_limit_is_rejected( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + monkeypatch: pytest.MonkeyPatch, + ): + monkeypatch.setattr(patch_endpoint, "ROM_PATCHER_MAX_FILE_SIZE_BYTES", 2) + game = _file(library, rom, "game.sfc", b"rom") + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(access_token), + files={"patch_file": ("fix.ips", b"+")}, + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"].startswith("ROM file is too large to patch") + + def test_needs_the_roms_read_scope( + self, client: TestClient, admin_user: User, rom: Rom, library: Path + ): + game = _file(library, rom, "game.sfc", b"rom") + token = oauth_handler.create_access_token( + data={"sub": admin_user.username, "iss": "romm:oauth", "scopes": ""}, + expires_delta=timedelta(seconds=OAUTH_ACCESS_TOKEN_EXPIRE_SECONDS), + ) + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(token), + files={"patch_file": ("fix.ips", b"+upload")}, + ) + + assert response.status_code == status.HTTP_403_FORBIDDEN + + +class TestLibraryPatchRejections: + def _post( + self, client: TestClient, headers: dict[str, str], game: RomFile, patch_id: int + ) -> httpx2.Response: + return client.post( + f"/api/roms/{game.id}/patch", + headers=headers, + data={"patch_file_id": str(patch_id)}, + ) + + def test_an_unknown_patch_is_not_found( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + scratch: Scratch, + ): + game = _file(library, rom, "game.sfc", b"rom") + + response = self._post(client, _auth(access_token), game, 999_999) + + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.json()["detail"] == "Patch file with id 999999 not found" + assert len(scratch.made) == 1 + assert scratch.all_removed() + + def test_a_patch_on_a_hidden_rom_is_not_found( + self, + client: TestClient, + viewer_user: User, + rom: Rom, + second_rom: Rom, + library: Path, + ): + game = _file(library, rom, "game.sfc", b"rom") + patch = _file( + library, second_rom, "translation.bps", b"+patch", RomFileCategory.PATCH + ) + _hide(second_rom, viewer_user) + + response = self._post(client, _user_auth(viewer_user), game, patch.id) + + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.json()["detail"] == f"Patch file with id {patch.id} not found" + + def test_a_patch_missing_from_disk_is_not_found( + self, client: TestClient, access_token: str, rom: Rom, library: Path + ): + game = _file(library, rom, "game.sfc", b"rom") + patch = _file( + library, + rom, + "translation.bps", + b"+patch", + RomFileCategory.PATCH, + missing_from_fs=True, + ) + + response = self._post(client, _auth(access_token), game, patch.id) + + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.json()["detail"] == ( + "Patch file 'translation.bps' is missing from filesystem" + ) + + def test_a_file_that_is_not_a_patch_is_rejected( + self, client: TestClient, access_token: str, rom: Rom, library: Path + ): + game = _file(library, rom, "game.sfc", b"rom") + manual = _file(library, rom, "manual.pdf", b"%PDF") + + response = self._post(client, _auth(access_token), game, manual.id) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"].startswith("Unsupported patch format '.pdf'") + + def test_a_patch_over_the_size_limit_is_rejected( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + monkeypatch: pytest.MonkeyPatch, + ): + monkeypatch.setattr(patch_endpoint, "ROM_PATCHER_MAX_FILE_SIZE_BYTES", 4) + game = _file(library, rom, "game.sfc", b"rom") + patch = _file(library, rom, "translation.bps", b"+patch", RomFileCategory.PATCH) + + response = self._post(client, _auth(access_token), game, patch.id) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"].startswith("Patch file is too large to patch") + + +class TestPatcherFailures: + def test_a_patcher_error_is_a_500_without_its_details( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + scratch: Scratch, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ): + install_fake_node( + tmp_path, + monkeypatch, + """ + sys.stderr.write(json.dumps({"error": "ROM file not found: /srv/secret"})) + sys.exit(2) + """, + ) + game = _file(library, rom, "game.sfc", b"rom") + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(access_token), + files={"patch_file": ("fix.ips", b"+upload")}, + ) + + assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR + assert response.json() == {"detail": "Patching failed"} + assert len(scratch.made) == 1 + assert scratch.all_removed() + + def test_an_unexpected_error_is_a_500( + self, + client: TestClient, + access_token: str, + rom: Rom, + library: Path, + scratch: Scratch, + monkeypatch: pytest.MonkeyPatch, + ): + async def explode(*_args: Any) -> bool: + raise RuntimeError("disk on fire") + + monkeypatch.setattr(patch_endpoint, "apply_patch", explode) + game = _file(library, rom, "game.sfc", b"rom") + + response = client.post( + f"/api/roms/{game.id}/patch", + headers=_auth(access_token), + files={"patch_file": ("fix.ips", b"+upload")}, + ) + + assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR + assert response.json() == {"detail": "Patching failed"} + assert len(scratch.made) == 1 + assert scratch.all_removed() diff --git a/backend/tests/rom_patcher_stubs.py b/backend/tests/rom_patcher_stubs.py new file mode 100644 index 000000000..79def6c92 --- /dev/null +++ b/backend/tests/rom_patcher_stubs.py @@ -0,0 +1,35 @@ +"""A stand-in `node` for the ROM patcher tests, as CI has no RomPatcher.js.""" + +import os +import sys +import textwrap +from pathlib import Path + +import pytest + +# Mimics patcher.js: appends the patch to the ROM and reports success on stdout. +APPEND_PATCH = """ +rom, patch, output = sys.argv[2:5] +data = open(rom, "rb").read() + open(patch, "rb").read() +open(output, "wb").write(data) +print(json.dumps({"success": True, "size": len(data), "validated": True})) +""" + + +def install_fake_node( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, body: str +) -> None: + """Put a `node` first on PATH that runs `body` as Python. + + Args: + body: Python run with `json`, `os`, `sys` and `time` imported; argv is + the script path, then the ROM, patch and output paths. + """ + bin_dir = tmp_path / "fake-node-bin" + bin_dir.mkdir(exist_ok=True) + node = bin_dir / "node" + node.write_text( + f"#!{sys.executable}\nimport json, os, sys, time\n{textwrap.dedent(body)}" + ) + node.chmod(0o755) + monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ.get('PATH', '')}") diff --git a/backend/tests/utils/test_rom_patcher.py b/backend/tests/utils/test_rom_patcher.py index eb21ab4af..6533f7a50 100644 --- a/backend/tests/utils/test_rom_patcher.py +++ b/backend/tests/utils/test_rom_patcher.py @@ -1,3 +1,5 @@ +import json +import os import zipfile from collections.abc import Awaitable, Callable, Iterator from pathlib import Path @@ -5,9 +7,12 @@ from unittest.mock import MagicMock import pytest from anyio import Path as AnyioPath +from tests.rom_patcher_stubs import APPEND_PATCH, install_fake_node from utils.rom_patcher import ( + PATCHER_SCRIPT, SUPPORTED_PATCH_EXTENSIONS, + PatcherError, PatcherInputError, apply_patch, ) @@ -349,3 +354,237 @@ async def test_apply_patch_rejects_unsupported_archive_format(tmp_path: Path): tmp_path / "patch.bps", tmp_path / "patched.7z", ) + + +@pytest.mark.asyncio +async def test_apply_patch_rejects_a_member_name_for_a_raw_rom(tmp_path: Path): + with pytest.raises(PatcherInputError, match="requires a ZIP"): + await apply_patch( + tmp_path / "game.sfc", + tmp_path / "patch.bps", + tmp_path / "patched.sfc", + archive_member_name="game.sfc", + ) + + +class TestPatcherProcess: + """The Node subprocess, run as a stand-in `node` on PATH.""" + + @pytest.fixture + def files(self, tmp_path: Path) -> tuple[Path, Path, Path]: + rom = tmp_path / "game.sfc" + patch = tmp_path / "translation.bps" + rom.write_bytes(b"rom") + patch.write_bytes(b"+patch") + return rom, patch, tmp_path / "out" / "patched.sfc" + + async def test_runs_the_patcher_script_on_the_three_paths( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + files: tuple[Path, Path, Path], + ): + rom, patch, output = files + output.parent.mkdir() + install_fake_node( + tmp_path, + monkeypatch, + f"open({str(tmp_path / 'argv.json')!r}, 'w').write(json.dumps(sys.argv[1:]))\n" + + APPEND_PATCH, + ) + + assert await apply_patch(rom, patch, output) is True + + assert output.read_bytes() == b"rom+patch" + argv = json.loads((tmp_path / "argv.json").read_text()) + assert argv == [str(PATCHER_SCRIPT), str(rom), str(patch), str(output)] + + @pytest.mark.parametrize( + "stdout", + [ + '{"success": true, "validated": false}', + # The PMSR format logs "a" and "b" while applying. + 'a\nb\n{"success": true, "validated": false}', + ], + ids=["json_only", "after_log_lines"], + ) + async def test_reports_a_source_checksum_mismatch( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + files: tuple[Path, Path, Path], + stdout: str, + ): + rom, patch, output = files + output.parent.mkdir() + install_fake_node( + tmp_path, + monkeypatch, + f""" + open(sys.argv[4], "wb").write(b"patched") + print({stdout!r}) + """, + ) + + assert await apply_patch(rom, patch, output) is False + + @pytest.mark.parametrize( + "stdout", + ["", "patched", "null", '{"success": true}'], + ids=["silent", "not_json", "not_an_object", "no_validated_key"], + ) + async def test_an_unreported_validation_counts_as_validated( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + files: tuple[Path, Path, Path], + stdout: str, + ): + rom, patch, output = files + output.parent.mkdir() + install_fake_node( + tmp_path, + monkeypatch, + f""" + open(sys.argv[4], "wb").write(b"patched") + print({stdout!r}, end="") + """, + ) + + assert await apply_patch(rom, patch, output) is True + + @pytest.mark.parametrize( + ("stderr", "message"), + [ + ('{"success": false, "error": "Invalid patch"}', "Invalid patch"), + ( + '(node:1) Warning: something\n{"success": false, "error": "Invalid patch"}', + "Invalid patch", + ), + ("Segmentation fault\n", "Segmentation fault"), + ('["not", "an", "object"]', '["not", "an", "object"]'), + ("", "Patching failed"), + ], + ids=["json_error", "after_a_warning", "plain_text", "json_list", "silent"], + ) + async def test_a_failed_run_raises_its_error( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + files: tuple[Path, Path, Path], + stderr: str, + message: str, + ): + rom, patch, output = files + install_fake_node( + tmp_path, + monkeypatch, + f""" + sys.stderr.write({stderr!r}) + sys.exit(2) + """, + ) + + with pytest.raises(PatcherError) as exc: + await apply_patch(rom, patch, output) + + assert str(exc.value) == message + assert not isinstance(exc.value, PatcherInputError) + + async def test_a_run_without_output_is_an_error( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + files: tuple[Path, Path, Path], + ): + rom, patch, output = files + install_fake_node(tmp_path, monkeypatch, 'print(json.dumps({"success": True}))') + + with pytest.raises(PatcherError, match="did not produce an output file"): + await apply_patch(rom, patch, output) + + async def test_a_hung_run_is_killed_at_the_timeout( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + files: tuple[Path, Path, Path], + ): + rom, patch, output = files + pid_file = tmp_path / "pid" + install_fake_node( + tmp_path, + monkeypatch, + f""" + open({str(pid_file)!r}, "w").write(str(os.getpid())) + time.sleep(30) + """, + ) + monkeypatch.setattr(rom_patcher, "ROM_PATCHER_TIMEOUT", 1) + + with pytest.raises(PatcherError, match="timed out after 1s"): + await apply_patch(rom, patch, output) + + with pytest.raises(ProcessLookupError): + os.kill(int(pid_file.read_text()), 0) + + async def test_a_zip_member_is_patched_through_the_process( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + source = tmp_path / "game.zip" + patch = tmp_path / "translation.bps" + output = tmp_path / "patched.zip" + _write_zip(source, {"game.sfc": b"rom", "readme.txt": b"hi"}) + patch.write_bytes(b"+patch") + install_fake_node(tmp_path, monkeypatch, APPEND_PATCH) + + assert await apply_patch(source, patch, output, "game.sfc") is True + + with zipfile.ZipFile(output) as archive: + assert archive.read("game.sfc") == b"rom+patch" + assert archive.read("readme.txt") == b"hi" + assert sorted([p.name async for p in AnyioPath(tmp_path).iterdir()]) == [ + "fake-node-bin", + "game.zip", + "patched.zip", + "translation.bps", + ] + + +@pytest.mark.asyncio +async def test_apply_patch_rejects_an_empty_zip(tmp_path: Path): + source = tmp_path / "game.zip" + _write_zip(source, {"saves/": b""}) + + with pytest.raises(PatcherInputError, match="contains no files"): + await apply_patch(source, tmp_path / "patch.bps", tmp_path / "patched.zip") + + +@pytest.mark.asyncio +async def test_apply_patch_rejects_a_zip_too_large_in_total( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + source = tmp_path / "game.zip" + _write_zip(source, {"game.sfc": b"rom", "extra.bin": b"x" * 8}) + monkeypatch.setattr(rom_patcher, "ROM_PATCHER_MAX_FILE_SIZE_BYTES", 8) + + with pytest.raises(PatcherInputError, match="archive is too large"): + await apply_patch( + source, tmp_path / "patch.bps", tmp_path / "patched.zip", "game.sfc" + ) + + +@pytest.mark.asyncio +async def test_apply_patch_keeps_the_archives_folders( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + source = tmp_path / "game.zip" + _write_zip(source, {"docs/": b"", "game.sfc": b"source"}) + monkeypatch.setattr( + rom_patcher, "_apply_binary_patch", _fake_patcher(b"source", b"patched") + ) + + await apply_patch(source, tmp_path / "patch.bps", tmp_path / "patched.zip") + + with zipfile.ZipFile(tmp_path / "patched.zip") as archive: + assert archive.namelist() == ["docs/", "game.sfc"] + assert archive.read("game.sfc") == b"patched" diff --git a/backend/utils/rom_patcher/patcher.py b/backend/utils/rom_patcher/patcher.py index 3dff6e1f0..ee35b3c25 100644 --- a/backend/utils/rom_patcher/patcher.py +++ b/backend/utils/rom_patcher/patcher.py @@ -12,6 +12,7 @@ import zipfile import zlib from collections.abc import Iterator from pathlib import Path +from typing import Any from anyio import Path as AnyioPath @@ -231,24 +232,28 @@ async def _apply_binary_patch( raise PatcherError(f"Patching timed out after {ROM_PATCHER_TIMEOUT}s") from e if proc.returncode != 0: - message = "Patching failed" - try: - err_data = json.loads(stderr.decode()) - message = err_data.get("error", message) - except json.JSONDecodeError, UnicodeDecodeError: - if stderr: - message = stderr.decode(errors="replace").strip() - raise PatcherError(message) + error = _last_json_object(stderr).get("error") + if isinstance(error, str) and error: + raise PatcherError(error) + raise PatcherError(stderr.decode(errors="replace").strip() or "Patching failed") if not await AnyioPath(output_path).exists(): raise PatcherError("Patcher did not produce an output file") # The script reports source-checksum validation in its JSON stdout. - try: - result = json.loads(stdout.decode()) - return bool(result.get("validated", True)) - except json.JSONDecodeError, UnicodeDecodeError: - return True + return bool(_last_json_object(stdout).get("validated", True)) + + +def _last_json_object(output: bytes) -> dict[str, Any]: + """Return the last JSON-object line or ``{}``, ignoring RomPatcher.js logs.""" + for line in reversed(output.decode(errors="replace").splitlines()): + try: + value = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(value, dict): + return value + return {} async def apply_patch( diff --git a/renovate.json b/renovate.json index 3e53df2fa..e171bded4 100644 --- a/renovate.json +++ b/renovate.json @@ -52,6 +52,12 @@ "matchManagers": ["npm", "pep621"], "rangeStrategy": "update-lockfile" }, + { + "description": "rom-patcher is pinned to a git tag, and its lockfile entry carries the version from upstream's package.json, stuck at 3.0.0. Compare the tag itself, or Renovate keeps proposing the tag already pinned.", + "matchManagers": ["npm"], + "matchPackageNames": ["rom-patcher"], + "rangeStrategy": "replace" + }, { "description": "One weekly PR per ecosystem for minor and patch updates, so app dependencies don't open a PR each.", "matchManagers": ["npm"],