refactor: compare archive members in one place

scripts/packverify.py

@@ -23,6 +23,7 @@ import functools
 import hashlib
 from common import filter_systems_by_target
 from hashing import parse_md5_list
+import io
 from common import load_emulator_profiles
 from common import load_platform_config
 import os
@@ -34,7 +35,6 @@ from packresolve import resolve_file
 from common import resolve_platform_cores
 from common import sanitize_pack_path
 import zipfile
-import io
 from collections.abc import Callable
 
 
@@ -100,6 +100,25 @@ def _settle_untracked(
     return "untracked", file_name
 
 
+def _member_crcs(archive: zipfile.ZipFile) -> dict[str, int]:
+    return {i.filename: i.CRC for i in archive.infolist() if not i.is_dir()}
+
+
+def _same_members(pack: zipfile.ZipFile, name: str, source: str) -> bool:
+    """Whether a ZIP inside the pack holds the members of a ZIP on disk.
+
+    A rebuilt archive (TorrentZip) differs byte for byte and is the same
+    set when every member and its CRC agree. The two checks that ask it
+    each carried their own copy.
+    """
+    try:
+        with zipfile.ZipFile(io.BytesIO(pack.read(name))) as packed, \
+                zipfile.ZipFile(source) as held:
+            return _member_crcs(packed) == _member_crcs(held)
+    except (zipfile.BadZipFile, OSError):
+        return False
+
+
 def verify_pack(
     zip_path: str, db: dict, data_registry: dict | None = None
 ) -> tuple[bool, dict]:
@@ -193,27 +212,10 @@ def verify_pack(
                     _src_path = files_db[_src_sha1]["path"]
                     if not os.path.exists(_src_path):
                         continue
-                    try:
-                        import io as _io
-
-                        with zipfile.ZipFile(_src_path) as _sz:
-                            _sc = {
-                                i.filename: i.CRC
-                                for i in _sz.infolist()
-                                if not i.is_dir()
-                            }
-                        with zipfile.ZipFile(_io.BytesIO(zf.read(name))) as _pz:
-                            _pc = {
-                                i.filename: i.CRC
-                                for i in _pz.infolist()
-                                if not i.is_dir()
-                            }
-                        if _sc == _pc:
-                            status = "verified_rebuild"
-                            file_name = _bn
-                            break
-                    except (zipfile.BadZipFile, OSError):
-                        continue
+                    if _same_members(zf, name, _src_path):
+                        status = "verified_rebuild"
+                        file_name = _bn
+                        break
 
             # Data directory: check against cached files
             if status == "untracked" and _data_index:
@@ -239,28 +241,13 @@ def verify_pack(
                         status = "verified_data"
                         file_name = _bn
                         break
-                    if name.endswith(".zip") and _dp.endswith(".zip"):
-                        try:
-                            import io as _io2
-
-                            with zipfile.ZipFile(_io2.BytesIO(zf.read(name))) as _pz2:
-                                _pc2 = {
-                                    i.filename: i.CRC
-                                    for i in _pz2.infolist()
-                                    if not i.is_dir()
-                                }
-                            with zipfile.ZipFile(_dp) as _dz:
-                                _dc = {
-                                    i.filename: i.CRC
-                                    for i in _dz.infolist()
-                                    if not i.is_dir()
-                                }
-                            if _pc2 == _dc:
-                                status = "verified_data"
-                                file_name = _bn
-                                break
-                        except (zipfile.BadZipFile, OSError):
-                            continue
+                    if (
+                        name.endswith(".zip") and _dp.endswith(".zip")
+                        and _same_members(zf, name, _dp)
+                    ):
+                        status = "verified_data"
+                        file_name = _bn
+                        break
 
             status, file_name = _settle_untracked(
                 status, name, zf, by_md5, held_inside, errors, file_name

tests/test_pack_member_identity.py

@@ -82,5 +82,40 @@ class SchemaAcceptsEveryStatus(unittest.TestCase):
         self.assertEqual(sorted(written - allowed), [])
 
 
+class OneComparisonOfArchiveMembers(unittest.TestCase):
+    """verify_pack compared member CRCs twice, each copy with its own local
+    import and names (_sc/_pc, _pc2/_dc)."""
+
+    def test_the_helper_answers_both_ways(self):
+        import io
+        import tempfile
+        import zipfile
+
+        sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
+        from packverify import _same_members  # noqa: PLC0415
+
+        def archive(members: dict) -> bytes:
+            buffer = io.BytesIO()
+            with zipfile.ZipFile(buffer, "w") as zf:
+                for member, data in members.items():
+                    zf.writestr(member, data)
+            return buffer.getvalue()
+
+        with tempfile.TemporaryDirectory() as directory:
+            held = Path(directory) / "held.zip"
+            held.write_bytes(archive({"a.rom": b"A", "b.rom": b"B"}))
+            pack = io.BytesIO()
+            with zipfile.ZipFile(pack, "w") as zf:
+                zf.writestr("same.zip", archive({"b.rom": b"B", "a.rom": b"A"}))
+                zf.writestr("other.zip", archive({"a.rom": b"X"}))
+                zf.writestr("broken.zip", b"not a zip")
+            with zipfile.ZipFile(pack) as zf:
+                self.assertTrue(_same_members(zf, "same.zip", str(held)))
+                self.assertFalse(_same_members(zf, "other.zip", str(held)))
+                self.assertFalse(_same_members(zf, "broken.zip", str(held)))
+        source = (Path(__file__).resolve().parents[1] / "scripts" / "packverify.py").read_text()
+        self.assertNotIn("import io as _io", source)
+
+
 if __name__ == "__main__":
     unittest.main()