using faster file hash checking

Author: Andrei Neagu

packages/aws-library/tests/test_s3_client.py

@@ -1526,6 +1526,7 @@ def _get_s3_object_keys(files: set[Path]) -> set[S3ObjectKey]:
 @pytest.mark.parametrize(
     "file_size, local_count, remote_count",
     [
+        pytest.param(TypeAdapter(ByteSize).validate_python("2Mib"), 10, 10, id="micro"),
         pytest.param(
             TypeAdapter(ByteSize).validate_python("10Mib"), 10, 10, id="small"
         ),

packages/pytest-simcore/src/pytest_simcore/helpers/comparing.py

@@ -1,10 +1,12 @@
+import asyncio
+import hashlib
 from collections.abc import Iterable
+from concurrent.futures import ProcessPoolExecutor
 from pathlib import Path
 from typing import TypeAlias
 
 import aiofiles
 from servicelib.file_utils import create_sha256_checksum
-from servicelib.utils import limited_gather
 
 _FilesInfo: TypeAlias = dict[str, Path]
 
@@ -28,13 +30,37 @@ def get_files_info_from_itrable(items: Iterable[Path]) -> _FilesInfo:
     return {f.name: f for f in items if f.is_file()}
 
 
+def _compute_hash(file_path: Path) -> tuple[Path, str]:
+    with Path.open(file_path, "rb") as file_to_hash:
+        file_hash = hashlib.md5()  # noqa: S324
+        chunk = file_to_hash.read(8192)
+        while chunk:
+            file_hash.update(chunk)
+            chunk = file_to_hash.read(8192)
+
+    return file_path, file_hash.hexdigest()
+
+
+async def compute_hashes(file_paths: list[Path]) -> dict[Path, str]:
+    """given a list of files computes hashes for the files on a process pool"""
+
+    loop = asyncio.get_event_loop()
+
+    with ProcessPoolExecutor() as prcess_pool_executor:
+        tasks = [
+            loop.run_in_executor(prcess_pool_executor, _compute_hash, file_path)
+            for file_path in file_paths
+        ]
+        # pylint: disable=unnecessary-comprehension
+        # see return value of _compute_hash it is a tuple, mapping list[Tuple[Path,str]] to Dict[Path, str] here
+        return dict(await asyncio.gather(*tasks))
+
+
 async def assert_same_contents(file_info1: _FilesInfo, file_info2: _FilesInfo) -> None:
     assert set(file_info1.keys()) == set(file_info2.keys())
 
-    await limited_gather(
-        *(
-            assert_same_file_content(file_info1[file_name], file_info2[file_name])
-            for file_name in file_info1
-        ),
-        limit=10,
-    )
+    hashes_1 = await compute_hashes(list(file_info1.values()))
+    hashes_2 = await compute_hashes(list(file_info2.values()))
+
+    for key in file_info1:
+        assert hashes_1[file_info1[key]] == hashes_2[file_info2[key]]

packages/service-library/src/servicelib/zip_stream/_zipper.py

@@ -6,7 +6,7 @@
 
 from ..progress_bar import ProgressBarData
 from ._constants import DEFAULT_READ_CHUNK_SIZE
-from ._types import ArchiveEntries, FileStream
+from ._types import ArchiveEntries, FileSize, FileStream
 
 
 async def _member_files_iter(
@@ -27,17 +27,20 @@ async def get_zip_archive_stream(
     archive_files: ArchiveEntries,
     *,
     progress_bar: ProgressBarData | None = None,
-    chunk_size: int = DEFAULT_READ_CHUNK_SIZE
+    chunk_size: int = DEFAULT_READ_CHUNK_SIZE,
 ) -> FileStream:
     # NOTE: this is CPU bound task, even though the loop is not blocked,
     # the CPU is still used for compressing the content.
     if progress_bar is None:
         progress_bar = ProgressBarData(num_steps=1, description="zip archive stream")
 
-    total_stream_lenth = sum(file_size for _, (file_size, _) in archive_files)
+    total_stream_lenth = FileSize(sum(file_size for _, (file_size, _) in archive_files))
+    description = (
+        f"STATS: count={len(archive_files)}, size={total_stream_lenth.human_readable()}"
+    )
 
     async with progress_bar.sub_progress(
-        steps=total_stream_lenth, description="streams_reader", progress_unit="Byte"
+        steps=total_stream_lenth, description=description, progress_unit="Byte"
     ) as sub_progress:
         # NOTE: do not disable compression or the streams will be
         # loaded fully in memory before yielding their content

packages/service-library/tests/archiving_utils/test_archiving_utils.py

@@ -4,19 +4,18 @@
 # pylint:disable=no-name-in-module
 
 import asyncio
-import hashlib
 import os
 import secrets
 import string
 import tempfile
 from collections.abc import Callable, Iterable
-from concurrent.futures import ProcessPoolExecutor
 from pathlib import Path
 
 import pytest
 from faker import Faker
 from pydantic import ByteSize, TypeAdapter
 from pytest_benchmark.plugin import BenchmarkFixture
+from pytest_simcore.helpers.comparing import compute_hashes
 from servicelib.archiving_utils import archive_dir, unarchive_dir
 
 
@@ -92,32 +91,6 @@ def get_all_files_in_dir(dir_path: Path) -> set[Path]:
     }
 
 
-def _compute_hash(file_path: Path) -> tuple[Path, str]:
-    with Path.open(file_path, "rb") as file_to_hash:
-        file_hash = hashlib.md5()  # noqa: S324
-        chunk = file_to_hash.read(8192)
-        while chunk:
-            file_hash.update(chunk)
-            chunk = file_to_hash.read(8192)
-
-    return file_path, file_hash.hexdigest()
-
-
-async def compute_hashes(file_paths: list[Path]) -> dict[Path, str]:
-    """given a list of files computes hashes for the files on a process pool"""
-
-    loop = asyncio.get_event_loop()
-
-    with ProcessPoolExecutor() as prcess_pool_executor:
-        tasks = [
-            loop.run_in_executor(prcess_pool_executor, _compute_hash, file_path)
-            for file_path in file_paths
-        ]
-        # pylint: disable=unnecessary-comprehension
-        # see return value of _compute_hash it is a tuple, mapping list[Tuple[Path,str]] to Dict[Path, str] here
-        return dict(await asyncio.gather(*tasks))
-
-
 def full_file_path_from_dir_and_subdirs(dir_path: Path) -> list[Path]:
     return [x for x in dir_path.rglob("*") if x.is_file()]