• 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/lint/buf/format_rules.py
1
# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2
# Licensed under the Apache License, Version 2.0 (see LICENSE).
UNCOV
3
from dataclasses import dataclass
×
4

UNCOV
5
from pants.backend.codegen.protobuf.lint.buf.skip_field import SkipBufFormatField
×
UNCOV
6
from pants.backend.codegen.protobuf.lint.buf.subsystem import BufSubsystem
×
UNCOV
7
from pants.backend.codegen.protobuf.target_types import (
×
8
    ProtobufDependenciesField,
9
    ProtobufSourceField,
10
)
UNCOV
11
from pants.core.goals.fmt import FmtResult, FmtTargetsRequest, Partitions
×
UNCOV
12
from pants.core.util_rules.external_tool import download_external_tool
×
UNCOV
13
from pants.core.util_rules.system_binaries import (
×
14
    BinaryShimsRequest,
15
    DiffBinary,
16
    create_binary_shims,
17
)
UNCOV
18
from pants.engine.fs import MergeDigests
×
UNCOV
19
from pants.engine.intrinsics import merge_digests
×
UNCOV
20
from pants.engine.platform import Platform
×
UNCOV
21
from pants.engine.process import Process, execute_process_or_raise
×
UNCOV
22
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
×
UNCOV
23
from pants.engine.target import FieldSet, Target
×
UNCOV
24
from pants.util.logging import LogLevel
×
UNCOV
25
from pants.util.meta import classproperty
×
UNCOV
26
from pants.util.strutil import pluralize
×
27

28

UNCOV
29
@dataclass(frozen=True)
×
UNCOV
30
class BufFieldSet(FieldSet):
×
UNCOV
31
    required_fields = (ProtobufSourceField,)
×
32

33
    sources: ProtobufSourceField
34
    dependencies: ProtobufDependenciesField
35

UNCOV
36
    @classmethod
×
UNCOV
37
    def opt_out(cls, tgt: Target) -> bool:
×
38
        return tgt.get(SkipBufFormatField).value
×
39

40

UNCOV
41
class BufFormatRequest(FmtTargetsRequest):
×
UNCOV
42
    field_set_type = BufFieldSet
×
UNCOV
43
    tool_subsystem = BufSubsystem  # type: ignore[assignment]
×
44

UNCOV
45
    @classproperty
×
UNCOV
46
    def tool_name(cls) -> str:
×
47
        return "buf format"
×
48

UNCOV
49
    @classproperty
×
UNCOV
50
    def tool_id(cls) -> str:
×
51
        return "buf-format"
×
52

53

UNCOV
54
@rule
×
UNCOV
55
async def partition_buf(
×
56
    request: BufFormatRequest.PartitionRequest, buf: BufSubsystem
57
) -> Partitions:
58
    return (
×
59
        Partitions()
60
        if buf.format_skip
61
        else Partitions.single_partition(
62
            field_set.sources.file_path for field_set in request.field_sets
63
        )
64
    )
65

66

UNCOV
67
@rule(desc="Format with buf format", level=LogLevel.DEBUG)
×
UNCOV
68
async def run_buf_format(
×
69
    request: BufFormatRequest.Batch, buf: BufSubsystem, diff_binary: DiffBinary, platform: Platform
70
) -> FmtResult:
71
    download_buf_get = download_external_tool(buf.get_request(platform))
×
72
    binary_shims_get = create_binary_shims(
×
73
        BinaryShimsRequest.for_paths(
74
            diff_binary,
75
            rationale="run `buf format`",
76
        ),
77
        **implicitly(),
78
    )
79
    downloaded_buf, binary_shims = await concurrently(download_buf_get, binary_shims_get)
×
80

81
    input_digest = await merge_digests(
×
82
        MergeDigests((request.snapshot.digest, downloaded_buf.digest))
83
    )
84

85
    argv = [
×
86
        downloaded_buf.exe,
87
        "format",
88
        "-w",
89
        *buf.format_args,
90
        "--path",
91
        ",".join(request.files),
92
    ]
93
    result = await execute_process_or_raise(
×
94
        **implicitly(
95
            Process(
96
                argv=argv,
97
                input_digest=input_digest,
98
                output_files=request.files,
99
                description=f"Run buf format on {pluralize(len(request.files), 'file')}.",
100
                level=LogLevel.DEBUG,
101
                env={"PATH": binary_shims.path_component},
102
                immutable_input_digests=binary_shims.immutable_input_digests,
103
            )
104
        ),
105
    )
106
    return await FmtResult.create(request, result)
×
107

108

UNCOV
109
def rules():
×
UNCOV
110
    return [
×
111
        *collect_rules(),
112
        *BufFormatRequest.rules(),
113
    ]
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