21832085155

Committed 09 Feb 2026 03:53PM UTC coverage: 70.203% (-10.1%) from 80.282%

Build # 21832085155

Build Type

push

github

Committed by

web-flow

Commit Message

Prepare 2.30.2rc0 (#23079)

Run Details

56068 of 79865 relevant lines covered (70.2%)

2.05 hits per line

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

57.14

/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
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,
)
from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name
from pants.engine.addresses import Address
from pants.engine.internals.graph import determine_explicitly_provided_dependencies, hydrate_sources
from pants.engine.intrinsics import get_digest_contents
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import (
    DependenciesRequest,
    FieldSet,
    HydrateSourcesRequest,
    InferDependenciesRequest,
    InferredDependencies,
)
from pants.engine.unions import 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 ProtobufMapping:
    """A mapping of stripped .proto file names to their owning file address."""

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


@rule(desc="Creating map of Protobuf file names to Protobuf targets", level=LogLevel.DEBUG)
async def map_protobuf_files(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(sorted(stripped_files_to_addresses.items())),
        ambiguous_modules=FrozenDict(
            (k, tuple(sorted(v))) for k, v in sorted(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


@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
    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()):
        unambiguous = protobuf_mapping.mapping.get(import_path)
        ambiguous = protobuf_mapping.ambiguous_modules.get(import_path)
        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	1✔
5
6	import re	1✔
7	from collections import defaultdict	1✔
8	from dataclasses import dataclass	1✔
9	from typing import DefaultDict	1✔
10
11	from pants.backend.codegen.protobuf.protoc import Protoc	1✔
12	from pants.backend.codegen.protobuf.target_types import (	1✔
13	AllProtobufTargets,
14	ProtobufDependenciesField,
15	ProtobufSourceField,
16	)
17	from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name	1✔
18	from pants.engine.addresses import Address	1✔
19	from pants.engine.internals.graph import determine_explicitly_provided_dependencies, hydrate_sources	1✔
20	from pants.engine.intrinsics import get_digest_contents	1✔
21	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	1✔
22	from pants.engine.target import (	1✔
23	DependenciesRequest,
24	FieldSet,
25	HydrateSourcesRequest,
26	InferDependenciesRequest,
27	InferredDependencies,
28	)
29	from pants.engine.unions import UnionRule	1✔
30	from pants.util.frozendict import FrozenDict	1✔
31	from pants.util.logging import LogLevel	1✔
32	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	1✔
33	from pants.util.strutil import softwrap	1✔
34
35
36	@dataclass(frozen=True)	1✔
37	class ProtobufMapping:	1✔
38	"""A mapping of stripped .proto file names to their owning file address."""
39
40	mapping: FrozenDict[str, Address]	1✔
41	ambiguous_modules: FrozenDict[str, tuple[Address, ...]]	1✔
42
43
44	@rule(desc="Creating map of Protobuf file names to Protobuf targets", level=LogLevel.DEBUG)	1✔
45	async def map_protobuf_files(protobuf_targets: AllProtobufTargets) -> ProtobufMapping:	1✔
46	stripped_file_per_target = await concurrently(	×
47	strip_file_name(StrippedFileNameRequest(tgt[ProtobufSourceField].file_path))
48	for tgt in protobuf_targets
49	)
50
51	stripped_files_to_addresses: dict[str, Address] = {}	×
52	stripped_files_with_multiple_owners: DefaultDict[str, set[Address]] = defaultdict(set)	×
53	for tgt, stripped_file in zip(protobuf_targets, stripped_file_per_target):	×
54	if stripped_file.value in stripped_files_to_addresses:	×
55	stripped_files_with_multiple_owners[stripped_file.value].update(	×
56	{stripped_files_to_addresses[stripped_file.value], tgt.address}
57	)
58	else:
59	stripped_files_to_addresses[stripped_file.value] = tgt.address	×
60
61	# Remove files with ambiguous owners.
62	for ambiguous_stripped_f in stripped_files_with_multiple_owners:	×
63	stripped_files_to_addresses.pop(ambiguous_stripped_f)	×
64
65	return ProtobufMapping(	×
66	mapping=FrozenDict(sorted(stripped_files_to_addresses.items())),
67	ambiguous_modules=FrozenDict(
68	(k, tuple(sorted(v))) for k, v in sorted(stripped_files_with_multiple_owners.items())
69	),
70	)
71
72
73	# See https://developers.google.com/protocol-buffers/docs/reference/proto3-spec for the Proto
74	# language spec.
75	QUOTE_CHAR = r"(?:'\|\")"	1✔
76	IMPORT_MODIFIERS = r"(?:\spublic\|\sweak)?"	1✔
77	FILE_NAME = r"(.+?\.proto)"	1✔
78	# NB: We don't specify what a valid file name looks like to avoid accidentally breaking unicode.
79	IMPORT_REGEX = re.compile(rf"import\s{IMPORT_MODIFIERS}\s{QUOTE_CHAR}{FILE_NAME}{QUOTE_CHAR}\s*;")	1✔
80
81
82	def parse_proto_imports(file_content: str) -> FrozenOrderedSet[str]:	1✔
83	return FrozenOrderedSet(IMPORT_REGEX.findall(file_content))	×
84
85
86	@dataclass(frozen=True)	1✔
87	class ProtobufDependencyInferenceFieldSet(FieldSet):	1✔
88	required_fields = (ProtobufSourceField, ProtobufDependenciesField)	1✔
89
90	source: ProtobufSourceField	1✔
91	dependencies: ProtobufDependenciesField	1✔
92
93
94	class InferProtobufDependencies(InferDependenciesRequest):	1✔
95	infer_from = ProtobufDependencyInferenceFieldSet	1✔
96
97
98	@rule(desc="Inferring Protobuf dependencies by analyzing imports")	1✔
99	async def infer_protobuf_dependencies(	1✔
100	request: InferProtobufDependencies, protobuf_mapping: ProtobufMapping, protoc: Protoc
101	) -> InferredDependencies:
102	if not protoc.dependency_inference:	×
103	return InferredDependencies([])	×
104
105	address = request.field_set.address	×
106	explicitly_provided_deps, hydrated_sources = await concurrently(	×
107	determine_explicitly_provided_dependencies(
108	**implicitly(DependenciesRequest(request.field_set.dependencies))
109	),
110	hydrate_sources(HydrateSourcesRequest(request.field_set.source), **implicitly()),
111	)
112	digest_contents = await get_digest_contents(hydrated_sources.snapshot.digest)	×
113	assert len(digest_contents) == 1	×
114	file_content = digest_contents[0]	×
115
116	result: OrderedSet[Address] = OrderedSet()	×
117	for import_path in parse_proto_imports(file_content.content.decode()):	×
118	unambiguous = protobuf_mapping.mapping.get(import_path)	×
119	ambiguous = protobuf_mapping.ambiguous_modules.get(import_path)	×
120	if unambiguous:	×
121	result.add(unambiguous)	×
122	elif ambiguous:	×
123	explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(	×
124	ambiguous,
125	address,
126	import_reference="file",
127	context=softwrap(
128	f"""
129	The target {address} imports `{import_path}` in the file
130	{file_content.path}
131	"""
132	),
133	)
134	maybe_disambiguated = explicitly_provided_deps.disambiguated(ambiguous)	×
135	if maybe_disambiguated:	×
136	result.add(maybe_disambiguated)	×
137	return InferredDependencies(sorted(result))	×
138
139
140	def rules():	1✔
141	return (*collect_rules(), UnionRule(InferDependenciesRequest, InferProtobufDependencies))	1✔

pantsbuild / pants / 21832085155

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