Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 67 additions & 0 deletions kinetic/utils/packager.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
"""

import os
import posixpath
import subprocess
import zipfile
from collections.abc import Callable
from typing import Any
Expand All @@ -18,6 +20,71 @@
PositionPath = tuple[str | int, ...]


def _list_git_files(base_dir: str) -> list[str] | None:
"""List tracked and non-ignored untracked files under ``base_dir``."""
try:
result = subprocess.run(
[
"git",
"-C",
base_dir,
"ls-files",
"--cached",
"--others",
"--exclude-standard",
"-z",
"--",
".",
],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
)
except (OSError, subprocess.CalledProcessError):
return None

return [os.fsdecode(path) for path in result.stdout.split(b"\0") if path]


def _path_is_excluded(path: str, exclude_paths: set[str]) -> bool:
if not exclude_paths:
return False
normalized_path = os.path.normpath(path)
return any(
normalized_path == excluded or normalized_path.startswith(excluded + os.sep)
for excluded in exclude_paths
)
Comment on lines +49 to +56

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

If base_dir is passed as a relative path (e.g., . or ./src), file_path will be constructed as a relative path. However, exclude_paths are absolute paths (as specified in the zip_working_dir docstring). Comparing a relative path with an absolute path using os.path.normpath will fail to match, causing exclusions to be silently ignored.

Converting path to an absolute path using os.path.abspath before performing the comparison ensures that exclusions are correctly respected regardless of whether base_dir is relative or absolute.

Suggested change
def _path_is_excluded(path: str, exclude_paths: set[str]) -> bool:
if not exclude_paths:
return False
normalized_path = os.path.normpath(path)
return any(
normalized_path == excluded or normalized_path.startswith(excluded + os.sep)
for excluded in exclude_paths
)
def _path_is_excluded(path: str, exclude_paths: set[str]) -> bool:
if not exclude_paths:
return False
abs_path = os.path.abspath(path)
return any(
abs_path == excluded or abs_path.startswith(excluded + os.sep)
for excluded in exclude_paths
)



def _write_git_files(
zipf: zipfile.ZipFile,
base_dir: str,
git_files: list[str],
exclude_paths: set[str],
archive_prefix: str = "",
) -> None:
for relative_path in git_files:
file_path = os.path.join(base_dir, relative_path)
if _path_is_excluded(file_path, exclude_paths) or not os.path.lexists(
file_path
):
continue

archive_name = posixpath.join(archive_prefix, relative_path)
if os.path.isdir(file_path) and not os.path.islink(file_path):
nested_files = _list_git_files(file_path)
if nested_files is not None:
_write_git_files(
zipf,
file_path,
nested_files,
exclude_paths,
archive_prefix=archive_name,
)
continue
zipf.write(file_path, archive_name)


def zip_working_dir(
base_dir: str, output_path: str, exclude_paths: set[str] | None = None
) -> None:
Expand Down
Loading