diff --git a/scripts/hashing.py b/scripts/hashing.py index c8f8fc7f..537cd165 100644 --- a/scripts/hashing.py +++ b/scripts/hashing.py @@ -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 diff --git a/tests/test_hashing.py b/tests/test_hashing.py new file mode 100644 index 00000000..d6c6b71e --- /dev/null +++ b/tests/test_hashing.py @@ -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()