diff --git a/backend/alembic/versions/0108_roms_locked_fields.py b/backend/alembic/versions/0108_roms_locked_fields.py new file mode 100644 index 000000000..c4373ca6b --- /dev/null +++ b/backend/alembic/versions/0108_roms_locked_fields.py @@ -0,0 +1,80 @@ +"""Record which rom fields a user supplied by hand + +Uploading artwork stores the file and clears `url_cover`, so a stored cover +path with no url was the only thing separating a hand-supplied cover from a +scraped one. That signal cannot survive: `get_cover` returns no path when the +file is missing, and the scan writes that straight back to `path_cover_s`, so a +single scan run while the resources volume is unavailable erases it. The next +scan then reads the row as having no cover at all, adopts the provider url, and +replaces the user's cover with provider art once storage returns. + +`locked_fields` records the same fact durably, independent of what is on disk. + +The backfill is the load-bearing part: existing uploads are recognisable only by +the old inferred marker, and reading it once here is the last chance to do so. +Without it the first scan after upgrading would replace every uploaded cover in +every library. + +Manuals are deliberately not backfilled. An uploaded manual and a scraped one +share a path and neither clears `url_manual`, so nothing distinguishes them and +any guess would be wrong for half the rows. Manuals stay pinned by the scan, and +uploads made from here on are marked as they happen. + +Revision ID: 0108_roms_locked_fields +Revises: 0107_roms_dedup_cover_index +Create Date: 2026-08-08 00:00:00.000000 + +""" + +import sqlalchemy as sa +from alembic import op + +from utils.database import CustomJSON + +# revision identifiers, used by Alembic. +revision = "0108_roms_locked_fields" +down_revision = "0107_roms_dedup_cover_index" +branch_labels = None +depends_on = None + + +def _roms_table() -> sa.TableClause: + return sa.table( + "roms", + sa.column("path_cover_s", sa.Text), + sa.column("url_cover", sa.Text), + sa.column("locked_fields", CustomJSON()), + ) + + +def upgrade() -> None: + with op.batch_alter_table("roms", schema=None) as batch_op: + batch_op.add_column( + sa.Column("locked_fields", CustomJSON(), nullable=True), + if_not_exists=True, + ) + + roms = _roms_table() + connection = op.get_bind() + + connection.execute(roms.update().values(locked_fields=[])) + + # A stored cover path with an empty url is the pre-migration marker for an + # upload. Values go through the JSON type rather than a literal so each + # dialect serialises its own way. + connection.execute( + roms.update() + .where( + sa.and_( + roms.c.path_cover_s.isnot(None), + roms.c.path_cover_s != "", + sa.or_(roms.c.url_cover.is_(None), roms.c.url_cover == ""), + ) + ) + .values(locked_fields=["url_cover"]) + ) + + +def downgrade() -> None: + with op.batch_alter_table("roms", schema=None) as batch_op: + batch_op.drop_column("locked_fields", if_exists=True) diff --git a/backend/endpoints/roms/__init__.py b/backend/endpoints/roms/__init__.py index e1375e5d5..c7c54f4ed 100644 --- a/backend/endpoints/roms/__init__.py +++ b/backend/endpoints/roms/__init__.py @@ -1845,9 +1845,17 @@ async def update_rom( } ) + # The cover and manual blocks below both take and release locks, so they + # share one running set that is written back once. Deriving each change from + # rom.locked_fields instead would let the later block discard the earlier + # block's change, since both would start from the same pre-update state. + locked_fields = set(rom.locked_fields or []) + if remove_cover: cleaned_data.update(await fs_resource_handler.remove_cover(rom)) + # Dropping the hand-supplied cover hands the slot back to the providers. cleaned_data.update({"url_cover": ""}) + locked_fields.discard("url_cover") else: if artwork is not None and artwork.filename is not None: file_ext = validate_image_upload(artwork, label="Artwork") @@ -1857,6 +1865,8 @@ async def update_rom( path_cover_s, ) = await fs_resource_handler.store_artwork(rom, artwork_content, file_ext) + # Supplying a file is the explicit act that locks the cover; the + # lock outlives the file, so losing it to a scan can't unlock it. cleaned_data.update( { "url_cover": "", @@ -1864,6 +1874,7 @@ async def update_rom( "path_cover_l": path_cover_l, } ) + locked_fields.add("url_cover") else: url_cover = ( form_data.url_cover if "url_cover" in provided_fields else rom.url_cover @@ -1881,6 +1892,12 @@ async def update_rom( "path_cover_l": path_cover_l, } ) + # Naming a different source url is a handover back to the + # providers. Testing that it changed, not merely that one was + # sent, matters because the client posts the stored url on every + # save, so a plain save must not release the lock. + if url_cover and url_cover != rom.url_cover: + locked_fields.discard("url_cover") except ValidationError as e: log.error(f"Invalid cover URL in update_rom: {str(e)}") raise HTTPException(status_code=400, detail=str(e)) from e @@ -1900,10 +1917,17 @@ async def update_rom( "path_manual": path_manual, } ) + # Same handover as the cover. An upload leaves url_manual untouched, so + # unlike the cover this url is often already set on a locked manual, + # which is exactly why only an actual change may release it. + if url_manual and url_manual != rom.url_manual: + locked_fields.discard("url_manual") except ValidationError as e: log.error(f"Invalid manual URL in update_rom: {str(e)}") raise HTTPException(status_code=400, detail=str(e)) from e + cleaned_data["locked_fields"] = sorted(locked_fields) + # Handle RetroAchievements badges when the ID has changed if cleaned_data["ra_id"] and int(cleaned_data["ra_id"]) != rom.ra_id: for ach in cleaned_data.get("ra_metadata", {}).get("achievements", []): diff --git a/backend/endpoints/roms/manual.py b/backend/endpoints/roms/manual.py index 94b9aecbc..34198b823 100644 --- a/backend/endpoints/roms/manual.py +++ b/backend/endpoints/roms/manual.py @@ -120,10 +120,13 @@ def cleanup_partial_file(): detail="There was an error uploading the manual", ) from exc + # An uploaded manual and a scraped one share this path and neither clears + # url_manual, so the lock is the only thing that will tell them apart. db_rom_handler.update_rom( id, { "path_manual": f"{manuals_path}/{rom.id}{ext}", + "locked_fields": rom.locked_fields_with("url_manual"), }, ) @@ -163,7 +166,14 @@ async def redownload_rom_manual( overwrite=True, url_manual=str(rom.url_manual), ) - db_rom_handler.update_rom(id, {"path_manual": path_manual}) + # Asking for the provider's manual back is a handover. + db_rom_handler.update_rom( + id, + { + "path_manual": path_manual, + "locked_fields": rom.locked_fields_without("url_manual"), + }, + ) log.info( f"Re-downloaded manual for {hl(rom.name or 'ROM', color=BLUE)} " f"[{hl(rom.fs_name)}]" @@ -384,6 +394,7 @@ async def delete_rom_manuals( { "path_manual": "", "url_manual": "", + "locked_fields": rom.locked_fields_without("url_manual"), }, ) @@ -400,6 +411,7 @@ async def delete_rom_manuals( { "path_manual": "", "url_manual": "", + "locked_fields": rom.locked_fields_without("url_manual"), }, ) except Exception as exc: diff --git a/backend/handler/filesystem/resources_handler.py b/backend/handler/filesystem/resources_handler.py index 376a576d8..660ea14d1 100644 --- a/backend/handler/filesystem/resources_handler.py +++ b/backend/handler/filesystem/resources_handler.py @@ -473,6 +473,11 @@ async def _store_screenshot(self, rom: Rom, url_screenhot: str, idx: int): log.error(f"Unable to write screenshot for {url_screenhot}: {str(exc)}") return None + def _stored_screenshot_count(self, rom: Rom) -> int: + """How many screenshot files this rom actually has on disk.""" + full_path = self.validate_path(f"{rom.fs_resources_path}/screenshots") + return sum(1 for _ in full_path.glob("*.jpg")) + def screenshots_exist(self, rom: Rom) -> bool: """Check if rom screenshots exist in filesystem @@ -507,16 +512,24 @@ async def get_rom_screenshots( Returns List of paths to screenshots """ - # Return existing screenshots if no URLs provided - # Or if not overwriting and screenshots already exist - if not url_screenshots or (not overwrite and self.screenshots_exist(rom)): + if not url_screenshots: + return rom.path_screenshots or [] + + # Count what is on disk rather than what was recorded: fewer files than + # urls means an earlier run lost some, and the url set alone would never + # signal that, so fall through and fetch them again. + if not overwrite and self._stored_screenshot_count(rom) >= len(url_screenshots): return rom.path_screenshots or [] # Download and store new screenshots path_screenshots: list[str] = [] for idx, url_screenshot in enumerate(url_screenshots): await self._store_screenshot(rom, url_screenshot, idx) - path_screenshots.append(self._get_screenshot_path(rom, str(idx))) + path = self._get_screenshot_path(rom, str(idx)) + # A failed download leaves nothing behind, and recording its path + # anyway points the database at a file that isn't there. + if await self.file_exists(path): + path_screenshots.append(path) return path_screenshots diff --git a/backend/handler/scan_handler.py b/backend/handler/scan_handler.py index 7b6d8e6ea..b6be3c7dd 100644 --- a/backend/handler/scan_handler.py +++ b/backend/handler/scan_handler.py @@ -442,6 +442,7 @@ async def scan_rom( "path_cover_l": rom.path_cover_l, "path_screenshots": rom.path_screenshots, "path_manual": rom.path_manual, + "locked_fields": rom.locked_fields, "igdb_id": rom.igdb_id, "moby_id": rom.moby_id, "ss_id": rom.ss_id, @@ -996,7 +997,10 @@ async def fetch_hasheous_rom(hasheous_rom: HasheousRom) -> HasheousRom: if fields["metadata_field"]: rom_attrs[fields["metadata_field"]] = {} - # Reset artwork fields so stale values are cleared when no source supplies them + # Reset artwork fields so stale values are cleared when no source supplies them. + # The locks go with them: a complete rescan deletes the resource files, so + # keeping one would leave a cover locked to a file that no longer exists and + # block any replacement from being fetched. rom_attrs.update( { "url_cover": "", @@ -1006,6 +1010,7 @@ async def fetch_hasheous_rom(hasheous_rom: HasheousRom) -> HasheousRom: "path_cover_l": "", "path_screenshots": [], "path_manual": "", + "locked_fields": [], } ) @@ -1064,16 +1069,22 @@ async def fetch_hasheous_rom(hasheous_rom: HasheousRom) -> HasheousRom: { "name": existing_name or matched_name or fs_name_no_tags or None, "summary": rom.summary or rom_attrs.get("summary") or None, - # Don't overwrite existing manually uploaded cover image + # A locked cover was supplied by hand and stays put. Anything + # else came from a provider, and the freshly resolved url + # replaces it so a changed source or source priority reaches the + # download step. "url_cover": ( - rom.url_cover - if rom.path_cover_s + "" + if rom.is_field_locked("url_cover") else rom_attrs.get("url_cover") or None ), + # A manual keeps its stored url because an uploaded manual and a + # scraped one share a path and neither clears the url, so there + # is nothing here to tell them apart. "url_manual": rom.url_manual or rom_attrs.get("url_manual") or None, - "url_screenshots": rom.url_screenshots - or rom_attrs.get("url_screenshots") - or [], + # Screenshots have no upload path, so every stored url came from + # a provider and the freshly resolved set wins. + "url_screenshots": rom_attrs.get("url_screenshots") or [], } ) diff --git a/backend/models/rom.py b/backend/models/rom.py index 105ed358a..3cb767736 100644 --- a/backend/models/rom.py +++ b/backend/models/rom.py @@ -482,6 +482,12 @@ class Rom(BaseModel): CustomJSON(), default=[], doc="URLs to screenshots stored in IGDB" ) + locked_fields: Mapped[list[str] | None] = mapped_column( + CustomJSON(), + default=[], + doc="Fields a user supplied by hand, which scans must not overwrite", + ) + revision: Mapped[str | None] = mapped_column(String(length=100)) version: Mapped[str | None] = mapped_column(String(length=100)) regions: Mapped[list[str] | None] = mapped_column(CustomJSON(), default=[]) @@ -651,6 +657,18 @@ def path_video(self) -> str | None: or (self.launchbox_metadata or {}).get("video_path") ) + def is_field_locked(self, field: str) -> bool: + """Whether a user supplied this field by hand, so scans must leave it.""" + return field in (self.locked_fields or []) + + def locked_fields_with(self, field: str) -> list[str]: + """This rom's locks plus ``field``, for handing to an update.""" + return sorted({*(self.locked_fields or []), field}) + + def locked_fields_without(self, field: str) -> list[str]: + """This rom's locks minus ``field``, for handing to an update.""" + return [f for f in (self.locked_fields or []) if f != field] + @property def is_unidentified(self) -> bool: return ( diff --git a/backend/tests/endpoints/roms/test_manual.py b/backend/tests/endpoints/roms/test_manual.py index 31e4888fe..8334ab0fa 100644 --- a/backend/tests/endpoints/roms/test_manual.py +++ b/backend/tests/endpoints/roms/test_manual.py @@ -89,6 +89,9 @@ def test_upload_manual_to_resources_success( assert written.read_bytes() == PDF_BYTES refreshed = db_rom_handler.get_rom(rom.id) assert refreshed.path_manual == f"{rom.fs_resources_path}/manual/{rom.id}.pdf" + # An uploaded manual and a scraped one land at the same path and neither + # clears url_manual, so the lock is the only thing telling them apart. + assert refreshed.locked_fields == ["url_manual"] def test_upload_markdown_manual_to_resources_preserves_extension( @@ -252,7 +255,11 @@ def test_redownload_manual_success( monkeypatch: pytest.MonkeyPatch, ): db_rom_handler.update_rom( - rom.id, {"url_manual": "https://screenscraper.fr/api/manual.pdf"} + rom.id, + { + "url_manual": "https://screenscraper.fr/api/manual.pdf", + "locked_fields": ["url_manual"], + }, ) fake_path = f"{rom.fs_resources_path}/manual/{rom.id}.pdf" monkeypatch.setattr( @@ -269,6 +276,8 @@ def test_redownload_manual_success( assert response.status_code == status.HTTP_200_OK refreshed = db_rom_handler.get_rom(rom.id) assert refreshed.path_manual == fake_path + # Asking for the provider's manual back is a handover. + assert refreshed.locked_fields == [] # ---------- DELETE /api/roms/{id}/manuals (resources) ---------- @@ -305,6 +314,7 @@ def test_delete_manual_success( { "path_manual": f"{rom.fs_resources_path}/manual/{rom.id}.pdf", "url_manual": "https://screenscraper.fr/api/manual.pdf", + "locked_fields": ["url_manual"], }, ) monkeypatch.setattr( @@ -325,6 +335,8 @@ def test_delete_manual_success( refreshed = db_rom_handler.get_rom(rom.id) assert refreshed.path_manual == "" assert refreshed.url_manual == "" + # Deleting the hand-supplied manual hands the slot back to the providers. + assert refreshed.locked_fields == [] # ---------- DELETE /api/roms/{id}/manuals/files/{file_id} ---------- diff --git a/backend/tests/endpoints/roms/test_rom.py b/backend/tests/endpoints/roms/test_rom.py index ff191ef73..16b2c4e14 100644 --- a/backend/tests/endpoints/roms/test_rom.py +++ b/backend/tests/endpoints/roms/test_rom.py @@ -779,6 +779,136 @@ def test_update_rom_artwork_uses_detected_extension( assert file_ext == "png" +@patch.object( + FSResourcesHandler, + "store_artwork", + new_callable=AsyncMock, + return_value=("path/to/big.png", "path/to/small.png"), +) +def test_update_rom_artwork_locks_the_cover( + store_artwork_mock: AsyncMock, + client: TestClient, + access_token: str, + rom: Rom, +): + # Supplying a file is the explicit act that locks the cover. Scans read this + # rather than inferring it from path_cover_s, which they themselves clear + # whenever the file is unreadable. + response = client.put( + f"/api/roms/{rom.id}", + headers={"Authorization": f"Bearer {access_token}"}, + files={"artwork": ("cover.png", _PNG_BYTES, "image/png")}, + ) + assert response.status_code == status.HTTP_200_OK + + assert db_rom_handler.get_rom(rom.id).locked_fields == ["url_cover"] + + +@patch.object( + FSResourcesHandler, + "remove_cover", + new_callable=AsyncMock, + return_value={"path_cover_s": "", "path_cover_l": ""}, +) +def test_remove_cover_releases_the_lock( + remove_cover_mock: AsyncMock, + client: TestClient, + access_token: str, + rom: Rom, +): + # Dropping the hand-supplied cover hands the slot back to the providers. + db_rom_handler.update_rom(rom.id, {"locked_fields": ["url_cover"]}) + + response = client.put( + f"/api/roms/{rom.id}?remove_cover=true", + headers={"Authorization": f"Bearer {access_token}"}, + ) + assert response.status_code == status.HTTP_200_OK + + assert db_rom_handler.get_rom(rom.id).locked_fields == [] + + +@patch.object( + FSResourcesHandler, + "get_cover", + new_callable=AsyncMock, + return_value=("path/to/small.png", "path/to/big.png"), +) +def test_saving_without_changing_urls_keeps_locks( + get_cover_mock: AsyncMock, + client: TestClient, + access_token: str, + rom: Rom, +): + # The client posts the stored urls on every save, so releasing a lock + # whenever a url is present would unlock hand-supplied artwork the first + # time anything else on the rom is edited. + db_rom_handler.update_rom( + rom.id, + { + "url_cover": "", + "url_manual": "https://ss.fr/manual?id=1", + "locked_fields": ["url_cover", "url_manual"], + }, + ) + + response = client.put( + f"/api/roms/{rom.id}", + headers={"Authorization": f"Bearer {access_token}"}, + data={"url_cover": "", "url_manual": "https://ss.fr/manual?id=1"}, + ) + assert response.status_code == status.HTTP_200_OK + + assert db_rom_handler.get_rom(rom.id).locked_fields == [ + "url_cover", + "url_manual", + ] + + +@patch.object( + FSResourcesHandler, + "get_manual", + new_callable=AsyncMock, + return_value="path/to/manual.pdf", +) +@patch.object( + FSResourcesHandler, + "get_cover", + new_callable=AsyncMock, + return_value=("path/to/small.png", "path/to/big.png"), +) +def test_naming_new_source_urls_releases_both_locks( + get_cover_mock: AsyncMock, + get_manual_mock: AsyncMock, + client: TestClient, + access_token: str, + rom: Rom, +): + # Choosing a provider's artwork is a handover. Both fields release in one + # request, which is why the locks are accumulated rather than each derived + # from the pre-update row. + db_rom_handler.update_rom( + rom.id, + { + "url_cover": "", + "url_manual": "https://ss.fr/manual?id=1", + "locked_fields": ["url_cover", "url_manual"], + }, + ) + + response = client.put( + f"/api/roms/{rom.id}", + headers={"Authorization": f"Bearer {access_token}"}, + data={ + "url_cover": "https://ss.fr/cover?id=2", + "url_manual": "https://ss.fr/manual?id=2", + }, + ) + assert response.status_code == status.HTTP_200_OK + + assert db_rom_handler.get_rom(rom.id).locked_fields == [] + + def test_delete_roms(client: TestClient, access_token: str, rom: Rom): response = client.post( "/api/roms/delete", diff --git a/backend/tests/handler/filesystem/test_resources_handler.py b/backend/tests/handler/filesystem/test_resources_handler.py index e63b3c685..4974783d8 100644 --- a/backend/tests/handler/filesystem/test_resources_handler.py +++ b/backend/tests/handler/filesystem/test_resources_handler.py @@ -391,15 +391,25 @@ async def test_get_rom_screenshots_no_urls( @pytest.mark.asyncio async def test_get_rom_screenshots_with_urls( - self, handler: FSResourcesHandler, rom + self, handler: FSResourcesHandler, rom, tmp_path ): """Test get_rom_screenshots with URLs""" + handler.base_path = tmp_path urls = [ "http://example.com/screenshot1.jpg", "http://example.com/screenshot2.jpg", ] - with patch.object(handler, "_store_screenshot") as mock_store: + # Only screenshots that reached the disk get a recorded path, so the + # stand-in has to write them. + async def store(_rom, _url, idx): + directory = tmp_path / f"{rom.fs_resources_path}/screenshots" + directory.mkdir(parents=True, exist_ok=True) + (directory / f"{idx}.jpg").write_bytes(b"jpeg") + + with patch.object( + handler, "_store_screenshot", side_effect=store + ) as mock_store: result = await handler.get_rom_screenshots(rom, True, urls) # Should call _store_screenshot for each URL @@ -1132,6 +1142,80 @@ async def test_cancelled_download_leaves_no_temp_file( assert list(cover_dir.iterdir()) == [] + @pytest.mark.asyncio + async def test_failed_screenshot_is_not_recorded( + self, handler: FSResourcesHandler, rom: Rom, tmp_path + ): + # Recording a path for a screenshot that never landed points the + # database at a missing file, and the gallery at a broken image. + handler.base_path = tmp_path + + async def store_only_the_first(_rom, _url, idx): + if idx != 0: + return None + path = tmp_path / "roms/1/1/screenshots" + path.mkdir(parents=True, exist_ok=True) + (path / "0.jpg").write_bytes(b"jpeg") + + handler._store_screenshot = store_only_the_first # type: ignore[method-assign] + + paths = await handler.get_rom_screenshots( + rom=rom, + overwrite=True, + url_screenshots=["http://x/a.jpg", "http://x/b.jpg"], + ) + + assert paths == ["roms/1/1/screenshots/0.jpg"] + + @pytest.mark.asyncio + async def test_short_screenshot_set_is_retried( + self, handler: FSResourcesHandler, rom: Rom, tmp_path + ): + # The url set is unchanged after a partial failure, so without this the + # missing screenshot would never be fetched again. + handler.base_path = tmp_path + rom.path_screenshots = ["roms/1/1/screenshots/0.jpg"] + screenshots = tmp_path / "roms/1/1/screenshots" + screenshots.mkdir(parents=True) + (screenshots / "0.jpg").write_bytes(b"jpeg") + + attempted: list[int] = [] + + async def record(_rom, _url, idx): + attempted.append(idx) + + handler._store_screenshot = record # type: ignore[method-assign] + + await handler.get_rom_screenshots( + rom=rom, + overwrite=False, + url_screenshots=["http://x/a.jpg", "http://x/b.jpg"], + ) + + assert attempted == [0, 1] + + @pytest.mark.asyncio + async def test_cover_with_no_source_url_is_rederived_from_disk( + self, handler: FSResourcesHandler, rom: Rom, tmp_path + ): + # Second half of what keeps a locked cover alive through an unmatch, + # which clears the stored paths but never deletes the files. A lock is + # not visible here: it resolves to an empty url upstream, and + # test_update_scan_keeps_uploaded_cover covers that half. + handler.base_path = tmp_path + cover = tmp_path / "roms/1/1/cover" + cover.mkdir(parents=True) + (cover / "big.png").write_bytes(b"uploaded") + (cover / "small.png").write_bytes(b"uploaded") + + path_s, path_l = await handler.get_cover( + entity=rom, overwrite=False, url_cover="" + ) + + assert path_s == "roms/1/1/cover/small.png" + assert path_l == "roms/1/1/cover/big.png" + assert (cover / "big.png").read_bytes() == b"uploaded" + @pytest.mark.asyncio async def test_interrupted_download_leaves_no_temp_file( self, handler: FSResourcesHandler, rom: Rom, tmp_path diff --git a/backend/tests/handler/test_fastapi.py b/backend/tests/handler/test_fastapi.py index 9953805a7..bd7bc9393 100644 --- a/backend/tests/handler/test_fastapi.py +++ b/backend/tests/handler/test_fastapi.py @@ -512,6 +512,162 @@ async def test_scan_rom_unmatched_no_match_uses_parsed_name( assert result.name == "Snow Brothers" +def _scraped_cover_rom(platform: Platform, **overrides) -> Rom: + attrs: dict = { + "platform_id": platform.id, + "fs_name": "game.sfc", + "fs_path": "snes", + "tags": [], + "ss_id": 321, + "name": "Game", + "url_cover": "https://ss.fr/media?media=box-2D&id=old", + "path_cover_s": "roms/1/1/cover/small.png", + "path_cover_l": "roms/1/1/cover/big.png", + } + attrs.update(overrides) + return db_rom_handler.add_rom(Rom(**attrs)) + + +async def _update_scan(platform: Platform, rom: Rom) -> Rom: + async with initialize_context(): + return await scan_rom( + platform=platform, + scan_type=ScanType.UPDATE, + rom=rom, + fs_rom=_ss_quota_fs_rom("game.sfc"), + metadata_sources=[MetadataSource.SS], + newly_added=False, + ) + + +@patch.object(meta_playmatch_handler, "is_enabled", return_value=False) +@patch.object(meta_ss_handler, "get_rom_by_id", new_callable=AsyncMock) +async def test_update_scan_replaces_scraped_cover_url( + mock_ss_get_by_id, mock_playmatch_enabled +): + """A cover that carries a source url came from a provider, so an UPDATE scan + hands the freshly resolved url downstream. Pinning it to the stored value is + what kept a changed source priority from ever reaching the download step.""" + mock_ss_get_by_id.return_value = SSRom( + ss_id=321, + name="Game", + url_cover="https://ss.fr/media?media=box-2D&id=new", + ) + + platform = _ss_quota_platform() + rom = _scraped_cover_rom(platform) + + result = await _update_scan(platform, rom) + + assert result.url_cover == "https://ss.fr/media?media=box-2D&id=new" + + +@patch.object(meta_playmatch_handler, "is_enabled", return_value=False) +@patch.object(meta_ss_handler, "get_rom_by_id", new_callable=AsyncMock) +async def test_update_scan_keeps_uploaded_cover( + mock_ss_get_by_id, mock_playmatch_enabled +): + """Uploading artwork locks the cover, so the provider url must not be adopted + over it.""" + mock_ss_get_by_id.return_value = SSRom( + ss_id=321, + name="Game", + url_cover="https://ss.fr/media?media=box-2D&id=new", + ) + + platform = _ss_quota_platform() + rom = _scraped_cover_rom(platform, url_cover="", locked_fields=["url_cover"]) + + result = await _update_scan(platform, rom) + + assert result.url_cover == "" + assert result.locked_fields == ["url_cover"] + + +@patch.object(meta_playmatch_handler, "is_enabled", return_value=False) +@patch.object(meta_ss_handler, "get_rom_by_id", new_callable=AsyncMock) +async def test_update_scan_keeps_locked_cover_with_no_stored_path( + mock_ss_get_by_id, mock_playmatch_enabled +): + """The lock has to outlive path_cover_s. That column tracks the filesystem and + a scan clears it whenever the file is unreadable, so inferring the lock from it + meant one scan against unavailable storage handed the cover to the provider.""" + mock_ss_get_by_id.return_value = SSRom( + ss_id=321, + name="Game", + url_cover="https://ss.fr/media?media=box-2D&id=new", + ) + + platform = _ss_quota_platform() + rom = _scraped_cover_rom( + platform, + url_cover="", + path_cover_s="", + path_cover_l="", + locked_fields=["url_cover"], + ) + + result = await _update_scan(platform, rom) + + assert result.url_cover == "" + + +@patch.object(meta_playmatch_handler, "is_enabled", return_value=False) +@patch.object(meta_ss_handler, "get_rom_by_id", new_callable=AsyncMock) +async def test_update_scan_replaces_screenshot_urls( + mock_ss_get_by_id, mock_playmatch_enabled +): + """Screenshots have no upload path, so a stored set is always provider-written + and the fresh set wins.""" + mock_ss_get_by_id.return_value = SSRom( + ss_id=321, + name="Game", + url_screenshots=["https://ss.fr/ss?id=new"], + ) + + platform = _ss_quota_platform() + rom = _scraped_cover_rom( + platform, + url_screenshots=["https://ss.fr/ss?id=old"], + path_screenshots=["roms/1/1/screenshots/0.png"], + ) + + result = await _update_scan(platform, rom) + + assert result.url_screenshots == ["https://ss.fr/ss?id=new"] + + +@patch.object(meta_playmatch_handler, "is_enabled", return_value=False) +@patch.object(meta_ss_handler, "get_rom_by_id", new_callable=AsyncMock) +async def test_update_scan_keeps_name_summary_and_manual( + mock_ss_get_by_id, mock_playmatch_enabled +): + """Text fields and manuals stay pinned. Neither can yet tell a hand-edited + value from a provider-written one, so freeing the artwork urls must not free + these too.""" + mock_ss_get_by_id.return_value = SSRom( + ss_id=321, + name="Provider Name", + summary="Provider summary", + url_manual="https://ss.fr/manual?id=new", + ) + + platform = _ss_quota_platform() + rom = _scraped_cover_rom( + platform, + name="My Title", + summary="My summary", + url_manual="https://ss.fr/manual?id=old", + path_manual="roms/1/1/manual/1.pdf", + ) + + result = await _update_scan(platform, rom) + + assert result.name == "My Title" + assert result.summary == "My summary" + assert result.url_manual == "https://ss.fr/manual?id=old" + + @patch.object(meta_playmatch_handler, "is_enabled", return_value=False) @patch.object(meta_hasheous_handler, "get_ra_game", new_callable=AsyncMock) @patch.object(meta_hasheous_handler, "get_igdb_game", new_callable=AsyncMock)