• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pantsbuild / pants / 19015773527

02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%
19015773527

Pull #22816

github

web-flow
Merge a12d75757 into 6c024e162
Pull Request #22816: Update Pants internal Python to 3.14

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

0.0
/src/python/pants/backend/codegen/protobuf/python/python_protobuf_module_mapper.py
1
# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2
# Licensed under the Apache License, Version 2.0 (see LICENSE).
3

UNCOV
4
from __future__ import annotations
×
5

UNCOV
6
from collections import defaultdict
×
UNCOV
7
from typing import DefaultDict
×
8

UNCOV
9
from pants.backend.codegen.protobuf.python.python_protobuf_subsystem import PythonProtobufSubsystem
×
UNCOV
10
from pants.backend.codegen.protobuf.target_types import (
×
11
    AllProtobufTargets,
12
    ProtobufGrpcToggleField,
13
    ProtobufSourceField,
14
)
UNCOV
15
from pants.backend.python.dependency_inference.module_mapper import (
×
16
    FirstPartyPythonMappingImpl,
17
    FirstPartyPythonMappingImplMarker,
18
    ModuleProvider,
19
    ModuleProviderType,
20
    ResolveName,
21
)
UNCOV
22
from pants.backend.python.subsystems.setup import PythonSetup
×
UNCOV
23
from pants.backend.python.target_types import PythonResolveField
×
UNCOV
24
from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name
×
UNCOV
25
from pants.engine.rules import collect_rules, concurrently, rule
×
UNCOV
26
from pants.engine.unions import UnionRule
×
UNCOV
27
from pants.util.logging import LogLevel
×
28

29

UNCOV
30
def proto_path_to_py_module(stripped_path: str, *, suffix: str) -> str:
×
31
    return stripped_path.replace(".proto", suffix).replace("/", ".")
×
32

33

34
# This is only used to register our implementation with the plugin hook via unions.
UNCOV
35
class PythonProtobufMappingMarker(FirstPartyPythonMappingImplMarker):
×
UNCOV
36
    pass
×
37

38

UNCOV
39
@rule(desc="Creating map of Protobuf targets to generated Python modules", level=LogLevel.DEBUG)
×
UNCOV
40
async def map_protobuf_to_python_modules(
×
41
    protobuf_targets: AllProtobufTargets,
42
    python_setup: PythonSetup,
43
    python_protobuf_subsystem: PythonProtobufSubsystem,
44
    _: PythonProtobufMappingMarker,
45
) -> FirstPartyPythonMappingImpl:
46
    grpc_suffixes = []
×
47
    if python_protobuf_subsystem.grpcio_plugin:
×
48
        grpc_suffixes.append("_pb2_grpc")
×
49
    if python_protobuf_subsystem.grpclib_plugin:
×
50
        grpc_suffixes.append("_grpc")
×
51

52
    stripped_file_per_target = await concurrently(
×
53
        strip_file_name(StrippedFileNameRequest(tgt[ProtobufSourceField].file_path))
54
        for tgt in protobuf_targets
55
    )
56

57
    resolves_to_modules_to_providers: DefaultDict[
×
58
        ResolveName, DefaultDict[str, list[ModuleProvider]]
59
    ] = defaultdict(lambda: defaultdict(list))
60
    for tgt, stripped_file in zip(protobuf_targets, stripped_file_per_target):
×
61
        resolve = tgt[PythonResolveField].normalized_value(python_setup)
×
62

63
        # NB: We don't consider the MyPy plugin, which generates `_pb2.pyi`. The stubs end up
64
        # sharing the same module as the implementation `_pb2.py`. Because both generated files
65
        # come from the same original Protobuf target, we're covered.
66
        module = proto_path_to_py_module(stripped_file.value, suffix="_pb2")
×
67
        resolves_to_modules_to_providers[resolve][module].append(
×
68
            ModuleProvider(tgt.address, ModuleProviderType.IMPL)
69
        )
70
        if tgt.get(ProtobufGrpcToggleField).value:
×
71
            for suffix in grpc_suffixes:
×
72
                module = proto_path_to_py_module(stripped_file.value, suffix=suffix)
×
73
                resolves_to_modules_to_providers[resolve][module].append(
×
74
                    ModuleProvider(tgt.address, ModuleProviderType.IMPL)
75
                )
76

77
    return FirstPartyPythonMappingImpl.create(resolves_to_modules_to_providers)
×
78

79

UNCOV
80
def rules():
×
UNCOV
81
    return (
×
82
        *collect_rules(),
83
        UnionRule(FirstPartyPythonMappingImplMarker, PythonProtobufMappingMarker),
84
    )
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc