23173035367

Committed 17 Mar 2026 12:47AM UTC coverage: 91.371% (-1.6%) from 92.933%

Build # 23173035367

Build Type

push

github

Committed by

web-flow

Commit Message

update helm (and friends) to a recent 3.x seies (#23143)

v4 is a major breaking change, so holding off on that. (There was also
just a 3.20, but 3.19 has this reassuring series of bug fixes and was
also quite recent.)

Run Details

2 of 2 new or added lines in 2 files covered. (100.0%)

1263 existing lines in 73 files now uncovered.

86196 of 94336 relevant lines covered (91.37%)

3.87 hits per line

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

94.24

/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	4✔
5
6	import re	4✔
7	import typing	4✔
8	from collections import defaultdict	4✔
9	from dataclasses import dataclass	4✔
10	from typing import DefaultDict	4✔
11
12	from pants.backend.codegen.protobuf.protoc import Protoc	4✔
13	from pants.backend.codegen.protobuf.target_types import (	4✔
14	AllProtobufTargets,
15	ProtobufDependenciesField,
16	ProtobufSourceField,
17	ProtobufSourceTarget,
18	)
19	from pants.core.target_types import (	4✔
20	ResolveLikeField,
21	ResolveLikeFieldToValueRequest,
22	get_resolve_from_resolve_like_field_request,
23	)
24	from pants.core.util_rules.stripped_source_files import (	4✔
25	StrippedFileName,
26	StrippedFileNameRequest,
27	strip_file_name,
28	)
29	from pants.engine.addresses import Address	4✔
30	from pants.engine.internals.graph import (	4✔
31	determine_explicitly_provided_dependencies,
32	hydrate_sources,
33	resolve_target,
34	)
35	from pants.engine.intrinsics import get_digest_contents	4✔
36	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	4✔
37	from pants.engine.target import (	4✔
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	4✔
48	from pants.util.frozendict import FrozenDict	4✔
49	from pants.util.logging import LogLevel	4✔
50	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	4✔
51	from pants.util.strutil import softwrap	4✔
52
53
54	@dataclass(frozen=True)	4✔
55	class ProtobufMappingResolveKey:	4✔
56	field_type: type[Field]	4✔
57	resolve: str	4✔
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(	4✔
64	field_type=ProtobufSourceField, resolve="<no-resolve>"
65	)
66
67
68	@dataclass(frozen=True)	4✔
69	class ProtobufMapping:	4✔
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]]	4✔
74	ambiguous_modules: FrozenDict[ProtobufMappingResolveKey, FrozenDict[str, tuple[Address, ...]]]	4✔
75
76
77	async def _map_single_pseudo_resolve(protobuf_targets: AllProtobufTargets) -> ProtobufMapping:	4✔
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)	4✔
117	async def map_protobuf_files(	4✔
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()	4✔
122	for field_type in ProtobufSourceTarget.class_field_types(union_membership):	4✔
123	if issubclass(field_type, ResolveLikeField):	4✔
124	resolve_like_field_types.add(field_type)	4✔
125	if not resolve_like_field_types:	4✔
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] = []	4✔
130	target_and_field_type_for_resolve_requests: list[tuple[Target, type[Field]]] = []	4✔
131	for tgt in protobuf_targets:	4✔
132	saw_at_least_one_field = False	4✔
133	for field_type in resolve_like_field_types:	4✔
134	if tgt.has_field(field_type):	4✔
135	resolve_request_type = typing.cast(	4✔
136	ResolveLikeField, tgt[field_type]
137	).get_resolve_like_field_to_value_request()
138	resolve_request = resolve_request_type(target=tgt)	4✔
139	resolve_requests.append(resolve_request)	4✔
140	target_and_field_type_for_resolve_requests.append((tgt, field_type))	4✔
141	saw_at_least_one_field = True	4✔
142
143	if not saw_at_least_one_field:	4✔
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(	4✔
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(	4✔
154	list
155	)
156	for resolve_result, (target, field_type) in zip(	4✔
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:	4✔
163	resolve_key = _NO_RESOLVE_LIKE_FIELDS_DEFINED	3✔
164	else:
165	resolve_key = ProtobufMappingResolveKey(	2✔
166	field_type=field_type, resolve=resolve_result.value
167	)
168	targets_partitioned_by_resolve[resolve_key].append(target)	4✔
169
170	stripped_file_per_target = await concurrently(	4✔
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(	4✔
176	zip(protobuf_targets, stripped_file_per_target)
177	)
178
179	stripped_files_to_addresses: dict[ProtobufMappingResolveKey, dict[str, Address]] = defaultdict(	4✔
180	dict
181	)
182	stripped_files_with_multiple_owners: dict[	4✔
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():	4✔
187	for tgt in targets_in_resolve:	4✔
188	stripped_file = target_to_stripped_file[tgt]	4✔
189	if stripped_file.value in stripped_files_to_addresses[resolve_key]:	4✔
UNCOV 190	stripped_files_with_multiple_owners[resolve_key][stripped_file.value].update(	×
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	4✔
195
196	# Remove files with ambiguous owners in each resolve.
197	for (	4✔
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:	×
UNCOV 202	stripped_files_to_addresses[resolve_key].pop(ambiguous_stripped_f)	×
203
204	return ProtobufMapping(	4✔
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"(?:'\|\")"	4✔
226	IMPORT_MODIFIERS = r"(?:\spublic\|\sweak)?"	4✔
227	FILE_NAME = r"(.+?\.proto)"	4✔
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*;")	4✔
230
231
232	def parse_proto_imports(file_content: str) -> FrozenOrderedSet[str]:	4✔
233	return FrozenOrderedSet(IMPORT_REGEX.findall(file_content))	4✔
234
235
236	@dataclass(frozen=True)	4✔
237	class ProtobufDependencyInferenceFieldSet(FieldSet):	4✔
238	required_fields = (ProtobufSourceField, ProtobufDependenciesField)	4✔
239
240	source: ProtobufSourceField	4✔
241	dependencies: ProtobufDependenciesField	4✔
242
243
244	class InferProtobufDependencies(InferDependenciesRequest):	4✔
245	infer_from = ProtobufDependencyInferenceFieldSet	4✔
246
247
248	async def get_resolve_key_from_target(address: Address) -> ProtobufMappingResolveKey:	4✔
249	wrapped_target = await resolve_target(	2✔
250	WrappedTargetRequest(address=address, description_of_origin="protobuf"), **implicitly()
251	)
252	resolve_field_type: type[Field] \| None = None	2✔
253	for field_type in wrapped_target.target.field_types:	2✔
254	if issubclass(field_type, ResolveLikeField):	2✔
255	if resolve_field_type is not None:	2✔
256	raise NotImplementedError(	×
257	f"TODO: Multiple resolve-like fields on target at address `{address}`."
258	)
259	resolve_field_type = field_type	2✔
260	if resolve_field_type is None:	2✔
261	raise ValueError(f"Failed to find resolve-like field on target at address `{address}.")	×
262
263	resolve_request_type = typing.cast(	2✔
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)	2✔
267	resolve_result = await get_resolve_from_resolve_like_field_request(	2✔
268	**implicitly({resolve_request: ResolveLikeFieldToValueRequest})
269	)
270
271	# When resolves are disabled, return the sentinel key
272	if resolve_result.value is None:	2✔
273	return _NO_RESOLVE_LIKE_FIELDS_DEFINED	×
274
275	return ProtobufMappingResolveKey(	2✔
276	field_type=resolve_field_type,
277	resolve=resolve_result.value,
278	)
279
280
281	@rule(desc="Inferring Protobuf dependencies by analyzing imports")	4✔
282	async def infer_protobuf_dependencies(	4✔
283	request: InferProtobufDependencies, protobuf_mapping: ProtobufMapping, protoc: Protoc
284	) -> InferredDependencies:
285	if not protoc.dependency_inference:	4✔
286	return InferredDependencies([])	×
287
288	address = request.field_set.address	4✔
289
290	resolve_key: ProtobufMappingResolveKey
291	if _NO_RESOLVE_LIKE_FIELDS_DEFINED in protobuf_mapping.mapping:	4✔
292	resolve_key = _NO_RESOLVE_LIKE_FIELDS_DEFINED	4✔
293	else:
294	resolve_key = await get_resolve_key_from_target(address)	2✔
295
296	explicitly_provided_deps, hydrated_sources = await concurrently(	4✔
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)	4✔
303	assert len(digest_contents) == 1	4✔
304	file_content = digest_contents[0]	4✔
305
306	result: OrderedSet[Address] = OrderedSet()	4✔
307	for import_path in parse_proto_imports(file_content.content.decode()):	4✔
308	mapping_in_resolve = protobuf_mapping.mapping.get(resolve_key)	2✔
309	unambiguous = mapping_in_resolve.get(import_path) if mapping_in_resolve else None	2✔
310
311	ambiguous_modules_in_resolve = protobuf_mapping.ambiguous_modules.get(resolve_key)	2✔
312	ambiguous = (	2✔
313	ambiguous_modules_in_resolve.get(import_path) if ambiguous_modules_in_resolve else None
314	)
315
316	if unambiguous:	2✔
317	result.add(unambiguous)	2✔
318	elif ambiguous:	1✔
319	explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(	1✔
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)	1✔
331	if maybe_disambiguated:	1✔
332	result.add(maybe_disambiguated)	1✔
333	return InferredDependencies(sorted(result))	4✔
334
335
336	def rules():	4✔
337	return (*collect_rules(), UnionRule(InferDependenciesRequest, InferProtobufDependencies))	4✔

pantsbuild / pants / 23173035367

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