This is an automated email from the ASF dual-hosted git repository.
ephraimbuddy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f31684af859 Add the manifest format for manifest-backed versioned Dag
bundles (#72476)
f31684af859 is described below
commit f31684af8596377142ecc587cd9b2030eac86a8b
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Wed Oct 14 12:43:51 2026 +0100
Add the manifest format for manifest-backed versioned Dag bundles (#72476)
* Add the manifest format for versioned Dag bundles
This is the first of several changes that add manifest-backed Dag bundles.
A manifest lists every file in a bundle with its checksum, and the bundle
version is a hash of that list.
Bundles get published and read on different machines, so both sides have to
agree on how the list is written and on what makes a version or a file path
valid. Keeping those rules in one place lets the storage backends that come
next share them instead of each keeping its own copy.
* Report unreadable bundle sources separately from changed ones
Reading a Dag bundle's source tree can fail for two different reasons. A
file
that vanished mid-walk means someone edited the Dag folder while we were
looking at it, and a publisher should simply try again on its next pass. A
permission problem or a stale network handle is a standing fault that an
operator has to see and fix.
Only the first was being caught, so the second escaped as a raw OSError
instead of a bundle error, and folding both into the "source changed" error
would leave a misconfigured Dag folder pinned to a stale version with
nothing
but a debug log to show for it.
* Let a manifest record the backend's object version for each file
An object store can keep several versions of the same key, so the key a
release
was built from is not necessarily the one a consumer fetches later.
Carrying the
backend's own identifier for each stored object, such as an S3 VersionId,
leaves
room to pin a fetch to the exact object instead of trusting whatever
currently
sits at the key.
Nothing populates it yet; it is optional so the local backend, which has no
such
identifier, is unaffected. It is deliberately excluded from the bundle
version:
republishing identical files to a versioned bucket mints new object
versions, and
that must still resolve to one bundle version.
* Fail a bundle source root the same way as the files inside it
Every other read of a bundle source now reports trouble as a manifest
error, but
a missing or unreadable root still escaped as a bare OSError, so a caller
that
guards a publish with the module's own error type would not catch it.
Unlike a
file vanishing mid-walk, a root we never saw is a configured path rather
than a
concurrent edit, so none of these say the source changed: a typo in the
configuration must not look like something worth quietly retrying.
Backend object versions are also checked before the hashing pass rather than
during it, so a bad identifier no longer costs a full read of the tree
first.
* Let a consumer check a manifest against its own contents
A digest only establishes that a manifest is the document it was taken
from, and a
consumer reading a manifest had no way to check the file entries against
anything.
Reading a malformed entry also failed as a stray KeyError rather than a
manifest error.
Re-deriving the version from the entries binds the two together, which is an
integrity guarantee only against a version the publisher did not supply,
such as an
operator's pin or one Airflow already recorded; checked against nothing but
itself, a
manifest is merely self-consistent. Making the caller state which case
applies keeps
the weaker one from being the default, and fixing the entry order keeps one
set of
files from being publishable under several versions.
---
.../src/airflow/dag_processing/bundles/manifest.py | 571 +++++++++++++
.../unit/dag_processing/bundles/test_manifest.py | 881 +++++++++++++++++++++
2 files changed, 1452 insertions(+)
diff --git a/airflow-core/src/airflow/dag_processing/bundles/manifest.py
b/airflow-core/src/airflow/dag_processing/bundles/manifest.py
new file mode 100644
index 00000000000..9a3e141f030
--- /dev/null
+++ b/airflow-core/src/airflow/dag_processing/bundles/manifest.py
@@ -0,0 +1,571 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import stat
+from collections.abc import Iterator, Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from typing import Any, NoReturn
+
+from airflow.exceptions import AirflowException
+
+MANIFEST_FILE_NAME = ".airflow-bundle-manifest.json"
+IGNORED_DIR_NAMES = frozenset({".git", "__pycache__"})
+# ".git" as a file name covers git-worktree checkouts, where .git is a pointer
file.
+IGNORED_FILE_NAMES = frozenset({MANIFEST_FILE_NAME, ".git"})
+IGNORED_FILE_SUFFIXES = frozenset({".pyc"})
+MANIFEST_SCHEMA_VERSION = 1
+# "sha256-" (not "sha256:") keeps a version usable as a path segment: Airflow
builds
+# cache and tracking paths from the raw value.
+SHA256_VERSION_PREFIX = "sha256-"
+
+
+class BundleManifestError(AirflowException):
+ """
+ Base class for bundle manifest errors.
+
+ Subclasses ``AirflowException`` because every Dag bundle entry point
Airflow calls
+ is expected to fail with one, so a bundle that cannot be read is reported
per
+ bundle rather than escaping as an unrelated error type.
+ """
+
+
+class BundleManifestSourceChangedError(BundleManifestError):
+ """Raised when bundle source files change while a manifest is being
built."""
+
+
+@dataclass(frozen=True)
+class BundleSourceFile:
+ """A source file and the stat metadata used to detect source changes
during publishing."""
+
+ path: Path
+ relative_path: str
+ size: int
+ mtime_ns: int
+ ctime_ns: int
+ mode: int
+
+ def build_signature_record(self) -> dict[str, Any]:
+ return {
+ "path": self.relative_path,
+ "size": self.size,
+ "mtime_ns": self.mtime_ns,
+ "ctime_ns": self.ctime_ns,
+ "mode": self.mode,
+ }
+
+
+@dataclass(frozen=True)
+class BundleSourceSnapshot:
+ """A deterministic snapshot of the source tree metadata."""
+
+ root: Path
+ files: tuple[BundleSourceFile, ...]
+ signature: str
+
+
+def is_ignored_bundle_file_name(file_name: str) -> bool:
+ """Return whether manifest collection excludes one file, judged by its
name alone."""
+ return file_name in IGNORED_FILE_NAMES or PurePosixPath(file_name).suffix
in IGNORED_FILE_SUFFIXES
+
+
+@dataclass(frozen=True)
+class BundleVersionManifest:
+ """Full manifest plus the compact release-reference payload written to
latest.json."""
+
+ version: str
+ manifest: dict[str, Any]
+ ref_payload: dict[str, Any]
+ source_snapshot: BundleSourceSnapshot
+
+
+def _raise_source_access_error(error: OSError, *, subject: str, phase: str) ->
NoReturn:
+ """
+ Re-raise an OSError from the bundle source as the matching manifest error.
+
+ A missing entry means the source really did change under us, which a
publisher
+ treats as "nothing stable to publish yet, try the next pass". Any other
OSError --
+ a permission problem, a stale network handle -- is a standing fault, so it
must not
+ be reported as a benign concurrent edit and then quietly retried forever.
+ """
+ if isinstance(error, FileNotFoundError):
+ raise BundleManifestSourceChangedError(
+ f"Bundle source {subject} disappeared while {phase}"
+ ) from error
+ raise BundleManifestError(f"Bundle source {subject} became unreadable
while {phase}") from error
+
+
+def _raise_source_walk_error(error: OSError) -> NoReturn:
+ _raise_source_access_error(
+ error, subject=f"directory {error.filename}", phase="collecting
manifest metadata"
+ )
+
+
+def _iter_manifest_file_paths(root: Path) -> Iterator[tuple[Path,
os.stat_result]]:
+ for dirpath, dirnames, filenames in os.walk(
+ root,
+ followlinks=False,
+ onerror=_raise_source_walk_error,
+ ):
+ retained_dirnames: list[str] = []
+ for dirname in sorted(dirnames):
+ if dirname in IGNORED_DIR_NAMES:
+ continue
+ path = Path(dirpath) / dirname
+ try:
+ file_stat = path.lstat()
+ except OSError as e:
+ _raise_source_access_error(
+ e, subject=f"directory {path}", phase="collecting manifest
metadata"
+ )
+ if stat.S_ISLNK(file_stat.st_mode):
+ raise BundleManifestError(f"Bundle source contains symlinked
directory: {path}")
+ if not stat.S_ISDIR(file_stat.st_mode):
+ raise BundleManifestError(f"Bundle source contains
non-directory entry: {path}")
+ retained_dirnames.append(dirname)
+ dirnames[:] = retained_dirnames
+
+ for filename in sorted(filenames):
+ if is_ignored_bundle_file_name(filename):
+ continue
+ path = Path(dirpath) / filename
+ try:
+ file_stat = path.lstat()
+ except OSError as e:
+ _raise_source_access_error(e, subject=f"file {path}",
phase="collecting manifest metadata")
+ if stat.S_ISLNK(file_stat.st_mode):
+ raise BundleManifestError(f"Bundle source contains symlinked
file: {path}")
+ if not stat.S_ISREG(file_stat.st_mode):
+ raise BundleManifestError(f"Bundle source contains non-regular
file: {path}")
+ yield path, file_stat
+
+
+def _build_source_file(root: Path, path: Path, file_stat: os.stat_result) ->
BundleSourceFile:
+ return BundleSourceFile(
+ path=path,
+ relative_path=path.relative_to(root).as_posix(),
+ size=file_stat.st_size,
+ mtime_ns=file_stat.st_mtime_ns,
+ ctime_ns=file_stat.st_ctime_ns,
+ mode=stat.S_IMODE(file_stat.st_mode),
+ )
+
+
+def compute_file_sha256(path: Path) -> tuple[str, int]:
+ """Compute the sha256 hex digest and byte size of a file."""
+ digest = hashlib.sha256()
+ size = 0
+ with path.open("rb") as file:
+ for chunk in iter(lambda: file.read(1024 * 1024), b""):
+ digest.update(chunk)
+ size += len(chunk)
+ return digest.hexdigest(), size
+
+
+def compute_bundle_version(files: Sequence[dict[str, Any]]) -> str:
+ """
+ Compute the content-addressed bundle version from manifest file entries.
+
+ Only the identity fields (path, sha256, size, executable) participate, so
the
+ version stays stable if file entries ever gain extra metadata.
+ """
+ try:
+ payload = {
+ "schema_version": MANIFEST_SCHEMA_VERSION,
+ "files": [
+ {
+ "path": file_info["path"],
+ "sha256": file_info["sha256"],
+ "size": file_info["size"],
+ "executable": file_info["executable"],
+ }
+ for file_info in files
+ ],
+ }
+ except (KeyError, TypeError) as e:
+ raise BundleManifestError(
+ "Bundle manifest file entries must each contain path, sha256, size
and executable"
+ ) from e
+ return
f"{SHA256_VERSION_PREFIX}{hashlib.sha256(_serialize_manifest_payload(payload)).hexdigest()}"
+
+
+def is_sha256_hex(value: str) -> bool:
+ """Return whether a string is exactly 64 lowercase hex characters."""
+ return len(value) == 64 and all(character in "0123456789abcdef" for
character in value)
+
+
+def is_valid_bundle_version(version: str) -> bool:
+ """Return whether a string has the exact shape ``compute_bundle_version``
produces."""
+ digest = version.removeprefix(SHA256_VERSION_PREFIX)
+ return digest != version and is_sha256_hex(digest)
+
+
+def validate_bundle_version(version: str, *, source: str) -> str:
+ """
+ Return ``version`` if it is a well-formed bundle version, else raise.
+
+ Versions arrive from published documents and from operator-pinned
configuration, and
+ Airflow joins them into cache and tracking paths, so they are checked
before use.
+ """
+ if not is_valid_bundle_version(version):
+ raise BundleManifestError(f"Bundle {source} is not a valid sha256
version: {version!r}")
+ return version
+
+
+def validate_bundle_relative_path(relative_path: str) -> Path:
+ """
+ Return ``relative_path`` as a ``Path`` if it is safe to join onto a bundle
root, else raise.
+
+ Manifest paths are written by whoever published the bundle, so a consumer
must not
+ join one onto a local root before rejecting absolute paths, parent
traversal, the
+ non-normalized spellings (``a//b``, ``a/./b``, trailing slash) that would
resolve
+ somewhere other than where the manifest says, and control characters that
no
+ filesystem can represent. A backslash is an ordinary character in a POSIX
file
+ name, so it is accepted -- rejecting it would make a legitimate Dag file
+ publishable but never materializable.
+
+ This checks the path in isolation. Whether the manifest may legitimately
name the
+ path at all -- that it is not the manifest file itself, not a duplicate,
and not
+ excluded from collection -- is the caller's decision.
+ """
+ if any(character < " " for character in relative_path):
+ raise BundleManifestError(f"Bundle manifest contains unsafe relative
path: {relative_path!r}")
+ path = Path(relative_path)
+ if (
+ path.is_absolute()
+ or any(segment in {"", ".", ".."} for segment in
relative_path.split("/"))
+ or path.as_posix() != relative_path
+ ):
+ raise BundleManifestError(f"Bundle manifest contains unsafe relative
path: {relative_path!r}")
+ return path
+
+
+def _serialize_manifest_payload(payload: dict[str, Any]) -> bytes:
+ return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
+
+
+def serialize_bundle_version_manifest(manifest: dict[str, Any]) -> bytes:
+ """Serialize a bundle manifest deterministically for storage and
verification."""
+ return _serialize_manifest_payload(manifest)
+
+
+def _validate_bundle_root(root: Path) -> Path:
+ # Unlike a file vanishing mid-walk, a root we never saw is a configured
path rather
+ # than a concurrent edit, so none of these are reported as a changed
source: a typo
+ # in the configuration must not look like something worth quietly retrying.
+ root = Path(root)
+ try:
+ root_stat = root.stat()
+ except FileNotFoundError as e:
+ raise BundleManifestError(f"Bundle source path does not exist:
{root}") from e
+ except OSError as e:
+ raise BundleManifestError(f"Bundle source path is unreadable: {root}")
from e
+ if not stat.S_ISDIR(root_stat.st_mode):
+ raise BundleManifestError(f"Bundle source path is not a directory:
{root}")
+ return root
+
+
+def collect_bundle_source_snapshot(root: Path) -> BundleSourceSnapshot:
+ """Collect deterministic source-file metadata without reading file
contents."""
+ root = _validate_bundle_root(root)
+ files = [_build_source_file(root, path, file_stat) for path, file_stat in
_iter_manifest_file_paths(root)]
+ files.sort(key=lambda source_file: source_file.relative_path)
+ signature_payload = {
+ "schema_version": MANIFEST_SCHEMA_VERSION,
+ "files": [file.build_signature_record() for file in files],
+ }
+ signature =
f"sha256:{hashlib.sha256(_serialize_manifest_payload(signature_payload)).hexdigest()}"
+ return BundleSourceSnapshot(root=root, files=tuple(files),
signature=signature)
+
+
+def _ensure_source_file_unchanged(source_file: BundleSourceFile) -> None:
+ try:
+ current_stat = source_file.path.lstat()
+ except OSError as e:
+ _raise_source_access_error(
+ e, subject=f"file {source_file.relative_path}", phase="verifying
manifest metadata"
+ )
+
+ current_metadata = (
+ current_stat.st_size,
+ current_stat.st_mtime_ns,
+ current_stat.st_ctime_ns,
+ stat.S_IMODE(current_stat.st_mode),
+ )
+ expected_metadata = (
+ source_file.size,
+ source_file.mtime_ns,
+ source_file.ctime_ns,
+ source_file.mode,
+ )
+ if current_metadata != expected_metadata:
+ raise BundleManifestSourceChangedError(
+ f"Bundle source file changed while building manifest:
{source_file.relative_path}"
+ )
+
+
+def build_ref_payload(
+ *,
+ bundle_name: str,
+ version: str,
+ backend: dict[str, Any],
+ manifest_sha256: str,
+ file_count: int,
+ total_size: int,
+) -> dict[str, Any]:
+ """Build the compact release-reference payload that defines the
``latest.json`` format."""
+ return {
+ "schema_version": MANIFEST_SCHEMA_VERSION,
+ "bundle_name": bundle_name,
+ "version": version,
+ "backend": backend,
+ "manifest": {
+ "path": MANIFEST_FILE_NAME,
+ "sha256": manifest_sha256,
+ },
+ "file_count": file_count,
+ "total_size": total_size,
+ }
+
+
+def _build_ref_payload_from_manifest(manifest: dict[str, Any]) -> dict[str,
Any]:
+ manifest_sha256 =
hashlib.sha256(serialize_bundle_version_manifest(manifest)).hexdigest()
+ return build_ref_payload(
+ bundle_name=manifest["bundle_name"],
+ version=manifest["version"],
+ backend=manifest["backend"],
+ manifest_sha256=f"sha256:{manifest_sha256}",
+ file_count=manifest["file_count"],
+ total_size=manifest["total_size"],
+ )
+
+
+def build_bundle_version_manifest(
+ *,
+ bundle_name: str,
+ root: Path,
+ backend_type: str,
+ source_snapshot: BundleSourceSnapshot | None = None,
+ precomputed_file_sha256: Mapping[str, str] | None = None,
+ file_version_ids: Mapping[str, str] | None = None,
+) -> BundleVersionManifest:
+ """
+ Build a deterministic content manifest for a materialized Dag bundle root.
+
+ ``file_version_ids`` records the backend's own identifier for each stored
object, such
+ as an S3 ``VersionId``, so a consumer can fetch the exact object a release
was built
+ from rather than whatever currently sits at the key. It is backend
bookkeeping, not
+ content, so it stays out of ``compute_bundle_version`` -- republishing
identical files
+ to a versioned bucket mints new object versions and must still yield one
bundle version.
+ """
+ source_snapshot = source_snapshot or collect_bundle_source_snapshot(root)
+ root = _validate_bundle_root(root)
+ if source_snapshot.root != root:
+ raise ValueError("source_snapshot root does not match manifest root")
+ expected_paths = {source_file.relative_path for source_file in
source_snapshot.files}
+ if precomputed_file_sha256 is not None and set(precomputed_file_sha256) !=
expected_paths:
+ raise BundleManifestError("Precomputed file hashes do not match the
bundle source snapshot")
+ if file_version_ids is not None:
+ if set(file_version_ids) != expected_paths:
+ raise BundleManifestError("File version ids do not match the
bundle source snapshot")
+ for relative_path, version_id in sorted(file_version_ids.items()):
+ if not isinstance(version_id, str) or not version_id:
+ raise BundleManifestError(f"File version id is invalid for
{relative_path!r}")
+
+ files: list[dict[str, Any]] = []
+ total_size = 0
+ for source_file in source_snapshot.files:
+ if precomputed_file_sha256 is None:
+ try:
+ file_digest, _ = compute_file_sha256(source_file.path)
+ except OSError as e:
+ _raise_source_access_error(
+ e, subject=f"file {source_file.relative_path}",
phase="building manifest"
+ )
+ else:
+ file_digest = precomputed_file_sha256[source_file.relative_path]
+ if not is_sha256_hex(file_digest):
+ raise BundleManifestError(
+ f"Precomputed file hash is invalid for
{source_file.relative_path!r}"
+ )
+ _ensure_source_file_unchanged(source_file)
+ file_info = {
+ "path": source_file.relative_path,
+ "sha256": file_digest,
+ "size": source_file.size,
+ "executable": bool(source_file.mode & 0o111),
+ }
+ if file_version_ids is not None:
+ file_info["version_id"] =
file_version_ids[source_file.relative_path]
+ files.append(file_info)
+ total_size += source_file.size
+
+ version = compute_bundle_version(files)
+ manifest = {
+ "schema_version": MANIFEST_SCHEMA_VERSION,
+ "bundle_name": bundle_name,
+ "version": version,
+ "backend": {"type": backend_type},
+ "file_count": len(files),
+ "total_size": total_size,
+ "files": files,
+ }
+ return BundleVersionManifest(
+ version=version,
+ manifest=manifest,
+ ref_payload=_build_ref_payload_from_manifest(manifest),
+ source_snapshot=source_snapshot,
+ )
+
+
+def _validate_manifest_file_entry(file_info: Any) -> str:
+ if not isinstance(file_info, dict):
+ raise BundleManifestError("Bundle manifest file entries must be
objects")
+ relative_path = file_info.get("path")
+ if not isinstance(relative_path, str):
+ raise BundleManifestError("Bundle manifest file entry does not contain
a path")
+ validate_bundle_relative_path(relative_path)
+ if relative_path == MANIFEST_FILE_NAME:
+ raise BundleManifestError(f"Bundle manifest must not list
{MANIFEST_FILE_NAME!r} as a bundle file")
+
+ digest = file_info.get("sha256")
+ if not isinstance(digest, str) or not is_sha256_hex(digest):
+ raise BundleManifestError(f"Bundle manifest entry {relative_path!r}
does not contain a valid sha256")
+ size = file_info.get("size")
+ # bool is an int subclass and True == 1, so it would slip through the
checks below.
+ if not isinstance(size, int) or isinstance(size, bool) or size < 0:
+ raise BundleManifestError(f"Bundle manifest entry {relative_path!r}
does not contain a valid size")
+ if not isinstance(file_info.get("executable"), bool):
+ raise BundleManifestError(
+ f"Bundle manifest entry {relative_path!r} does not contain a valid
executable flag"
+ )
+ version_id = file_info.get("version_id")
+ if version_id is not None and (not isinstance(version_id, str) or not
version_id):
+ raise BundleManifestError(
+ f"Bundle manifest entry {relative_path!r} does not contain a valid
version id"
+ )
+ return relative_path
+
+
+def validate_bundle_version_manifest_structure(
+ manifest: Any,
+ *,
+ expected_version: str | None,
+ expected_bundle_name: str | None = None,
+ expected_backend_type: str | None = None,
+) -> None:
+ """
+ Check that a manifest is well formed and that its version commits to its
file entries.
+
+ Re-deriving the version from the entries binds the two together, but that
is only as
+ strong as the version it is compared with: a publisher who edits the
entries can
+ recompute the version as well, so a manifest checked against nothing but
itself is
+ merely self-consistent. Pass ``expected_version`` from somewhere the
publisher does not
+ control -- an operator's pin, or a version Airflow has already recorded --
to make this
+ an integrity check. It is required so that passing ``None`` is a
deliberate statement
+ that no such version is available.
+ """
+ if not isinstance(manifest, dict):
+ raise BundleManifestError("Bundle manifest must be an object")
+
+ schema_version = manifest.get("schema_version")
+ if schema_version != MANIFEST_SCHEMA_VERSION:
+ raise BundleManifestError(f"Bundle manifest has unsupported
schema_version {schema_version!r}")
+
+ bundle_name = manifest.get("bundle_name")
+ if not isinstance(bundle_name, str) or not bundle_name:
+ raise BundleManifestError("Bundle manifest does not contain a bundle
name")
+ if expected_bundle_name is not None and bundle_name !=
expected_bundle_name:
+ raise BundleManifestError(
+ f"Bundle manifest is for bundle {bundle_name!r}, expected
{expected_bundle_name!r}"
+ )
+
+ version = manifest.get("version")
+ if not isinstance(version, str):
+ raise BundleManifestError("Bundle manifest does not contain a version")
+ validate_bundle_version(version, source="manifest version")
+ if expected_version is not None and version != expected_version:
+ raise BundleManifestError(
+ f"Bundle manifest contains version {version!r}, expected
{expected_version!r}"
+ )
+
+ backend = manifest.get("backend")
+ if not isinstance(backend, dict) or not isinstance(backend.get("type"),
str):
+ raise BundleManifestError("Bundle manifest does not contain a backend
type")
+ if expected_backend_type is not None and backend["type"] !=
expected_backend_type:
+ # Worded to match the manifest bundle's existing "not for a local
backend" checks.
+ raise BundleManifestError(
+ f"Bundle manifest is not for a {expected_backend_type} backend:
got {backend['type']!r}"
+ )
+
+ files = manifest.get("files")
+ if not isinstance(files, list):
+ raise BundleManifestError("Bundle manifest files must be a list")
+
+ seen_paths: set[str] = set()
+ paths: list[str] = []
+ total_size = 0
+ for file_info in files:
+ relative_path = _validate_manifest_file_entry(file_info)
+ if relative_path in seen_paths:
+ raise BundleManifestError(f"Bundle manifest contains duplicate
path {relative_path!r}")
+ seen_paths.add(relative_path)
+ paths.append(relative_path)
+ total_size += file_info["size"]
+ # The version hashes entries in order, so without a fixed order the same
files could
+ # be published under several versions.
+ if paths != sorted(paths):
+ raise BundleManifestError("Bundle manifest file entries are not sorted
by path")
+
+ if manifest.get("file_count") != len(files):
+ raise BundleManifestError(
+ f"Bundle manifest file_count {manifest.get('file_count')!r} does
not match its "
+ f"{len(files)} file entries"
+ )
+ if manifest.get("total_size") != total_size:
+ raise BundleManifestError(
+ f"Bundle manifest total_size {manifest.get('total_size')!r} does
not match its "
+ f"entries, which total {total_size}"
+ )
+
+ computed_version = compute_bundle_version(files)
+ if version != computed_version:
+ raise BundleManifestError(
+ f"Bundle manifest version {version!r} does not match its contents,
"
+ f"which hash to {computed_version!r}"
+ )
+
+
+def verify_bundle_version_manifest(manifest: dict[str, Any], expected_sha256:
str) -> None:
+ """
+ Check that a manifest serializes to ``expected_sha256``.
+
+ This proves only that the manifest is the document the digest was taken
from, which
+ says nothing about the contents when the digest came from the same
publisher. Pair it
+ with ``validate_bundle_version_manifest_structure``, which carries its own
caveat.
+ """
+ actual_sha256 =
hashlib.sha256(serialize_bundle_version_manifest(manifest)).hexdigest()
+ expected_sha256 = expected_sha256.removeprefix("sha256:")
+ if actual_sha256 != expected_sha256:
+ raise BundleManifestError(
+ f"Bundle manifest digest mismatch: expected
sha256:{expected_sha256}, got sha256:{actual_sha256}"
+ )
diff --git a/airflow-core/tests/unit/dag_processing/bundles/test_manifest.py
b/airflow-core/tests/unit/dag_processing/bundles/test_manifest.py
new file mode 100644
index 00000000000..feee579e98e
--- /dev/null
+++ b/airflow-core/tests/unit/dag_processing/bundles/test_manifest.py
@@ -0,0 +1,881 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import stat
+from dataclasses import replace
+from pathlib import Path
+
+import pytest
+
+from airflow.dag_processing.bundles import manifest as manifest_module
+from airflow.dag_processing.bundles.manifest import (
+ MANIFEST_FILE_NAME,
+ SHA256_VERSION_PREFIX,
+ BundleManifestError,
+ BundleManifestSourceChangedError,
+ build_bundle_version_manifest,
+ collect_bundle_source_snapshot,
+ compute_bundle_version,
+ is_ignored_bundle_file_name,
+ is_sha256_hex,
+ is_valid_bundle_version,
+ serialize_bundle_version_manifest,
+ validate_bundle_relative_path,
+ validate_bundle_version,
+ validate_bundle_version_manifest_structure,
+ verify_bundle_version_manifest,
+)
+
+
+def _build_manifest(**kwargs):
+ return build_bundle_version_manifest(**kwargs).manifest
+
+
+def _write_file(root, relative_path, content):
+ path = root / relative_path
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(content)
+ return path
+
+
+def test_manifest_hash_is_stable_across_filesystem_creation_order(tmp_path):
+ first = tmp_path / "first"
+ second = tmp_path / "second"
+
+ _write_file(first, "dags/a.py", "print('a')")
+ _write_file(first, "dags/nested/b.py", "print('b')")
+ _write_file(second, "dags/nested/b.py", "print('b')")
+ _write_file(second, "dags/a.py", "print('a')")
+
+ first_manifest = _build_manifest(bundle_name="manifest-local", root=first,
backend_type="local")
+ second_manifest = _build_manifest(bundle_name="manifest-local",
root=second, backend_type="local")
+
+ assert first_manifest["version"] == second_manifest["version"]
+ assert first_manifest["files"] == second_manifest["files"]
+
+
+def test_manifest_hash_changes_when_file_content_changes(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('first')")
+ first_manifest = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ _write_file(source, "dags/example.py", "print('second')")
+ second_manifest = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ assert first_manifest["version"] != second_manifest["version"]
+
+
+def test_bundle_version_only_uses_file_identity_fields():
+ file_info = {
+ "path": "example.py",
+ "sha256": "0" * 64,
+ "size": 1,
+ "executable": False,
+ "future_metadata": "ignored",
+ }
+
+ assert compute_bundle_version([file_info]) == compute_bundle_version(
+ [{key: value for key, value in file_info.items() if key !=
"future_metadata"}]
+ )
+
+
+def test_manifest_uses_complete_precomputed_file_hashes(tmp_path, monkeypatch):
+ source = tmp_path / "source"
+ content = "print('dag')"
+ _write_file(source, "dags/example.py", content)
+ source_snapshot = collect_bundle_source_snapshot(source)
+ expected_digest = hashlib.sha256(content.encode()).hexdigest()
+
+ def fail_hash(_):
+ raise AssertionError("precomputed hashes must skip a second source
read")
+
+ monkeypatch.setattr(manifest_module, "compute_file_sha256", fail_hash)
+ result = build_bundle_version_manifest(
+ bundle_name="manifest-s3",
+ root=source,
+ backend_type="local",
+ source_snapshot=source_snapshot,
+ precomputed_file_sha256={"dags/example.py": expected_digest},
+ )
+
+ assert result.manifest["files"][0]["sha256"] == expected_digest
+
+
[email protected](
+ ("precomputed_hashes", "expected_message"),
+ [
+ ({}, "Precomputed file hashes do not match"),
+ ({"other.py": "0" * 64}, "Precomputed file hashes do not match"),
+ ({"dags/example.py": "not-a-sha256"}, "Precomputed file hash is
invalid"),
+ ({"dags/example.py": "z" * 64}, "Precomputed file hash is invalid"),
+ ({"dags/example.py": "A" * 64}, "Precomputed file hash is invalid"),
+ ],
+)
+def test_manifest_rejects_invalid_precomputed_file_hashes(tmp_path,
precomputed_hashes, expected_message):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+
+ with pytest.raises(BundleManifestError, match=expected_message):
+ build_bundle_version_manifest(
+ bundle_name="manifest-s3",
+ root=source,
+ backend_type="local",
+ precomputed_file_sha256=precomputed_hashes,
+ )
+
+
+def test_manifest_hash_changes_when_executable_bit_changes(tmp_path):
+ source = tmp_path / "source"
+ script = _write_file(source, "scripts/run.sh", "#!/bin/sh\n")
+ first_manifest = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ script.chmod(stat.S_IMODE(script.stat().st_mode) | stat.S_IXUSR)
+ second_manifest = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ assert first_manifest["version"] != second_manifest["version"]
+ assert first_manifest["files"][0]["executable"] is False
+ assert second_manifest["files"][0]["executable"] is True
+
+
+def test_manifest_rejects_source_symlink(tmp_path):
+ source = tmp_path / "source"
+ source.mkdir()
+ target = tmp_path / "outside.py"
+ target.write_text("print('outside')")
+ try:
+ (source / "linked.py").symlink_to(target)
+ except OSError as e:
+ pytest.skip(f"Symlinks are not supported in this test environment:
{e}")
+
+ with pytest.raises(BundleManifestError, match="symlinked file"):
+ _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+
+
+def test_manifest_rejects_source_symlinked_directory(tmp_path):
+ source = tmp_path / "source"
+ source.mkdir()
+ target = tmp_path / "outside"
+ target.mkdir()
+ try:
+ (source / "linked").symlink_to(target, target_is_directory=True)
+ except OSError as e:
+ pytest.skip(f"Symlinks are not supported in this test environment:
{e}")
+
+ with pytest.raises(BundleManifestError, match="symlinked directory"):
+ _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+
+
[email protected](
+ ("file_name", "expected"),
+ [
+ (".airflow-bundle-manifest.json", True),
+ (".git", True),
+ ("example.pyc", True),
+ ("example.py", False),
+ (".pyc", False),
+ ],
+)
+def test_ignored_bundle_file_names(file_name, expected):
+ assert is_ignored_bundle_file_name(file_name) is expected
+
+
+# A vanished entry is a benign concurrent edit a publisher retries past;
anything else
+# is a standing fault that must not be reported as one.
+SOURCE_ACCESS_ERROR_KINDS = [
+ pytest.param(
+ FileNotFoundError(2, "No such file or directory"),
+ BundleManifestSourceChangedError,
+ "disappeared while",
+ id="missing",
+ ),
+ pytest.param(
+ PermissionError(13, "Permission denied"),
+ BundleManifestError,
+ "became unreadable while",
+ id="unreadable",
+ ),
+ pytest.param(
+ OSError(116, "Stale file handle"),
+ BundleManifestError,
+ "became unreadable while",
+ id="stale-handle",
+ ),
+]
+
+
[email protected](("error", "expected_type", "expected_message"),
SOURCE_ACCESS_ERROR_KINDS)
+def test_manifest_maps_source_walk_errors_by_kind(
+ tmp_path, monkeypatch, error, expected_type, expected_message
+):
+ source = tmp_path / "source"
+ source.mkdir()
+ error.filename = str(source / "unreadable")
+
+ def fail_walk(*args, onerror, **kwargs):
+ onerror(error)
+
+ monkeypatch.setattr(manifest_module.os, "walk", fail_walk)
+
+ with pytest.raises(BundleManifestError, match=expected_message) as excinfo:
+ _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+
+ assert type(excinfo.value) is expected_type
+ assert "collecting manifest metadata" in str(excinfo.value)
+
+
[email protected]("entry_name", ["nested", "example.py"])
[email protected](("error", "expected_type", "expected_message"),
SOURCE_ACCESS_ERROR_KINDS)
+def test_manifest_maps_source_entry_stat_errors_by_kind(
+ tmp_path, monkeypatch, entry_name, error, expected_type, expected_message
+):
+ source = tmp_path / "source"
+ _write_file(source, "nested/example.py", "print('dag')")
+ real_lstat = Path.lstat
+
+ def failing_lstat(self, *args, **kwargs):
+ if self.name == entry_name:
+ raise error
+ return real_lstat(self, *args, **kwargs)
+
+ monkeypatch.setattr(Path, "lstat", failing_lstat)
+
+ with pytest.raises(BundleManifestError, match=expected_message) as excinfo:
+ _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+
+ assert type(excinfo.value) is expected_type
+ assert "collecting manifest metadata" in str(excinfo.value)
+
+
+def test_manifest_reports_an_unreadable_source_file_during_hashing(tmp_path,
monkeypatch):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+
+ def unreadable(_):
+ raise PermissionError(13, "Permission denied")
+
+ monkeypatch.setattr(manifest_module, "compute_file_sha256", unreadable)
+
+ with pytest.raises(BundleManifestError, match="became unreadable while
building manifest") as excinfo:
+ _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+
+ assert type(excinfo.value) is BundleManifestError
+
+
+def
test_manifest_reports_an_unreadable_source_file_during_verification(tmp_path,
monkeypatch):
+ source = tmp_path / "source"
+ _write_file(source, "example.py", "content")
+ source_snapshot = collect_bundle_source_snapshot(source)
+ digest = hashlib.sha256(b"content").hexdigest()
+
+ def unreadable(self, *args, **kwargs):
+ raise PermissionError(13, "Permission denied")
+
+ # Precomputed hashes skip the read, so verification is the first access to
the file.
+ monkeypatch.setattr(Path, "lstat", unreadable)
+
+ with pytest.raises(
+ BundleManifestError, match="became unreadable while verifying manifest
metadata"
+ ) as excinfo:
+ build_bundle_version_manifest(
+ bundle_name="manifest-local",
+ root=source,
+ backend_type="local",
+ source_snapshot=source_snapshot,
+ precomputed_file_sha256={"example.py": digest},
+ )
+
+ assert type(excinfo.value) is BundleManifestError
+
+
+def test_manifest_ignores_cache_and_vcs_files(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+ manifest_before = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ _write_file(source, ".git/config", "[core]")
+ _write_file(source, MANIFEST_FILE_NAME, "{}")
+ _write_file(source, "dags/__pycache__/example.cpython-312.pyc", "compiled")
+ manifest_after = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ assert manifest_after["version"] == manifest_before["version"]
+ assert [file_info["path"] for file_info in manifest_after["files"]] ==
["dags/example.py"]
+
+
+def test_manifest_ignores_symlinked_ignored_directory(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+ manifest_before = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ real_git_dir = tmp_path / "real-git"
+ real_git_dir.mkdir()
+ try:
+ (source / ".git").symlink_to(real_git_dir, target_is_directory=True)
+ except OSError as e:
+ pytest.skip(f"Symlinks are not supported in this test environment:
{e}")
+ manifest_after = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ assert manifest_after["version"] == manifest_before["version"]
+
+
+def test_manifest_ignores_git_worktree_pointer_file(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+ manifest_before = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ _write_file(source, ".git", "gitdir: /repos/main/.git/worktrees/dags")
+ manifest_after = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ assert manifest_after["version"] == manifest_before["version"]
+ assert [file_info["path"] for file_info in manifest_after["files"]] ==
["dags/example.py"]
+
+
+def test_manifest_paths_are_relative_and_sorted(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "z.py", "print('z')")
+ _write_file(source, "a/nested.py", "print('nested')")
+ _write_file(source, "a.py", "print('a')")
+
+ manifest = _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+ paths = [file_info["path"] for file_info in manifest["files"]]
+
+ assert paths == ["a.py", "a/nested.py", "z.py"]
+ assert str(source) not in json.dumps(manifest)
+
+
+def test_ref_payload_is_compact_and_points_to_manifest(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+
+ result = build_bundle_version_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ assert result.ref_payload == {
+ "schema_version": 1,
+ "bundle_name": "manifest-local",
+ "version": result.version,
+ "backend": {"type": "local"},
+ "manifest": {
+ "path": MANIFEST_FILE_NAME,
+ "sha256": result.ref_payload["manifest"]["sha256"],
+ },
+ "file_count": 1,
+ "total_size": len("print('dag')"),
+ }
+ assert result.ref_payload["manifest"]["sha256"].startswith("sha256:")
+ assert "files" not in result.ref_payload
+ assert str(source) not in json.dumps(result.ref_payload)
+
+
[email protected](
+ ("source_name", "source_factory", "expected_message"),
+ [
+ ("missing", lambda path: None, "does not exist"),
+ ("file", lambda path: path.write_text("not a directory"), "is not a
directory"),
+ ],
+)
+def test_manifest_rejects_invalid_source_roots(tmp_path, source_name,
source_factory, expected_message):
+ source = tmp_path / source_name
+ source_factory(source)
+
+ with pytest.raises(BundleManifestError, match=expected_message) as excinfo:
+ collect_bundle_source_snapshot(source)
+
+ # A configured root that is wrong is a standing fault, never a concurrent
edit.
+ assert type(excinfo.value) is BundleManifestError
+
+
+def test_manifest_rejects_an_unreadable_source_root(tmp_path, monkeypatch):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+
+ def unreadable(self, *args, **kwargs):
+ raise PermissionError(13, "Permission denied")
+
+ monkeypatch.setattr(Path, "stat", unreadable)
+
+ with pytest.raises(BundleManifestError, match="is unreadable") as excinfo:
+ collect_bundle_source_snapshot(source)
+
+ assert type(excinfo.value) is BundleManifestError
+
+
+def test_manifest_rejects_snapshot_from_another_root(tmp_path):
+ first = tmp_path / "first"
+ second = tmp_path / "second"
+ _write_file(first, "example.py", "first")
+ _write_file(second, "example.py", "second")
+
+ with pytest.raises(ValueError, match="source_snapshot root"):
+ build_bundle_version_manifest(
+ bundle_name="manifest-local",
+ root=second,
+ backend_type="local",
+ source_snapshot=collect_bundle_source_snapshot(first),
+ )
+
+
+def test_manifest_rejects_file_removed_after_source_snapshot(tmp_path):
+ source = tmp_path / "source"
+ source_file = _write_file(source, "example.py", "content")
+ source_snapshot = collect_bundle_source_snapshot(source)
+ source_file.unlink()
+
+ with pytest.raises(BundleManifestSourceChangedError, match="disappeared
while building manifest"):
+ build_bundle_version_manifest(
+ bundle_name="manifest-local",
+ root=source,
+ backend_type="local",
+ source_snapshot=source_snapshot,
+ )
+
+
+def
test_manifest_rejects_file_metadata_changed_after_source_snapshot(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "example.py", "content")
+ source_snapshot = collect_bundle_source_snapshot(source)
+ changed_file = replace(source_snapshot.files[0],
size=source_snapshot.files[0].size + 1)
+
+ with pytest.raises(BundleManifestSourceChangedError, match="changed while
building manifest"):
+ build_bundle_version_manifest(
+ bundle_name="manifest-local",
+ root=source,
+ backend_type="local",
+ source_snapshot=replace(source_snapshot, files=(changed_file,)),
+ )
+
+
+def test_manifest_verification_accepts_prefixed_and_bare_digests(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "example.py", "content")
+ manifest = _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+ digest =
hashlib.sha256(serialize_bundle_version_manifest(manifest)).hexdigest()
+
+ verify_bundle_version_manifest(manifest, digest)
+ verify_bundle_version_manifest(manifest, f"sha256:{digest}")
+
+
+def test_manifest_verification_rejects_digest_mismatch(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "example.py", "content")
+ manifest = _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+
+ with pytest.raises(BundleManifestError, match="manifest digest mismatch"):
+ verify_bundle_version_manifest(manifest, "0" * 64)
+
+
+def test_computed_bundle_version_passes_its_own_validator(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+
+ version = _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")["version"]
+
+ assert version.startswith(SHA256_VERSION_PREFIX)
+ assert is_valid_bundle_version(version)
+ assert validate_bundle_version(version, source="test") == version
+
+
[email protected](
+ "version",
+ [
+ "",
+ "../outside",
+ "sha256-bad",
+ f"sha256-{'A' * 64}",
+ f"sha256-{'a' * 63}",
+ f"sha256-{'a' * 65}",
+ f"sha256:{'a' * 64}",
+ "a" * 64,
+ f"sha256-{'a' * 63}/",
+ f"sha256-sha256-{'a' * 50}",
+ ],
+)
+def test_bundle_version_validation_rejects_malformed_versions(version):
+ assert is_valid_bundle_version(version) is False
+ with pytest.raises(BundleManifestError, match="valid sha256 version"):
+ validate_bundle_version(version, source="pinned bundle version")
+
+
[email protected]("relative_path", ["a.py", "dags/example.py",
"dags/nested/example.py"])
+def
test_bundle_relative_path_validation_keeps_accepted_paths_under_the_root(tmp_path,
relative_path):
+ resolved = (tmp_path /
validate_bundle_relative_path(relative_path)).resolve()
+
+ assert resolved.is_relative_to(tmp_path.resolve())
+ assert resolved != tmp_path.resolve()
+
+
[email protected](
+ "relative_path",
+ [
+ "",
+ "..",
+ "../evil.py",
+ "dags/../../evil.py",
+ "/tmp/evil.py",
+ "dags//example.py",
+ "dags/./example.py",
+ "dags/",
+ ".",
+ "./example.py",
+ "/",
+ "dags/ex\x00ample.py",
+ "dags/ex\nample.py",
+ "dags/ex\tample.py",
+ ],
+)
+def test_bundle_relative_path_validation_rejects_unsafe_paths(relative_path):
+ with pytest.raises(BundleManifestError, match="unsafe relative path"):
+ validate_bundle_relative_path(relative_path)
+
+
+def test_source_verification_does_not_follow_symlinks(tmp_path, monkeypatch):
+ source = tmp_path / "source"
+ source_file = _write_file(source, "example.py", "content")
+ source_snapshot = collect_bundle_source_snapshot(source)
+
+ followed: list[Path] = []
+ real_stat = Path.stat
+
+ # Path.lstat() delegates to Path.stat(follow_symlinks=False), so only the
+ # following calls distinguish "verified the file itself" from "verified
its target".
+ def recording_stat(self, *, follow_symlinks=True):
+ if follow_symlinks:
+ followed.append(Path(self))
+ return real_stat(self, follow_symlinks=follow_symlinks)
+
+ monkeypatch.setattr(Path, "stat", recording_stat)
+ build_bundle_version_manifest(
+ bundle_name="manifest-local",
+ root=source,
+ backend_type="local",
+ source_snapshot=source_snapshot,
+ )
+
+ assert source_file not in followed
+
+
+def test_bundle_version_validation_names_the_source_in_the_message():
+ with pytest.raises(BundleManifestError, match="pinned bundle version"):
+ validate_bundle_version("sha256-bad", source="pinned bundle version")
+
+
[email protected](
+ ("value", "expected"),
+ [("a" * 64, True), ("0123456789abcdef" * 4, True), ("A" * 64, False), ("z"
* 64, False)],
+)
+def test_sha256_hex_recognition(value, expected):
+ assert is_sha256_hex(value) is expected
+
+
+def test_manifest_ignores_non_bytecode_files_under_pycache(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+ manifest_before = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ _write_file(source, "dags/__pycache__/notes.txt", "not bytecode")
+ manifest_after = _build_manifest(bundle_name="manifest-local",
root=source, backend_type="local")
+
+ assert manifest_after["version"] == manifest_before["version"]
+ assert [file_info["path"] for file_info in manifest_after["files"]] ==
["dags/example.py"]
+
+
+def
test_manifest_serialization_is_independent_of_key_insertion_order(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+ manifest = _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+ reordered = dict(reversed(list(manifest.items())))
+
+ assert list(reordered) != list(manifest)
+ assert serialize_bundle_version_manifest(reordered) ==
serialize_bundle_version_manifest(manifest)
+
+
+def test_source_snapshot_signature_covers_every_metadata_field(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "example.py", "content")
+ source_snapshot = collect_bundle_source_snapshot(source)
+
+ assert source_snapshot.signature.startswith("sha256:")
+ for field, value in [("size", 1), ("mtime_ns", 1), ("ctime_ns", 1),
("mode", 0o600)]:
+ changed = replace(source_snapshot.files[0], **{field: value})
+ assert changed.build_signature_record() !=
source_snapshot.files[0].build_signature_record()
+
+
[email protected](
+ ("field", "value"),
+ [("size", 999), ("mtime_ns", 1), ("ctime_ns", 1), ("mode", 0o600)],
+)
+def test_manifest_rejects_any_source_metadata_field_changing(tmp_path, field,
value):
+ source = tmp_path / "source"
+ _write_file(source, "example.py", "content")
+ source_snapshot = collect_bundle_source_snapshot(source)
+ changed_file = replace(source_snapshot.files[0], **{field: value})
+
+ with pytest.raises(BundleManifestSourceChangedError, match="changed while
building manifest"):
+ build_bundle_version_manifest(
+ bundle_name="manifest-local",
+ root=source,
+ backend_type="local",
+ source_snapshot=replace(source_snapshot, files=(changed_file,)),
+ )
+
+
+def
test_manifest_rejects_file_removed_between_hashing_and_verification(tmp_path):
+ source = tmp_path / "source"
+ source_file = _write_file(source, "example.py", "content")
+ source_snapshot = collect_bundle_source_snapshot(source)
+ digest = hashlib.sha256(b"content").hexdigest()
+ source_file.unlink()
+
+ # Precomputed hashes skip the read, so verification is the first access to
the file.
+ with pytest.raises(BundleManifestSourceChangedError, match="disappeared
while verifying"):
+ build_bundle_version_manifest(
+ bundle_name="manifest-local",
+ root=source,
+ backend_type="local",
+ source_snapshot=source_snapshot,
+ precomputed_file_sha256={"example.py": digest},
+ )
+
+
+def
test_backslash_in_a_posix_file_name_survives_publish_and_validation(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/weird\\name.py", "print('dag')")
+
+ manifest = _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+ recorded_path = manifest["files"][0]["path"]
+
+ assert recorded_path == "dags/weird\\name.py"
+ assert validate_bundle_relative_path(recorded_path) == Path(recorded_path)
+
+
+def test_manifest_records_backend_file_version_ids(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/a.py", "print('a')")
+ _write_file(source, "dags/b.py", "print('b')")
+
+ manifest = _build_manifest(
+ bundle_name="manifest-s3",
+ root=source,
+ backend_type="s3",
+ file_version_ids={"dags/a.py": "3HL4kqtJlcpXroDTDmJ+rm", "dags/b.py":
"nspShmM1rXMPGZ4qkCwTPw"},
+ )
+
+ assert [file_info["version_id"] for file_info in manifest["files"]] == [
+ "3HL4kqtJlcpXroDTDmJ+rm",
+ "nspShmM1rXMPGZ4qkCwTPw",
+ ]
+
+
+def test_manifest_omits_version_id_when_the_backend_supplies_none(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+
+ manifest = _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local")
+
+ assert "version_id" not in manifest["files"][0]
+
+
+def test_bundle_version_ignores_backend_file_version_ids(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+ kwargs = {"bundle_name": "manifest-s3", "root": source, "backend_type":
"s3"}
+
+ without = _build_manifest(**kwargs)
+ first = _build_manifest(**kwargs, file_version_ids={"dags/example.py":
"objver-one"})
+ second = _build_manifest(**kwargs, file_version_ids={"dags/example.py":
"objver-two"})
+
+ assert without["version"] == first["version"] == second["version"]
+ assert first["files"][0]["version_id"] != second["files"][0]["version_id"]
+
+
[email protected](
+ ("file_version_ids", "expected_message"),
+ [
+ ({}, "File version ids do not match"),
+ ({"other.py": "objver"}, "File version ids do not match"),
+ ({"dags/example.py": ""}, "File version id is invalid"),
+ ({"dags/example.py": 1}, "File version id is invalid"),
+ ],
+)
+def test_manifest_rejects_invalid_file_version_ids(tmp_path, file_version_ids,
expected_message):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+
+ with pytest.raises(BundleManifestError, match=expected_message):
+ _build_manifest(
+ bundle_name="manifest-s3",
+ root=source,
+ backend_type="s3",
+ file_version_ids=file_version_ids,
+ )
+
+
+def
test_manifest_rejects_bad_file_version_ids_before_reading_the_tree(tmp_path,
monkeypatch):
+ source = tmp_path / "source"
+ for index in range(3):
+ _write_file(source, f"dags/f{index}.py", "print('dag')")
+ hashed: list[str] = []
+
+ def recording_sha256(path):
+ hashed.append(path.name)
+ raise AssertionError("invalid file version ids must be rejected before
hashing")
+
+ monkeypatch.setattr(manifest_module, "compute_file_sha256",
recording_sha256)
+
+ with pytest.raises(BundleManifestError, match="File version id is invalid
for 'dags/f2.py'"):
+ _build_manifest(
+ bundle_name="manifest-s3",
+ root=source,
+ backend_type="s3",
+ file_version_ids={"dags/f0.py": "ok", "dags/f1.py": "ok",
"dags/f2.py": ""},
+ )
+
+ assert hashed == []
+
+
+def _published_manifest(tmp_path, **kwargs):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+ _write_file(source, "dags/nested/other.py", "print('other')")
+ return _build_manifest(bundle_name="manifest-local", root=source,
backend_type="local", **kwargs)
+
+
+def test_structure_validation_accepts_a_freshly_built_manifest(tmp_path):
+ manifest = _published_manifest(tmp_path)
+
+ validate_bundle_version_manifest_structure(
+ manifest,
+ expected_bundle_name="manifest-local",
+ expected_backend_type="local",
+ expected_version=manifest["version"],
+ )
+
+
+def test_structure_validation_accepts_entries_carrying_a_version_id(tmp_path):
+ source = tmp_path / "source"
+ _write_file(source, "dags/example.py", "print('dag')")
+ manifest = _build_manifest(
+ bundle_name="manifest-s3",
+ root=source,
+ backend_type="s3",
+ file_version_ids={"dags/example.py": "objver-1"},
+ )
+
+ validate_bundle_version_manifest_structure(manifest,
expected_version=None, expected_backend_type="s3")
+
+
+def
test_structure_validation_is_an_integrity_check_only_against_a_trusted_version(tmp_path):
+ manifest = _published_manifest(tmp_path)
+ trusted_version = manifest["version"]
+
+ edited = copy.deepcopy(manifest)
+ edited["files"][0]["sha256"] = "f" * 64
+ verify_bundle_version_manifest(
+ edited,
hashlib.sha256(serialize_bundle_version_manifest(edited)).hexdigest()
+ )
+ with pytest.raises(BundleManifestError, match="does not match its
contents"):
+ validate_bundle_version_manifest_structure(edited,
expected_version=None)
+
+ forged = copy.deepcopy(edited)
+ forged["version"] = compute_bundle_version(forged["files"])
+ validate_bundle_version_manifest_structure(forged, expected_version=None)
+ with pytest.raises(BundleManifestError, match="expected 'sha256-"):
+ validate_bundle_version_manifest_structure(forged,
expected_version=trusted_version)
+
+
+def test_structure_validation_rejects_entries_out_of_path_order(tmp_path):
+ manifest = _published_manifest(tmp_path)
+ manifest["files"].reverse()
+ manifest["version"] = compute_bundle_version(manifest["files"])
+
+ with pytest.raises(BundleManifestError, match="not sorted by path"):
+ validate_bundle_version_manifest_structure(manifest,
expected_version=None)
+
+
+def test_structure_validation_requires_an_explicit_expected_version(tmp_path):
+ manifest = _published_manifest(tmp_path)
+
+ with pytest.raises(TypeError, match="expected_version"):
+ validate_bundle_version_manifest_structure(manifest)
+
+
[email protected](
+ ("mutate", "expected_message"),
+ [
+ (lambda m: m.update({"schema_version": 2}), "unsupported
schema_version"),
+ (lambda m: m.update({"bundle_name": None}), "does not contain a bundle
name"),
+ (lambda m: m.update({"version": None}), "does not contain a version"),
+ (lambda m: m.update({"version": "sha256-nope"}), "valid sha256
version"),
+ (lambda m: m.update({"backend": "local"}), "does not contain a backend
type"),
+ (lambda m: m.update({"files": {}}), "files must be a list"),
+ (lambda m: m.update({"file_count": 99}), "file_count"),
+ (lambda m: m.update({"total_size": 99}), "total_size"),
+ (lambda m: m["files"].append(dict(m["files"][0])), "duplicate path"),
+ (lambda m: m["files"].__setitem__(0, "not-an-object"), "entries must
be objects"),
+ (lambda m: m["files"][0].update({"path": 1}), "does not contain a
path"),
+ (lambda m: m["files"][0].update({"path": "../evil.py"}), "unsafe
relative path"),
+ (lambda m: m["files"][0].update({"path": MANIFEST_FILE_NAME}), "must
not list"),
+ (lambda m: m["files"][0].update({"sha256": "nope"}), "valid sha256"),
+ (lambda m: m["files"][0].update({"size": -1}), "valid size"),
+ (lambda m: m["files"][0].update({"size": True}), "valid size"),
+ (lambda m: m["files"][0].update({"executable": "yes"}), "valid
executable flag"),
+ (lambda m: m["files"][0].update({"version_id": ""}), "valid version
id"),
+ ],
+)
+def test_structure_validation_rejects_malformed_manifests(tmp_path, mutate,
expected_message):
+ manifest = _published_manifest(tmp_path)
+ mutate(manifest)
+
+ with pytest.raises(BundleManifestError, match=expected_message):
+ validate_bundle_version_manifest_structure(manifest,
expected_version=None)
+
+
[email protected](
+ ("kwargs", "expected_message"),
+ [
+ ({"expected_bundle_name": "other"}, "expected 'other'"),
+ ({"expected_backend_type": "s3"}, "not for a s3 backend: got 'local'"),
+ ({"expected_version": "sha256-" + "a" * 64}, "expected 'sha256-"),
+ ],
+)
+def test_structure_validation_rejects_a_manifest_for_something_else(tmp_path,
kwargs, expected_message):
+ manifest = _published_manifest(tmp_path)
+
+ with pytest.raises(BundleManifestError, match=expected_message):
+ validate_bundle_version_manifest_structure(manifest,
**{"expected_version": None, **kwargs})
+
+
+def test_structure_validation_rejects_a_manifest_that_is_not_an_object():
+ with pytest.raises(BundleManifestError, match="must be an object"):
+ validate_bundle_version_manifest_structure(["not", "a", "manifest"],
expected_version=None)
+
+
[email protected](
+ "files",
+ [
+ [{"path": "a.py", "sha256": "0" * 64, "size": 1}],
+ [{"sha256": "0" * 64, "size": 1, "executable": False}],
+ ["not-an-object"],
+ [None],
+ ],
+)
+def test_bundle_version_rejects_incomplete_file_entries(files):
+ with pytest.raises(BundleManifestError, match="must each contain path,
sha256, size and executable"):
+ compute_bundle_version(files)
