18730153336

Committed 22 Oct 2025 09:17PM UTC coverage: 80.279% (+0.02%) from 80.262%

Build # 18730153336

Build Type

Pull #22756

github

Committed by

web-flow

Commit Message

Merge c2314d839 into eff55710f

Pull Request Pull Request #22756: TypeScript and TSX test support

Run Details

18 of 18 new or added lines in 2 files covered. (100.0%)

65 existing lines in 5 files now uncovered.

77875 of 97005 relevant lines covered (80.28%)

3.08 hits per line

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

52.0

/src/python/pants/backend/python/goals/lockfile.py

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

from __future__ import annotations

import itertools
import os.path
from collections import defaultdict
from dataclasses import dataclass
from operator import itemgetter

from pants.backend.python.subsystems.python_tool_base import PythonToolBase
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.target_types import (
    PythonRequirementFindLinksField,
    PythonRequirementResolveField,
    PythonRequirementsField,
)
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.lockfile_diff import _generate_python_lockfile_diff
from pants.backend.python.util_rules.lockfile_metadata import PythonLockfileMetadata
from pants.backend.python.util_rules.pex import find_interpreter
from pants.backend.python.util_rules.pex_cli import PexCliProcess, maybe_log_pex_stderr
from pants.backend.python.util_rules.pex_environment import PexSubsystem
from pants.backend.python.util_rules.pex_requirements import (
    PexRequirements,
    ResolvePexConfig,
    ResolvePexConfigRequest,
    determine_resolve_pex_config,
)
from pants.core.goals.generate_lockfiles import (
    DEFAULT_TOOL_LOCKFILE,
    GenerateLockfile,
    GenerateLockfileResult,
    GenerateLockfilesSubsystem,
    KnownUserResolveNames,
    KnownUserResolveNamesRequest,
    RequestedUserResolveNames,
    UserGenerateLockfiles,
    WrappedGenerateLockfile,
)
from pants.core.goals.resolves import ExportableTool
from pants.core.util_rules.lockfile_metadata import calculate_invalidation_digest
from pants.engine.fs import CreateDigest, Digest, FileContent, MergeDigests
from pants.engine.internals.synthetic_targets import SyntheticAddressMaps, SyntheticTargetsRequest
from pants.engine.internals.target_adaptor import TargetAdaptor
from pants.engine.intrinsics import create_digest, get_digest_contents, merge_digests
from pants.engine.process import ProcessCacheScope, fallible_to_exec_result_or_raise
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import AllTargets
from pants.engine.unions import UnionMembership, UnionRule
from pants.option.subsystem import _construct_subsystem
from pants.util.docutil import bin_name
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet
from pants.util.pip_requirement import PipRequirement


@dataclass(frozen=True)
class GeneratePythonLockfile(GenerateLockfile):
    requirements: FrozenOrderedSet[str]
    find_links: FrozenOrderedSet[str]
    interpreter_constraints: InterpreterConstraints

    @property
    def requirements_hex_digest(self) -> str:
        """Produces a hex digest of the requirements input for this lockfile."""
        return calculate_invalidation_digest(self.requirements)


@rule
async def wrap_python_lockfile_request(request: GeneratePythonLockfile) -> WrappedGenerateLockfile:
    return WrappedGenerateLockfile(request)


@dataclass(frozen=True)
class _PipArgsAndConstraintsSetup:
    resolve_config: ResolvePexConfig
    args: tuple[str, ...]
    digest: Digest


async def _setup_pip_args_and_constraints_file(resolve_name: str) -> _PipArgsAndConstraintsSetup:
    resolve_config = await determine_resolve_pex_config(
        ResolvePexConfigRequest(resolve_name), **implicitly()
    )

    args = list(resolve_config.pex_args())
    digests: list[Digest] = []

    if resolve_config.constraints_file:
        args.append(f"--constraints={resolve_config.constraints_file.path}")
        digests.append(resolve_config.constraints_file.digest)

    input_digest = await merge_digests(MergeDigests(digests))
    return _PipArgsAndConstraintsSetup(resolve_config, tuple(args), input_digest)


@rule(desc="Generate Python lockfile", level=LogLevel.DEBUG)
async def generate_lockfile(
    req: GeneratePythonLockfile,
    generate_lockfiles_subsystem: GenerateLockfilesSubsystem,
    python_setup: PythonSetup,
    pex_subsystem: PexSubsystem,
) -> GenerateLockfileResult:
    if not req.requirements:
        raise ValueError(
            f"Cannot generate lockfile with no requirements. Please add some requirements to {req.resolve_name}."
        )

    pip_args_setup = await _setup_pip_args_and_constraints_file(req.resolve_name)
    header_delimiter = "//"

    python = await find_interpreter(req.interpreter_constraints, **implicitly())

    result = await fallible_to_exec_result_or_raise(
        **implicitly(
            PexCliProcess(
                subcommand=("lock", "create"),
                extra_args=(
                    f"--output={req.lockfile_dest}",
                    # See https://github.com/pantsbuild/pants/issues/12458. For now, we always
                    # generate universal locks because they have the best compatibility. We may
                    # want to let users change this, as `style=strict` is safer.
                    "--style=universal",
                    "--pip-version",
                    python_setup.pip_version,
                    "--resolver-version",
                    "pip-2020-resolver",
                    "--preserve-pip-download-log",
                    "pex-pip-download.log",
                    # PEX files currently only run on Linux and Mac machines; so we hard code this
                    # limit on lock universality to avoid issues locking due to irrelevant
                    # Windows-only dependency issues. See this Pex issue that originated from a
                    # Pants user issue presented in Slack:
                    #   https://github.com/pex-tool/pex/issues/1821
                    #
                    # At some point it will probably make sense to expose `--target-system` for
                    # configuration.
                    "--target-system",
                    "linux",
                    "--target-system",
                    "mac",
                    # This makes diffs more readable when lockfiles change.
                    "--indent=2",
                    f"--python-path={python.path}",
                    *(f"--find-links={link}" for link in req.find_links),
                    *pip_args_setup.args,
                    *req.interpreter_constraints.generate_pex_arg_list(),
                    *req.requirements,
                ),
                additional_input_digest=pip_args_setup.digest,
                output_files=(req.lockfile_dest,),
                description=f"Generate lockfile for {req.resolve_name}",
                # Instead of caching lockfile generation with LMDB, we instead use the invalidation
                # scheme from `lockfile_metadata.py` to check for stale/invalid lockfiles. This is
                # necessary so that our invalidation is resilient to deleting LMDB or running on a
                # new machine.
                #
                # We disable caching with LMDB so that when you generate a lockfile, you always get
                # the most up-to-date snapshot of the world. This is generally desirable and also
                # necessary to avoid an awkward edge case where different developers generate
                # different lockfiles even when generating at the same time. See
                # https://github.com/pantsbuild/pants/issues/12591.
                cache_scope=ProcessCacheScope.PER_SESSION,
            )
        )
    )
    maybe_log_pex_stderr(result.stderr, pex_subsystem.verbosity)

    metadata = PythonLockfileMetadata.new(
        valid_for_interpreter_constraints=req.interpreter_constraints,
        requirements={
            PipRequirement.parse(
                i,
                description_of_origin=f"the lockfile {req.lockfile_dest} for the resolve {req.resolve_name}",
            )
            for i in req.requirements
        },
        manylinux=pip_args_setup.resolve_config.manylinux,
        requirement_constraints=(
            set(pip_args_setup.resolve_config.constraints_file.constraints)
            if pip_args_setup.resolve_config.constraints_file
            else set()
        ),
        only_binary=set(pip_args_setup.resolve_config.only_binary),
        no_binary=set(pip_args_setup.resolve_config.no_binary),
        excludes=set(pip_args_setup.resolve_config.excludes),
        overrides=set(pip_args_setup.resolve_config.overrides),
        sources=set(pip_args_setup.resolve_config.sources),
    )
    regenerate_command = (
        generate_lockfiles_subsystem.custom_command
        or f"{bin_name()} generate-lockfiles --resolve={req.resolve_name}"
    )
    if python_setup.separate_lockfile_metadata_file:
        descr = f"This lockfile was generated by Pants. To regenerate, run: {regenerate_command}"
        metadata_digest = await create_digest(
            CreateDigest(
                [
                    FileContent(
                        PythonLockfileMetadata.metadata_location_for_lockfile(req.lockfile_dest),
                        metadata.to_json(with_description=descr).encode(),
                    ),
                ]
            )
        )
        final_lockfile_digest = await merge_digests(
            MergeDigests([metadata_digest, result.output_digest])
        )
    else:
        initial_lockfile_digest_contents = await get_digest_contents(result.output_digest)
        lockfile_with_header = metadata.add_header_to_lockfile(
            initial_lockfile_digest_contents[0].content,
            regenerate_command=regenerate_command,
            delimeter=header_delimiter,
        )
        final_lockfile_digest = await create_digest(
            CreateDigest(
                [
                    FileContent(req.lockfile_dest, lockfile_with_header),
                ]
            )
        )

    if req.diff:
        diff = await _generate_python_lockfile_diff(
            final_lockfile_digest, req.resolve_name, req.lockfile_dest
        )
    else:
        diff = None

    return GenerateLockfileResult(final_lockfile_digest, req.resolve_name, req.lockfile_dest, diff)


class RequestedPythonUserResolveNames(RequestedUserResolveNames):
    pass


class KnownPythonUserResolveNamesRequest(KnownUserResolveNamesRequest):
    pass


@rule
async def determine_python_user_resolves(
    _: KnownPythonUserResolveNamesRequest,
    python_setup: PythonSetup,
    union_membership: UnionMembership,
) -> KnownUserResolveNames:
    """Find all know Python resolves, from both user-created resolves and internal tools."""
    python_tool_resolves = ExportableTool.filter_for_subclasses(union_membership, PythonToolBase)

    tools_using_default_resolve = [
        resolve_name
        for resolve_name, subsystem_cls in python_tool_resolves.items()
        if (await _construct_subsystem(subsystem_cls)).install_from_resolve is None
    ]

    return KnownUserResolveNames(
        names=(
            *python_setup.resolves.keys(),
            *tools_using_default_resolve,
        ),  # the order of the keys doesn't matter since shadowing is done in `setup_user_lockfile_requests`
        option_name="[python].resolves",
        requested_resolve_names_cls=RequestedPythonUserResolveNames,
    )


@rule
async def setup_user_lockfile_requests(
    requested: RequestedPythonUserResolveNames,
    all_targets: AllTargets,
    python_setup: PythonSetup,
    union_membership: UnionMembership,
) -> UserGenerateLockfiles:
    """Transform the names of resolves requested into the `GeneratePythonLockfile` request object.

    Shadowing is done here by only checking internal resolves if the resolve is not a user-created
    resolve.
    """
    if not (python_setup.enable_resolves and python_setup.resolves_generate_lockfiles):
        return UserGenerateLockfiles()

    resolve_to_requirements_fields = defaultdict(set)
    find_links: set[str] = set()
    for tgt in all_targets:
        if not tgt.has_fields((PythonRequirementResolveField, PythonRequirementsField)):
            continue
        resolve = tgt[PythonRequirementResolveField].normalized_value(python_setup)
        resolve_to_requirements_fields[resolve].add(tgt[PythonRequirementsField])
        find_links.update(tgt[PythonRequirementFindLinksField].value or ())

    tools = ExportableTool.filter_for_subclasses(union_membership, PythonToolBase)

    out = set()
    for resolve in requested:
        if resolve in python_setup.resolves:
            out.add(
                GeneratePythonLockfile(
                    requirements=PexRequirements.req_strings_from_requirement_fields(
                        resolve_to_requirements_fields[resolve]
                    ),
                    find_links=FrozenOrderedSet(find_links),
                    interpreter_constraints=InterpreterConstraints(
                        python_setup.resolves_to_interpreter_constraints.get(
                            resolve, python_setup.interpreter_constraints
                        )
                    ),
                    resolve_name=resolve,
                    lockfile_dest=python_setup.resolves[resolve],
                    diff=False,
                )
            )
        else:
            tool_cls: type[PythonToolBase] = tools[resolve]
            tool = await _construct_subsystem(tool_cls)

            # TODO: we shouldn't be managing default ICs in lockfile identification.
            #   We should find a better place to do this or a better way to default
            if tool.register_interpreter_constraints:
                ic = tool.interpreter_constraints
            else:
                ic = InterpreterConstraints(tool.default_interpreter_constraints)

            out.add(
                GeneratePythonLockfile(
                    requirements=FrozenOrderedSet(sorted(tool.requirements)),
                    find_links=FrozenOrderedSet(find_links),
                    interpreter_constraints=ic,
                    resolve_name=resolve,
                    lockfile_dest=DEFAULT_TOOL_LOCKFILE,
                    diff=False,
                )
            )

    return UserGenerateLockfiles(out)


@dataclass(frozen=True)
class PythonSyntheticLockfileTargetsRequest(SyntheticTargetsRequest):
    """Register the type used to create synthetic targets for Python lockfiles.

    As the paths for all lockfiles are known up-front, we set the `path` field to
    `SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS` so that we get a single request for all
    our synthetic targets rather than one request per directory.
    """

    path: str = SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS


def synthetic_lockfile_target_name(resolve: str) -> str:
    return f"_{resolve}_lockfile"


@rule
async def python_lockfile_synthetic_targets(
    request: PythonSyntheticLockfileTargetsRequest,
    python_setup: PythonSetup,
) -> SyntheticAddressMaps:
    if not python_setup.enable_synthetic_lockfiles:
        return SyntheticAddressMaps()

    resolves = [
        (os.path.dirname(lockfile), os.path.basename(lockfile), name)
        for name, lockfile in python_setup.resolves.items()
    ]

    return SyntheticAddressMaps.for_targets_request(
        request,
        [
            (
                os.path.join(spec_path, "BUILD.python-lockfiles"),
                tuple(
                    TargetAdaptor(
                        "_lockfiles",
                        name=synthetic_lockfile_target_name(name),
                        sources=(lockfile,),
                        __description_of_origin__=f"the [python].resolves option {name!r}",
                    )
                    for _, lockfile, name in lockfiles
                ),
            )
            for spec_path, lockfiles in itertools.groupby(sorted(resolves), key=itemgetter(0))
        ],
    )


def rules():
    return (
        *collect_rules(),
        UnionRule(GenerateLockfile, GeneratePythonLockfile),
        UnionRule(KnownUserResolveNamesRequest, KnownPythonUserResolveNamesRequest),
        UnionRule(RequestedUserResolveNames, RequestedPythonUserResolveNames),
        UnionRule(SyntheticTargetsRequest, PythonSyntheticLockfileTargetsRequest),
    )

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	11✔
5
6	import itertools	11✔
7	import os.path	11✔
8	from collections import defaultdict	11✔
9	from dataclasses import dataclass	11✔
10	from operator import itemgetter	11✔
11
12	from pants.backend.python.subsystems.python_tool_base import PythonToolBase	11✔
13	from pants.backend.python.subsystems.setup import PythonSetup	11✔
14	from pants.backend.python.target_types import (	11✔
15	PythonRequirementFindLinksField,
16	PythonRequirementResolveField,
17	PythonRequirementsField,
18	)
19	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	11✔
20	from pants.backend.python.util_rules.lockfile_diff import _generate_python_lockfile_diff	11✔
21	from pants.backend.python.util_rules.lockfile_metadata import PythonLockfileMetadata	11✔
22	from pants.backend.python.util_rules.pex import find_interpreter	11✔
23	from pants.backend.python.util_rules.pex_cli import PexCliProcess, maybe_log_pex_stderr	11✔
24	from pants.backend.python.util_rules.pex_environment import PexSubsystem	11✔
25	from pants.backend.python.util_rules.pex_requirements import (	11✔
26	PexRequirements,
27	ResolvePexConfig,
28	ResolvePexConfigRequest,
29	determine_resolve_pex_config,
30	)
31	from pants.core.goals.generate_lockfiles import (	11✔
32	DEFAULT_TOOL_LOCKFILE,
33	GenerateLockfile,
34	GenerateLockfileResult,
35	GenerateLockfilesSubsystem,
36	KnownUserResolveNames,
37	KnownUserResolveNamesRequest,
38	RequestedUserResolveNames,
39	UserGenerateLockfiles,
40	WrappedGenerateLockfile,
41	)
42	from pants.core.goals.resolves import ExportableTool	11✔
43	from pants.core.util_rules.lockfile_metadata import calculate_invalidation_digest	11✔
44	from pants.engine.fs import CreateDigest, Digest, FileContent, MergeDigests	11✔
45	from pants.engine.internals.synthetic_targets import SyntheticAddressMaps, SyntheticTargetsRequest	11✔
46	from pants.engine.internals.target_adaptor import TargetAdaptor	11✔
47	from pants.engine.intrinsics import create_digest, get_digest_contents, merge_digests	11✔
48	from pants.engine.process import ProcessCacheScope, fallible_to_exec_result_or_raise	11✔
49	from pants.engine.rules import collect_rules, implicitly, rule	11✔
50	from pants.engine.target import AllTargets	11✔
51	from pants.engine.unions import UnionMembership, UnionRule	11✔
52	from pants.option.subsystem import _construct_subsystem	11✔
53	from pants.util.docutil import bin_name	11✔
54	from pants.util.logging import LogLevel	11✔
55	from pants.util.ordered_set import FrozenOrderedSet	11✔
56	from pants.util.pip_requirement import PipRequirement	11✔
57
58
59	@dataclass(frozen=True)	11✔
60	class GeneratePythonLockfile(GenerateLockfile):	11✔
61	requirements: FrozenOrderedSet[str]	11✔
62	find_links: FrozenOrderedSet[str]	11✔
63	interpreter_constraints: InterpreterConstraints	11✔
64
65	@property	11✔
66	def requirements_hex_digest(self) -> str:	11✔
67	"""Produces a hex digest of the requirements input for this lockfile."""
68	return calculate_invalidation_digest(self.requirements)	×
69
70
71	@rule	11✔
72	async def wrap_python_lockfile_request(request: GeneratePythonLockfile) -> WrappedGenerateLockfile:	11✔
73	return WrappedGenerateLockfile(request)	×
74
75
76	@dataclass(frozen=True)	11✔
77	class _PipArgsAndConstraintsSetup:	11✔
78	resolve_config: ResolvePexConfig	11✔
79	args: tuple[str, ...]	11✔
80	digest: Digest	11✔
81
82
83	async def _setup_pip_args_and_constraints_file(resolve_name: str) -> _PipArgsAndConstraintsSetup:	11✔
84	resolve_config = await determine_resolve_pex_config(	×
85	ResolvePexConfigRequest(resolve_name), **implicitly()
86	)
87
88	args = list(resolve_config.pex_args())	×
89	digests: list[Digest] = []	×
90
91	if resolve_config.constraints_file:	×
92	args.append(f"--constraints={resolve_config.constraints_file.path}")	×
93	digests.append(resolve_config.constraints_file.digest)	×
94
95	input_digest = await merge_digests(MergeDigests(digests))	×
96	return _PipArgsAndConstraintsSetup(resolve_config, tuple(args), input_digest)	×
97
98
99	@rule(desc="Generate Python lockfile", level=LogLevel.DEBUG)	11✔
100	async def generate_lockfile(	11✔
101	req: GeneratePythonLockfile,
102	generate_lockfiles_subsystem: GenerateLockfilesSubsystem,
103	python_setup: PythonSetup,
104	pex_subsystem: PexSubsystem,
105	) -> GenerateLockfileResult:
106	if not req.requirements:	×
107	raise ValueError(	×
108	f"Cannot generate lockfile with no requirements. Please add some requirements to {req.resolve_name}."
109	)
110
111	pip_args_setup = await _setup_pip_args_and_constraints_file(req.resolve_name)	×
112	header_delimiter = "//"	×
113
114	python = await find_interpreter(req.interpreter_constraints, **implicitly())	×
115
116	result = await fallible_to_exec_result_or_raise(	×
117	**implicitly(
118	PexCliProcess(
119	subcommand=("lock", "create"),
120	extra_args=(
121	f"--output={req.lockfile_dest}",
122	# See https://github.com/pantsbuild/pants/issues/12458. For now, we always
123	# generate universal locks because they have the best compatibility. We may
124	# want to let users change this, as `style=strict` is safer.
125	"--style=universal",
126	"--pip-version",
127	python_setup.pip_version,
128	"--resolver-version",
129	"pip-2020-resolver",
130	"--preserve-pip-download-log",
131	"pex-pip-download.log",
132	# PEX files currently only run on Linux and Mac machines; so we hard code this
133	# limit on lock universality to avoid issues locking due to irrelevant
134	# Windows-only dependency issues. See this Pex issue that originated from a
135	# Pants user issue presented in Slack:
136	# https://github.com/pex-tool/pex/issues/1821
137	#
138	# At some point it will probably make sense to expose `--target-system` for
139	# configuration.
140	"--target-system",
141	"linux",
142	"--target-system",
143	"mac",
144	# This makes diffs more readable when lockfiles change.
145	"--indent=2",
146	f"--python-path={python.path}",
147	*(f"--find-links={link}" for link in req.find_links),
148	*pip_args_setup.args,
149	*req.interpreter_constraints.generate_pex_arg_list(),
150	*req.requirements,
151	),
152	additional_input_digest=pip_args_setup.digest,
153	output_files=(req.lockfile_dest,),
154	description=f"Generate lockfile for {req.resolve_name}",
155	# Instead of caching lockfile generation with LMDB, we instead use the invalidation
156	# scheme from `lockfile_metadata.py` to check for stale/invalid lockfiles. This is
157	# necessary so that our invalidation is resilient to deleting LMDB or running on a
158	# new machine.
159	#
160	# We disable caching with LMDB so that when you generate a lockfile, you always get
161	# the most up-to-date snapshot of the world. This is generally desirable and also
162	# necessary to avoid an awkward edge case where different developers generate
163	# different lockfiles even when generating at the same time. See
164	# https://github.com/pantsbuild/pants/issues/12591.
165	cache_scope=ProcessCacheScope.PER_SESSION,
166	)
167	)
168	)
169	maybe_log_pex_stderr(result.stderr, pex_subsystem.verbosity)	×
170
171	metadata = PythonLockfileMetadata.new(	×
172	valid_for_interpreter_constraints=req.interpreter_constraints,
173	requirements={
174	PipRequirement.parse(
175	i,
176	description_of_origin=f"the lockfile {req.lockfile_dest} for the resolve {req.resolve_name}",
177	)
178	for i in req.requirements
179	},
180	manylinux=pip_args_setup.resolve_config.manylinux,
181	requirement_constraints=(
182	set(pip_args_setup.resolve_config.constraints_file.constraints)
183	if pip_args_setup.resolve_config.constraints_file
184	else set()
185	),
186	only_binary=set(pip_args_setup.resolve_config.only_binary),
187	no_binary=set(pip_args_setup.resolve_config.no_binary),
188	excludes=set(pip_args_setup.resolve_config.excludes),
189	overrides=set(pip_args_setup.resolve_config.overrides),
190	sources=set(pip_args_setup.resolve_config.sources),
191	)
UNCOV 192	regenerate_command = (	×
193	generate_lockfiles_subsystem.custom_command
194	or f"{bin_name()} generate-lockfiles --resolve={req.resolve_name}"
195	)
196	if python_setup.separate_lockfile_metadata_file:	×
197	descr = f"This lockfile was generated by Pants. To regenerate, run: {regenerate_command}"	×
UNCOV 198	metadata_digest = await create_digest(	×
199	CreateDigest(
200	[
201	FileContent(
202	PythonLockfileMetadata.metadata_location_for_lockfile(req.lockfile_dest),
203	metadata.to_json(with_description=descr).encode(),
204	),
205	]
206	)
207	)
UNCOV 208	final_lockfile_digest = await merge_digests(	×
209	MergeDigests([metadata_digest, result.output_digest])
210	)
211	else:
212	initial_lockfile_digest_contents = await get_digest_contents(result.output_digest)	×
UNCOV 213	lockfile_with_header = metadata.add_header_to_lockfile(	×
214	initial_lockfile_digest_contents[0].content,
215	regenerate_command=regenerate_command,
216	delimeter=header_delimiter,
217	)
UNCOV 218	final_lockfile_digest = await create_digest(	×
219	CreateDigest(
220	[
221	FileContent(req.lockfile_dest, lockfile_with_header),
222	]
223	)
224	)
225
226	if req.diff:	×
UNCOV 227	diff = await _generate_python_lockfile_diff(	×
228	final_lockfile_digest, req.resolve_name, req.lockfile_dest
229	)
230	else:
UNCOV 231	diff = None	×
232
UNCOV 233	return GenerateLockfileResult(final_lockfile_digest, req.resolve_name, req.lockfile_dest, diff)	×
234
235
236	class RequestedPythonUserResolveNames(RequestedUserResolveNames):	11✔
237	pass	11✔
238
239
240	class KnownPythonUserResolveNamesRequest(KnownUserResolveNamesRequest):	11✔
241	pass	11✔
242
243
244	@rule	11✔
245	async def determine_python_user_resolves(	11✔
246	_: KnownPythonUserResolveNamesRequest,
247	python_setup: PythonSetup,
248	union_membership: UnionMembership,
249	) -> KnownUserResolveNames:
250	"""Find all know Python resolves, from both user-created resolves and internal tools."""
UNCOV 251	python_tool_resolves = ExportableTool.filter_for_subclasses(union_membership, PythonToolBase)	×
252
UNCOV 253	tools_using_default_resolve = [	×
254	resolve_name
255	for resolve_name, subsystem_cls in python_tool_resolves.items()
256	if (await _construct_subsystem(subsystem_cls)).install_from_resolve is None
257	]
258
UNCOV 259	return KnownUserResolveNames(	×
260	names=(
261	*python_setup.resolves.keys(),
262	*tools_using_default_resolve,
263	), # the order of the keys doesn't matter since shadowing is done in `setup_user_lockfile_requests`
264	option_name="[python].resolves",
265	requested_resolve_names_cls=RequestedPythonUserResolveNames,
266	)
267
268
269	@rule	11✔
270	async def setup_user_lockfile_requests(	11✔
271	requested: RequestedPythonUserResolveNames,
272	all_targets: AllTargets,
273	python_setup: PythonSetup,
274	union_membership: UnionMembership,
275	) -> UserGenerateLockfiles:
276	"""Transform the names of resolves requested into the `GeneratePythonLockfile` request object.
277
278	Shadowing is done here by only checking internal resolves if the resolve is not a user-created
279	resolve.
280	"""
281	if not (python_setup.enable_resolves and python_setup.resolves_generate_lockfiles):	×
UNCOV 282	return UserGenerateLockfiles()	×
283
284	resolve_to_requirements_fields = defaultdict(set)	×
285	find_links: set[str] = set()	×
286	for tgt in all_targets:	×
287	if not tgt.has_fields((PythonRequirementResolveField, PythonRequirementsField)):	×
288	continue	×
289	resolve = tgt[PythonRequirementResolveField].normalized_value(python_setup)	×
290	resolve_to_requirements_fields[resolve].add(tgt[PythonRequirementsField])	×
UNCOV 291	find_links.update(tgt[PythonRequirementFindLinksField].value or ())	×
292
UNCOV 293	tools = ExportableTool.filter_for_subclasses(union_membership, PythonToolBase)	×
294
295	out = set()	×
296	for resolve in requested:	×
297	if resolve in python_setup.resolves:	×
UNCOV 298	out.add(	×
299	GeneratePythonLockfile(
300	requirements=PexRequirements.req_strings_from_requirement_fields(
301	resolve_to_requirements_fields[resolve]
302	),
303	find_links=FrozenOrderedSet(find_links),
304	interpreter_constraints=InterpreterConstraints(
305	python_setup.resolves_to_interpreter_constraints.get(
306	resolve, python_setup.interpreter_constraints
307	)
308	),
309	resolve_name=resolve,
310	lockfile_dest=python_setup.resolves[resolve],
311	diff=False,
312	)
313	)
314	else:
315	tool_cls: type[PythonToolBase] = tools[resolve]	×
UNCOV 316	tool = await _construct_subsystem(tool_cls)	×
317
318	# TODO: we shouldn't be managing default ICs in lockfile identification.
319	# We should find a better place to do this or a better way to default
320	if tool.register_interpreter_constraints:	×
UNCOV 321	ic = tool.interpreter_constraints	×
322	else:
UNCOV 323	ic = InterpreterConstraints(tool.default_interpreter_constraints)	×
324
UNCOV 325	out.add(	×
326	GeneratePythonLockfile(
327	requirements=FrozenOrderedSet(sorted(tool.requirements)),
328	find_links=FrozenOrderedSet(find_links),
329	interpreter_constraints=ic,
330	resolve_name=resolve,
331	lockfile_dest=DEFAULT_TOOL_LOCKFILE,
332	diff=False,
333	)
334	)
335
UNCOV 336	return UserGenerateLockfiles(out)	×
337
338
339	@dataclass(frozen=True)	11✔
340	class PythonSyntheticLockfileTargetsRequest(SyntheticTargetsRequest):	11✔
341	"""Register the type used to create synthetic targets for Python lockfiles.
342
343	As the paths for all lockfiles are known up-front, we set the `path` field to
344	`SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS` so that we get a single request for all
345	our synthetic targets rather than one request per directory.
346	"""
347
348	path: str = SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS	11✔
349
350
351	def synthetic_lockfile_target_name(resolve: str) -> str:	11✔
UNCOV 352	return f"_{resolve}_lockfile"	×
353
354
355	@rule	11✔
356	async def python_lockfile_synthetic_targets(	11✔
357	request: PythonSyntheticLockfileTargetsRequest,
358	python_setup: PythonSetup,
359	) -> SyntheticAddressMaps:
360	if not python_setup.enable_synthetic_lockfiles:	×
UNCOV 361	return SyntheticAddressMaps()	×
362
UNCOV 363	resolves = [	×
364	(os.path.dirname(lockfile), os.path.basename(lockfile), name)
365	for name, lockfile in python_setup.resolves.items()
366	]
367
UNCOV 368	return SyntheticAddressMaps.for_targets_request(	×
369	request,
370	[
371	(
372	os.path.join(spec_path, "BUILD.python-lockfiles"),
373	tuple(
374	TargetAdaptor(
375	"_lockfiles",
376	name=synthetic_lockfile_target_name(name),
377	sources=(lockfile,),
378	__description_of_origin__=f"the [python].resolves option {name!r}",
379	)
380	for _, lockfile, name in lockfiles
381	),
382	)
383	for spec_path, lockfiles in itertools.groupby(sorted(resolves), key=itemgetter(0))
384	],
385	)
386
387
388	def rules():	11✔
389	return (	10✔
390	*collect_rules(),
391	UnionRule(GenerateLockfile, GeneratePythonLockfile),
392	UnionRule(KnownUserResolveNamesRequest, KnownPythonUserResolveNamesRequest),
393	UnionRule(RequestedUserResolveNames, RequestedPythonUserResolveNames),
394	UnionRule(SyntheticTargetsRequest, PythonSyntheticLockfileTargetsRequest),
395	)

pantsbuild / pants / 18730153336

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