19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

43.53

/src/python/pants/core/util_rules/archive.py

# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import logging
import os
import shlex
from dataclasses import dataclass
from pathlib import PurePath

from pants.core.util_rules import system_binaries
from pants.core.util_rules.adhoc_binaries import find_gunzip
from pants.core.util_rules.system_binaries import ArchiveFormat as ArchiveFormat
from pants.core.util_rules.system_binaries import (
    SystemBinariesSubsystem,
    find_tar,
    find_unzip,
    find_zip,
    get_bash,
)
from pants.engine.fs import (
    CreateDigest,
    Digest,
    Directory,
    FileContent,
    MergeDigests,
    RemovePrefix,
    Snapshot,
)
from pants.engine.intrinsics import create_digest, digest_to_snapshot, merge_digests, remove_prefix
from pants.engine.process import Process, execute_process_or_raise
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class CreateArchive:
    """A request to create an archive.

    All files in the input snapshot will be included in the resulting archive.
    """

    snapshot: Snapshot
    output_filename: str
    format: ArchiveFormat


@rule(desc="Creating an archive file", level=LogLevel.DEBUG)
async def create_archive(
    request: CreateArchive, system_binaries_environment: SystemBinariesSubsystem.EnvironmentAware
) -> Digest:
    # #16091 -- if an arg list is really long, archive utilities tend to get upset.
    # passing a list of filenames into the utilities fixes this.
    FILE_LIST_FILENAME = "__pants_archive_filelist__"
    file_list_file = FileContent(
        FILE_LIST_FILENAME, "\n".join(request.snapshot.files).encode("utf-8")
    )
    file_list_file_digest = await create_digest(CreateDigest([file_list_file]))
    files_digests = [file_list_file_digest, request.snapshot.digest]
    input_digests = []

    if request.format == ArchiveFormat.ZIP:
        zip_binary, bash_binary = await concurrently(
            find_zip(**implicitly()), get_bash(**implicitly())
        )
        env = {}
        argv: tuple[str, ...] = (
            bash_binary.path,
            "-c",
            # Note: The -A (--adjust-sfx) option causes zip to treat the given archive name as-is.
            # This works even when archive isn't created as a self-extracting archive
            #  see https://unix.stackexchange.com/a/557812
            softwrap(
                f"""
                {zip_binary.path} --adjust-sfx --names-stdin {shlex.quote(request.output_filename)}
                < {FILE_LIST_FILENAME}
                """
            ),
        )
    else:
        tar_binary = await find_tar(**implicitly())
        argv = tar_binary.create_archive_argv(
            request.output_filename,
            request.format,
            input_file_list_filename=FILE_LIST_FILENAME,
        )
        # `tar` expects to find a couple binaries like `gzip` and `xz` by looking on the PATH.
        env = {"PATH": os.pathsep.join(system_binaries_environment.system_binary_paths)}

        # `tar` requires that the output filename's parent directory exists, so if the caller
        # wants the output in a directory we explicitly create it here.
        # We have to guard this path as the Rust code will crash if we give it empty paths.
        output_dir = os.path.dirname(request.output_filename)
        if output_dir != "":
            output_dir_digest = await create_digest(CreateDigest([Directory(output_dir)]))
            input_digests.append(output_dir_digest)

    input_digest = await merge_digests(MergeDigests([*files_digests, *input_digests]))

    result = await execute_process_or_raise(
        **implicitly(
            Process(
                argv=argv,
                env=env,
                input_digest=input_digest,
                description=f"Create {request.output_filename}",
                level=LogLevel.DEBUG,
                output_files=(request.output_filename,),
            )
        )
    )
    return result.output_digest


@dataclass(frozen=True)
class MaybeExtractArchiveRequest:
    """A request to extract a single archive file (otherwise returns the input digest).

    :param digest: The digest of the archive to maybe extract. If the archive contains a single file
        which matches a known suffix, the `ExtractedArchive` will contain the extracted digest.
        Otherwise the `ExtractedArchive` will contain this digest.
    :param use_suffix: If provided, extracts the single file archive as if it had this suffix.
        Useful in situations where the file is archived then renamed.
        (E.g. A Python `.whl` is a renamed `.zip`, so the client should provide `".zip"`)
    """

    digest: Digest
    use_suffix: str | None = None


@dataclass(frozen=True)
class ExtractedArchive:
    """The result of extracting an archive."""

    digest: Digest


@rule
async def convert_digest_to_MaybeExtractArchiveRequest(
    digest: Digest,
) -> MaybeExtractArchiveRequest:
    """Backwards-compatibility helper."""
    return MaybeExtractArchiveRequest(digest)


@rule(desc="Extracting an archive file", level=LogLevel.DEBUG)
async def maybe_extract_archive(
    request: MaybeExtractArchiveRequest,
    system_binaries_environment: SystemBinariesSubsystem.EnvironmentAware,
) -> ExtractedArchive:
    """If digest contains a single archive file, extract it, otherwise return the input digest."""
    extract_archive_dir = "__extract_archive_dir"
    snapshot, output_dir_digest = await concurrently(
        digest_to_snapshot(request.digest),
        create_digest(CreateDigest([Directory(extract_archive_dir)])),
    )
    if len(snapshot.files) != 1:
        return ExtractedArchive(request.digest)

    archive_path = snapshot.files[0]
    archive_suffix = request.use_suffix or "".join(PurePath(archive_path).suffixes)
    is_zip = archive_suffix.endswith(".zip")
    is_tar = archive_suffix.endswith(
        (".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tbz2", ".tar.xz", ".txz", ".tar.lz4")
    )
    is_gz = not is_tar and archive_suffix.endswith(".gz")
    if not is_zip and not is_tar and not is_gz:
        return ExtractedArchive(request.digest)

    merge_digest_get = merge_digests(MergeDigests((request.digest, output_dir_digest)))
    env = {}
    append_only_caches: FrozenDict[str, str] = FrozenDict({})
    if is_zip:
        input_digest, unzip_binary = await concurrently(
            merge_digest_get, find_unzip(**implicitly())
        )
        argv = unzip_binary.extract_archive_argv(archive_path, extract_archive_dir)
    elif is_tar:
        input_digest, tar_binary = await concurrently(merge_digest_get, find_tar(**implicitly()))
        argv = tar_binary.extract_archive_argv(
            archive_path, extract_archive_dir, archive_suffix=archive_suffix
        )
        # `tar` expects to find a couple binaries like `gzip` and `xz` by looking on the PATH.
        env = {"PATH": os.pathsep.join(system_binaries_environment.system_binary_paths)}
    else:
        input_digest, gunzip = await concurrently(merge_digest_get, find_gunzip(**implicitly()))
        argv = gunzip.extract_archive_argv(archive_path, extract_archive_dir)
        append_only_caches = gunzip.python_binary.APPEND_ONLY_CACHES

    result = await execute_process_or_raise(
        **implicitly(
            Process(
                argv=argv,
                env=env,
                input_digest=input_digest,
                description=f"Extract {archive_path}",
                level=LogLevel.DEBUG,
                output_directories=(extract_archive_dir,),
                append_only_caches=append_only_caches,
            )
        )
    )
    resulting_digest = await remove_prefix(RemovePrefix(result.output_digest, extract_archive_dir))
    return ExtractedArchive(resulting_digest)


def rules():
    return (*collect_rules(), *system_binaries.rules())

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	1✔
5
6	import logging	1✔
7	import os	1✔
8	import shlex	1✔
9	from dataclasses import dataclass	1✔
10	from pathlib import PurePath	1✔
11
12	from pants.core.util_rules import system_binaries	1✔
13	from pants.core.util_rules.adhoc_binaries import find_gunzip	1✔
14	from pants.core.util_rules.system_binaries import ArchiveFormat as ArchiveFormat	1✔
15	from pants.core.util_rules.system_binaries import (	1✔
16	SystemBinariesSubsystem,
17	find_tar,
18	find_unzip,
19	find_zip,
20	get_bash,
21	)
22	from pants.engine.fs import (	1✔
23	CreateDigest,
24	Digest,
25	Directory,
26	FileContent,
27	MergeDigests,
28	RemovePrefix,
29	Snapshot,
30	)
31	from pants.engine.intrinsics import create_digest, digest_to_snapshot, merge_digests, remove_prefix	1✔
32	from pants.engine.process import Process, execute_process_or_raise	1✔
33	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	1✔
34	from pants.util.frozendict import FrozenDict	1✔
35	from pants.util.logging import LogLevel	1✔
36	from pants.util.strutil import softwrap	1✔
37
38	logger = logging.getLogger(__name__)	1✔
39
40
41	@dataclass(frozen=True)	1✔
42	class CreateArchive:	1✔
43	"""A request to create an archive.
44
45	All files in the input snapshot will be included in the resulting archive.
46	"""
47
48	snapshot: Snapshot	1✔
49	output_filename: str	1✔
50	format: ArchiveFormat	1✔
51
52
53	@rule(desc="Creating an archive file", level=LogLevel.DEBUG)	1✔
54	async def create_archive(	1✔
55	request: CreateArchive, system_binaries_environment: SystemBinariesSubsystem.EnvironmentAware
56	) -> Digest:
57	# #16091 -- if an arg list is really long, archive utilities tend to get upset.
58	# passing a list of filenames into the utilities fixes this.
59	FILE_LIST_FILENAME = "__pants_archive_filelist__"	×
60	file_list_file = FileContent(	×
61	FILE_LIST_FILENAME, "\n".join(request.snapshot.files).encode("utf-8")
62	)
63	file_list_file_digest = await create_digest(CreateDigest([file_list_file]))	×
64	files_digests = [file_list_file_digest, request.snapshot.digest]	×
65	input_digests = []	×
66
67	if request.format == ArchiveFormat.ZIP:	×
68	zip_binary, bash_binary = await concurrently(	×
69	find_zip(implicitly()), get_bash(implicitly())
70	)
71	env = {}	×
72	argv: tuple[str, ...] = (	×
73	bash_binary.path,
74	"-c",
75	# Note: The -A (--adjust-sfx) option causes zip to treat the given archive name as-is.
76	# This works even when archive isn't created as a self-extracting archive
77	# see https://unix.stackexchange.com/a/557812
78	softwrap(
79	f"""
80	{zip_binary.path} --adjust-sfx --names-stdin {shlex.quote(request.output_filename)}
81	< {FILE_LIST_FILENAME}
82	"""
83	),
84	)
85	else:
86	tar_binary = await find_tar(**implicitly())	×
87	argv = tar_binary.create_archive_argv(	×
88	request.output_filename,
89	request.format,
90	input_file_list_filename=FILE_LIST_FILENAME,
91	)
92	# `tar` expects to find a couple binaries like `gzip` and `xz` by looking on the PATH.
93	env = {"PATH": os.pathsep.join(system_binaries_environment.system_binary_paths)}	×
94
95	# `tar` requires that the output filename's parent directory exists, so if the caller
96	# wants the output in a directory we explicitly create it here.
97	# We have to guard this path as the Rust code will crash if we give it empty paths.
98	output_dir = os.path.dirname(request.output_filename)	×
99	if output_dir != "":	×
100	output_dir_digest = await create_digest(CreateDigest([Directory(output_dir)]))	×
101	input_digests.append(output_dir_digest)	×
102
103	input_digest = await merge_digests(MergeDigests([files_digests, input_digests]))	×
104
105	result = await execute_process_or_raise(	×
106	**implicitly(
107	Process(
108	argv=argv,
109	env=env,
110	input_digest=input_digest,
111	description=f"Create {request.output_filename}",
112	level=LogLevel.DEBUG,
113	output_files=(request.output_filename,),
114	)
115	)
116	)
117	return result.output_digest	×
118
119
120	@dataclass(frozen=True)	1✔
121	class MaybeExtractArchiveRequest:	1✔
122	"""A request to extract a single archive file (otherwise returns the input digest).
123
124	:param digest: The digest of the archive to maybe extract. If the archive contains a single file
125	which matches a known suffix, the `ExtractedArchive` will contain the extracted digest.
126	Otherwise the `ExtractedArchive` will contain this digest.
127	:param use_suffix: If provided, extracts the single file archive as if it had this suffix.
128	Useful in situations where the file is archived then renamed.
129	(E.g. A Python `.whl` is a renamed `.zip`, so the client should provide `".zip"`)
130	"""
131
132	digest: Digest	1✔
133	use_suffix: str \| None = None	1✔
134
135
136	@dataclass(frozen=True)	1✔
137	class ExtractedArchive:	1✔
138	"""The result of extracting an archive."""
139
140	digest: Digest	1✔
141
142
143	@rule	1✔
144	async def convert_digest_to_MaybeExtractArchiveRequest(	1✔
145	digest: Digest,
146	) -> MaybeExtractArchiveRequest:
147	"""Backwards-compatibility helper."""
148	return MaybeExtractArchiveRequest(digest)	×
149
150
151	@rule(desc="Extracting an archive file", level=LogLevel.DEBUG)	1✔
152	async def maybe_extract_archive(	1✔
153	request: MaybeExtractArchiveRequest,
154	system_binaries_environment: SystemBinariesSubsystem.EnvironmentAware,
155	) -> ExtractedArchive:
156	"""If digest contains a single archive file, extract it, otherwise return the input digest."""
157	extract_archive_dir = "__extract_archive_dir"	×
158	snapshot, output_dir_digest = await concurrently(	×
159	digest_to_snapshot(request.digest),
160	create_digest(CreateDigest([Directory(extract_archive_dir)])),
161	)
162	if len(snapshot.files) != 1:	×
163	return ExtractedArchive(request.digest)	×
164
165	archive_path = snapshot.files[0]	×
166	archive_suffix = request.use_suffix or "".join(PurePath(archive_path).suffixes)	×
167	is_zip = archive_suffix.endswith(".zip")	×
168	is_tar = archive_suffix.endswith(	×
169	(".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tbz2", ".tar.xz", ".txz", ".tar.lz4")
170	)
171	is_gz = not is_tar and archive_suffix.endswith(".gz")	×
172	if not is_zip and not is_tar and not is_gz:	×
173	return ExtractedArchive(request.digest)	×
174
175	merge_digest_get = merge_digests(MergeDigests((request.digest, output_dir_digest)))	×
176	env = {}	×
177	append_only_caches: FrozenDict[str, str] = FrozenDict({})	×
178	if is_zip:	×
179	input_digest, unzip_binary = await concurrently(	×
180	merge_digest_get, find_unzip(**implicitly())
181	)
182	argv = unzip_binary.extract_archive_argv(archive_path, extract_archive_dir)	×
183	elif is_tar:	×
184	input_digest, tar_binary = await concurrently(merge_digest_get, find_tar(**implicitly()))	×
185	argv = tar_binary.extract_archive_argv(	×
186	archive_path, extract_archive_dir, archive_suffix=archive_suffix
187	)
188	# `tar` expects to find a couple binaries like `gzip` and `xz` by looking on the PATH.
189	env = {"PATH": os.pathsep.join(system_binaries_environment.system_binary_paths)}	×
190	else:
191	input_digest, gunzip = await concurrently(merge_digest_get, find_gunzip(**implicitly()))	×
192	argv = gunzip.extract_archive_argv(archive_path, extract_archive_dir)	×
193	append_only_caches = gunzip.python_binary.APPEND_ONLY_CACHES	×
194
195	result = await execute_process_or_raise(	×
196	**implicitly(
197	Process(
198	argv=argv,
199	env=env,
200	input_digest=input_digest,
201	description=f"Extract {archive_path}",
202	level=LogLevel.DEBUG,
203	output_directories=(extract_archive_dir,),
204	append_only_caches=append_only_caches,
205	)
206	)
207	)
208	resulting_digest = await remove_prefix(RemovePrefix(result.output_digest, extract_archive_dir))	×
209	return ExtractedArchive(resulting_digest)	×
210
211
212	def rules():	1✔
UNCOV 213	return (collect_rules(), system_binaries.rules())	×

pantsbuild / pants / 19015773527

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous