feat: partially implement multipart its

pyproject.toml

@@ -21,17 +21,18 @@ classifiers = [
 ]
 # Runtime dependencies (for running tests/examples)
 dependencies = [
-  "boto3>=1.40.35",
-  "requests>=2.32.3",
+  "boto3>=1.42.34",
+  "requests>=2.32.5",
+  "awscrt>=0.31.1",
 ]
 
 [dependency-groups]
 dev = [
-  "pytest>=8.4.2",
-  "testcontainers>=4.13.3",
-  "boto3-stubs[s3]>=1.40.35",
-  "ruff>=0.13.1",
-  "ty>=0.0.1-alpha.33"
+  "pytest>=9.0.2",
+  "testcontainers>=4.14.0",
+  "boto3-stubs[s3]>=1.42.34",
+  "ruff>=0.14.14",
+  "ty>=0.0.13"
 ]
 
 [project.urls]

s3mock_test.py

@@ -5,6 +5,7 @@
 import re
 import time
 import uuid
+import zlib
 from enum import Enum
 from pathlib import Path
 from typing import Optional
@@ -13,6 +14,7 @@
 import boto3
 import pytest
 from _pytest.fixtures import FixtureRequest
+from awscrt import checksums as awscrt_checksums
 from boto3.s3.transfer import TransferConfig
 from botocore.client import Config
 from botocore.exceptions import ClientError
@@ -416,3 +418,51 @@ def checksum_algorithms() -> list[ChecksumAlgorithm]:
         ChecksumAlgorithm.CRC32C,
         ChecksumAlgorithm.CRC64NVME,
     ]
+
+def crc32(data: bytes) -> bytes:
+    crc = zlib.crc32(data) & 0xFFFFFFFF
+    return crc.to_bytes(4, byteorder="big")
+
+def crc32_b64(data: bytes) -> str:
+    return base64.b64encode(crc32(data)).decode("ascii")
+
+def crc64nvme(data: bytes) -> bytes:
+    checksum = awscrt_checksums.crc64nvme(data)
+    return checksum.to_bytes(8, byteorder="big")
+
+def crc64nvme_b64(data: bytes) -> str:
+    return base64.b64encode(crc64nvme(data)).decode("ascii")
+
+def hex_digest(path: str) -> str:
+    h = hashlib.sha256()
+    with open(path, "rb") as f:
+        while True:
+            chunk = f.read(8192)
+            if not chunk:
+                break
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def multipart_etag_hex(parts: list[bytes]) -> str:
+    digests = [hashlib.md5(part).digest() for part in parts]
+    combined = hashlib.md5(b"".join(digests)).hexdigest()
+    return f"{combined}-{len(parts)}"
+
+
+def multipart_crc32_checksum(parts: list[bytes]) -> str:
+    part_checksums = [
+        crc32(part)
+        for part in parts
+    ]
+    checksum_b64 = crc32_b64(b"".join(part_checksums))
+    return f"{checksum_b64}-{len(parts)}"
+
+
+def multipart_crc64nvme_checksum(parts: list[bytes]) -> str:
+    part_checksums = [
+        crc64nvme(part)
+        for part in parts
+    ]
+    checksum_b64 = crc64nvme_b64(b"".join(part_checksums))
+    return f"{checksum_b64}-{len(parts)}"