fix(patcher): read the patcher's report from its last JSON line; test the patch endpoint (#5120)

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()

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', '')}")

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"

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(

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"],