26342152999

Committed 23 May 2026 07:59PM UTC coverage: 91.165% (-1.6%) from 92.792%

Build # 26342152999

Build Type

push

github

Committed by

web-flow

Commit Message

Run Linux ARM CI on Depot runners (#23363)

RunsOn is deprecating their v2 stack, and rather than migrate
to v3 we should use the resources kindly donated by Depot.

GitHub also now has Linux ARM runners, should we need them.

Coverage Stats

87305 of 95766 relevant lines covered (91.16%)

3.87 hits per line

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

89.52

/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	11✔
5
6	import logging	11✔
7	import os	11✔
8	from abc import ABCMeta	11✔
9	from collections import defaultdict, deque	11✔
10	from collections.abc import Iterable, Iterator, Sequence	11✔
11	from dataclasses import dataclass	11✔
12	from enum import Enum, auto	11✔
13	from typing import ClassVar	11✔
14
15	from pants.core.target_types import (	11✔
16	FilesGeneratingSourcesField,
17	FileSourceField,
18	RelocatedFilesOriginalTargetsField,
19	)
20	from pants.engine.collection import Collection	11✔
21	from pants.engine.engine_aware import EngineAwareReturnType	11✔
22	from pants.engine.environment import EnvironmentName	11✔
23	from pants.engine.fs import Digest	11✔
24	from pants.engine.process import FallibleProcessResult	11✔
25	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	11✔
26	from pants.engine.target import (	11✔
27	CoarsenedTarget,
28	Field,
29	FieldSet,
30	GenerateSourcesRequest,
31	SourcesField,
32	Target,
33	TargetFilesGenerator,
34	)
35	from pants.engine.unions import UnionMembership, union	11✔
36	from pants.jvm.resolve.key import CoursierResolveKey	11✔
37	from pants.util.frozendict import FrozenDict	11✔
38	from pants.util.logging import LogLevel	11✔
39	from pants.util.ordered_set import FrozenOrderedSet	11✔
40	from pants.util.strutil import Simplifier	11✔
41
42	logger = logging.getLogger(__name__)	11✔
43
44
45	class ClasspathSourceMissing(Exception):	11✔
46	"""No compiler instances were compatible with a CoarsenedTarget."""
47
48
49	class ClasspathSourceAmbiguity(Exception):	11✔
50	"""Too many compiler instances were compatible with a CoarsenedTarget."""
51
52
53	class ClasspathRootOnlyWasInner(Exception):	11✔
54	"""A root_only request type was used as an inner node in a compile graph."""
55
56
57	class _ClasspathEntryRequestClassification(Enum):	11✔
58	COMPATIBLE = auto()	11✔
59	PARTIAL = auto()	11✔
60	CONSUME_ONLY = auto()	11✔
61	INCOMPATIBLE = auto()	11✔
62
63
64	@union(in_scope_types=[EnvironmentName])	11✔
65	@dataclass(frozen=True)	11✔
66	class ClasspathEntryRequest(metaclass=ABCMeta):	11✔
67	"""A request for a ClasspathEntry for the given CoarsenedTarget and resolve.
68
69	TODO: Move to `classpath.py`.
70	"""
71
72	component: CoarsenedTarget	11✔
73	resolve: CoursierResolveKey	11✔
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	11✔
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], ...]]	11✔
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], ...]] = ()	11✔
88
89	# True if this request type is only valid at the root of a compile graph.
90	root_only: ClassVar[bool] = False	11✔
91
92
93	@dataclass(frozen=True)	11✔
94	class ClasspathEntryRequestFactory:	11✔
95	impls: tuple[type[ClasspathEntryRequest], ...]	11✔
96	generator_sources: FrozenDict[type[ClasspathEntryRequest], frozenset[type[SourcesField]]]	11✔
97
98	def for_targets(	11✔
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
111	compatible = []	9✔
112	partial = []	9✔
113	consume_only = []	9✔
114	impls = self.impls	9✔
115	for impl in impls:	9✔
116	classification = self.classify_impl(impl, component)	9✔
117	if classification == _ClasspathEntryRequestClassification.INCOMPATIBLE:	9✔
118	continue	9✔
119	elif classification == _ClasspathEntryRequestClassification.COMPATIBLE:	9✔
120	compatible.append(impl)	9✔
121	elif classification == _ClasspathEntryRequestClassification.PARTIAL:	×
122	partial.append(impl)	×
123	elif classification == _ClasspathEntryRequestClassification.CONSUME_ONLY:	×
124	consume_only.append(impl)	×
125
126	if len(compatible) == 1:	9✔
127	if not root and compatible[0].root_only:	9✔
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	)
132	return compatible[0](component, resolve, None)	9✔
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.
136	if not compatible and len(partial) == 1 and len(consume_only) == 1:	×
137	# TODO: Precompute which requests might be partial for others?
138	if set(partial[0].field_sets).issubset(set(consume_only[0].field_sets_consume_only)):	×
139	return partial[0](component, resolve, consume_only[0](component, resolve, None))	×
140
141	if compatible:	×
142	impls_str = ", ".join(sorted(impl.__name__ for impl in compatible))	×
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).
151	impls_str = ", ".join(sorted(impl.__name__ for impl in impls))	×
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(	11✔
158	self, impl: type[ClasspathEntryRequest], component: CoarsenedTarget
159	) -> _ClasspathEntryRequestClassification:
160	targets = component.members	9✔
161	generator_sources = self.generator_sources.get(impl) or frozenset()	9✔
162
163	def is_compatible(target: Target) -> bool:	9✔
164	return (	9✔
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
181	compatible = sum(1 for t in targets if is_compatible(t))	9✔
182	if compatible == 0:	9✔
183	return _ClasspathEntryRequestClassification.INCOMPATIBLE	9✔
184	if compatible == len(targets):	9✔
185	return _ClasspathEntryRequestClassification.COMPATIBLE	9✔
186
187	consume_only = sum(	×
188	1 for t in targets for fs in impl.field_sets_consume_only if fs.is_applicable(t)
189	)
190	if compatible + consume_only == len(targets):	×
191	return _ClasspathEntryRequestClassification.CONSUME_ONLY	×
192	return _ClasspathEntryRequestClassification.PARTIAL	×
193
194
195	@rule	11✔
196	async def calculate_jvm_request_types(	11✔
197	union_membership: UnionMembership,
198	) -> ClasspathEntryRequestFactory:
199	cpe_impls = union_membership.get(ClasspathEntryRequest)	9✔
200
201	impls_by_source: dict[type[Field], type[ClasspathEntryRequest]] = {}	9✔
202	for impl in cpe_impls:	9✔
203	for field_set in impl.field_sets:	9✔
204	for field in field_set.required_fields:	9✔
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	9✔
208
209	# Classify code generator sources by their CPE impl
210	sources_by_impl_: dict[type[ClasspathEntryRequest], list[type[SourcesField]]] = defaultdict(	9✔
211	list
212	)
213
214	for g in union_membership.get(GenerateSourcesRequest):	9✔
215	if g.output in impls_by_source:	7✔
216	sources_by_impl_[impls_by_source[g.output]].append(g.input)	7✔
217	sources_by_impl = FrozenDict((key, frozenset(value)) for key, value in sources_by_impl_.items())	9✔
218
219	return ClasspathEntryRequestFactory(tuple(cpe_impls), sources_by_impl)	9✔
220
221
222	@dataclass(frozen=True)	11✔
223	class ClasspathEntry:	11✔
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	11✔
245	filenames: tuple[str, ...]	11✔
246	dependencies: FrozenOrderedSet[ClasspathEntry]	11✔
247
248	def __init__(	11✔
249	self,
250	digest: Digest,
251	filenames: Iterable[str] = (),
252	dependencies: Iterable[ClasspathEntry] = (),
253	):
254	object.__setattr__(self, "digest", digest)	11✔
255	object.__setattr__(self, "filenames", tuple(filenames))	11✔
256	object.__setattr__(self, "dependencies", FrozenOrderedSet(dependencies))	11✔
257
258	@classmethod	11✔
259	def merge(cls, digest: Digest, entries: Iterable[ClasspathEntry]) -> ClasspathEntry:	11✔
260	"""After merging the Digests for entries, merge their filenames and dependencies."""
261	return cls(	9✔
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	11✔
268	def args(cls, entries: Iterable[ClasspathEntry], *, prefix: str = "") -> Iterator[str]:	11✔
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	"""
276	return (os.path.join(prefix, f) for cpe in entries for f in cpe.filenames)	4✔
277
278	@classmethod	11✔
279	def immutable_inputs(	11✔
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)	9✔
287
288	@classmethod	11✔
289	def immutable_inputs_args(	11✔
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:	9✔
297	fingerprint_prefix = cpe.digest.fingerprint[:12]	9✔
298	for filename in cpe.filenames:	9✔
299	yield os.path.join(prefix, fingerprint_prefix, filename)	9✔
300
301	@classmethod	11✔
302	def closure(cls, roots: Iterable[ClasspathEntry]) -> Iterator[ClasspathEntry]:	11✔
303	"""All ClasspathEntries reachable from the given roots."""
304
305	visited = set()	9✔
306	queue = deque(roots)	9✔
307	while queue:	9✔
308	ct = queue.popleft()	9✔
309	if ct in visited:	9✔
310	continue	3✔
311	visited.add(ct)	9✔
312	yield ct	9✔
313	queue.extend(ct.dependencies)	9✔
314
315	def __repr__(self):	11✔
316	return f"ClasspathEntry({self.filenames}, dependencies={len(self.dependencies)})"	×
317
318	def __str__(self) -> str:	11✔
319	return repr(self)	×
320
321
322	class CompileResult(Enum):	11✔
323	SUCCEEDED = "succeeded"	11✔
324	FAILED = "failed"	11✔
325	DEPENDENCY_FAILED = "dependency failed"	11✔
326
327
328	@dataclass(frozen=True)	11✔
329	class FallibleClasspathEntry(EngineAwareReturnType):	11✔
330	description: str	11✔
331	result: CompileResult	11✔
332	output: ClasspathEntry \| None	11✔
333	exit_code: int	11✔
334	stdout: str \| None = None	11✔
335	stderr: str \| None = None	11✔
336
337	@classmethod	11✔
338	def from_fallible_process_result(	11✔
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	9✔
347	stderr = output_simplifier.simplify(process_result.stderr)	9✔
348	return cls(	9✔
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	11✔
358	def if_all_succeeded(	11✔
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)	9✔
363	if len(classpath_entries) != len(fallible_classpath_entries):	9✔
364	return None	×
365	return classpath_entries	9✔
366
367	def level(self) -> LogLevel:	11✔
368	return LogLevel.ERROR if self.result == CompileResult.FAILED else LogLevel.DEBUG	9✔
369
370	def message(self) -> str:	11✔
371	message = self.description	9✔
372	message += (	9✔
373	" succeeded." if self.exit_code == 0 else f" failed (exit code {self.exit_code})."
374	)
375	if self.stdout:	9✔
376	message += f"\n{self.stdout}"	3✔
377	if self.stderr:	9✔
378	message += f"\n{self.stderr}"	6✔
379	return message	9✔
380
381	def cacheable(self) -> bool:	11✔
382	# Failed compile outputs should be re-rendered in every run.
383	return self.exit_code == 0	9✔
384
385
386	@rule(polymorphic=True)	11✔
387	async def get_fallible_classpath_entry(	11✔
388	req: ClasspathEntryRequest,
389	environment_name: EnvironmentName,
390	) -> FallibleClasspathEntry:
391	raise NotImplementedError()	×
392
393
394	class ClasspathEntryRequests(Collection[ClasspathEntryRequest]):	11✔
395	pass	11✔
396
397
398	class FallibleClasspathEntries(Collection[FallibleClasspathEntry]):	11✔
399	def if_all_succeeded(self) -> tuple[ClasspathEntry, ...] \| None:	11✔
400	return FallibleClasspathEntry.if_all_succeeded(self)	9✔
401
402
403	@dataclass(frozen=True)	11✔
404	class ClasspathDependenciesRequest:	11✔
405	request: ClasspathEntryRequest	11✔
406	ignore_generated: bool = False	11✔
407
408
409	@rule	11✔
410	async def required_classfiles(fallible_result: FallibleClasspathEntry) -> ClasspathEntry:	11✔
411	if fallible_result.result == CompileResult.SUCCEEDED:	9✔
412	assert fallible_result.output	9✔
413	return fallible_result.output	9✔
414	# NB: The compile outputs will already have been streamed as FallibleClasspathEntries finish.
415	raise Exception(	2✔
416	f"Compile failed:\nstdout:\n{fallible_result.stdout}\nstderr:\n{fallible_result.stderr}"
417	)
418
419
420	@rule	11✔
421	async def classpath_dependency_requests(	11✔
422	classpath_entry_request: ClasspathEntryRequestFactory, request: ClasspathDependenciesRequest
423	) -> ClasspathEntryRequests:
424	def ignore_because_generated(coarsened_dep: CoarsenedTarget) -> bool:	9✔
425	if not request.ignore_generated:	8✔
426	return False	8✔
427	if len(coarsened_dep.members) != 1:	1✔
428	# Do not ignore a dependency which is involved in a cycle.
429	return False	×
430	us = request.request.component.representative.address	1✔
431	them = coarsened_dep.representative.address	1✔
432	return us.spec_path == them.spec_path and us.target_name == them.target_name	1✔
433
434	def ignore_because_file(coarsened_dep: CoarsenedTarget) -> bool:	9✔
435	return sum(	8✔
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(	9✔
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	11✔
453	async def compile_classpath_entries(requests: ClasspathEntryRequests) -> FallibleClasspathEntries:	11✔
454	return FallibleClasspathEntries(	9✔
455	await concurrently(
456	get_fallible_classpath_entry(**implicitly({request: ClasspathEntryRequest}))
457	for request in requests
458	)
459	)
460
461
462	def rules():	11✔
463	return collect_rules()	11✔

pantsbuild / pants / 26342152999

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