|
| 1 | +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. |
| 2 | +# SPDX-License-Identifier: Apache-2.0 |
| 3 | + |
| 4 | +"""Recognize one temporary Pixi 0.73.0 source-pruning defect, failing closed.""" |
| 5 | + |
| 6 | +from __future__ import annotations |
| 7 | + |
| 8 | +import argparse |
| 9 | +import re |
| 10 | +import subprocess |
| 11 | +import sys |
| 12 | +from pathlib import Path |
| 13 | + |
| 14 | +import tomllib |
| 15 | + |
| 16 | +LOCAL_PACKAGES = { |
| 17 | + "cuda-bindings": "../cuda_bindings", |
| 18 | + "cuda-core": ".", |
| 19 | + "cuda-pathfinder": "../cuda_pathfinder", |
| 20 | +} |
| 21 | +PLATFORMS = {"p1": "linux-64", "p2": "linux-aarch64", "p3": "win-64"} |
| 22 | +SOURCE_REFERENCE = re.compile( |
| 23 | + r" - conda_source: (cuda-bindings|cuda-core|cuda-pathfinder)\[([0-9a-f]{8})\] @ (\S+)\n" |
| 24 | +) |
| 25 | + |
| 26 | + |
| 27 | +def is_known_samples_source_pruning(original: str, repaired: str) -> bool: |
| 28 | + """Accept only removal of the nine known samples references, with nothing else changed. |
| 29 | +
|
| 30 | + This deliberately understands only the pinned version's exact serialization. |
| 31 | + Source record definitions remain checked byte-for-byte along with the rest of |
| 32 | + the lockfile; missing definitions and the reverse (addition) cycle are rejected. |
| 33 | + """ |
| 34 | + lines = original.splitlines(keepends=True) |
| 35 | + if not lines or lines[0] != "version: 7\n" or lines.count("environments:\n") != 1: |
| 36 | + return False |
| 37 | + for alias, platform in PLATFORMS.items(): |
| 38 | + if original.count(f"- name: {alias}\n subdir: {platform}\n") != 1: |
| 39 | + return False |
| 40 | + |
| 41 | + environment_start = lines.index("environments:\n") + 1 |
| 42 | + environment_end = next( |
| 43 | + (index for index in range(environment_start, len(lines)) if not lines[index].startswith(" ")), |
| 44 | + len(lines), |
| 45 | + ) |
| 46 | + sample_headers = [index for index in range(environment_start, environment_end) if lines[index] == " samples:\n"] |
| 47 | + if len(sample_headers) != 1: |
| 48 | + return False |
| 49 | + samples_start = sample_headers[0] + 1 |
| 50 | + samples_end = next( |
| 51 | + (index for index in range(samples_start, environment_end) if not lines[index].startswith(" ")), |
| 52 | + environment_end, |
| 53 | + ) |
| 54 | + if any(lines[samples_start:samples_end].count(f" {alias}:\n") != 1 for alias in PLATFORMS): |
| 55 | + return False |
| 56 | + |
| 57 | + platform = None |
| 58 | + seen: set[tuple[str, str]] = set() |
| 59 | + removed: set[int] = set() |
| 60 | + for index in range(samples_start, samples_end): |
| 61 | + line = lines[index] |
| 62 | + platform_match = re.fullmatch(r" ([^\s:]+):\n", line) |
| 63 | + if platform_match: |
| 64 | + platform = platform_match[1] if platform_match[1] in PLATFORMS else None |
| 65 | + reference = SOURCE_REFERENCE.fullmatch(line) |
| 66 | + if reference is None: |
| 67 | + continue |
| 68 | + name, identifier, path = reference.groups() |
| 69 | + if platform is None or path != LOCAL_PACKAGES[name] or (platform, name) in seen: |
| 70 | + return False |
| 71 | + definition = f"- conda_source: {name}[{identifier}] @ {path}\n" |
| 72 | + if lines[environment_end:].count(definition) != 1: |
| 73 | + return False |
| 74 | + seen.add((platform, name)) |
| 75 | + removed.add(index) |
| 76 | + |
| 77 | + expected = {(platform, name) for platform in PLATFORMS for name in LOCAL_PACKAGES} |
| 78 | + return seen == expected and "".join(line for index, line in enumerate(lines) if index not in removed) == repaired |
| 79 | + |
| 80 | + |
| 81 | +def _manifest_matches(manifest: dict) -> bool: |
| 82 | + features = manifest.get("feature", {}) |
| 83 | + if not isinstance(features, dict): |
| 84 | + return False |
| 85 | + samples = features.get("samples", {}) |
| 86 | + local_deps = features.get("local-deps", {}) |
| 87 | + environments = manifest.get("environments", {}) |
| 88 | + if not all(isinstance(table, dict) for table in (samples, local_deps, environments)): |
| 89 | + return False |
| 90 | + options = samples.get("pypi-options", {}) |
| 91 | + sample_dependencies = samples.get("dependencies", {}) |
| 92 | + local_dependencies = local_deps.get("dependencies", {}) |
| 93 | + if not all(isinstance(table, dict) for table in (options, sample_dependencies, local_dependencies)): |
| 94 | + return False |
| 95 | + overrides = options.get("dependency-overrides", {}) |
| 96 | + if overrides != {name: {"version": "*", "env-markers": "python_version < '0'"} for name in LOCAL_PACKAGES}: |
| 97 | + return False |
| 98 | + if sample_dependencies.get("cuda-core") != {"path": "."}: |
| 99 | + return False |
| 100 | + if any( |
| 101 | + local_dependencies.get(name) != {"path": path} for name, path in LOCAL_PACKAGES.items() if name != "cuda-core" |
| 102 | + ): |
| 103 | + return False |
| 104 | + environment = environments.get("samples", {}) |
| 105 | + return isinstance(environment, dict) and environment.get("features") == ["cu13", "samples", "local-deps"] |
| 106 | + |
| 107 | + |
| 108 | +def main(argv: list[str] | None = None) -> int: |
| 109 | + """Exit 0 for the known defect, 1 otherwise, and 2 for an operational error.""" |
| 110 | + parser = argparse.ArgumentParser(description=__doc__) |
| 111 | + parser.add_argument("--original-blob", required=True) |
| 112 | + parser.add_argument("--lockfile", type=Path, required=True) |
| 113 | + parser.add_argument("--manifest-path", type=Path, required=True) |
| 114 | + parser.add_argument("--pixi-version", required=True) |
| 115 | + parser.add_argument("--check-status", type=int, required=True) |
| 116 | + args = parser.parse_args(argv) |
| 117 | + if args.pixi_version.removeprefix("v") != "0.73.0" or args.check_status != 1: |
| 118 | + return 1 |
| 119 | + if re.fullmatch(r"[0-9a-f]{40}", args.original_blob) is None: |
| 120 | + print("Source-pruning check: original blob must be a full Git SHA-1.", file=sys.stderr) |
| 121 | + return 2 |
| 122 | + try: |
| 123 | + original = subprocess.run( # noqa: S603 - validated SHA, argv passed without a shell. |
| 124 | + ["git", "cat-file", "blob", args.original_blob], # noqa: S607 |
| 125 | + check=True, |
| 126 | + capture_output=True, |
| 127 | + ).stdout.decode("utf-8") |
| 128 | + repaired = args.lockfile.read_bytes().decode("utf-8") |
| 129 | + manifest = tomllib.loads(args.manifest_path.read_bytes().decode("utf-8")) |
| 130 | + except (OSError, UnicodeError, tomllib.TOMLDecodeError, subprocess.CalledProcessError) as error: |
| 131 | + print(f"Source-pruning check could not read its inputs: {error}", file=sys.stderr) |
| 132 | + return 2 |
| 133 | + if not _manifest_matches(manifest): |
| 134 | + return 1 |
| 135 | + return 0 if is_known_samples_source_pruning(original, repaired) else 1 |
| 136 | + |
| 137 | + |
| 138 | +if __name__ == "__main__": |
| 139 | + sys.exit(main()) |
0 commit comments