fix: read each file's digests once per run

This commit is contained in:
Abdessamad Derraz committed 2026-10-18 16:08:16 +02:00
1 parent 8001b9d86d
commit 4e7abebfa1
2 files changed
+72 -1

No files matched your search

+27 -1
View File
@@ -6,6 +6,7 @@ frontends check against."""
from __future__ import annotations
import hashlib
import os
import zipfile
import zlib
from pathlib import Path
@@ -13,12 +14,37 @@ from pathlib import Path
_ALL_ALGORITHMS = frozenset({"sha1", "md5", "sha256", "crc32", "adler32"})
# (path, inode, size, mtime_ns, ctime_ns) -> the digests read so far.
# verify --all hashed GB_dmg.bin 13 times, once per platform and entry; the
# bytes are still read from disk, once per file and change.
_hash_cache: dict[tuple, dict[str, str]] = {}
def compute_hashes(
filepath: str | Path,
algorithms: frozenset[str] | None = None,
) -> dict[str, str]:
"""Compute file hashes. Pass *algorithms* to limit which are computed."""
algos = algorithms or _ALL_ALGORITHMS
algos = frozenset(algorithms) if algorithms else _ALL_ALGORITHMS
try:
stat = os.stat(filepath)
key: tuple | None = (
os.path.realpath(filepath), stat.st_ino, stat.st_size,
stat.st_mtime_ns, stat.st_ctime_ns,
)
except OSError:
key = None
known = _hash_cache.get(key) if key else None
if known is not None and algos <= known.keys():
return {name: known[name] for name in algos}
result = _compute_hashes(filepath, algos)
if key is not None:
_hash_cache.setdefault(key, {}).update(result)
return result
def _compute_hashes(filepath: str | Path, algos: frozenset[str]) -> dict[str, str]:
"""Read the file once and compute the requested digests."""
sha1 = hashlib.sha1() if "sha1" in algos else None
md5 = hashlib.md5() if "md5" in algos else None
sha256 = hashlib.sha256() if "sha256" in algos else None
+45
View File
@@ -0,0 +1,45 @@
"""compute_hashes reads a file once per process and change."""
from __future__ import annotations
import shutil
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "scripts"))
import hashing # noqa: E402
class HashesAreReadOncePerContent(unittest.TestCase):
"""verify --all hashed GB_dmg.bin 13 times and scph5501.bin 12, once
per platform and entry, in one process."""
def setUp(self):
self.dir = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp"))
self.addCleanup(shutil.rmtree, self.dir, True)
self.file = self.dir / "rom.bin"
self.file.write_bytes(b"a" * 64)
def test_a_second_ask_reads_nothing(self):
first = hashing.compute_hashes(self.file)
with mock.patch.object(hashing, "_compute_hashes", side_effect=AssertionError):
self.assertEqual(hashing.compute_hashes(self.file), first)
self.assertEqual(
hashing.compute_hashes(self.file, frozenset({"md5"})),
{"md5": first["md5"]},
)
def test_a_rewrite_in_place_is_read_again(self):
first = hashing.compute_hashes(self.file)
with open(self.file, "r+b") as handle:
handle.write(b"b" * 64)
self.assertNotEqual(hashing.compute_hashes(self.file)["sha1"], first["sha1"])
if __name__ == "__main__":
unittest.main()