[INFRA-6259] feat: implement reproducible zip

package_python_function/packager.py

@@ -1,15 +1,21 @@
+from __future__ import annotations
+
+import logging
+import os
+import shutil
+import time
+import zipfile
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-import zipfile
-import shutil
-import logging
+from typing import TYPE_CHECKING
 
 from .python_project import PythonProject
 
+if TYPE_CHECKING:
+    from typing import Tuple
 
 logger = logging.getLogger(__name__)
 
-
 class Packager:
     AWS_LAMBDA_MAX_UNZIP_SIZE = 262144000
 
@@ -40,14 +46,34 @@ def package(self) -> None:
     def zip_all_dependencies(self, target_path: Path) -> None:
         logger.info(f"Zipping to {target_path}...")
 
-        with zipfile.ZipFile(target_path, 'w', zipfile.ZIP_DEFLATED) as zip_file:
+        def date_time() -> Tuple[int, int, int, int, int, int]:
+            """Returns date_time value used to force overwrite on all ZipInfo objects. Defaults to
+            1980-01-01 00:00:00. You can set this with the environment variable SOURCE_DATE_EPOCH as an
+            integer value representing seconds since Epoch.
+            """
+            source_date_epoch = os.environ.get("SOURCE_DATE_EPOCH", None)
+            if source_date_epoch is not None:
+                return time.gmtime(int(source_date_epoch))[:6]
+            return (1980, 1, 1, 0, 0, 0)
+
+        with zipfile.ZipFile(target_path, "w", zipfile.ZIP_DEFLATED) as zip_file:
+
             def zip_dir(path: Path) -> None:
                 for item in path.iterdir():
                     if item.is_dir():
                         zip_dir(item)
                     else:
+                        zinfo = zipfile.ZipInfo.from_file(
+                            item, item.relative_to(self.input_path)
+                        )
+                        zinfo.date_time = date_time()
+                        zinfo.external_attr = 0o644 << 16
                         self._uncompressed_bytes += item.stat().st_size
-                        zip_file.write(item, item.relative_to(self.input_path))
+                        with (
+                            open(item, "rb") as src,
+                            zip_file.open(zinfo, "w") as dest,
+                        ):
+                            shutil.copyfileobj(src, dest, 1024 * 8)
 
             zip_dir(self.input_path)
 
@@ -61,7 +87,7 @@ def zip_dir(path: Path) -> None:
                 logger.info(f"The compressed size ({compressed_bytes:,}) is less than the AWS limit, so the nested-zip strategy will be used.")
                 self.generate_nested_zip(target_path)
             else:
-                print(f"TODO Error.  The unzipped size it too large for AWS Lambda.")
+                print("TODO Error.  The unzipped size it too large for AWS Lambda.")
         else:
             logger.info(f"Copying '{target_path}' to '{self.output_file}'")
             shutil.copy(str(target_path), str(self.output_file))
@@ -80,4 +106,4 @@ def generate_nested_zip(self, inner_zip_path: Path) -> None:
                 str(entrypoint_dir / "__init__.py"),
                 Path(__file__).parent.joinpath("nested_zip_loader.py").read_text(),
                 compresslevel=zipfile.ZIP_DEFLATED
-            )
+            )

tests/projects/project-1/pyproject.toml

@@ -5,7 +5,7 @@ description = "project-1"
 authors = [{ name = "Brandon White", email = "brandonlwhite@gmail.com" }]
 license = "MIT"
 readme = "README.md"
-requires-python = "^3.10"
+requires-python = ">=3.10,<4.0"
 
 
 [build-system]

tests/test_package_python_function.py

@@ -1,12 +1,15 @@
-from pathlib import Path
 import sys
-from package_python_function.main import main
+import zipfile
+from pathlib import Path
 
+from package_python_function.main import main
 
 PROJECTS_DIR_PATH = Path(__file__).parent / 'projects'
 
-
 def test_package_python_function(tmp_path: Path) -> None:
+    EXPECTED_FILE_MODE = 0o644
+    EXPECTED_FILE_DATE_TIME = (1980, 1, 1, 0, 0, 0)
+
     project_file_path = PROJECTS_DIR_PATH / 'project-1' / 'pyproject.toml'
 
     venv_dir_path = tmp_path / 'venv'
@@ -34,4 +37,19 @@ def test_package_python_function(tmp_path: Path) -> None:
     ]
     main()
 
-    assert (output_dir_path / 'project_1.zip').exists()
+    zip_file = output_dir_path / "project_1.zip"
+    assert zip_file.exists()
+
+    verify_dir = tmp_path / "verify"
+    verify_dir.mkdir()
+    with zipfile.ZipFile(zip_file, "r") as zip:
+        zip.extractall(verify_dir)
+        for file_info in zip.infolist():
+            mode = (file_info.external_attr >> 16) & 0xFFFF
+            assert mode == EXPECTED_FILE_MODE
+            assert file_info.date_time == EXPECTED_FILE_DATE_TIME
+
+        assert (verify_dir / "project_1" / "__init__.py").exists()
+        assert (verify_dir / "project_1" / "project1.py").exists()
+        assert (verify_dir / "small_dependency" / "__init__.py").exists()
+        assert (verify_dir / "small_dependency" / "small_dependency.py").exists()