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

0.0

/src/python/pants/backend/project_info/paths.py

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

from __future__ import annotations

import json
from collections import deque
from collections.abc import Iterable
from dataclasses import dataclass

from pants.base.specs import Specs
from pants.base.specs_parser import SpecsParser
from pants.engine.addresses import Address
from pants.engine.console import Console
from pants.engine.goal import Goal, GoalSubsystem, Outputting
from pants.engine.internals.graph import resolve_targets
from pants.engine.internals.graph import transitive_targets as transitive_targets_get
from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly, rule
from pants.engine.target import (
    AlwaysTraverseDeps,
    Dependencies,
    DependenciesRequest,
    Target,
    Targets,
    TransitiveTargetsRequest,
)
from pants.option.option_types import StrOption


class PathsSubsystem(Outputting, GoalSubsystem):
    name = "paths"
    help = (
        "List the paths between two addresses. "
        "Either address may represent a group of targets, e.g. `--from=src/app/main.py --to=src/library::`."
    )

    from_ = StrOption(
        default=None,
        help="The path starting address",
    )

    to = StrOption(
        default=None,
        help="The path end address",
    )


class PathsGoal(Goal):
    subsystem_cls = PathsSubsystem
    environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY


def find_paths_breadth_first(
    adjacency_lists: dict[Address, Targets], from_target: Address, to_target: Address
) -> Iterable[list[Address]]:
    """Yields the paths between from_target to to_target if they exist.

    The paths are returned ordered by length, shortest first. If there are cycles, it checks visited
    edges to prevent recrossing them.
    """

    if from_target == to_target:
        yield [from_target]
        return

    visited_edges = set()
    to_walk_paths = deque([[from_target]])

    while len(to_walk_paths) > 0:
        cur_path = to_walk_paths.popleft()
        target = cur_path[-1]

        if len(cur_path) > 1:
            prev_target: Address | None = cur_path[-2]
        else:
            prev_target = None
        current_edge = (prev_target, target)

        if current_edge not in visited_edges:
            for dep in adjacency_lists.get(target, []):
                dep_path = cur_path + [dep.address]
                if dep.address == to_target:
                    yield dep_path
                else:
                    to_walk_paths.append(dep_path)
            visited_edges.add(current_edge)


@dataclass
class SpecsPaths:
    paths: list[list[str]]


@dataclass
class SpecsPathsCollection:
    spec_paths: list[SpecsPaths]


@dataclass(frozen=True)
class RootDestinationPair:
    root: Target
    destination: Target


@dataclass(frozen=True)
class RootDestinationsPair:
    root: Target
    destinations: Targets


@rule(desc="Get paths between root and destination.")
async def get_paths_between_root_and_destination(pair: RootDestinationPair) -> SpecsPaths:
    transitive_targets = await transitive_targets_get(
        TransitiveTargetsRequest(
            [pair.root.address], should_traverse_deps_predicate=AlwaysTraverseDeps()
        ),
        **implicitly(),
    )

    adjacent_targets_per_target = await concurrently(
        resolve_targets(
            **implicitly(
                DependenciesRequest(
                    tgt.get(Dependencies), should_traverse_deps_predicate=AlwaysTraverseDeps()
                )
            )
        )
        for tgt in transitive_targets.closure
    )

    transitive_targets_closure_addresses = (t.address for t in transitive_targets.closure)
    adjacency_lists = dict(zip(transitive_targets_closure_addresses, adjacent_targets_per_target))

    spec_paths = []
    for path in find_paths_breadth_first(
        adjacency_lists, pair.root.address, pair.destination.address
    ):
        spec_path = [address.spec for address in path]
        spec_paths.append(spec_path)

    return SpecsPaths(paths=spec_paths)


@rule(desc="Get paths between root and multiple destinations.")
async def get_paths_between_root_and_destinations(
    pair: RootDestinationsPair,
) -> SpecsPathsCollection:
    spec_paths = await concurrently(
        get_paths_between_root_and_destination(
            RootDestinationPair(destination=destination, root=pair.root)
        )
        for destination in pair.destinations
    )
    return SpecsPathsCollection(spec_paths=list(spec_paths))


@goal_rule
async def paths(console: Console, paths_subsystem: PathsSubsystem) -> PathsGoal:
    path_from = paths_subsystem.from_
    path_to = paths_subsystem.to

    if path_from is None:
        raise ValueError("Must set --from")

    if path_to is None:
        raise ValueError("Must set --to")

    specs_parser = SpecsParser()

    from_tgts, to_tgts = await concurrently(
        resolve_targets(
            **implicitly(
                {
                    specs_parser.parse_specs(
                        [path_from],
                        description_of_origin="the option `--paths-from`",
                    ): Specs
                }
            )
        ),
        resolve_targets(
            **implicitly(
                {
                    specs_parser.parse_specs(
                        [path_to],
                        description_of_origin="the option `--paths-to`",
                    ): Specs
                }
            )
        ),
    )

    all_spec_paths = []
    spec_paths = await concurrently(
        get_paths_between_root_and_destinations(
            RootDestinationsPair(root=root, destinations=to_tgts)
        )
        for root in from_tgts
    )

    for spec_path in spec_paths:
        for path in (p.paths for p in spec_path.spec_paths):
            all_spec_paths.extend(path)

    with paths_subsystem.output(console) as write_stdout:
        write_stdout(json.dumps(all_spec_paths, indent=2) + "\n")

    return PathsGoal(exit_code=0)


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
6	import json	×
7	from collections import deque	×
8	from collections.abc import Iterable	×
9	from dataclasses import dataclass	×
10
11	from pants.base.specs import Specs	×
12	from pants.base.specs_parser import SpecsParser	×
13	from pants.engine.addresses import Address	×
14	from pants.engine.console import Console	×
15	from pants.engine.goal import Goal, GoalSubsystem, Outputting	×
16	from pants.engine.internals.graph import resolve_targets	×
17	from pants.engine.internals.graph import transitive_targets as transitive_targets_get	×
18	from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly, rule	×
19	from pants.engine.target import (	×
20	AlwaysTraverseDeps,
21	Dependencies,
22	DependenciesRequest,
23	Target,
24	Targets,
25	TransitiveTargetsRequest,
26	)
27	from pants.option.option_types import StrOption	×
28
29
30	class PathsSubsystem(Outputting, GoalSubsystem):	×
31	name = "paths"	×
32	help = (	×
33	"List the paths between two addresses. "
34	"Either address may represent a group of targets, e.g. `--from=src/app/main.py --to=src/library::`."
35	)
36
37	from_ = StrOption(	×
38	default=None,
39	help="The path starting address",
40	)
41
42	to = StrOption(	×
43	default=None,
44	help="The path end address",
45	)
46
47
48	class PathsGoal(Goal):	×
49	subsystem_cls = PathsSubsystem	×
50	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	×
51
52
53	def find_paths_breadth_first(	×
54	adjacency_lists: dict[Address, Targets], from_target: Address, to_target: Address
55	) -> Iterable[list[Address]]:
56	"""Yields the paths between from_target to to_target if they exist.
57
58	The paths are returned ordered by length, shortest first. If there are cycles, it checks visited
59	edges to prevent recrossing them.
60	"""
61
62	if from_target == to_target:	×
63	yield [from_target]	×
64	return	×
65
66	visited_edges = set()	×
67	to_walk_paths = deque([[from_target]])	×
68
69	while len(to_walk_paths) > 0:	×
70	cur_path = to_walk_paths.popleft()	×
71	target = cur_path[-1]	×
72
73	if len(cur_path) > 1:	×
74	prev_target: Address \| None = cur_path[-2]	×
75	else:
76	prev_target = None	×
77	current_edge = (prev_target, target)	×
78
79	if current_edge not in visited_edges:	×
80	for dep in adjacency_lists.get(target, []):	×
81	dep_path = cur_path + [dep.address]	×
82	if dep.address == to_target:	×
83	yield dep_path	×
84	else:
85	to_walk_paths.append(dep_path)	×
86	visited_edges.add(current_edge)	×
87
88
89	@dataclass	×
90	class SpecsPaths:	×
91	paths: list[list[str]]	×
92
93
94	@dataclass	×
95	class SpecsPathsCollection:	×
96	spec_paths: list[SpecsPaths]	×
97
98
99	@dataclass(frozen=True)	×
100	class RootDestinationPair:	×
101	root: Target	×
102	destination: Target	×
103
104
105	@dataclass(frozen=True)	×
106	class RootDestinationsPair:	×
107	root: Target	×
108	destinations: Targets	×
109
110
111	@rule(desc="Get paths between root and destination.")	×
112	async def get_paths_between_root_and_destination(pair: RootDestinationPair) -> SpecsPaths:	×
113	transitive_targets = await transitive_targets_get(	×
114	TransitiveTargetsRequest(
115	[pair.root.address], should_traverse_deps_predicate=AlwaysTraverseDeps()
116	),
117	**implicitly(),
118	)
119
120	adjacent_targets_per_target = await concurrently(	×
121	resolve_targets(
122	**implicitly(
123	DependenciesRequest(
124	tgt.get(Dependencies), should_traverse_deps_predicate=AlwaysTraverseDeps()
125	)
126	)
127	)
128	for tgt in transitive_targets.closure
129	)
130
131	transitive_targets_closure_addresses = (t.address for t in transitive_targets.closure)	×
132	adjacency_lists = dict(zip(transitive_targets_closure_addresses, adjacent_targets_per_target))	×
133
134	spec_paths = []	×
135	for path in find_paths_breadth_first(	×
136	adjacency_lists, pair.root.address, pair.destination.address
137	):
138	spec_path = [address.spec for address in path]	×
139	spec_paths.append(spec_path)	×
140
141	return SpecsPaths(paths=spec_paths)	×
142
143
144	@rule(desc="Get paths between root and multiple destinations.")	×
145	async def get_paths_between_root_and_destinations(	×
146	pair: RootDestinationsPair,
147	) -> SpecsPathsCollection:
148	spec_paths = await concurrently(	×
149	get_paths_between_root_and_destination(
150	RootDestinationPair(destination=destination, root=pair.root)
151	)
152	for destination in pair.destinations
153	)
154	return SpecsPathsCollection(spec_paths=list(spec_paths))	×
155
156
157	@goal_rule	×
158	async def paths(console: Console, paths_subsystem: PathsSubsystem) -> PathsGoal:	×
159	path_from = paths_subsystem.from_	×
160	path_to = paths_subsystem.to	×
161
162	if path_from is None:	×
163	raise ValueError("Must set --from")	×
164
165	if path_to is None:	×
166	raise ValueError("Must set --to")	×
167
168	specs_parser = SpecsParser()	×
169
170	from_tgts, to_tgts = await concurrently(	×
171	resolve_targets(
172	**implicitly(
173	{
174	specs_parser.parse_specs(
175	[path_from],
176	description_of_origin="the option `--paths-from`",
177	): Specs
178	}
179	)
180	),
181	resolve_targets(
182	**implicitly(
183	{
184	specs_parser.parse_specs(
185	[path_to],
186	description_of_origin="the option `--paths-to`",
187	): Specs
188	}
189	)
190	),
191	)
192
193	all_spec_paths = []	×
194	spec_paths = await concurrently(	×
195	get_paths_between_root_and_destinations(
196	RootDestinationsPair(root=root, destinations=to_tgts)
197	)
198	for root in from_tgts
199	)
200
201	for spec_path in spec_paths:	×
202	for path in (p.paths for p in spec_path.spec_paths):	×
203	all_spec_paths.extend(path)	×
204
205	with paths_subsystem.output(console) as write_stdout:	×
206	write_stdout(json.dumps(all_spec_paths, indent=2) + "\n")	×
207
208	return PathsGoal(exit_code=0)	×
209
210
211	def rules():	×
212	return collect_rules()	×

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