25443604553

Committed 06 May 2026 03:05PM UTC coverage: 92.879% (-0.04%) from 92.915%

Build # 25443604553

Build Type

push

github

Committed by

web-flow

Commit Message

[pants_ng] Scaffolding for a pants_ng mode. (#23319)

In this mode the command line is parsed as an
NG invocation, and dispatched appropriately.

Of course at the moment there are no
implementations to dispatch to. That will follow.

This does expose a new option, `pants_ng` to users. 
There is a big warning not to set it, but we're not trying
to hide that we're working on a new thing, so I am
comfortable with this.

Coverage Stats

25 of 76 new or added lines in 9 files covered. (32.89%)

1294 existing lines in 76 files now uncovered.

92234 of 99306 relevant lines covered (92.88%)

4.05 hits per line

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

96.4

/src/python/pants/backend/codegen/protobuf/protobuf_dependency_inference.py

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

from __future__ import annotations

import re
import typing
from collections import defaultdict
from dataclasses import dataclass
from typing import DefaultDict

from pants.backend.codegen.protobuf.protoc import Protoc
from pants.backend.codegen.protobuf.target_types import (
    AllProtobufTargets,
    ProtobufDependenciesField,
    ProtobufSourceField,
    ProtobufSourceTarget,
)
from pants.core.target_types import (
    ResolveLikeField,
    ResolveLikeFieldToValueRequest,
    get_resolve_from_resolve_like_field_request,
)
from pants.core.util_rules.stripped_source_files import (
    StrippedFileName,
    StrippedFileNameRequest,
    strip_file_name,
)
from pants.engine.addresses import Address
from pants.engine.internals.graph import (
    determine_explicitly_provided_dependencies,
    hydrate_sources,
    resolve_target,
)
from pants.engine.intrinsics import get_digest_contents
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import (
    DependenciesRequest,
    Field,
    FieldSet,
    HydrateSourcesRequest,
    InferDependenciesRequest,
    InferredDependencies,
    Target,
    WrappedTargetRequest,
)
from pants.engine.unions import UnionMembership, UnionRule
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet, OrderedSet
from pants.util.strutil import softwrap


@dataclass(frozen=True)
class ProtobufMappingResolveKey:
    field_type: type[Field]
    resolve: str


# Sentinel value for when:
#   1. No resolve-like fields are registered on protobuf_source targets.
# .  2. Resolve-like fields exist but resolves are disabled for a language backend.
_NO_RESOLVE_LIKE_FIELDS_DEFINED = ProtobufMappingResolveKey(
    field_type=ProtobufSourceField, resolve="<no-resolve>"
)


@dataclass(frozen=True)
class ProtobufMapping:
    """A mapping of stripped .proto file names to their owning file address indirectly mapped by
    resolve-like fields."""

    mapping: FrozenDict[ProtobufMappingResolveKey, FrozenDict[str, Address]]
    ambiguous_modules: FrozenDict[ProtobufMappingResolveKey, FrozenDict[str, tuple[Address, ...]]]


async def _map_single_pseudo_resolve(protobuf_targets: AllProtobufTargets) -> ProtobufMapping:
    stripped_file_per_target = await concurrently(
        strip_file_name(StrippedFileNameRequest(tgt[ProtobufSourceField].file_path))
        for tgt in protobuf_targets
    )

    stripped_files_to_addresses: dict[str, Address] = {}
    stripped_files_with_multiple_owners: DefaultDict[str, set[Address]] = defaultdict(set)
    for tgt, stripped_file in zip(protobuf_targets, stripped_file_per_target):
        if stripped_file.value in stripped_files_to_addresses:
            stripped_files_with_multiple_owners[stripped_file.value].update(
                {stripped_files_to_addresses[stripped_file.value], tgt.address}
            )
        else:
            stripped_files_to_addresses[stripped_file.value] = tgt.address

    # Remove files with ambiguous owners.
    for ambiguous_stripped_f in stripped_files_with_multiple_owners:
        stripped_files_to_addresses.pop(ambiguous_stripped_f)

    return ProtobufMapping(
        mapping=FrozenDict(
            {
                _NO_RESOLVE_LIKE_FIELDS_DEFINED: FrozenDict(
                    sorted(stripped_files_to_addresses.items())
                )
            }
        ),
        ambiguous_modules=FrozenDict(
            {
                _NO_RESOLVE_LIKE_FIELDS_DEFINED: FrozenDict(
                    (k, tuple(sorted(v)))
                    for k, v in sorted(stripped_files_with_multiple_owners.items())
                )
            }
        ),
    )


@rule(desc="Creating map of Protobuf file names to Protobuf targets", level=LogLevel.DEBUG)
async def map_protobuf_files(
    protobuf_targets: AllProtobufTargets, union_membership: UnionMembership
) -> ProtobufMapping:
    # Determine the resolve-like fields installed on the `protobuf_source` target type.
    resolve_like_field_types: set[type[Field]] = set()
    for field_type in ProtobufSourceTarget.class_field_types(union_membership):
        if issubclass(field_type, ResolveLikeField):
            resolve_like_field_types.add(field_type)
    if not resolve_like_field_types:
        return await _map_single_pseudo_resolve(protobuf_targets)

    # Discover which resolves are present in the protobuf_source targets.
    resolve_requests: list[ResolveLikeFieldToValueRequest] = []
    target_and_field_type_for_resolve_requests: list[tuple[Target, type[Field]]] = []
    for tgt in protobuf_targets:
        saw_at_least_one_field = False
        for field_type in resolve_like_field_types:
            if tgt.has_field(field_type):
                resolve_request_type = typing.cast(
                    ResolveLikeField, tgt[field_type]
                ).get_resolve_like_field_to_value_request()
                resolve_request = resolve_request_type(target=tgt)
                resolve_requests.append(resolve_request)
                target_and_field_type_for_resolve_requests.append((tgt, field_type))
                saw_at_least_one_field = True

        if not saw_at_least_one_field:
            raise ValueError(f"Did not find a resolve field on target at address `{tgt.address}`.")

    # Obtain the resolves for each target and then partition.
    resolve_results = await concurrently(
        get_resolve_from_resolve_like_field_request(
            **implicitly({resolve_request: ResolveLikeFieldToValueRequest})
        )
        for resolve_request in resolve_requests
    )
    targets_partitioned_by_resolve: dict[ProtobufMappingResolveKey, list[Target]] = defaultdict(
        list
    )
    for resolve_result, (target, field_type) in zip(
        resolve_results, target_and_field_type_for_resolve_requests
    ):
        # When a resolve field returns None (resolves disabled), canonicalize to
        # _NO_RESOLVE_LIKE_FIELDS_DEFINED to ensure all "resolves disabled" targets share
        # the same partition regardless of which resolve-like field they have.
        if resolve_result.value is None:
            resolve_key = _NO_RESOLVE_LIKE_FIELDS_DEFINED
        else:
            resolve_key = ProtobufMappingResolveKey(
                field_type=field_type, resolve=resolve_result.value
            )
        targets_partitioned_by_resolve[resolve_key].append(target)

    stripped_file_per_target = await concurrently(
        strip_file_name(StrippedFileNameRequest(tgt[ProtobufSourceField].file_path))
        for tgt in protobuf_targets
    )

    target_to_stripped_file: dict[Target, StrippedFileName] = dict(
        zip(protobuf_targets, stripped_file_per_target)
    )

    stripped_files_to_addresses: dict[ProtobufMappingResolveKey, dict[str, Address]] = defaultdict(
        dict
    )
    stripped_files_with_multiple_owners: dict[
        ProtobufMappingResolveKey, dict[str, set[Address]]
    ] = defaultdict(lambda: defaultdict(set))

    for resolve_key, targets_in_resolve in targets_partitioned_by_resolve.items():
        for tgt in targets_in_resolve:
            stripped_file = target_to_stripped_file[tgt]
            if stripped_file.value in stripped_files_to_addresses[resolve_key]:
                stripped_files_with_multiple_owners[resolve_key][stripped_file.value].update(
                    {stripped_files_to_addresses[resolve_key][stripped_file.value], tgt.address}
                )
            else:
                stripped_files_to_addresses[resolve_key][stripped_file.value] = tgt.address

    # Remove files with ambiguous owners in each resolve.
    for (
        resolve_key,
        stripped_files_with_multiple_owners_in_resolve,
    ) in stripped_files_with_multiple_owners.items():
        for ambiguous_stripped_f in stripped_files_with_multiple_owners_in_resolve:
            stripped_files_to_addresses[resolve_key].pop(ambiguous_stripped_f)

    return ProtobufMapping(
        mapping=FrozenDict(
            {
                resolve_key: FrozenDict(sorted(stripped_files_to_addresses_in_resolve.items()))
                for resolve_key, stripped_files_to_addresses_in_resolve in stripped_files_to_addresses.items()
            }
        ),
        ambiguous_modules=FrozenDict(
            {
                resolve_key: FrozenDict(
                    (k, tuple(sorted(v)))
                    for k, v in sorted(stripped_files_with_multiple_owners_in_resolve.items())
                )
                for resolve_key, stripped_files_with_multiple_owners_in_resolve in stripped_files_with_multiple_owners.items()
            }
        ),
    )


# See https://developers.google.com/protocol-buffers/docs/reference/proto3-spec for the Proto
# language spec.
QUOTE_CHAR = r"(?:'|\")"
IMPORT_MODIFIERS = r"(?:\spublic|\sweak)?"
FILE_NAME = r"(.+?\.proto)"
# NB: We don't specify what a valid file name looks like to avoid accidentally breaking unicode.
IMPORT_REGEX = re.compile(rf"import\s*{IMPORT_MODIFIERS}\s*{QUOTE_CHAR}{FILE_NAME}{QUOTE_CHAR}\s*;")


def parse_proto_imports(file_content: str) -> FrozenOrderedSet[str]:
    return FrozenOrderedSet(IMPORT_REGEX.findall(file_content))


@dataclass(frozen=True)
class ProtobufDependencyInferenceFieldSet(FieldSet):
    required_fields = (ProtobufSourceField, ProtobufDependenciesField)

    source: ProtobufSourceField
    dependencies: ProtobufDependenciesField


class InferProtobufDependencies(InferDependenciesRequest):
    infer_from = ProtobufDependencyInferenceFieldSet


async def get_resolve_key_from_target(address: Address) -> ProtobufMappingResolveKey:
    wrapped_target = await resolve_target(
        WrappedTargetRequest(address=address, description_of_origin="protobuf"), **implicitly()
    )
    resolve_field_type: type[Field] | None = None
    for field_type in wrapped_target.target.field_types:
        if issubclass(field_type, ResolveLikeField):
            if resolve_field_type is not None:
                raise NotImplementedError(
                    f"TODO: Multiple resolve-like fields on target at address `{address}`."
                )
            resolve_field_type = field_type
    if resolve_field_type is None:
        raise ValueError(f"Failed to find resolve-like field on target at address `{address}.")

    resolve_request_type = typing.cast(
        ResolveLikeField, wrapped_target.target[resolve_field_type]
    ).get_resolve_like_field_to_value_request()
    resolve_request = resolve_request_type(target=wrapped_target.target)
    resolve_result = await get_resolve_from_resolve_like_field_request(
        **implicitly({resolve_request: ResolveLikeFieldToValueRequest})
    )

    # When resolves are disabled, return the sentinel key
    if resolve_result.value is None:
        return _NO_RESOLVE_LIKE_FIELDS_DEFINED

    return ProtobufMappingResolveKey(
        field_type=resolve_field_type,
        resolve=resolve_result.value,
    )


@rule(desc="Inferring Protobuf dependencies by analyzing imports")
async def infer_protobuf_dependencies(
    request: InferProtobufDependencies, protobuf_mapping: ProtobufMapping, protoc: Protoc
) -> InferredDependencies:
    if not protoc.dependency_inference:
        return InferredDependencies([])

    address = request.field_set.address

    resolve_key: ProtobufMappingResolveKey
    if _NO_RESOLVE_LIKE_FIELDS_DEFINED in protobuf_mapping.mapping:
        resolve_key = _NO_RESOLVE_LIKE_FIELDS_DEFINED
    else:
        resolve_key = await get_resolve_key_from_target(address)

    explicitly_provided_deps, hydrated_sources = await concurrently(
        determine_explicitly_provided_dependencies(
            **implicitly(DependenciesRequest(request.field_set.dependencies))
        ),
        hydrate_sources(HydrateSourcesRequest(request.field_set.source), **implicitly()),
    )
    digest_contents = await get_digest_contents(hydrated_sources.snapshot.digest)
    assert len(digest_contents) == 1
    file_content = digest_contents[0]

    result: OrderedSet[Address] = OrderedSet()
    for import_path in parse_proto_imports(file_content.content.decode()):
        mapping_in_resolve = protobuf_mapping.mapping.get(resolve_key)
        unambiguous = mapping_in_resolve.get(import_path) if mapping_in_resolve else None

        ambiguous_modules_in_resolve = protobuf_mapping.ambiguous_modules.get(resolve_key)
        ambiguous = (
            ambiguous_modules_in_resolve.get(import_path) if ambiguous_modules_in_resolve else None
        )

        if unambiguous:
            result.add(unambiguous)
        elif ambiguous:
            explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(
                ambiguous,
                address,
                import_reference="file",
                context=softwrap(
                    f"""
                    The target {address} imports `{import_path}` in the file
                    {file_content.path}
                    """
                ),
            )
            maybe_disambiguated = explicitly_provided_deps.disambiguated(ambiguous)
            if maybe_disambiguated:
                result.add(maybe_disambiguated)
    return InferredDependencies(sorted(result))


def rules():
    return (*collect_rules(), UnionRule(InferDependenciesRequest, InferProtobufDependencies))

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	5✔
5
6	import re	5✔
7	import typing	5✔
8	from collections import defaultdict	5✔
9	from dataclasses import dataclass	5✔
10	from typing import DefaultDict	5✔
11
12	from pants.backend.codegen.protobuf.protoc import Protoc	5✔
13	from pants.backend.codegen.protobuf.target_types import (	5✔
14	AllProtobufTargets,
15	ProtobufDependenciesField,
16	ProtobufSourceField,
17	ProtobufSourceTarget,
18	)
19	from pants.core.target_types import (	5✔
20	ResolveLikeField,
21	ResolveLikeFieldToValueRequest,
22	get_resolve_from_resolve_like_field_request,
23	)
24	from pants.core.util_rules.stripped_source_files import (	5✔
25	StrippedFileName,
26	StrippedFileNameRequest,
27	strip_file_name,
28	)
29	from pants.engine.addresses import Address	5✔
30	from pants.engine.internals.graph import (	5✔
31	determine_explicitly_provided_dependencies,
32	hydrate_sources,
33	resolve_target,
34	)
35	from pants.engine.intrinsics import get_digest_contents	5✔
36	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	5✔
37	from pants.engine.target import (	5✔
38	DependenciesRequest,
39	Field,
40	FieldSet,
41	HydrateSourcesRequest,
42	InferDependenciesRequest,
43	InferredDependencies,
44	Target,
45	WrappedTargetRequest,
46	)
47	from pants.engine.unions import UnionMembership, UnionRule	5✔
48	from pants.util.frozendict import FrozenDict	5✔
49	from pants.util.logging import LogLevel	5✔
50	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	5✔
51	from pants.util.strutil import softwrap	5✔
52
53
54	@dataclass(frozen=True)	5✔
55	class ProtobufMappingResolveKey:	5✔
56	field_type: type[Field]	5✔
57	resolve: str	5✔
58
59
60	# Sentinel value for when:
61	# 1. No resolve-like fields are registered on protobuf_source targets.
62	# . 2. Resolve-like fields exist but resolves are disabled for a language backend.
63	_NO_RESOLVE_LIKE_FIELDS_DEFINED = ProtobufMappingResolveKey(	5✔
64	field_type=ProtobufSourceField, resolve="<no-resolve>"
65	)
66
67
68	@dataclass(frozen=True)	5✔
69	class ProtobufMapping:	5✔
70	"""A mapping of stripped .proto file names to their owning file address indirectly mapped by
71	resolve-like fields."""
72
73	mapping: FrozenDict[ProtobufMappingResolveKey, FrozenDict[str, Address]]	5✔
74	ambiguous_modules: FrozenDict[ProtobufMappingResolveKey, FrozenDict[str, tuple[Address, ...]]]	5✔
75
76
77	async def _map_single_pseudo_resolve(protobuf_targets: AllProtobufTargets) -> ProtobufMapping:	5✔
78	stripped_file_per_target = await concurrently(	2✔
79	strip_file_name(StrippedFileNameRequest(tgt[ProtobufSourceField].file_path))
80	for tgt in protobuf_targets
81	)
82
83	stripped_files_to_addresses: dict[str, Address] = {}	2✔
84	stripped_files_with_multiple_owners: DefaultDict[str, set[Address]] = defaultdict(set)	2✔
85	for tgt, stripped_file in zip(protobuf_targets, stripped_file_per_target):	2✔
86	if stripped_file.value in stripped_files_to_addresses:	2✔
87	stripped_files_with_multiple_owners[stripped_file.value].update(	1✔
88	{stripped_files_to_addresses[stripped_file.value], tgt.address}
89	)
90	else:
91	stripped_files_to_addresses[stripped_file.value] = tgt.address	2✔
92
93	# Remove files with ambiguous owners.
94	for ambiguous_stripped_f in stripped_files_with_multiple_owners:	2✔
95	stripped_files_to_addresses.pop(ambiguous_stripped_f)	1✔
96
97	return ProtobufMapping(	2✔
98	mapping=FrozenDict(
99	{
100	_NO_RESOLVE_LIKE_FIELDS_DEFINED: FrozenDict(
101	sorted(stripped_files_to_addresses.items())
102	)
103	}
104	),
105	ambiguous_modules=FrozenDict(
106	{
107	_NO_RESOLVE_LIKE_FIELDS_DEFINED: FrozenDict(
108	(k, tuple(sorted(v)))
109	for k, v in sorted(stripped_files_with_multiple_owners.items())
110	)
111	}
112	),
113	)
114
115
116	@rule(desc="Creating map of Protobuf file names to Protobuf targets", level=LogLevel.DEBUG)	5✔
117	async def map_protobuf_files(	5✔
118	protobuf_targets: AllProtobufTargets, union_membership: UnionMembership
119	) -> ProtobufMapping:
120	# Determine the resolve-like fields installed on the `protobuf_source` target type.
121	resolve_like_field_types: set[type[Field]] = set()	5✔
122	for field_type in ProtobufSourceTarget.class_field_types(union_membership):	5✔
123	if issubclass(field_type, ResolveLikeField):	5✔
124	resolve_like_field_types.add(field_type)	5✔
125	if not resolve_like_field_types:	5✔
126	return await _map_single_pseudo_resolve(protobuf_targets)	2✔
127
128	# Discover which resolves are present in the protobuf_source targets.
129	resolve_requests: list[ResolveLikeFieldToValueRequest] = []	5✔
130	target_and_field_type_for_resolve_requests: list[tuple[Target, type[Field]]] = []	5✔
131	for tgt in protobuf_targets:	5✔
132	saw_at_least_one_field = False	5✔
133	for field_type in resolve_like_field_types:	5✔
134	if tgt.has_field(field_type):	5✔
135	resolve_request_type = typing.cast(	5✔
136	ResolveLikeField, tgt[field_type]
137	).get_resolve_like_field_to_value_request()
138	resolve_request = resolve_request_type(target=tgt)	5✔
139	resolve_requests.append(resolve_request)	5✔
140	target_and_field_type_for_resolve_requests.append((tgt, field_type))	5✔
141	saw_at_least_one_field = True	5✔
142
143	if not saw_at_least_one_field:	5✔
144	raise ValueError(f"Did not find a resolve field on target at address `{tgt.address}`.")	×
145
146	# Obtain the resolves for each target and then partition.
147	resolve_results = await concurrently(	5✔
148	get_resolve_from_resolve_like_field_request(
149	**implicitly({resolve_request: ResolveLikeFieldToValueRequest})
150	)
151	for resolve_request in resolve_requests
152	)
153	targets_partitioned_by_resolve: dict[ProtobufMappingResolveKey, list[Target]] = defaultdict(	5✔
154	list
155	)
156	for resolve_result, (target, field_type) in zip(	5✔
157	resolve_results, target_and_field_type_for_resolve_requests
158	):
159	# When a resolve field returns None (resolves disabled), canonicalize to
160	# _NO_RESOLVE_LIKE_FIELDS_DEFINED to ensure all "resolves disabled" targets share
161	# the same partition regardless of which resolve-like field they have.
162	if resolve_result.value is None:	5✔
163	resolve_key = _NO_RESOLVE_LIKE_FIELDS_DEFINED	3✔
164	else:
165	resolve_key = ProtobufMappingResolveKey(	3✔
166	field_type=field_type, resolve=resolve_result.value
167	)
168	targets_partitioned_by_resolve[resolve_key].append(target)	5✔
169
170	stripped_file_per_target = await concurrently(	5✔
171	strip_file_name(StrippedFileNameRequest(tgt[ProtobufSourceField].file_path))
172	for tgt in protobuf_targets
173	)
174
175	target_to_stripped_file: dict[Target, StrippedFileName] = dict(	5✔
176	zip(protobuf_targets, stripped_file_per_target)
177	)
178
179	stripped_files_to_addresses: dict[ProtobufMappingResolveKey, dict[str, Address]] = defaultdict(	5✔
180	dict
181	)
182	stripped_files_with_multiple_owners: dict[	5✔
183	ProtobufMappingResolveKey, dict[str, set[Address]]
184	] = defaultdict(lambda: defaultdict(set))
185
186	for resolve_key, targets_in_resolve in targets_partitioned_by_resolve.items():	5✔
187	for tgt in targets_in_resolve:	5✔
188	stripped_file = target_to_stripped_file[tgt]	5✔
189	if stripped_file.value in stripped_files_to_addresses[resolve_key]:	5✔
UNCOV 190	stripped_files_with_multiple_owners[resolve_key][stripped_file.value].update(	1✔
191	{stripped_files_to_addresses[resolve_key][stripped_file.value], tgt.address}
192	)
193	else:
194	stripped_files_to_addresses[resolve_key][stripped_file.value] = tgt.address	5✔
195
196	# Remove files with ambiguous owners in each resolve.
197	for (	5✔
198	resolve_key,
199	stripped_files_with_multiple_owners_in_resolve,
200	) in stripped_files_with_multiple_owners.items():
UNCOV 201	for ambiguous_stripped_f in stripped_files_with_multiple_owners_in_resolve:	1✔
UNCOV 202	stripped_files_to_addresses[resolve_key].pop(ambiguous_stripped_f)	1✔
203
204	return ProtobufMapping(	5✔
205	mapping=FrozenDict(
206	{
207	resolve_key: FrozenDict(sorted(stripped_files_to_addresses_in_resolve.items()))
208	for resolve_key, stripped_files_to_addresses_in_resolve in stripped_files_to_addresses.items()
209	}
210	),
211	ambiguous_modules=FrozenDict(
212	{
213	resolve_key: FrozenDict(
214	(k, tuple(sorted(v)))
215	for k, v in sorted(stripped_files_with_multiple_owners_in_resolve.items())
216	)
217	for resolve_key, stripped_files_with_multiple_owners_in_resolve in stripped_files_with_multiple_owners.items()
218	}
219	),
220	)
221
222
223	# See https://developers.google.com/protocol-buffers/docs/reference/proto3-spec for the Proto
224	# language spec.
225	QUOTE_CHAR = r"(?:'\|\")"	5✔
226	IMPORT_MODIFIERS = r"(?:\spublic\|\sweak)?"	5✔
227	FILE_NAME = r"(.+?\.proto)"	5✔
228	# NB: We don't specify what a valid file name looks like to avoid accidentally breaking unicode.
229	IMPORT_REGEX = re.compile(rf"import\s{IMPORT_MODIFIERS}\s{QUOTE_CHAR}{FILE_NAME}{QUOTE_CHAR}\s*;")	5✔
230
231
232	def parse_proto_imports(file_content: str) -> FrozenOrderedSet[str]:	5✔
233	return FrozenOrderedSet(IMPORT_REGEX.findall(file_content))	5✔
234
235
236	@dataclass(frozen=True)	5✔
237	class ProtobufDependencyInferenceFieldSet(FieldSet):	5✔
238	required_fields = (ProtobufSourceField, ProtobufDependenciesField)	5✔
239
240	source: ProtobufSourceField	5✔
241	dependencies: ProtobufDependenciesField	5✔
242
243
244	class InferProtobufDependencies(InferDependenciesRequest):	5✔
245	infer_from = ProtobufDependencyInferenceFieldSet	5✔
246
247
248	async def get_resolve_key_from_target(address: Address) -> ProtobufMappingResolveKey:	5✔
249	wrapped_target = await resolve_target(	3✔
250	WrappedTargetRequest(address=address, description_of_origin="protobuf"), **implicitly()
251	)
252	resolve_field_type: type[Field] \| None = None	3✔
253	for field_type in wrapped_target.target.field_types:	3✔
254	if issubclass(field_type, ResolveLikeField):	3✔
255	if resolve_field_type is not None:	3✔
256	raise NotImplementedError(	×
257	f"TODO: Multiple resolve-like fields on target at address `{address}`."
258	)
259	resolve_field_type = field_type	3✔
260	if resolve_field_type is None:	3✔
261	raise ValueError(f"Failed to find resolve-like field on target at address `{address}.")	×
262
263	resolve_request_type = typing.cast(	3✔
264	ResolveLikeField, wrapped_target.target[resolve_field_type]
265	).get_resolve_like_field_to_value_request()
266	resolve_request = resolve_request_type(target=wrapped_target.target)	3✔
267	resolve_result = await get_resolve_from_resolve_like_field_request(	3✔
268	**implicitly({resolve_request: ResolveLikeFieldToValueRequest})
269	)
270
271	# When resolves are disabled, return the sentinel key
272	if resolve_result.value is None:	3✔
273	return _NO_RESOLVE_LIKE_FIELDS_DEFINED	×
274
275	return ProtobufMappingResolveKey(	3✔
276	field_type=resolve_field_type,
277	resolve=resolve_result.value,
278	)
279
280
281	@rule(desc="Inferring Protobuf dependencies by analyzing imports")	5✔
282	async def infer_protobuf_dependencies(	5✔
283	request: InferProtobufDependencies, protobuf_mapping: ProtobufMapping, protoc: Protoc
284	) -> InferredDependencies:
285	if not protoc.dependency_inference:	5✔
286	return InferredDependencies([])	×
287
288	address = request.field_set.address	5✔
289
290	resolve_key: ProtobufMappingResolveKey
291	if _NO_RESOLVE_LIKE_FIELDS_DEFINED in protobuf_mapping.mapping:	5✔
292	resolve_key = _NO_RESOLVE_LIKE_FIELDS_DEFINED	4✔
293	else:
294	resolve_key = await get_resolve_key_from_target(address)	3✔
295
296	explicitly_provided_deps, hydrated_sources = await concurrently(	5✔
297	determine_explicitly_provided_dependencies(
298	**implicitly(DependenciesRequest(request.field_set.dependencies))
299	),
300	hydrate_sources(HydrateSourcesRequest(request.field_set.source), **implicitly()),
301	)
302	digest_contents = await get_digest_contents(hydrated_sources.snapshot.digest)	5✔
303	assert len(digest_contents) == 1	5✔
304	file_content = digest_contents[0]	5✔
305
306	result: OrderedSet[Address] = OrderedSet()	5✔
307	for import_path in parse_proto_imports(file_content.content.decode()):	5✔
308	mapping_in_resolve = protobuf_mapping.mapping.get(resolve_key)	3✔
309	unambiguous = mapping_in_resolve.get(import_path) if mapping_in_resolve else None	3✔
310
311	ambiguous_modules_in_resolve = protobuf_mapping.ambiguous_modules.get(resolve_key)	3✔
312	ambiguous = (	3✔
313	ambiguous_modules_in_resolve.get(import_path) if ambiguous_modules_in_resolve else None
314	)
315
316	if unambiguous:	3✔
317	result.add(unambiguous)	3✔
318	elif ambiguous:	2✔
319	explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(	2✔
320	ambiguous,
321	address,
322	import_reference="file",
323	context=softwrap(
324	f"""
325	The target {address} imports `{import_path}` in the file
326	{file_content.path}
327	"""
328	),
329	)
330	maybe_disambiguated = explicitly_provided_deps.disambiguated(ambiguous)	2✔
331	if maybe_disambiguated:	2✔
332	result.add(maybe_disambiguated)	1✔
333	return InferredDependencies(sorted(result))	5✔
334
335
336	def rules():	5✔
337	return (*collect_rules(), UnionRule(InferDependenciesRequest, InferProtobufDependencies))	5✔

pantsbuild / pants / 25443604553

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