22285099215

Committed 22 Feb 2026 08:52PM UTC coverage: 75.854% (-17.1%) from 92.936%

Build # 22285099215

Build Type

Pull #23121

github

Committed by

web-flow

Commit Message

Merge c7299df9c into ba8359840

Pull Request Pull Request #23121: fix issue with optional fields in dependency validator

Run Details

28 of 29 new or added lines in 2 files covered. (96.55%)

11174 existing lines in 400 files now uncovered.

53694 of 70786 relevant lines covered (75.85%)

1.88 hits per line

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

50.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	1✔
5
6	import json	1✔
7	from collections import deque	1✔
8	from collections.abc import Iterable	1✔
9	from dataclasses import dataclass	1✔
10
11	from pants.base.specs import Specs	1✔
12	from pants.base.specs_parser import SpecsParser	1✔
13	from pants.engine.addresses import Address	1✔
14	from pants.engine.console import Console	1✔
15	from pants.engine.goal import Goal, GoalSubsystem, Outputting	1✔
16	from pants.engine.internals.graph import resolve_targets	1✔
17	from pants.engine.internals.graph import transitive_targets as transitive_targets_get	1✔
18	from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly, rule	1✔
19	from pants.engine.target import (	1✔
20	AlwaysTraverseDeps,
21	Dependencies,
22	DependenciesRequest,
23	Target,
24	Targets,
25	TransitiveTargetsRequest,
26	)
27	from pants.option.option_types import StrOption	1✔
28
29
30	class PathsSubsystem(Outputting, GoalSubsystem):	1✔
31	name = "paths"	1✔
32	help = (	1✔
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(	1✔
38	default=None,
39	help="The path starting address",
40	)
41
42	to = StrOption(	1✔
43	default=None,
44	help="The path end address",
45	)
46
47
48	class PathsGoal(Goal):	1✔
49	subsystem_cls = PathsSubsystem	1✔
50	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	1✔
51
52
53	def find_paths_breadth_first(	1✔
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
UNCOV 62	if from_target == to_target:	×
UNCOV 63	yield [from_target]	×
UNCOV 64	return	×
65
UNCOV 66	visited_edges = set()	×
UNCOV 67	to_walk_paths = deque([[from_target]])	×
68
UNCOV 69	while len(to_walk_paths) > 0:	×
UNCOV 70	cur_path = to_walk_paths.popleft()	×
UNCOV 71	target = cur_path[-1]	×
72
UNCOV 73	if len(cur_path) > 1:	×
UNCOV 74	prev_target: Address \| None = cur_path[-2]	×
75	else:
UNCOV 76	prev_target = None	×
UNCOV 77	current_edge = (prev_target, target)	×
78
UNCOV 79	if current_edge not in visited_edges:	×
UNCOV 80	for dep in adjacency_lists.get(target, []):	×
UNCOV 81	dep_path = cur_path + [dep.address]	×
UNCOV 82	if dep.address == to_target:	×
UNCOV 83	yield dep_path	×
84	else:
UNCOV 85	to_walk_paths.append(dep_path)	×
UNCOV 86	visited_edges.add(current_edge)	×
87
88
89	@dataclass	1✔
90	class SpecsPaths:	1✔
91	paths: list[list[str]]	1✔
92
93
94	@dataclass	1✔
95	class SpecsPathsCollection:	1✔
96	spec_paths: list[SpecsPaths]	1✔
97
98
99	@dataclass(frozen=True)	1✔
100	class RootDestinationPair:	1✔
101	root: Target	1✔
102	destination: Target	1✔
103
104
105	@dataclass(frozen=True)	1✔
106	class RootDestinationsPair:	1✔
107	root: Target	1✔
108	destinations: Targets	1✔
109
110
111	@rule(desc="Get paths between root and destination.")	1✔
112	async def get_paths_between_root_and_destination(pair: RootDestinationPair) -> SpecsPaths:	1✔
UNCOV 113	transitive_targets = await transitive_targets_get(	×
114	TransitiveTargetsRequest(
115	[pair.root.address], should_traverse_deps_predicate=AlwaysTraverseDeps()
116	),
117	**implicitly(),
118	)
119
UNCOV 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
UNCOV 131	transitive_targets_closure_addresses = (t.address for t in transitive_targets.closure)	×
UNCOV 132	adjacency_lists = dict(zip(transitive_targets_closure_addresses, adjacent_targets_per_target))	×
133
UNCOV 134	spec_paths = []	×
UNCOV 135	for path in find_paths_breadth_first(	×
136	adjacency_lists, pair.root.address, pair.destination.address
137	):
UNCOV 138	spec_path = [address.spec for address in path]	×
UNCOV 139	spec_paths.append(spec_path)	×
140
UNCOV 141	return SpecsPaths(paths=spec_paths)	×
142
143
144	@rule(desc="Get paths between root and multiple destinations.")	1✔
145	async def get_paths_between_root_and_destinations(	1✔
146	pair: RootDestinationsPair,
147	) -> SpecsPathsCollection:
UNCOV 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	)
UNCOV 154	return SpecsPathsCollection(spec_paths=list(spec_paths))	×
155
156
157	@goal_rule	1✔
158	async def paths(console: Console, paths_subsystem: PathsSubsystem) -> PathsGoal:	1✔
UNCOV 159	path_from = paths_subsystem.from_	×
UNCOV 160	path_to = paths_subsystem.to	×
161
UNCOV 162	if path_from is None:	×
UNCOV 163	raise ValueError("Must set --from")	×
164
UNCOV 165	if path_to is None:	×
UNCOV 166	raise ValueError("Must set --to")	×
167
UNCOV 168	specs_parser = SpecsParser()	×
169
UNCOV 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
UNCOV 193	all_spec_paths = []	×
UNCOV 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
UNCOV 201	for spec_path in spec_paths:	×
UNCOV 202	for path in (p.paths for p in spec_path.spec_paths):	×
UNCOV 203	all_spec_paths.extend(path)	×
204
UNCOV 205	with paths_subsystem.output(console) as write_stdout:	×
UNCOV 206	write_stdout(json.dumps(all_spec_paths, indent=2) + "\n")	×
207
UNCOV 208	return PathsGoal(exit_code=0)	×
209
210
211	def rules():	1✔
212	return collect_rules()	1✔

pantsbuild / pants / 22285099215

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