20332790708

Committed 18 Dec 2025 09:48AM UTC coverage: 64.992% (-15.3%) from 80.295%

Build # 20332790708

Build Type

Pull #22949

github

Committed by

web-flow

Commit Message

Merge f730a56cd into 407284c67

Pull Request Pull Request #22949: Add experimental uv resolver for Python lockfiles

Run Details

54 of 97 new or added lines in 5 files covered. (55.67%)

8270 existing lines in 295 files now uncovered.

48990 of 75379 relevant lines covered (64.99%)

1.81 hits per line

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

53.33

/src/python/pants/jvm/compile.py

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

from __future__ import annotations

import logging
import os
from abc import ABCMeta
from collections import defaultdict, deque
from collections.abc import Iterable, Iterator, Sequence
from dataclasses import dataclass
from enum import Enum, auto
from typing import ClassVar

from pants.core.target_types import (
    FilesGeneratingSourcesField,
    FileSourceField,
    RelocatedFilesOriginalTargetsField,
)
from pants.engine.collection import Collection
from pants.engine.engine_aware import EngineAwareReturnType
from pants.engine.environment import EnvironmentName
from pants.engine.fs import Digest
from pants.engine.process import FallibleProcessResult
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import (
    CoarsenedTarget,
    Field,
    FieldSet,
    GenerateSourcesRequest,
    SourcesField,
    Target,
    TargetFilesGenerator,
)
from pants.engine.unions import UnionMembership, union
from pants.jvm.resolve.key import CoursierResolveKey
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet
from pants.util.strutil import Simplifier

logger = logging.getLogger(__name__)


class ClasspathSourceMissing(Exception):
    """No compiler instances were compatible with a CoarsenedTarget."""


class ClasspathSourceAmbiguity(Exception):
    """Too many compiler instances were compatible with a CoarsenedTarget."""


class ClasspathRootOnlyWasInner(Exception):
    """A root_only request type was used as an inner node in a compile graph."""


class _ClasspathEntryRequestClassification(Enum):
    COMPATIBLE = auto()
    PARTIAL = auto()
    CONSUME_ONLY = auto()
    INCOMPATIBLE = auto()


@union(in_scope_types=[EnvironmentName])
@dataclass(frozen=True)
class ClasspathEntryRequest(metaclass=ABCMeta):
    """A request for a ClasspathEntry for the given CoarsenedTarget and resolve.

    TODO: Move to `classpath.py`.
    """

    component: CoarsenedTarget
    resolve: CoursierResolveKey
    # If this request contains some FieldSets which do _not_ match this request class's
    # FieldSets a prerequisite request will be set. When set, the provider of the
    # ClasspathEntry should recurse with this request first, and include it as a dependency.
    prerequisite: ClasspathEntryRequest | None = None

    # The FieldSet types that this request subclass can produce a ClasspathEntry for. A request
    # will only be constructed if it is compatible with all of the members of the CoarsenedTarget,
    # or if a `prerequisite` request will provide an entry for the rest of the members.
    field_sets: ClassVar[tuple[type[FieldSet], ...]]

    # Additional FieldSet types that this request subclass may consume (but not produce a
    # ClasspathEntry for) iff they are contained in a component with FieldSets matching
    # `cls.field_sets`.
    field_sets_consume_only: ClassVar[tuple[type[FieldSet], ...]] = ()

    # True if this request type is only valid at the root of a compile graph.
    root_only: ClassVar[bool] = False


@dataclass(frozen=True)
class ClasspathEntryRequestFactory:
    impls: tuple[type[ClasspathEntryRequest], ...]
    generator_sources: FrozenDict[type[ClasspathEntryRequest], frozenset[type[SourcesField]]]

    def for_targets(
        self,
        component: CoarsenedTarget,
        resolve: CoursierResolveKey,
        *,
        root: bool = False,
    ) -> ClasspathEntryRequest:
        """Constructs a subclass compatible with the members of the CoarsenedTarget.

        If the CoarsenedTarget is a root of a compile graph, pass `root=True` to allow usage of
        request types which are marked `root_only`.
        """

        compatible = []
        partial = []
        consume_only = []
        impls = self.impls
        for impl in impls:
            classification = self.classify_impl(impl, component)
            if classification == _ClasspathEntryRequestClassification.INCOMPATIBLE:
                continue
            elif classification == _ClasspathEntryRequestClassification.COMPATIBLE:
                compatible.append(impl)
            elif classification == _ClasspathEntryRequestClassification.PARTIAL:
                partial.append(impl)
            elif classification == _ClasspathEntryRequestClassification.CONSUME_ONLY:
                consume_only.append(impl)

        if len(compatible) == 1:
            if not root and compatible[0].root_only:
                raise ClasspathRootOnlyWasInner(
                    "The following targets had dependents, but can only be used as roots in a "
                    f"build graph:\n{component.bullet_list()}"
                )
            return compatible[0](component, resolve, None)

        # No single request can handle the entire component: see whether there are exactly one
        # partial and consume_only impl to handle it together.
        if not compatible and len(partial) == 1 and len(consume_only) == 1:
            # TODO: Precompute which requests might be partial for others?
            if set(partial[0].field_sets).issubset(set(consume_only[0].field_sets_consume_only)):
                return partial[0](component, resolve, consume_only[0](component, resolve, None))

        if compatible:
            impls_str = ", ".join(sorted(impl.__name__ for impl in compatible))
            raise ClasspathSourceAmbiguity(
                f"More than one JVM classpath provider ({impls_str}) was compatible with "
                f"the inputs:\n{component.bullet_list()}"
            )
        else:
            # TODO: There is more subtlety of error messages possible here if there are multiple
            # partial providers, but can cross that bridge when we have them (multiple Scala or Java
            # compiler implementations, for example).
            impls_str = ", ".join(sorted(impl.__name__ for impl in impls))
            raise ClasspathSourceMissing(
                f"No JVM classpath providers (from: {impls_str}) were compatible with the "
                f"combination of inputs:\n{component.bullet_list()}"
            )

    def classify_impl(
        self, impl: type[ClasspathEntryRequest], component: CoarsenedTarget
    ) -> _ClasspathEntryRequestClassification:
        targets = component.members
        generator_sources = self.generator_sources.get(impl) or frozenset()

        def is_compatible(target: Target) -> bool:
            return (
                # Is directly applicable.
                any(fs.is_applicable(target) for fs in impl.field_sets)
                or
                # Is applicable via generated sources.
                any(target.has_field(g) for g in generator_sources)
                or
                # Is applicable via a generator.
                (
                    isinstance(target, TargetFilesGenerator)
                    and any(
                        field in target.generated_target_cls.core_fields
                        for field in generator_sources
                    )
                )
            )

        compatible = sum(1 for t in targets if is_compatible(t))
        if compatible == 0:
            return _ClasspathEntryRequestClassification.INCOMPATIBLE
        if compatible == len(targets):
            return _ClasspathEntryRequestClassification.COMPATIBLE

        consume_only = sum(
            1 for t in targets for fs in impl.field_sets_consume_only if fs.is_applicable(t)
        )
        if compatible + consume_only == len(targets):
            return _ClasspathEntryRequestClassification.CONSUME_ONLY
        return _ClasspathEntryRequestClassification.PARTIAL


@rule
async def calculate_jvm_request_types(
    union_membership: UnionMembership,
) -> ClasspathEntryRequestFactory:
    cpe_impls = union_membership.get(ClasspathEntryRequest)

    impls_by_source: dict[type[Field], type[ClasspathEntryRequest]] = {}
    for impl in cpe_impls:
        for field_set in impl.field_sets:
            for field in field_set.required_fields:
                # Assume only one impl per field (normally sound)
                # (note that subsequently, we only check for `SourceFields`, so no need to filter)
                impls_by_source[field] = impl

    # Classify code generator sources by their CPE impl
    sources_by_impl_: dict[type[ClasspathEntryRequest], list[type[SourcesField]]] = defaultdict(
        list
    )

    for g in union_membership.get(GenerateSourcesRequest):
        if g.output in impls_by_source:
            sources_by_impl_[impls_by_source[g.output]].append(g.input)
    sources_by_impl = FrozenDict((key, frozenset(value)) for key, value in sources_by_impl_.items())

    return ClasspathEntryRequestFactory(tuple(cpe_impls), sources_by_impl)


@dataclass(frozen=True)
class ClasspathEntry:
    """A JVM classpath entry represented as a series of JAR files, and their dependencies.

    This is a series of JAR files in order to account for "exported" dependencies, when a node
    and some of its dependencies are indistinguishable (such as for aliases, or potentially
    explicitly declared or inferred `exported=` lists in the future).

    This class additionally keeps filenames in order to preserve classpath ordering for the
    `classpath_arg` method: although Digests encode filenames, they are stored sorted.

    If `[jvm].reproducible_jars`, then all JARs in a classpath entry must have had timestamps
    stripped -- either natively, or via the `pants.jvm.strip_jar` rules.

    TODO: Move to `classpath.py`.
    TODO: Generalize via https://github.com/pantsbuild/pants/issues/13112.

    Note: Non-jar artifacts (e.g., executables with "exe" packaging) may end up on the classpath if
    they are dependencies.
    TODO: Does there need to be a filtering mechanism to exclude non-jar artifacts in the default case?
    """

    digest: Digest
    filenames: tuple[str, ...]
    dependencies: FrozenOrderedSet[ClasspathEntry]

    def __init__(
        self,
        digest: Digest,
        filenames: Iterable[str] = (),
        dependencies: Iterable[ClasspathEntry] = (),
    ):
        object.__setattr__(self, "digest", digest)
        object.__setattr__(self, "filenames", tuple(filenames))
        object.__setattr__(self, "dependencies", FrozenOrderedSet(dependencies))

    @classmethod
    def merge(cls, digest: Digest, entries: Iterable[ClasspathEntry]) -> ClasspathEntry:
        """After merging the Digests for entries, merge their filenames and dependencies."""
        return cls(
            digest,
            (f for cpe in entries for f in cpe.filenames),
            (d for cpe in entries for d in cpe.dependencies),
        )

    @classmethod
    def args(cls, entries: Iterable[ClasspathEntry], *, prefix: str = "") -> Iterator[str]:
        """Returns the filenames for the given entries.

        TODO: See whether this method can be completely eliminated in favor of
        `immutable_inputs(_args)`.

        To compute transitive filenames, first expand the entries with `cls.closure()`.
        """
        return (os.path.join(prefix, f) for cpe in entries for f in cpe.filenames)

    @classmethod
    def immutable_inputs(
        cls, entries: Iterable[ClasspathEntry], *, prefix: str = ""
    ) -> Iterator[tuple[str, Digest]]:
        """Returns (relpath, Digest) tuples for use with `Process.immutable_input_digests`.

        To compute transitive input tuples, first expand the entries with `cls.closure()`.
        """
        return ((os.path.join(prefix, cpe.digest.fingerprint[:12]), cpe.digest) for cpe in entries)

    @classmethod
    def immutable_inputs_args(
        cls, entries: Iterable[ClasspathEntry], *, prefix: str = ""
    ) -> Iterator[str]:
        """Returns the relative filenames for the given entries to be used as immutable_inputs.

        To compute transitive input tuples, first expand the entries with `cls.closure()`.
        """
        for cpe in entries:
            fingerprint_prefix = cpe.digest.fingerprint[:12]
            for filename in cpe.filenames:
                yield os.path.join(prefix, fingerprint_prefix, filename)

    @classmethod
    def closure(cls, roots: Iterable[ClasspathEntry]) -> Iterator[ClasspathEntry]:
        """All ClasspathEntries reachable from the given roots."""

        visited = set()
        queue = deque(roots)
        while queue:
            ct = queue.popleft()
            if ct in visited:
                continue
            visited.add(ct)
            yield ct
            queue.extend(ct.dependencies)

    def __repr__(self):
        return f"ClasspathEntry({self.filenames}, dependencies={len(self.dependencies)})"

    def __str__(self) -> str:
        return repr(self)


class CompileResult(Enum):
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    DEPENDENCY_FAILED = "dependency failed"


@dataclass(frozen=True)
class FallibleClasspathEntry(EngineAwareReturnType):
    description: str
    result: CompileResult
    output: ClasspathEntry | None
    exit_code: int
    stdout: str | None = None
    stderr: str | None = None

    @classmethod
    def from_fallible_process_result(
        cls,
        description: str,
        process_result: FallibleProcessResult,
        output: ClasspathEntry | None,
        *,
        output_simplifier: Simplifier = Simplifier(),
    ) -> FallibleClasspathEntry:
        exit_code = process_result.exit_code
        stderr = output_simplifier.simplify(process_result.stderr)
        return cls(
            description=description,
            result=(CompileResult.SUCCEEDED if exit_code == 0 else CompileResult.FAILED),
            output=output,
            exit_code=exit_code,
            stdout=output_simplifier.simplify(process_result.stdout),
            stderr=stderr,
        )

    @classmethod
    def if_all_succeeded(
        cls, fallible_classpath_entries: Sequence[FallibleClasspathEntry]
    ) -> tuple[ClasspathEntry, ...] | None:
        """If all given FallibleClasspathEntries succeeded, return them as ClasspathEntries."""
        classpath_entries = tuple(fcc.output for fcc in fallible_classpath_entries if fcc.output)
        if len(classpath_entries) != len(fallible_classpath_entries):
            return None
        return classpath_entries

    def level(self) -> LogLevel:
        return LogLevel.ERROR if self.result == CompileResult.FAILED else LogLevel.DEBUG

    def message(self) -> str:
        message = self.description
        message += (
            " succeeded." if self.exit_code == 0 else f" failed (exit code {self.exit_code})."
        )
        if self.stdout:
            message += f"\n{self.stdout}"
        if self.stderr:
            message += f"\n{self.stderr}"
        return message

    def cacheable(self) -> bool:
        # Failed compile outputs should be re-rendered in every run.
        return self.exit_code == 0


@rule(polymorphic=True)
async def get_fallible_classpath_entry(
    req: ClasspathEntryRequest,
    environment_name: EnvironmentName,
) -> FallibleClasspathEntry:
    raise NotImplementedError()


class ClasspathEntryRequests(Collection[ClasspathEntryRequest]):
    pass


class FallibleClasspathEntries(Collection[FallibleClasspathEntry]):
    def if_all_succeeded(self) -> tuple[ClasspathEntry, ...] | None:
        return FallibleClasspathEntry.if_all_succeeded(self)


@dataclass(frozen=True)
class ClasspathDependenciesRequest:
    request: ClasspathEntryRequest
    ignore_generated: bool = False


@rule
async def required_classfiles(fallible_result: FallibleClasspathEntry) -> ClasspathEntry:
    if fallible_result.result == CompileResult.SUCCEEDED:
        assert fallible_result.output
        return fallible_result.output
    # NB: The compile outputs will already have been streamed as FallibleClasspathEntries finish.
    raise Exception(
        f"Compile failed:\nstdout:\n{fallible_result.stdout}\nstderr:\n{fallible_result.stderr}"
    )


@rule
async def classpath_dependency_requests(
    classpath_entry_request: ClasspathEntryRequestFactory, request: ClasspathDependenciesRequest
) -> ClasspathEntryRequests:
    def ignore_because_generated(coarsened_dep: CoarsenedTarget) -> bool:
        if not request.ignore_generated:
            return False
        if len(coarsened_dep.members) != 1:
            # Do not ignore a dependency which is involved in a cycle.
            return False
        us = request.request.component.representative.address
        them = coarsened_dep.representative.address
        return us.spec_path == them.spec_path and us.target_name == them.target_name

    def ignore_because_file(coarsened_dep: CoarsenedTarget) -> bool:
        return sum(
            1
            for t in coarsened_dep.members
            if t.has_field(FileSourceField)
            or t.has_field(FilesGeneratingSourcesField)
            or t.has_field(RelocatedFilesOriginalTargetsField)
        ) == len(coarsened_dep.members)

    return ClasspathEntryRequests(
        classpath_entry_request.for_targets(
            component=coarsened_dep, resolve=request.request.resolve
        )
        for coarsened_dep in request.request.component.dependencies
        if not ignore_because_generated(coarsened_dep) and not ignore_because_file(coarsened_dep)
    )


@rule
async def compile_classpath_entries(requests: ClasspathEntryRequests) -> FallibleClasspathEntries:
    return FallibleClasspathEntries(
        await concurrently(
            get_fallible_classpath_entry(**implicitly({request: ClasspathEntryRequest}))
            for request in requests
        )
    )


def rules():
    return collect_rules()

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	5✔
5
6	import logging	5✔
7	import os	5✔
8	from abc import ABCMeta	5✔
9	from collections import defaultdict, deque	5✔
10	from collections.abc import Iterable, Iterator, Sequence	5✔
11	from dataclasses import dataclass	5✔
12	from enum import Enum, auto	5✔
13	from typing import ClassVar	5✔
14
15	from pants.core.target_types import (	5✔
16	FilesGeneratingSourcesField,
17	FileSourceField,
18	RelocatedFilesOriginalTargetsField,
19	)
20	from pants.engine.collection import Collection	5✔
21	from pants.engine.engine_aware import EngineAwareReturnType	5✔
22	from pants.engine.environment import EnvironmentName	5✔
23	from pants.engine.fs import Digest	5✔
24	from pants.engine.process import FallibleProcessResult	5✔
25	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	5✔
26	from pants.engine.target import (	5✔
27	CoarsenedTarget,
28	Field,
29	FieldSet,
30	GenerateSourcesRequest,
31	SourcesField,
32	Target,
33	TargetFilesGenerator,
34	)
35	from pants.engine.unions import UnionMembership, union	5✔
36	from pants.jvm.resolve.key import CoursierResolveKey	5✔
37	from pants.util.frozendict import FrozenDict	5✔
38	from pants.util.logging import LogLevel	5✔
39	from pants.util.ordered_set import FrozenOrderedSet	5✔
40	from pants.util.strutil import Simplifier	5✔
41
42	logger = logging.getLogger(__name__)	5✔
43
44
45	class ClasspathSourceMissing(Exception):	5✔
46	"""No compiler instances were compatible with a CoarsenedTarget."""
47
48
49	class ClasspathSourceAmbiguity(Exception):	5✔
50	"""Too many compiler instances were compatible with a CoarsenedTarget."""
51
52
53	class ClasspathRootOnlyWasInner(Exception):	5✔
54	"""A root_only request type was used as an inner node in a compile graph."""
55
56
57	class _ClasspathEntryRequestClassification(Enum):	5✔
58	COMPATIBLE = auto()	5✔
59	PARTIAL = auto()	5✔
60	CONSUME_ONLY = auto()	5✔
61	INCOMPATIBLE = auto()	5✔
62
63
64	@union(in_scope_types=[EnvironmentName])	5✔
65	@dataclass(frozen=True)	5✔
66	class ClasspathEntryRequest(metaclass=ABCMeta):	5✔
67	"""A request for a ClasspathEntry for the given CoarsenedTarget and resolve.
68
69	TODO: Move to `classpath.py`.
70	"""
71
72	component: CoarsenedTarget	5✔
73	resolve: CoursierResolveKey	5✔
74	# If this request contains some FieldSets which do _not_ match this request class's
75	# FieldSets a prerequisite request will be set. When set, the provider of the
76	# ClasspathEntry should recurse with this request first, and include it as a dependency.
77	prerequisite: ClasspathEntryRequest \| None = None	5✔
78
79	# The FieldSet types that this request subclass can produce a ClasspathEntry for. A request
80	# will only be constructed if it is compatible with all of the members of the CoarsenedTarget,
81	# or if a `prerequisite` request will provide an entry for the rest of the members.
82	field_sets: ClassVar[tuple[type[FieldSet], ...]]	5✔
83
84	# Additional FieldSet types that this request subclass may consume (but not produce a
85	# ClasspathEntry for) iff they are contained in a component with FieldSets matching
86	# `cls.field_sets`.
87	field_sets_consume_only: ClassVar[tuple[type[FieldSet], ...]] = ()	5✔
88
89	# True if this request type is only valid at the root of a compile graph.
90	root_only: ClassVar[bool] = False	5✔
91
92
93	@dataclass(frozen=True)	5✔
94	class ClasspathEntryRequestFactory:	5✔
95	impls: tuple[type[ClasspathEntryRequest], ...]	5✔
96	generator_sources: FrozenDict[type[ClasspathEntryRequest], frozenset[type[SourcesField]]]	5✔
97
98	def for_targets(	5✔
99	self,
100	component: CoarsenedTarget,
101	resolve: CoursierResolveKey,
102	*,
103	root: bool = False,
104	) -> ClasspathEntryRequest:
105	"""Constructs a subclass compatible with the members of the CoarsenedTarget.
106
107	If the CoarsenedTarget is a root of a compile graph, pass `root=True` to allow usage of
108	request types which are marked `root_only`.
109	"""
110
UNCOV 111	compatible = []	×
UNCOV 112	partial = []	×
UNCOV 113	consume_only = []	×
UNCOV 114	impls = self.impls	×
UNCOV 115	for impl in impls:	×
UNCOV 116	classification = self.classify_impl(impl, component)	×
UNCOV 117	if classification == _ClasspathEntryRequestClassification.INCOMPATIBLE:	×
UNCOV 118	continue	×
UNCOV 119	elif classification == _ClasspathEntryRequestClassification.COMPATIBLE:	×
UNCOV 120	compatible.append(impl)	×
UNCOV 121	elif classification == _ClasspathEntryRequestClassification.PARTIAL:	×
UNCOV 122	partial.append(impl)	×
UNCOV 123	elif classification == _ClasspathEntryRequestClassification.CONSUME_ONLY:	×
UNCOV 124	consume_only.append(impl)	×
125
UNCOV 126	if len(compatible) == 1:	×
UNCOV 127	if not root and compatible[0].root_only:	×
128	raise ClasspathRootOnlyWasInner(	×
129	"The following targets had dependents, but can only be used as roots in a "
130	f"build graph:\n{component.bullet_list()}"
131	)
UNCOV 132	return compatible[0](component, resolve, None)	×
133
134	# No single request can handle the entire component: see whether there are exactly one
135	# partial and consume_only impl to handle it together.
UNCOV 136	if not compatible and len(partial) == 1 and len(consume_only) == 1:	×
137	# TODO: Precompute which requests might be partial for others?
UNCOV 138	if set(partial[0].field_sets).issubset(set(consume_only[0].field_sets_consume_only)):	×
UNCOV 139	return partial[0](component, resolve, consume_only[0](component, resolve, None))	×
140
UNCOV 141	if compatible:	×
UNCOV 142	impls_str = ", ".join(sorted(impl.__name__ for impl in compatible))	×
UNCOV 143	raise ClasspathSourceAmbiguity(	×
144	f"More than one JVM classpath provider ({impls_str}) was compatible with "
145	f"the inputs:\n{component.bullet_list()}"
146	)
147	else:
148	# TODO: There is more subtlety of error messages possible here if there are multiple
149	# partial providers, but can cross that bridge when we have them (multiple Scala or Java
150	# compiler implementations, for example).
UNCOV 151	impls_str = ", ".join(sorted(impl.__name__ for impl in impls))	×
UNCOV 152	raise ClasspathSourceMissing(	×
153	f"No JVM classpath providers (from: {impls_str}) were compatible with the "
154	f"combination of inputs:\n{component.bullet_list()}"
155	)
156
157	def classify_impl(	5✔
158	self, impl: type[ClasspathEntryRequest], component: CoarsenedTarget
159	) -> _ClasspathEntryRequestClassification:
UNCOV 160	targets = component.members	×
UNCOV 161	generator_sources = self.generator_sources.get(impl) or frozenset()	×
162
UNCOV 163	def is_compatible(target: Target) -> bool:	×
UNCOV 164	return (	×
165	# Is directly applicable.
166	any(fs.is_applicable(target) for fs in impl.field_sets)
167	or
168	# Is applicable via generated sources.
169	any(target.has_field(g) for g in generator_sources)
170	or
171	# Is applicable via a generator.
172	(
173	isinstance(target, TargetFilesGenerator)
174	and any(
175	field in target.generated_target_cls.core_fields
176	for field in generator_sources
177	)
178	)
179	)
180
UNCOV 181	compatible = sum(1 for t in targets if is_compatible(t))	×
UNCOV 182	if compatible == 0:	×
UNCOV 183	return _ClasspathEntryRequestClassification.INCOMPATIBLE	×
UNCOV 184	if compatible == len(targets):	×
UNCOV 185	return _ClasspathEntryRequestClassification.COMPATIBLE	×
186
UNCOV 187	consume_only = sum(	×
188	1 for t in targets for fs in impl.field_sets_consume_only if fs.is_applicable(t)
189	)
UNCOV 190	if compatible + consume_only == len(targets):	×
UNCOV 191	return _ClasspathEntryRequestClassification.CONSUME_ONLY	×
UNCOV 192	return _ClasspathEntryRequestClassification.PARTIAL	×
193
194
195	@rule	5✔
196	async def calculate_jvm_request_types(	5✔
197	union_membership: UnionMembership,
198	) -> ClasspathEntryRequestFactory:
199	cpe_impls = union_membership.get(ClasspathEntryRequest)	×
200
201	impls_by_source: dict[type[Field], type[ClasspathEntryRequest]] = {}	×
202	for impl in cpe_impls:	×
203	for field_set in impl.field_sets:	×
204	for field in field_set.required_fields:	×
205	# Assume only one impl per field (normally sound)
206	# (note that subsequently, we only check for `SourceFields`, so no need to filter)
207	impls_by_source[field] = impl	×
208
209	# Classify code generator sources by their CPE impl
210	sources_by_impl_: dict[type[ClasspathEntryRequest], list[type[SourcesField]]] = defaultdict(	×
211	list
212	)
213
214	for g in union_membership.get(GenerateSourcesRequest):	×
215	if g.output in impls_by_source:	×
216	sources_by_impl_[impls_by_source[g.output]].append(g.input)	×
217	sources_by_impl = FrozenDict((key, frozenset(value)) for key, value in sources_by_impl_.items())	×
218
219	return ClasspathEntryRequestFactory(tuple(cpe_impls), sources_by_impl)	×
220
221
222	@dataclass(frozen=True)	5✔
223	class ClasspathEntry:	5✔
224	"""A JVM classpath entry represented as a series of JAR files, and their dependencies.
225
226	This is a series of JAR files in order to account for "exported" dependencies, when a node
227	and some of its dependencies are indistinguishable (such as for aliases, or potentially
228	explicitly declared or inferred `exported=` lists in the future).
229
230	This class additionally keeps filenames in order to preserve classpath ordering for the
231	`classpath_arg` method: although Digests encode filenames, they are stored sorted.
232
233	If `[jvm].reproducible_jars`, then all JARs in a classpath entry must have had timestamps
234	stripped -- either natively, or via the `pants.jvm.strip_jar` rules.
235
236	TODO: Move to `classpath.py`.
237	TODO: Generalize via https://github.com/pantsbuild/pants/issues/13112.
238
239	Note: Non-jar artifacts (e.g., executables with "exe" packaging) may end up on the classpath if
240	they are dependencies.
241	TODO: Does there need to be a filtering mechanism to exclude non-jar artifacts in the default case?
242	"""
243
244	digest: Digest	5✔
245	filenames: tuple[str, ...]	5✔
246	dependencies: FrozenOrderedSet[ClasspathEntry]	5✔
247
248	def __init__(	5✔
249	self,
250	digest: Digest,
251	filenames: Iterable[str] = (),
252	dependencies: Iterable[ClasspathEntry] = (),
253	):
254	object.__setattr__(self, "digest", digest)	×
255	object.__setattr__(self, "filenames", tuple(filenames))	×
256	object.__setattr__(self, "dependencies", FrozenOrderedSet(dependencies))	×
257
258	@classmethod	5✔
259	def merge(cls, digest: Digest, entries: Iterable[ClasspathEntry]) -> ClasspathEntry:	5✔
260	"""After merging the Digests for entries, merge their filenames and dependencies."""
261	return cls(	×
262	digest,
263	(f for cpe in entries for f in cpe.filenames),
264	(d for cpe in entries for d in cpe.dependencies),
265	)
266
267	@classmethod	5✔
268	def args(cls, entries: Iterable[ClasspathEntry], *, prefix: str = "") -> Iterator[str]:	5✔
269	"""Returns the filenames for the given entries.
270
271	TODO: See whether this method can be completely eliminated in favor of
272	`immutable_inputs(_args)`.
273
274	To compute transitive filenames, first expand the entries with `cls.closure()`.
275	"""
UNCOV 276	return (os.path.join(prefix, f) for cpe in entries for f in cpe.filenames)	×
277
278	@classmethod	5✔
279	def immutable_inputs(	5✔
280	cls, entries: Iterable[ClasspathEntry], *, prefix: str = ""
281	) -> Iterator[tuple[str, Digest]]:
282	"""Returns (relpath, Digest) tuples for use with `Process.immutable_input_digests`.
283
284	To compute transitive input tuples, first expand the entries with `cls.closure()`.
285	"""
286	return ((os.path.join(prefix, cpe.digest.fingerprint[:12]), cpe.digest) for cpe in entries)	×
287
288	@classmethod	5✔
289	def immutable_inputs_args(	5✔
290	cls, entries: Iterable[ClasspathEntry], *, prefix: str = ""
291	) -> Iterator[str]:
292	"""Returns the relative filenames for the given entries to be used as immutable_inputs.
293
294	To compute transitive input tuples, first expand the entries with `cls.closure()`.
295	"""
296	for cpe in entries:	×
297	fingerprint_prefix = cpe.digest.fingerprint[:12]	×
298	for filename in cpe.filenames:	×
299	yield os.path.join(prefix, fingerprint_prefix, filename)	×
300
301	@classmethod	5✔
302	def closure(cls, roots: Iterable[ClasspathEntry]) -> Iterator[ClasspathEntry]:	5✔
303	"""All ClasspathEntries reachable from the given roots."""
304
305	visited = set()	1✔
306	queue = deque(roots)	1✔
307	while queue:	1✔
308	ct = queue.popleft()	1✔
309	if ct in visited:	1✔
310	continue	×
311	visited.add(ct)	1✔
312	yield ct	1✔
313	queue.extend(ct.dependencies)	1✔
314
315	def __repr__(self):	5✔
316	return f"ClasspathEntry({self.filenames}, dependencies={len(self.dependencies)})"	×
317
318	def __str__(self) -> str:	5✔
319	return repr(self)	×
320
321
322	class CompileResult(Enum):	5✔
323	SUCCEEDED = "succeeded"	5✔
324	FAILED = "failed"	5✔
325	DEPENDENCY_FAILED = "dependency failed"	5✔
326
327
328	@dataclass(frozen=True)	5✔
329	class FallibleClasspathEntry(EngineAwareReturnType):	5✔
330	description: str	5✔
331	result: CompileResult	5✔
332	output: ClasspathEntry \| None	5✔
333	exit_code: int	5✔
334	stdout: str \| None = None	5✔
335	stderr: str \| None = None	5✔
336
337	@classmethod	5✔
338	def from_fallible_process_result(	5✔
339	cls,
340	description: str,
341	process_result: FallibleProcessResult,
342	output: ClasspathEntry \| None,
343	*,
344	output_simplifier: Simplifier = Simplifier(),
345	) -> FallibleClasspathEntry:
346	exit_code = process_result.exit_code	×
347	stderr = output_simplifier.simplify(process_result.stderr)	×
348	return cls(	×
349	description=description,
350	result=(CompileResult.SUCCEEDED if exit_code == 0 else CompileResult.FAILED),
351	output=output,
352	exit_code=exit_code,
353	stdout=output_simplifier.simplify(process_result.stdout),
354	stderr=stderr,
355	)
356
357	@classmethod	5✔
358	def if_all_succeeded(	5✔
359	cls, fallible_classpath_entries: Sequence[FallibleClasspathEntry]
360	) -> tuple[ClasspathEntry, ...] \| None:
361	"""If all given FallibleClasspathEntries succeeded, return them as ClasspathEntries."""
362	classpath_entries = tuple(fcc.output for fcc in fallible_classpath_entries if fcc.output)	×
363	if len(classpath_entries) != len(fallible_classpath_entries):	×
364	return None	×
365	return classpath_entries	×
366
367	def level(self) -> LogLevel:	5✔
368	return LogLevel.ERROR if self.result == CompileResult.FAILED else LogLevel.DEBUG	×
369
370	def message(self) -> str:	5✔
371	message = self.description	×
372	message += (	×
373	" succeeded." if self.exit_code == 0 else f" failed (exit code {self.exit_code})."
374	)
375	if self.stdout:	×
376	message += f"\n{self.stdout}"	×
377	if self.stderr:	×
378	message += f"\n{self.stderr}"	×
379	return message	×
380
381	def cacheable(self) -> bool:	5✔
382	# Failed compile outputs should be re-rendered in every run.
383	return self.exit_code == 0	×
384
385
386	@rule(polymorphic=True)	5✔
387	async def get_fallible_classpath_entry(	5✔
388	req: ClasspathEntryRequest,
389	environment_name: EnvironmentName,
390	) -> FallibleClasspathEntry:
391	raise NotImplementedError()	×
392
393
394	class ClasspathEntryRequests(Collection[ClasspathEntryRequest]):	5✔
395	pass	5✔
396
397
398	class FallibleClasspathEntries(Collection[FallibleClasspathEntry]):	5✔
399	def if_all_succeeded(self) -> tuple[ClasspathEntry, ...] \| None:	5✔
400	return FallibleClasspathEntry.if_all_succeeded(self)	×
401
402
403	@dataclass(frozen=True)	5✔
404	class ClasspathDependenciesRequest:	5✔
405	request: ClasspathEntryRequest	5✔
406	ignore_generated: bool = False	5✔
407
408
409	@rule	5✔
410	async def required_classfiles(fallible_result: FallibleClasspathEntry) -> ClasspathEntry:	5✔
411	if fallible_result.result == CompileResult.SUCCEEDED:	×
412	assert fallible_result.output	×
413	return fallible_result.output	×
414	# NB: The compile outputs will already have been streamed as FallibleClasspathEntries finish.
415	raise Exception(	×
416	f"Compile failed:\nstdout:\n{fallible_result.stdout}\nstderr:\n{fallible_result.stderr}"
417	)
418
419
420	@rule	5✔
421	async def classpath_dependency_requests(	5✔
422	classpath_entry_request: ClasspathEntryRequestFactory, request: ClasspathDependenciesRequest
423	) -> ClasspathEntryRequests:
424	def ignore_because_generated(coarsened_dep: CoarsenedTarget) -> bool:	×
425	if not request.ignore_generated:	×
426	return False	×
427	if len(coarsened_dep.members) != 1:	×
428	# Do not ignore a dependency which is involved in a cycle.
429	return False	×
430	us = request.request.component.representative.address	×
431	them = coarsened_dep.representative.address	×
432	return us.spec_path == them.spec_path and us.target_name == them.target_name	×
433
434	def ignore_because_file(coarsened_dep: CoarsenedTarget) -> bool:	×
435	return sum(	×
436	1
437	for t in coarsened_dep.members
438	if t.has_field(FileSourceField)
439	or t.has_field(FilesGeneratingSourcesField)
440	or t.has_field(RelocatedFilesOriginalTargetsField)
441	) == len(coarsened_dep.members)
442
443	return ClasspathEntryRequests(	×
444	classpath_entry_request.for_targets(
445	component=coarsened_dep, resolve=request.request.resolve
446	)
447	for coarsened_dep in request.request.component.dependencies
448	if not ignore_because_generated(coarsened_dep) and not ignore_because_file(coarsened_dep)
449	)
450
451
452	@rule	5✔
453	async def compile_classpath_entries(requests: ClasspathEntryRequests) -> FallibleClasspathEntries:	5✔
454	return FallibleClasspathEntries(	×
455	await concurrently(
456	get_fallible_classpath_entry(**implicitly({request: ClasspathEntryRequest}))
457	for request in requests
458	)
459	)
460
461
462	def rules():	5✔
463	return collect_rules()	5✔

pantsbuild / pants / 20332790708

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