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/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	×
5
6	import re	×
7	from collections import defaultdict	×
8	from dataclasses import dataclass	×
9	from typing import DefaultDict	×
10
11	from pants.backend.codegen.protobuf.protoc import Protoc	×
12	from pants.backend.codegen.protobuf.target_types import (	×
13	AllProtobufTargets,
14	ProtobufDependenciesField,
15	ProtobufSourceField,
16	)
17	from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name	×
18	from pants.engine.addresses import Address	×
19	from pants.engine.internals.graph import determine_explicitly_provided_dependencies, hydrate_sources	×
20	from pants.engine.intrinsics import get_digest_contents	×
21	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
22	from pants.engine.target import (	×
23	DependenciesRequest,
24	FieldSet,
25	HydrateSourcesRequest,
26	InferDependenciesRequest,
27	InferredDependencies,
28	)
29	from pants.engine.unions import UnionRule	×
30	from pants.util.frozendict import FrozenDict	×
31	from pants.util.logging import LogLevel	×
32	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	×
33	from pants.util.strutil import softwrap	×
34
35
36	@dataclass(frozen=True)	×
37	class ProtobufMapping:	×
38	"""A mapping of stripped .proto file names to their owning file address."""
39
40	mapping: FrozenDict[str, Address]	×
41	ambiguous_modules: FrozenDict[str, tuple[Address, ...]]	×
42
43
44	@rule(desc="Creating map of Protobuf file names to Protobuf targets", level=LogLevel.DEBUG)	×
45	async def map_protobuf_files(protobuf_targets: AllProtobufTargets) -> ProtobufMapping:	×
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"(?:'\|\")"	×
76	IMPORT_MODIFIERS = r"(?:\spublic\|\sweak)?"	×
77	FILE_NAME = r"(.+?\.proto)"	×
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*;")	×
80
81
82	def parse_proto_imports(file_content: str) -> FrozenOrderedSet[str]:	×
83	return FrozenOrderedSet(IMPORT_REGEX.findall(file_content))	×
84
85
86	@dataclass(frozen=True)	×
87	class ProtobufDependencyInferenceFieldSet(FieldSet):	×
88	required_fields = (ProtobufSourceField, ProtobufDependenciesField)	×
89
90	source: ProtobufSourceField	×
91	dependencies: ProtobufDependenciesField	×
92
93
94	class InferProtobufDependencies(InferDependenciesRequest):	×
95	infer_from = ProtobufDependencyInferenceFieldSet	×
96
97
98	@rule(desc="Inferring Protobuf dependencies by analyzing imports")	×
99	async def infer_protobuf_dependencies(	×
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():	×
141	return (*collect_rules(), UnionRule(InferDependenciesRequest, InferProtobufDependencies))	×

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