18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 hits per line

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

37.67

/src/python/pants/engine/internals/specs_rules.py

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

from __future__ import annotations

import dataclasses
import itertools
import logging
import os
from collections import defaultdict
from collections.abc import Iterable

# TODO: This is a very fishy import. `internals` should not be importing from a backend.
from pants.backend.project_info.filter_targets import FilterSubsystem
from pants.base.specs import (
    AddressLiteralSpec,
    AncestorGlobSpec,
    DirGlobSpec,
    DirLiteralSpec,
    RawSpecs,
    RawSpecsWithOnlyFileOwners,
    RawSpecsWithoutFileOwners,
    RecursiveGlobSpec,
    Specs,
)
from pants.engine.addresses import Address, Addresses, AddressInput
from pants.engine.environment import ChosenLocalEnvironmentName, EnvironmentName
from pants.engine.fs import SpecsPaths
from pants.engine.internals.build_files import resolve_address
from pants.engine.internals.graph import (
    Owners,
    OwnersRequest,
    address_families_from_raw_specs_without_file_owners,
    filter_targets,
    find_owners,
    find_valid_field_sets,
    resolve_source_paths,
    resolve_target,
    resolve_target_parametrizations,
    resolve_targets,
)
from pants.engine.internals.mapper import SpecsFilter
from pants.engine.internals.parametrize import _TargetParametrizationsRequest
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import path_globs_to_paths
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import (
    FieldSet,
    FieldSetsPerTargetRequest,
    NoApplicableTargetsBehavior,
    RegisteredTargetTypes,
    SourcesField,
    SourcesPathsRequest,
    Target,
    TargetGenerator,
    TargetRootsToFieldSets,
    TargetRootsToFieldSetsRequest,
    WrappedTarget,
    WrappedTargetRequest,
)
from pants.engine.unions import UnionMembership
from pants.option.global_options import GlobalOptions
from pants.util.dirutil import recursive_dirname
from pants.util.docutil import bin_name
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet, OrderedSet
from pants.util.strutil import bullet_list

logger = logging.getLogger(__name__)


# -----------------------------------------------------------------------------------------------
# RawSpecsWithoutFileOwners -> Targets
# -----------------------------------------------------------------------------------------------


async def _determine_literal_addresses_from_raw_specs(
    literal_specs: tuple[AddressLiteralSpec, ...],
    local_environment_name: ChosenLocalEnvironmentName,
    *,
    description_of_origin: str,
) -> tuple[WrappedTarget, ...]:
    literal_addresses = await concurrently(
        resolve_address(
            **implicitly(
                AddressInput(
                    str(spec),
                    spec.path_component,
                    description_of_origin=description_of_origin,
                    target_component=spec.target_component,
                    generated_component=spec.generated_component,
                    parameters=dict(spec.parameters),
                )
            )
        )
        for spec in literal_specs
    )

    # We replace references to parametrized target templates with all their created targets. For
    # example:
    #  - dir:tgt -> (dir:tgt@k=v1, dir:tgt@k=v2)
    #  - dir:tgt@k=v -> (dir:tgt@k=v,another=a, dir:tgt@k=v,another=b), but not anything
    #       where @k=v is not true.
    literal_parametrizations = await concurrently(
        resolve_target_parametrizations(
            **implicitly(
                {
                    _TargetParametrizationsRequest(
                        address.maybe_convert_to_target_generator(),
                        description_of_origin=description_of_origin,
                    ): _TargetParametrizationsRequest,
                    local_environment_name.val: EnvironmentName,
                }
            )
        )
        for address in literal_addresses
    )

    # Note that if the address is not in the _TargetParametrizations, we must fall back to that
    # address's value. This will allow us to error that the address is invalid.
    all_candidate_addresses = itertools.chain.from_iterable(
        list(params.get_all_superset_targets(address)) or [address]
        for address, params in zip(literal_addresses, literal_parametrizations)
    )

    # We eagerly call the `WrappedTarget` rule because it will validate that every final address
    # actually exists, such as with generated target addresses.
    return await concurrently(
        resolve_target(
            WrappedTargetRequest(addr, description_of_origin=description_of_origin), **implicitly()
        )
        for addr in all_candidate_addresses
    )


@rule(_masked_types=[EnvironmentName])
async def addresses_from_raw_specs_without_file_owners(
    specs: RawSpecsWithoutFileOwners,
    specs_filter: SpecsFilter,
    local_environment_name: ChosenLocalEnvironmentName,
) -> Addresses:
    matched_addresses: OrderedSet[Address] = OrderedSet()
    filtering_disabled = specs.filter_by_global_options is False

    literal_wrapped_targets = await _determine_literal_addresses_from_raw_specs(
        specs.address_literals,
        local_environment_name,
        description_of_origin=specs.description_of_origin,
    )
    matched_addresses.update(
        wrapped_tgt.target.address
        for wrapped_tgt in literal_wrapped_targets
        if filtering_disabled or specs_filter.matches(wrapped_tgt.target)
    )

    address_families = await address_families_from_raw_specs_without_file_owners(
        specs, **implicitly()
    )
    if not address_families:
        return Addresses(matched_addresses)

    base_addresses = address_families.addresses()

    target_parametrizations_list = await concurrently(
        resolve_target_parametrizations(
            **implicitly(
                {
                    _TargetParametrizationsRequest(
                        base_address, description_of_origin=specs.description_of_origin
                    ): _TargetParametrizationsRequest,
                    local_environment_name.val: EnvironmentName,
                }
            )
        )
        for base_address in base_addresses
    )
    residence_dir_to_targets = defaultdict(list)
    for target_parametrizations in target_parametrizations_list:
        for tgt in target_parametrizations.all:
            residence_dir_to_targets[tgt.residence_dir].append(tgt)

    def valid_tgt(
        tgt: Target, spec: DirLiteralSpec | DirGlobSpec | RecursiveGlobSpec | AncestorGlobSpec
    ) -> bool:
        if not spec.matches_target_generators and isinstance(tgt, TargetGenerator):
            return False
        return filtering_disabled or specs_filter.matches(tgt)

    for glob_spec in specs.glob_specs():
        for residence_dir in residence_dir_to_targets:
            if not glob_spec.matches_target_residence_dir(residence_dir):
                continue
            matched_addresses.update(
                tgt.address
                for tgt in residence_dir_to_targets[residence_dir]
                if valid_tgt(tgt, glob_spec)
            )

    return Addresses(sorted(matched_addresses))


# -----------------------------------------------------------------------------------------------
# RawSpecsWithOnlyFileOwners -> Targets
# -----------------------------------------------------------------------------------------------


@rule(_masked_types=[EnvironmentName])
async def addresses_from_raw_specs_with_only_file_owners(
    specs: RawSpecsWithOnlyFileOwners,
) -> Owners:
    """Find the owner(s) for each spec."""
    paths_per_include = await concurrently(
        path_globs_to_paths(specs.path_globs_for_spec(spec)) for spec in specs.all_specs()
    )
    all_files = tuple(itertools.chain.from_iterable(paths.files for paths in paths_per_include))
    owners = await find_owners(
        OwnersRequest(
            all_files,
            filter_by_global_options=specs.filter_by_global_options,
            # Specifying a BUILD file should not expand to all the targets it defines.
            match_if_owning_build_file_included_in_sources=False,
        ),
        **implicitly(),
    )
    return owners


@rule(_masked_types=[EnvironmentName])
async def addresses_from_owners(owners: Owners) -> Addresses:
    return Addresses(sorted(owners))


# -----------------------------------------------------------------------------------------------
# RawSpecs & Specs -> Targets
# -----------------------------------------------------------------------------------------------


@rule(_masked_types=[EnvironmentName])
async def resolve_addresses_from_raw_specs(specs: RawSpecs) -> Addresses:
    without_file_owners, with_file_owners = await concurrently(
        addresses_from_raw_specs_without_file_owners(
            RawSpecsWithoutFileOwners.from_raw_specs(specs), **implicitly()
        ),
        addresses_from_owners(**implicitly(RawSpecsWithOnlyFileOwners.from_raw_specs(specs))),
    )
    # Use a set to dedupe.
    return Addresses(sorted({*without_file_owners, *with_file_owners}))


@rule(desc="Find targets from input specs", level=LogLevel.DEBUG, _masked_types=[EnvironmentName])
async def resolve_addresses_from_specs(specs: Specs) -> Addresses:
    includes, ignores = await concurrently(
        resolve_addresses_from_raw_specs(specs.includes),
        resolve_addresses_from_raw_specs(specs.ignores),
    )
    # No matter what, ignores win out over includes. This avoids "specificity wars" and keeps our
    # semantics simple/predictable.
    return Addresses(FrozenOrderedSet(includes) - FrozenOrderedSet(ignores))


@rule
async def setup_specs_filter(
    global_options: GlobalOptions,
    filter_subsystem: FilterSubsystem,
    registered_target_types: RegisteredTargetTypes,
) -> SpecsFilter:
    return SpecsFilter.create(filter_subsystem, registered_target_types, tags=global_options.tag)


# -----------------------------------------------------------------------------------------------
# SpecsPaths
# -----------------------------------------------------------------------------------------------


@rule(desc="Find all sources from input specs", level=LogLevel.DEBUG)
async def resolve_specs_paths(specs: Specs) -> SpecsPaths:
    """Resolve all files matching the given specs.

    All matched targets will use their `sources` field. Certain specs like FileLiteralSpec will
    also match against all their files, regardless of if a target owns them.

    Ignores win out over includes, with these edge cases:

    * Ignored paths: the resolved paths should be excluded.
    * Ignored targets: their `sources` should be excluded.
    * File owned by a target that gets filtered out, e.g. via `--tag`. See
      https://github.com/pantsbuild/pants/issues/15478.
    """

    unfiltered_include_targets, ignore_targets, include_paths, ignore_paths = await concurrently(
        resolve_targets(
            **implicitly(
                {dataclasses.replace(specs.includes, filter_by_global_options=False): RawSpecs}
            )
        ),
        resolve_targets(**implicitly(specs.ignores)),
        path_globs_to_paths(specs.includes.to_specs_paths_path_globs()),
        path_globs_to_paths(specs.ignores.to_specs_paths_path_globs()),
    )

    filtered_include_targets = await filter_targets(unfiltered_include_targets, **implicitly())

    include_targets_sources_paths = await concurrently(
        resolve_source_paths(SourcesPathsRequest(tgt[SourcesField]), **implicitly())
        for tgt in filtered_include_targets
        if tgt.has_field(SourcesField)
    )

    ignore_targets_sources_paths = await concurrently(
        resolve_source_paths(SourcesPathsRequest(tgt[SourcesField]), **implicitly())
        for tgt in ignore_targets
        if tgt.has_field(SourcesField)
    )

    result_paths = OrderedSet(
        itertools.chain.from_iterable(paths.files for paths in include_targets_sources_paths),
    )
    result_paths.update(include_paths.files)
    result_paths.difference_update(
        itertools.chain.from_iterable(paths.files for paths in ignore_targets_sources_paths)
    )
    result_paths.difference_update(ignore_paths.files)

    # If include paths were given, we need to also remove any paths from filtered out targets
    # (e.g. via `--tag`), per https://github.com/pantsbuild/pants/issues/15478.
    if include_paths.files:
        filtered_out_include_targets = FrozenOrderedSet(unfiltered_include_targets).difference(
            FrozenOrderedSet(filtered_include_targets)
        )
        filtered_include_targets_sources_paths = await concurrently(
            resolve_source_paths(SourcesPathsRequest(tgt[SourcesField]), **implicitly())
            for tgt in filtered_out_include_targets
            if tgt.has_field(SourcesField)
        )
        result_paths.difference_update(
            itertools.chain.from_iterable(
                paths.files for paths in filtered_include_targets_sources_paths
            )
        )

    dirs = OrderedSet(
        itertools.chain.from_iterable(recursive_dirname(os.path.dirname(f)) for f in result_paths)
    ) - {""}
    return SpecsPaths(tuple(sorted(result_paths)), tuple(sorted(dirs)))


# -----------------------------------------------------------------------------------------------
# RawSpecs -> FieldSets
# -----------------------------------------------------------------------------------------------


class NoApplicableTargetsException(Exception):
    def __init__(
        self,
        targets: Iterable[Target],
        specs: Specs,
        union_membership: UnionMembership,
        *,
        applicable_target_types: Iterable[type[Target]],
        goal_description: str,
    ) -> None:
        applicable_target_aliases = sorted(
            {target_type.alias for target_type in applicable_target_types}
        )
        inapplicable_target_aliases = sorted({tgt.alias for tgt in targets})
        msg = (
            "No applicable files or targets matched."
            if inapplicable_target_aliases
            else "No files or targets specified."
        )
        msg += (
            f" {goal_description.capitalize()} works "
            f"with these target types:\n\n"
            f"{bullet_list(applicable_target_aliases)}\n\n"
        )

        # Explain what was specified, if relevant.
        if inapplicable_target_aliases:
            specs_description = specs.arguments_provided_description() or ""
            if specs_description:
                specs_description = f" {specs_description} with"
            msg += (
                f"However, you only specified{specs_description} these target types:\n\n"
                f"{bullet_list(inapplicable_target_aliases)}\n\n"
            )

        # Add a remedy.
        #
        # We sometimes suggest using `./pants filedeps` to find applicable files. However, this
        # command only works if at least one of the targets has a SourcesField field.
        filedeps_goal_works = any(
            tgt.class_has_field(SourcesField, union_membership) for tgt in applicable_target_types
        )
        pants_filter_command = (
            f"{bin_name()} --filter-target-type={','.join(applicable_target_aliases)}"
        )
        remedy = (
            f"Please specify relevant file and/or target arguments. Run `{pants_filter_command} "
            f"list ::` to find all applicable targets in your project"
        )
        if filedeps_goal_works:
            remedy += f", or run `{pants_filter_command} filedeps ::` to find all applicable files."
        else:
            remedy += "."
        msg += remedy
        super().__init__(msg)

    @classmethod
    def create_from_field_sets(
        cls,
        targets: Iterable[Target],
        specs: Specs,
        union_membership: UnionMembership,
        registered_target_types: RegisteredTargetTypes,
        *,
        field_set_types: Iterable[type[FieldSet]],
        goal_description: str,
    ) -> NoApplicableTargetsException:
        applicable_target_types = {
            target_type
            for field_set_type in field_set_types
            for target_type in field_set_type.applicable_target_types(
                registered_target_types.types, union_membership
            )
        }
        return cls(
            targets,
            specs,
            union_membership,
            applicable_target_types=applicable_target_types,
            goal_description=goal_description,
        )


class TooManyTargetsException(Exception):
    def __init__(self, targets: Iterable[Target], *, goal_description: str) -> None:
        addresses = sorted(tgt.address.spec for tgt in targets)
        super().__init__(
            f"{goal_description.capitalize()} only works with one valid target, but was given "
            f"multiple valid targets:\n\n{bullet_list(addresses)}\n\n"
            "Please select one of these targets to run."
        )


class AmbiguousImplementationsException(Exception):
    """A target has multiple valid FieldSets, but a goal expects there to be one FieldSet."""

    def __init__(
        self,
        target: Target,
        field_sets: Iterable[FieldSet],
        *,
        goal_description: str,
    ) -> None:
        # TODO: improve this error message. A better error message would explain to users how they
        #  can resolve the issue.
        possible_field_sets_types = sorted(field_set.__class__.__name__ for field_set in field_sets)
        super().__init__(
            f"Multiple of the registered implementations for {goal_description} work for "
            f"{target.address} (target type {repr(target.alias)}). It is ambiguous which "
            "implementation to use.\n\nPossible implementations:\n\n"
            f"{bullet_list(possible_field_sets_types)}"
        )


@rule
async def find_valid_field_sets_for_target_roots(
    request: TargetRootsToFieldSetsRequest,
    specs: Specs,
    union_membership: UnionMembership,
    registered_target_types: RegisteredTargetTypes,
) -> TargetRootsToFieldSets:
    # NB: This must be in an `await Get`, rather than the rule signature, to avoid a rule graph
    # issue.
    targets = await filter_targets(**implicitly({specs: Specs}))
    field_sets_per_target = await find_valid_field_sets(
        FieldSetsPerTargetRequest(request.field_set_superclass, targets), **implicitly()
    )
    targets_to_applicable_field_sets = {}
    for tgt, field_sets in zip(targets, field_sets_per_target.collection):
        if field_sets:
            targets_to_applicable_field_sets[tgt] = field_sets

    # Possibly warn or error if no targets were applicable.
    if not targets_to_applicable_field_sets:
        no_applicable_exception = NoApplicableTargetsException.create_from_field_sets(
            targets,
            specs,
            union_membership,
            registered_target_types,
            field_set_types=union_membership.get(request.field_set_superclass),
            goal_description=request.goal_description,
        )
        if request.no_applicable_targets_behavior == NoApplicableTargetsBehavior.error:
            raise no_applicable_exception

        # We squelch the warning if the specs came from change detection or only from globs,
        # since in that case we interpret the user's intent as "if there are relevant matching
        # targets, act on them". But we still want to warn if the specs were literal, or empty.
        #
        # No need to check `specs.ignores` here, as change detection will not set that. Likewise,
        # we don't want an ignore spec to trigger this warning, even if it was a literal.
        empty_ok = specs.includes.from_change_detection or (
            specs.includes
            and not specs.includes.address_literals
            and not specs.includes.file_literals
        )
        if (
            request.no_applicable_targets_behavior == NoApplicableTargetsBehavior.warn
            and not empty_ok
        ):
            logger.warning(str(no_applicable_exception))

    if request.num_shards > 0:
        sharded_targets_to_applicable_field_sets = {
            tgt: value
            for tgt, value in targets_to_applicable_field_sets.items()
            if request.is_in_shard(tgt.address.spec)
        }
        return TargetRootsToFieldSets(sharded_targets_to_applicable_field_sets)
    return TargetRootsToFieldSets(targets_to_applicable_field_sets)


def rules():
    return collect_rules()

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	3✔
5
6	import dataclasses	3✔
7	import itertools	3✔
8	import logging	3✔
9	import os	3✔
10	from collections import defaultdict	3✔
11	from collections.abc import Iterable	3✔
12
13	# TODO: This is a very fishy import. `internals` should not be importing from a backend.
14	from pants.backend.project_info.filter_targets import FilterSubsystem	3✔
15	from pants.base.specs import (	3✔
16	AddressLiteralSpec,
17	AncestorGlobSpec,
18	DirGlobSpec,
19	DirLiteralSpec,
20	RawSpecs,
21	RawSpecsWithOnlyFileOwners,
22	RawSpecsWithoutFileOwners,
23	RecursiveGlobSpec,
24	Specs,
25	)
26	from pants.engine.addresses import Address, Addresses, AddressInput	3✔
27	from pants.engine.environment import ChosenLocalEnvironmentName, EnvironmentName	3✔
28	from pants.engine.fs import SpecsPaths	3✔
29	from pants.engine.internals.build_files import resolve_address	3✔
30	from pants.engine.internals.graph import (	3✔
31	Owners,
32	OwnersRequest,
33	address_families_from_raw_specs_without_file_owners,
34	filter_targets,
35	find_owners,
36	find_valid_field_sets,
37	resolve_source_paths,
38	resolve_target,
39	resolve_target_parametrizations,
40	resolve_targets,
41	)
42	from pants.engine.internals.mapper import SpecsFilter	3✔
43	from pants.engine.internals.parametrize import _TargetParametrizationsRequest	3✔
44	from pants.engine.internals.selectors import concurrently	3✔
45	from pants.engine.intrinsics import path_globs_to_paths	3✔
46	from pants.engine.rules import collect_rules, implicitly, rule	3✔
47	from pants.engine.target import (	3✔
48	FieldSet,
49	FieldSetsPerTargetRequest,
50	NoApplicableTargetsBehavior,
51	RegisteredTargetTypes,
52	SourcesField,
53	SourcesPathsRequest,
54	Target,
55	TargetGenerator,
56	TargetRootsToFieldSets,
57	TargetRootsToFieldSetsRequest,
58	WrappedTarget,
59	WrappedTargetRequest,
60	)
61	from pants.engine.unions import UnionMembership	3✔
62	from pants.option.global_options import GlobalOptions	3✔
63	from pants.util.dirutil import recursive_dirname	3✔
64	from pants.util.docutil import bin_name	3✔
65	from pants.util.logging import LogLevel	3✔
66	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	3✔
67	from pants.util.strutil import bullet_list	3✔
68
69	logger = logging.getLogger(__name__)	3✔
70
71
72	# -----------------------------------------------------------------------------------------------
73	# RawSpecsWithoutFileOwners -> Targets
74	# -----------------------------------------------------------------------------------------------
75
76
77	async def _determine_literal_addresses_from_raw_specs(	3✔
78	literal_specs: tuple[AddressLiteralSpec, ...],
79	local_environment_name: ChosenLocalEnvironmentName,
80	*,
81	description_of_origin: str,
82	) -> tuple[WrappedTarget, ...]:
83	literal_addresses = await concurrently(	×
84	resolve_address(
85	**implicitly(
86	AddressInput(
87	str(spec),
88	spec.path_component,
89	description_of_origin=description_of_origin,
90	target_component=spec.target_component,
91	generated_component=spec.generated_component,
92	parameters=dict(spec.parameters),
93	)
94	)
95	)
96	for spec in literal_specs
97	)
98
99	# We replace references to parametrized target templates with all their created targets. For
100	# example:
101	# - dir:tgt -> (dir:tgt@k=v1, dir:tgt@k=v2)
102	# - dir:tgt@k=v -> (dir:tgt@k=v,another=a, dir:tgt@k=v,another=b), but not anything
103	# where @k=v is not true.
104	literal_parametrizations = await concurrently(	×
105	resolve_target_parametrizations(
106	**implicitly(
107	{
108	_TargetParametrizationsRequest(
109	address.maybe_convert_to_target_generator(),
110	description_of_origin=description_of_origin,
111	): _TargetParametrizationsRequest,
112	local_environment_name.val: EnvironmentName,
113	}
114	)
115	)
116	for address in literal_addresses
117	)
118
119	# Note that if the address is not in the _TargetParametrizations, we must fall back to that
120	# address's value. This will allow us to error that the address is invalid.
121	all_candidate_addresses = itertools.chain.from_iterable(	×
122	list(params.get_all_superset_targets(address)) or [address]
123	for address, params in zip(literal_addresses, literal_parametrizations)
124	)
125
126	# We eagerly call the `WrappedTarget` rule because it will validate that every final address
127	# actually exists, such as with generated target addresses.
128	return await concurrently(	×
129	resolve_target(
130	WrappedTargetRequest(addr, description_of_origin=description_of_origin), **implicitly()
131	)
132	for addr in all_candidate_addresses
133	)
134
135
136	@rule(_masked_types=[EnvironmentName])	3✔
137	async def addresses_from_raw_specs_without_file_owners(	3✔
138	specs: RawSpecsWithoutFileOwners,
139	specs_filter: SpecsFilter,
140	local_environment_name: ChosenLocalEnvironmentName,
141	) -> Addresses:
142	matched_addresses: OrderedSet[Address] = OrderedSet()	×
143	filtering_disabled = specs.filter_by_global_options is False	×
144
145	literal_wrapped_targets = await _determine_literal_addresses_from_raw_specs(	×
146	specs.address_literals,
147	local_environment_name,
148	description_of_origin=specs.description_of_origin,
149	)
150	matched_addresses.update(	×
151	wrapped_tgt.target.address
152	for wrapped_tgt in literal_wrapped_targets
153	if filtering_disabled or specs_filter.matches(wrapped_tgt.target)
154	)
155
156	address_families = await address_families_from_raw_specs_without_file_owners(	×
157	specs, **implicitly()
158	)
159	if not address_families:	×
160	return Addresses(matched_addresses)	×
161
162	base_addresses = address_families.addresses()	×
163
164	target_parametrizations_list = await concurrently(	×
165	resolve_target_parametrizations(
166	**implicitly(
167	{
168	_TargetParametrizationsRequest(
169	base_address, description_of_origin=specs.description_of_origin
170	): _TargetParametrizationsRequest,
171	local_environment_name.val: EnvironmentName,
172	}
173	)
174	)
175	for base_address in base_addresses
176	)
177	residence_dir_to_targets = defaultdict(list)	×
178	for target_parametrizations in target_parametrizations_list:	×
179	for tgt in target_parametrizations.all:	×
180	residence_dir_to_targets[tgt.residence_dir].append(tgt)	×
181
182	def valid_tgt(	×
183	tgt: Target, spec: DirLiteralSpec \| DirGlobSpec \| RecursiveGlobSpec \| AncestorGlobSpec
184	) -> bool:
185	if not spec.matches_target_generators and isinstance(tgt, TargetGenerator):	×
186	return False	×
187	return filtering_disabled or specs_filter.matches(tgt)	×
188
189	for glob_spec in specs.glob_specs():	×
190	for residence_dir in residence_dir_to_targets:	×
191	if not glob_spec.matches_target_residence_dir(residence_dir):	×
192	continue	×
193	matched_addresses.update(	×
194	tgt.address
195	for tgt in residence_dir_to_targets[residence_dir]
196	if valid_tgt(tgt, glob_spec)
197	)
198
199	return Addresses(sorted(matched_addresses))	×
200
201
202	# -----------------------------------------------------------------------------------------------
203	# RawSpecsWithOnlyFileOwners -> Targets
204	# -----------------------------------------------------------------------------------------------
205
206
207	@rule(_masked_types=[EnvironmentName])	3✔
208	async def addresses_from_raw_specs_with_only_file_owners(	3✔
209	specs: RawSpecsWithOnlyFileOwners,
210	) -> Owners:
211	"""Find the owner(s) for each spec."""
212	paths_per_include = await concurrently(	×
213	path_globs_to_paths(specs.path_globs_for_spec(spec)) for spec in specs.all_specs()
214	)
215	all_files = tuple(itertools.chain.from_iterable(paths.files for paths in paths_per_include))	×
216	owners = await find_owners(	×
217	OwnersRequest(
218	all_files,
219	filter_by_global_options=specs.filter_by_global_options,
220	# Specifying a BUILD file should not expand to all the targets it defines.
221	match_if_owning_build_file_included_in_sources=False,
222	),
223	**implicitly(),
224	)
225	return owners	×
226
227
228	@rule(_masked_types=[EnvironmentName])	3✔
229	async def addresses_from_owners(owners: Owners) -> Addresses:	3✔
230	return Addresses(sorted(owners))	×
231
232
233	# -----------------------------------------------------------------------------------------------
234	# RawSpecs & Specs -> Targets
235	# -----------------------------------------------------------------------------------------------
236
237
238	@rule(_masked_types=[EnvironmentName])	3✔
239	async def resolve_addresses_from_raw_specs(specs: RawSpecs) -> Addresses:	3✔
240	without_file_owners, with_file_owners = await concurrently(	×
241	addresses_from_raw_specs_without_file_owners(
242	RawSpecsWithoutFileOwners.from_raw_specs(specs), **implicitly()
243	),
244	addresses_from_owners(**implicitly(RawSpecsWithOnlyFileOwners.from_raw_specs(specs))),
245	)
246	# Use a set to dedupe.
247	return Addresses(sorted({without_file_owners, with_file_owners}))	×
248
249
250	@rule(desc="Find targets from input specs", level=LogLevel.DEBUG, _masked_types=[EnvironmentName])	3✔
251	async def resolve_addresses_from_specs(specs: Specs) -> Addresses:	3✔
252	includes, ignores = await concurrently(	×
253	resolve_addresses_from_raw_specs(specs.includes),
254	resolve_addresses_from_raw_specs(specs.ignores),
255	)
256	# No matter what, ignores win out over includes. This avoids "specificity wars" and keeps our
257	# semantics simple/predictable.
258	return Addresses(FrozenOrderedSet(includes) - FrozenOrderedSet(ignores))	×
259
260
261	@rule	3✔
262	async def setup_specs_filter(	3✔
263	global_options: GlobalOptions,
264	filter_subsystem: FilterSubsystem,
265	registered_target_types: RegisteredTargetTypes,
266	) -> SpecsFilter:
267	return SpecsFilter.create(filter_subsystem, registered_target_types, tags=global_options.tag)	×
268
269
270	# -----------------------------------------------------------------------------------------------
271	# SpecsPaths
272	# -----------------------------------------------------------------------------------------------
273
274
275	@rule(desc="Find all sources from input specs", level=LogLevel.DEBUG)	3✔
276	async def resolve_specs_paths(specs: Specs) -> SpecsPaths:	3✔
277	"""Resolve all files matching the given specs.
278
279	All matched targets will use their `sources` field. Certain specs like FileLiteralSpec will
280	also match against all their files, regardless of if a target owns them.
281
282	Ignores win out over includes, with these edge cases:
283
284	* Ignored paths: the resolved paths should be excluded.
285	* Ignored targets: their `sources` should be excluded.
286	* File owned by a target that gets filtered out, e.g. via `--tag`. See
287	https://github.com/pantsbuild/pants/issues/15478.
288	"""
289
290	unfiltered_include_targets, ignore_targets, include_paths, ignore_paths = await concurrently(	×
291	resolve_targets(
292	**implicitly(
293	{dataclasses.replace(specs.includes, filter_by_global_options=False): RawSpecs}
294	)
295	),
296	resolve_targets(**implicitly(specs.ignores)),
297	path_globs_to_paths(specs.includes.to_specs_paths_path_globs()),
298	path_globs_to_paths(specs.ignores.to_specs_paths_path_globs()),
299	)
300
301	filtered_include_targets = await filter_targets(unfiltered_include_targets, **implicitly())	×
302
303	include_targets_sources_paths = await concurrently(	×
304	resolve_source_paths(SourcesPathsRequest(tgt[SourcesField]), **implicitly())
305	for tgt in filtered_include_targets
306	if tgt.has_field(SourcesField)
307	)
308
309	ignore_targets_sources_paths = await concurrently(	×
310	resolve_source_paths(SourcesPathsRequest(tgt[SourcesField]), **implicitly())
311	for tgt in ignore_targets
312	if tgt.has_field(SourcesField)
313	)
314
315	result_paths = OrderedSet(	×
316	itertools.chain.from_iterable(paths.files for paths in include_targets_sources_paths),
317	)
318	result_paths.update(include_paths.files)	×
319	result_paths.difference_update(	×
320	itertools.chain.from_iterable(paths.files for paths in ignore_targets_sources_paths)
321	)
322	result_paths.difference_update(ignore_paths.files)	×
323
324	# If include paths were given, we need to also remove any paths from filtered out targets
325	# (e.g. via `--tag`), per https://github.com/pantsbuild/pants/issues/15478.
326	if include_paths.files:	×
327	filtered_out_include_targets = FrozenOrderedSet(unfiltered_include_targets).difference(	×
328	FrozenOrderedSet(filtered_include_targets)
329	)
330	filtered_include_targets_sources_paths = await concurrently(	×
331	resolve_source_paths(SourcesPathsRequest(tgt[SourcesField]), **implicitly())
332	for tgt in filtered_out_include_targets
333	if tgt.has_field(SourcesField)
334	)
335	result_paths.difference_update(	×
336	itertools.chain.from_iterable(
337	paths.files for paths in filtered_include_targets_sources_paths
338	)
339	)
340
341	dirs = OrderedSet(	×
342	itertools.chain.from_iterable(recursive_dirname(os.path.dirname(f)) for f in result_paths)
343	) - {""}
344	return SpecsPaths(tuple(sorted(result_paths)), tuple(sorted(dirs)))	×
345
346
347	# -----------------------------------------------------------------------------------------------
348	# RawSpecs -> FieldSets
349	# -----------------------------------------------------------------------------------------------
350
351
352	class NoApplicableTargetsException(Exception):	3✔
353	def __init__(	3✔
354	self,
355	targets: Iterable[Target],
356	specs: Specs,
357	union_membership: UnionMembership,
358	*,
359	applicable_target_types: Iterable[type[Target]],
360	goal_description: str,
361	) -> None:
362	applicable_target_aliases = sorted(	×
363	{target_type.alias for target_type in applicable_target_types}
364	)
365	inapplicable_target_aliases = sorted({tgt.alias for tgt in targets})	×
366	msg = (	×
367	"No applicable files or targets matched."
368	if inapplicable_target_aliases
369	else "No files or targets specified."
370	)
371	msg += (	×
372	f" {goal_description.capitalize()} works "
373	f"with these target types:\n\n"
374	f"{bullet_list(applicable_target_aliases)}\n\n"
375	)
376
377	# Explain what was specified, if relevant.
378	if inapplicable_target_aliases:	×
379	specs_description = specs.arguments_provided_description() or ""	×
380	if specs_description:	×
381	specs_description = f" {specs_description} with"	×
382	msg += (	×
383	f"However, you only specified{specs_description} these target types:\n\n"
384	f"{bullet_list(inapplicable_target_aliases)}\n\n"
385	)
386
387	# Add a remedy.
388	#
389	# We sometimes suggest using `./pants filedeps` to find applicable files. However, this
390	# command only works if at least one of the targets has a SourcesField field.
391	filedeps_goal_works = any(	×
392	tgt.class_has_field(SourcesField, union_membership) for tgt in applicable_target_types
393	)
394	pants_filter_command = (	×
395	f"{bin_name()} --filter-target-type={','.join(applicable_target_aliases)}"
396	)
397	remedy = (	×
398	f"Please specify relevant file and/or target arguments. Run `{pants_filter_command} "
399	f"list ::` to find all applicable targets in your project"
400	)
401	if filedeps_goal_works:	×
402	remedy += f", or run `{pants_filter_command} filedeps ::` to find all applicable files."	×
403	else:
404	remedy += "."	×
405	msg += remedy	×
406	super().__init__(msg)	×
407
408	@classmethod	3✔
409	def create_from_field_sets(	3✔
410	cls,
411	targets: Iterable[Target],
412	specs: Specs,
413	union_membership: UnionMembership,
414	registered_target_types: RegisteredTargetTypes,
415	*,
416	field_set_types: Iterable[type[FieldSet]],
417	goal_description: str,
418	) -> NoApplicableTargetsException:
419	applicable_target_types = {	×
420	target_type
421	for field_set_type in field_set_types
422	for target_type in field_set_type.applicable_target_types(
423	registered_target_types.types, union_membership
424	)
425	}
426	return cls(	×
427	targets,
428	specs,
429	union_membership,
430	applicable_target_types=applicable_target_types,
431	goal_description=goal_description,
432	)
433
434
435	class TooManyTargetsException(Exception):	3✔
436	def __init__(self, targets: Iterable[Target], *, goal_description: str) -> None:	3✔
437	addresses = sorted(tgt.address.spec for tgt in targets)	×
438	super().__init__(	×
439	f"{goal_description.capitalize()} only works with one valid target, but was given "
440	f"multiple valid targets:\n\n{bullet_list(addresses)}\n\n"
441	"Please select one of these targets to run."
442	)
443
444
445	class AmbiguousImplementationsException(Exception):	3✔
446	"""A target has multiple valid FieldSets, but a goal expects there to be one FieldSet."""
447
448	def __init__(	3✔
449	self,
450	target: Target,
451	field_sets: Iterable[FieldSet],
452	*,
453	goal_description: str,
454	) -> None:
455	# TODO: improve this error message. A better error message would explain to users how they
456	# can resolve the issue.
457	possible_field_sets_types = sorted(field_set.__class__.__name__ for field_set in field_sets)	×
458	super().__init__(	×
459	f"Multiple of the registered implementations for {goal_description} work for "
460	f"{target.address} (target type {repr(target.alias)}). It is ambiguous which "
461	"implementation to use.\n\nPossible implementations:\n\n"
462	f"{bullet_list(possible_field_sets_types)}"
463	)
464
465
466	@rule	3✔
467	async def find_valid_field_sets_for_target_roots(	3✔
468	request: TargetRootsToFieldSetsRequest,
469	specs: Specs,
470	union_membership: UnionMembership,
471	registered_target_types: RegisteredTargetTypes,
472	) -> TargetRootsToFieldSets:
473	# NB: This must be in an `await Get`, rather than the rule signature, to avoid a rule graph
474	# issue.
475	targets = await filter_targets(**implicitly({specs: Specs}))	×
476	field_sets_per_target = await find_valid_field_sets(	×
477	FieldSetsPerTargetRequest(request.field_set_superclass, targets), **implicitly()
478	)
479	targets_to_applicable_field_sets = {}	×
480	for tgt, field_sets in zip(targets, field_sets_per_target.collection):	×
481	if field_sets:	×
482	targets_to_applicable_field_sets[tgt] = field_sets	×
483
484	# Possibly warn or error if no targets were applicable.
485	if not targets_to_applicable_field_sets:	×
486	no_applicable_exception = NoApplicableTargetsException.create_from_field_sets(	×
487	targets,
488	specs,
489	union_membership,
490	registered_target_types,
491	field_set_types=union_membership.get(request.field_set_superclass),
492	goal_description=request.goal_description,
493	)
494	if request.no_applicable_targets_behavior == NoApplicableTargetsBehavior.error:	×
495	raise no_applicable_exception	×
496
497	# We squelch the warning if the specs came from change detection or only from globs,
498	# since in that case we interpret the user's intent as "if there are relevant matching
499	# targets, act on them". But we still want to warn if the specs were literal, or empty.
500	#
501	# No need to check `specs.ignores` here, as change detection will not set that. Likewise,
502	# we don't want an ignore spec to trigger this warning, even if it was a literal.
503	empty_ok = specs.includes.from_change_detection or (	×
504	specs.includes
505	and not specs.includes.address_literals
506	and not specs.includes.file_literals
507	)
508	if (	×
509	request.no_applicable_targets_behavior == NoApplicableTargetsBehavior.warn
510	and not empty_ok
511	):
512	logger.warning(str(no_applicable_exception))	×
513
514	if request.num_shards > 0:	×
515	sharded_targets_to_applicable_field_sets = {	×
516	tgt: value
517	for tgt, value in targets_to_applicable_field_sets.items()
518	if request.is_in_shard(tgt.address.spec)
519	}
520	return TargetRootsToFieldSets(sharded_targets_to_applicable_field_sets)	×
521	return TargetRootsToFieldSets(targets_to_applicable_field_sets)	×
522
523
524	def rules():	3✔
525	return collect_rules()	3✔

pantsbuild / pants / 18252174847

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