26689585807

Committed 30 May 2026 04:55PM UTC coverage: 92.742% (-0.05%) from 92.792%

Build # 26689585807

Build Type

Pull #23343

github

Committed by

web-flow

Commit Message

Merge c42efe377 into c8127c1f4

Pull Request Pull Request #23343: Add buf as an alternate Python protobuf code generator

Coverage Stats

767 of 807 new or added lines in 17 files covered. (95.04%)

69 existing lines in 3 files now uncovered.

93753 of 101090 relevant lines covered (92.74%)

4.01 hits per line

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

85.98

/src/python/pants/backend/codegen/protobuf/buf/config.py

# Copyright 2026 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

"""Shared, language-agnostic helpers for working with `buf.yaml` and `buf.gen.yaml`.

These primitives are used by the per-language buf integrations so each language
only owns its own suffix conventions and module-name math, not yaml parsing
or plugin matching.
"""

from __future__ import annotations

import os
from collections.abc import Mapping, Sequence
from dataclasses import dataclass

import yaml

from pants.backend.codegen.protobuf.buf.fields import BufGenTemplateField
from pants.backend.codegen.protobuf.buf.subsystem import BufSubsystem
from pants.core.util_rules.config_files import ConfigFilesRequest, find_config_file
from pants.engine.intrinsics import get_digest_contents
from pants.engine.rules import concurrently
from pants.engine.target import Target

# ---- Errors ----------------------------------------------------------------


class UnpinnedBufPluginError(Exception):
    """Raised when a `remote:` plugin entry is missing a version+revision pin and
    isn't in the user's `DEFAULT_PLUGIN_PINS` registry."""


class MissingBufLockError(Exception):
    """Raised when `buf.yaml` declares `deps:` but no sibling `buf.lock` exists."""


# ---- Default plugin-pin registry -------------------------------------------


# Default `(version, revision)` pin Pants will fill in for known `remote:`
# plugins that the user wrote without a pin or revision. The synthesized
# `buf.gen.yaml` entry has the form:
#
#     - remote: <id>:<version>
#       revision: <revision>
#       out: ...
#
# These values are the latest as of writing, fetched from the BSR's
# `PluginCurationService.GetLatestCuratedPlugin` endpoint. To refresh:
#
#     curl -X POST \
#       https://buf.build/buf.alpha.registry.v1alpha1.PluginCurationService/GetLatestCuratedPlugin \
#       -H 'Content-Type: application/json' -H 'Connect-Protocol-Version: 1' \
#       -d '{"owner":"<owner>","name":"<plugin>"}'
#
# The browseable equivalent is https://buf.build/<owner>/<plugin> — each
# plugin's "Versions" tab lists every `(version, revision)` pair available.
# Bumping a pin is a one-line change here; cache invalidation is automatic
# via the synthesized `buf.gen.yaml`'s content hash.
DEFAULT_PLUGIN_PINS: Mapping[str, tuple[str, int]] = {
    "buf.build/protocolbuffers/python": ("v34.1", 1),
    "buf.build/protocolbuffers/pyi": ("v34.1", 1),
    "buf.build/connectrpc/python": ("v0.10.0", 1),
    "buf.build/grpc/python": ("v1.80.0", 1),
}


# ---- buf.yaml parsers ------------------------------------------------------


def parse_buf_yaml_module_paths(content: bytes) -> tuple[str, ...]:
    """Module paths declared in a v2 `buf.yaml`, relative to the file.

    Returns `()` for v1 `buf.yaml` (no `modules:` block); callers should treat the
    file's own directory as the implicit module root in that case.
    """
    try:
        data = yaml.safe_load(content)
    except yaml.YAMLError:
        return ()
    if not isinstance(data, dict):
        return ()
    modules = data.get("modules")
    if not isinstance(modules, list):
        return ()
    paths: list[str] = []
    for entry in modules:
        if isinstance(entry, dict):
            p = entry.get("path")
            if isinstance(p, str) and p:
                paths.append(p)
    return tuple(paths)


def parse_buf_yaml_deps(content: bytes) -> tuple[str, ...]:
    """BSR module IDs declared in a `buf.yaml`'s `deps:` list."""
    try:
        data = yaml.safe_load(content)
    except yaml.YAMLError:
        return ()
    if not isinstance(data, dict):
        return ()
    deps = data.get("deps")
    if not isinstance(deps, list):
        return ()
    return tuple(d for d in deps if isinstance(d, str) and d)


def resolve_buf_module_root(
    proto_path: str, buf_yaml_dir: str, module_paths: tuple[str, ...]
) -> str:
    """Resolve the buf module root for a proto file.

    `buf_yaml_dir` is the directory holding `buf.yaml`. `module_paths` are the
    entries from its `modules:` list (relative to that directory). The returned
    root is the longest configured module path that contains the proto, normalized
    to the repo root.
    """
    if not module_paths:
        return buf_yaml_dir
    candidates = sorted(
        (os.path.normpath(os.path.join(buf_yaml_dir, p)) for p in module_paths),
        key=len,
        reverse=True,
    )
    for root in candidates:
        if root == "." or root == "":
            return ""
        if proto_path == root or proto_path.startswith(root + os.sep):
            return root
    return candidates[0]


# ---- buf.gen.yaml parsers --------------------------------------------------


def _plugin_identifier(plugin: dict) -> str | None:
    """Return the registry key (`"<kind>:<ident>"`) for a `buf.gen.yaml` plugin entry.

    `kind` is one of `protoc_builtin`, `local`, or `remote` — matching the field
    name in the entry. `ident` strips any `:vX.Y` version pin from `remote:` so
    callers can match against the registry without knowing the version. Returns
    `None` if the entry declares none of these fields.
    """
    for key in ("protoc_builtin", "local", "remote"):
        val = plugin.get(key)
        if val is None:
            continue
        ident = " ".join(str(x) for x in val) if isinstance(val, list) else str(val)
        if key == "remote":
            base, _, _ = ident.partition(":")
            return f"remote:{base}"
        return f"{key}:{ident}"
    return None


def parse_plugin_outs(content: bytes, suffixes: Mapping[str, str]) -> dict[str, str]:
    """Walk `buf.gen.yaml` plugins and return `suffix -> out:` for matching entries.

    `suffixes` is a `<kind>:<ident> -> suffix` dict supplied by the calling
    language backend, where `<kind>` is `remote`, `protoc_builtin`, or `local`
    and `suffix` is the language's module/file-naming suffix. The first matching
    plugin per suffix wins.
    """
    try:
        data = yaml.safe_load(content)
    except yaml.YAMLError:
        return {}
    if not isinstance(data, dict):
        return {}
    plugins = data.get("plugins")
    if not isinstance(plugins, list):
        return {}

    result: dict[str, str] = {}
    for plugin in plugins:
        if not isinstance(plugin, dict):
            continue
        out = plugin.get("out")
        if not isinstance(out, str) or not out:
            continue
        key = _plugin_identifier(plugin)
        if key is None:
            continue
        suffix = suffixes.get(key)
        if suffix is not None and suffix not in result:
            result[suffix] = out
    return result


def suffix_plugin_includes_imports(
    content: bytes, suffix: str, suffixes: Mapping[str, str]
) -> bool:
    """True if the `buf.gen.yaml` plugin emitting `suffix` has `include_imports:
    true` set — meaning buf will materialize generated artifacts for
    transitively-imported BSR-dep protos into the digest.

    `suffixes` is the same `<kind>:<ident> -> suffix` mapping passed to
    `parse_plugin_outs`; the first plugin entry whose registry key maps to
    `suffix` wins.
    """
    try:
        data = yaml.safe_load(content)
    except yaml.YAMLError:
        return False
    if not isinstance(data, dict):
        return False
    plugins = data.get("plugins")
    if not isinstance(plugins, list):
        return False
    for plugin in plugins:
        if not isinstance(plugin, dict):
            continue
        out = plugin.get("out")
        if not isinstance(out, str) or not out:
            continue
        key = _plugin_identifier(plugin)
        if key is None:
            continue
        if suffixes.get(key) != suffix:
            continue
        return plugin.get("include_imports") is True
    return False


# ---- buf.gen.yaml pin synthesis --------------------------------------------


def _split_remote_ident(ident: str) -> tuple[str, str | None]:
    """Split a `remote:` value into `(base_id, version_or_None)`.

    `buf.build/foo/bar:v1.2` → (`buf.build/foo/bar`, `v1.2`).
    `buf.build/foo/bar` → (`buf.build/foo/bar`, None).
    """
    base, sep, suffix = ident.partition(":")
    return (base, suffix) if sep else (base, None)


def _parse_pin_string(pin: str) -> tuple[str, int] | None:
    """Parse `"vX.Y:N"` (Pants-internal pin format) into `(version, revision)`.
    Returns `None` if the format is invalid."""
    parts = pin.split(":")
    if len(parts) != 2 or not parts[0]:
        return None
    try:
        return parts[0], int(parts[1])
    except ValueError:
        return None


def synthesize_pinned_buf_gen_yaml(
    content: bytes,
    source_path: str,
    *,
    extra_pins: Mapping[str, str] | None = None,
) -> bytes:
    """Return the user's `buf.gen.yaml` with `remote:` plugin pins resolved.

    For each `remote:` entry, the buf-recognized pinned form is:

        - remote: <id>:<version>
          revision: <int>
          out: ...

    Pants requires both fields. If a plugin entry is missing either, it must be
    in `DEFAULT_PLUGIN_PINS` (or the user's `extra_pins`) for Pants to fill in
    defaults. `extra_pins` values are `"vX.Y:N"` strings, parsed here.
    `protoc_builtin:` and `local:` plugins are not subject to pin enforcement.
    """
    try:
        data = yaml.safe_load(content)
    except yaml.YAMLError:
        return content
    if not isinstance(data, dict):
        return content
    plugins = data.get("plugins")
    if not isinstance(plugins, list):
        return content

    parsed_extra: dict[str, tuple[str, int]] = {}
    for k, v in (extra_pins or {}).items():
        parsed = _parse_pin_string(v)
        if parsed is not None:
            parsed_extra[k] = parsed
    pins: Mapping[str, tuple[str, int]] = {**DEFAULT_PLUGIN_PINS, **parsed_extra}

    unresolvable: list[str] = []
    rewrote = False
    for plugin in plugins:
        if not isinstance(plugin, dict):
            continue
        val = plugin.get("remote")
        if val is None:
            continue
        ident = " ".join(str(x) for x in val) if isinstance(val, list) else str(val)
        base, version = _split_remote_ident(ident)
        revision = plugin.get("revision")
        if version is not None and isinstance(revision, int):
            continue  # already fully pinned
        default = pins.get(base)
        if default is None:
            unresolvable.append(ident)
            continue
        default_version, default_revision = default
        plugin["remote"] = f"{base}:{default_version}"
        plugin["revision"] = default_revision
        rewrote = True

    if unresolvable:
        bullets = "\n".join(f"  - remote: {ident}" for ident in unresolvable)
        known = ", ".join(sorted(DEFAULT_PLUGIN_PINS)) or "(none)"
        raise UnpinnedBufPluginError(
            f"`{source_path}` has `remote:` plugin entries that are missing a "
            f"version or `revision:`:\n{bullets}\n\n"
            "Pin both fields explicitly:\n\n"
            "    - remote: buf.build/owner/plugin:vX.Y\n"
            "      revision: N\n"
            "      out: ...\n\n"
            "Alternatively, for a plugin in Pants's built-in registry, leave both "
            "fields unset and Pants will fill in defaults. To extend the registry "
            "for your own plugins, use `[buf].extra_plugin_pins`.\n\n"
            f"Built-in registry: {known}."
        )

    if not rewrote:
        return content
    return yaml.safe_dump(data, sort_keys=False).encode("utf-8")


def check_pinned_remote_plugins(
    content: bytes,
    source_path: str,
    *,
    extra_pins: Mapping[str, str] | None = None,
) -> None:
    """Raise if `remote:` plugin entries can't be resolved to a full pin, without
    returning the synthesized content."""
    synthesize_pinned_buf_gen_yaml(content, source_path, extra_pins=extra_pins)


# ---- Per-target template request resolvers --------------------------------


def gen_template_request_from_fields(
    *,
    spec_path: str,
    address_str: str,
    override: str | None,
    buf: BufSubsystem,
) -> ConfigFilesRequest:
    """Resolve the `buf.gen.yaml` request from already-extracted field values.

    Precedence: per-target `buf_gen_template` (`override`) → `[buf].gen_template`
    subsystem option → `[buf].gen_template_discovery`.
    """
    if override is None:
        return buf.gen_template_request
    path = os.path.normpath(os.path.join(spec_path, override))
    return ConfigFilesRequest(
        specified=path,
        specified_option_name=f"`{BufGenTemplateField.alias}` field on {address_str}",
        discovery=False,
        check_existence=(path,),
    )


def gen_template_request_for_target(tgt: Target, buf: BufSubsystem) -> ConfigFilesRequest:
    """Convenience wrapper around `gen_template_request_from_fields` for a Target."""
    return gen_template_request_from_fields(
        spec_path=tgt.address.spec_path,
        address_str=str(tgt.address),
        override=tgt.get(BufGenTemplateField).value,
        buf=buf,
    )


def resolved_template_path(tgt: Target, buf: BufSubsystem) -> str | None:
    """Path to pass to `buf generate --template`, or None to rely on discovery."""
    override = tgt.get(BufGenTemplateField).value
    if override is not None:
        return os.path.normpath(os.path.join(tgt.address.spec_path, override))
    return buf.gen_template


# ---- Async fetchers + result types ----------------------------------------


@dataclass(frozen=True)
class BufLayout:
    """Module layout derived from `buf.yaml`."""

    buf_yaml_dir: str
    module_paths: tuple[str, ...]
    deps: tuple[str, ...]  # BSR module ids (e.g. `buf.build/bufbuild/protovalidate`)

    def root_for_proto(self, proto_path: str) -> str:
        return resolve_buf_module_root(proto_path, self.buf_yaml_dir, self.module_paths)


async def fetch_buf_layout(buf: BufSubsystem) -> BufLayout:
    """Read `buf.yaml` and return the parsed module layout. Empty if not found.

    `config_request` may also surface `buf.lock` (it's listed in `check_existence`
    so codegen invalidates on lock changes), so we filter to `buf.yaml` here.
    """
    files = await find_config_file(buf.config_request)
    yaml_paths = [p for p in files.snapshot.files if os.path.basename(p) == "buf.yaml"]
    if not yaml_paths:
        return BufLayout("", (), ())
    path = yaml_paths[0]
    contents = await get_digest_contents(files.snapshot.digest)
    content = next((dc.content for dc in contents if dc.path == path), b"")
    return BufLayout(
        os.path.dirname(path),
        parse_buf_yaml_module_paths(content),
        parse_buf_yaml_deps(content),
    )


@dataclass(frozen=True)
class BufGenContent:
    """Per-target `buf.gen.yaml` resolution.

    `template_path` is `None` when no template was found (callers should fall back
    to source-root path arithmetic). When set, `content` is the raw yaml.
    """

    target: Target
    template_path: str | None
    content: bytes


async def fetch_buf_gen_contents(
    targets: Sequence[Target], buf: BufSubsystem
) -> tuple[BufGenContent, ...]:
    """Resolve and read each target's effective `buf.gen.yaml`."""
    if not targets:
        return ()
    template_files_per_target = await concurrently(
        find_config_file(gen_template_request_for_target(t, buf)) for t in targets
    )
    contents_per_target = await concurrently(
        get_digest_contents(tf.snapshot.digest) for tf in template_files_per_target
    )
    out: list[BufGenContent] = []
    for tgt, files, dcs in zip(targets, template_files_per_target, contents_per_target):
        if not files.snapshot.files:
            out.append(BufGenContent(tgt, None, b""))
            continue
        path = files.snapshot.files[0]
        content = next((dc.content for dc in dcs if dc.path == path), b"")
        out.append(BufGenContent(tgt, path, content))
    return tuple(out)

1	# Copyright 2026 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	"""Shared, language-agnostic helpers for working with `buf.yaml` and `buf.gen.yaml`.
5
6	These primitives are used by the per-language buf integrations so each language
7	only owns its own suffix conventions and module-name math, not yaml parsing
8	or plugin matching.
9	"""
10
11	from __future__ import annotations	8✔
12
13	import os	8✔
14	from collections.abc import Mapping, Sequence	8✔
15	from dataclasses import dataclass	8✔
16
17	import yaml	8✔
18
19	from pants.backend.codegen.protobuf.buf.fields import BufGenTemplateField	8✔
20	from pants.backend.codegen.protobuf.buf.subsystem import BufSubsystem	8✔
21	from pants.core.util_rules.config_files import ConfigFilesRequest, find_config_file	8✔
22	from pants.engine.intrinsics import get_digest_contents	8✔
23	from pants.engine.rules import concurrently	8✔
24	from pants.engine.target import Target	8✔
25
26	# ---- Errors ----------------------------------------------------------------
27
28
29	class UnpinnedBufPluginError(Exception):	8✔
30	"""Raised when a `remote:` plugin entry is missing a version+revision pin and
31	isn't in the user's `DEFAULT_PLUGIN_PINS` registry."""
32
33
34	class MissingBufLockError(Exception):	8✔
35	"""Raised when `buf.yaml` declares `deps:` but no sibling `buf.lock` exists."""
36
37
38	# ---- Default plugin-pin registry -------------------------------------------
39
40
41	# Default `(version, revision)` pin Pants will fill in for known `remote:`
42	# plugins that the user wrote without a pin or revision. The synthesized
43	# `buf.gen.yaml` entry has the form:
44	#
45	# - remote: <id>:<version>
46	# revision: <revision>
47	# out: ...
48	#
49	# These values are the latest as of writing, fetched from the BSR's
50	# `PluginCurationService.GetLatestCuratedPlugin` endpoint. To refresh:
51	#
52	# curl -X POST \
53	# https://buf.build/buf.alpha.registry.v1alpha1.PluginCurationService/GetLatestCuratedPlugin \
54	# -H 'Content-Type: application/json' -H 'Connect-Protocol-Version: 1' \
55	# -d '{"owner":"<owner>","name":"<plugin>"}'
56	#
57	# The browseable equivalent is https://buf.build/<owner>/<plugin> — each
58	# plugin's "Versions" tab lists every `(version, revision)` pair available.
59	# Bumping a pin is a one-line change here; cache invalidation is automatic
60	# via the synthesized `buf.gen.yaml`'s content hash.
61	DEFAULT_PLUGIN_PINS: Mapping[str, tuple[str, int]] = {	8✔
62	"buf.build/protocolbuffers/python": ("v34.1", 1),
63	"buf.build/protocolbuffers/pyi": ("v34.1", 1),
64	"buf.build/connectrpc/python": ("v0.10.0", 1),
65	"buf.build/grpc/python": ("v1.80.0", 1),
66	}
67
68
69	# ---- buf.yaml parsers ------------------------------------------------------
70
71
72	def parse_buf_yaml_module_paths(content: bytes) -> tuple[str, ...]:	8✔
73	"""Module paths declared in a v2 `buf.yaml`, relative to the file.
74
75	Returns `()` for v1 `buf.yaml` (no `modules:` block); callers should treat the
76	file's own directory as the implicit module root in that case.
77	"""
78	try:	1✔
79	data = yaml.safe_load(content)	1✔
NEW 80	except yaml.YAMLError:	×
NEW 81	return ()	×
82	if not isinstance(data, dict):	1✔
NEW 83	return ()	×
84	modules = data.get("modules")	1✔
85	if not isinstance(modules, list):	1✔
NEW 86	return ()	×
87	paths: list[str] = []	1✔
88	for entry in modules:	1✔
89	if isinstance(entry, dict):	1✔
90	p = entry.get("path")	1✔
91	if isinstance(p, str) and p:	1✔
92	paths.append(p)	1✔
93	return tuple(paths)	1✔
94
95
96	def parse_buf_yaml_deps(content: bytes) -> tuple[str, ...]:	8✔
97	"""BSR module IDs declared in a `buf.yaml`'s `deps:` list."""
98	try:	4✔
99	data = yaml.safe_load(content)	4✔
100	except yaml.YAMLError:	1✔
101	return ()	1✔
102	if not isinstance(data, dict):	4✔
103	return ()	1✔
104	deps = data.get("deps")	4✔
105	if not isinstance(deps, list):	4✔
106	return ()	4✔
107	return tuple(d for d in deps if isinstance(d, str) and d)	4✔
108
109
110	def resolve_buf_module_root(	8✔
111	proto_path: str, buf_yaml_dir: str, module_paths: tuple[str, ...]
112	) -> str:
113	"""Resolve the buf module root for a proto file.
114
115	`buf_yaml_dir` is the directory holding `buf.yaml`. `module_paths` are the
116	entries from its `modules:` list (relative to that directory). The returned
117	root is the longest configured module path that contains the proto, normalized
118	to the repo root.
119	"""
120	if not module_paths:	1✔
121	return buf_yaml_dir	1✔
122	candidates = sorted(	1✔
123	(os.path.normpath(os.path.join(buf_yaml_dir, p)) for p in module_paths),
124	key=len,
125	reverse=True,
126	)
127	for root in candidates:	1✔
128	if root == "." or root == "":	1✔
129	return ""	1✔
130	if proto_path == root or proto_path.startswith(root + os.sep):	1✔
131	return root	1✔
NEW 132	return candidates[0]	×
133
134
135	# ---- buf.gen.yaml parsers --------------------------------------------------
136
137
138	def _plugin_identifier(plugin: dict) -> str \| None:	8✔
139	"""Return the registry key (`"<kind>:<ident>"`) for a `buf.gen.yaml` plugin entry.
140
141	`kind` is one of `protoc_builtin`, `local`, or `remote` — matching the field
142	name in the entry. `ident` strips any `:vX.Y` version pin from `remote:` so
143	callers can match against the registry without knowing the version. Returns
144	`None` if the entry declares none of these fields.
145	"""
146	for key in ("protoc_builtin", "local", "remote"):	3✔
147	val = plugin.get(key)	3✔
148	if val is None:	3✔
149	continue	3✔
150	ident = " ".join(str(x) for x in val) if isinstance(val, list) else str(val)	3✔
151	if key == "remote":	3✔
152	base, _, _ = ident.partition(":")	3✔
153	return f"remote:{base}"	3✔
154	return f"{key}:{ident}"	2✔
NEW 155	return None	×
156
157
158	def parse_plugin_outs(content: bytes, suffixes: Mapping[str, str]) -> dict[str, str]:	8✔
159	"""Walk `buf.gen.yaml` plugins and return `suffix -> out:` for matching entries.
160
161	`suffixes` is a `<kind>:<ident> -> suffix` dict supplied by the calling
162	language backend, where `<kind>` is `remote`, `protoc_builtin`, or `local`
163	and `suffix` is the language's module/file-naming suffix. The first matching
164	plugin per suffix wins.
165	"""
166	try:	3✔
167	data = yaml.safe_load(content)	3✔
NEW 168	except yaml.YAMLError:	×
NEW 169	return {}	×
170	if not isinstance(data, dict):	3✔
NEW 171	return {}	×
172	plugins = data.get("plugins")	3✔
173	if not isinstance(plugins, list):	3✔
NEW 174	return {}	×
175
176	result: dict[str, str] = {}	3✔
177	for plugin in plugins:	3✔
178	if not isinstance(plugin, dict):	3✔
NEW 179	continue	×
180	out = plugin.get("out")	3✔
181	if not isinstance(out, str) or not out:	3✔
NEW 182	continue	×
183	key = _plugin_identifier(plugin)	3✔
184	if key is None:	3✔
NEW 185	continue	×
186	suffix = suffixes.get(key)	3✔
187	if suffix is not None and suffix not in result:	3✔
188	result[suffix] = out	3✔
189	return result	3✔
190
191
192	def suffix_plugin_includes_imports(	8✔
193	content: bytes, suffix: str, suffixes: Mapping[str, str]
194	) -> bool:
195	"""True if the `buf.gen.yaml` plugin emitting `suffix` has `include_imports:
196	true` set — meaning buf will materialize generated artifacts for
197	transitively-imported BSR-dep protos into the digest.
198
199	`suffixes` is the same `<kind>:<ident> -> suffix` mapping passed to
200	`parse_plugin_outs`; the first plugin entry whose registry key maps to
201	`suffix` wins.
202	"""
203	try:	2✔
204	data = yaml.safe_load(content)	2✔
NEW 205	except yaml.YAMLError:	×
NEW 206	return False	×
207	if not isinstance(data, dict):	2✔
NEW 208	return False	×
209	plugins = data.get("plugins")	2✔
210	if not isinstance(plugins, list):	2✔
NEW 211	return False	×
212	for plugin in plugins:	2✔
213	if not isinstance(plugin, dict):	2✔
NEW 214	continue	×
215	out = plugin.get("out")	2✔
216	if not isinstance(out, str) or not out:	2✔
NEW 217	continue	×
218	key = _plugin_identifier(plugin)	2✔
219	if key is None:	2✔
NEW 220	continue	×
221	if suffixes.get(key) != suffix:	2✔
222	continue	1✔
223	return plugin.get("include_imports") is True	2✔
NEW 224	return False	×
225
226
227	# ---- buf.gen.yaml pin synthesis --------------------------------------------
228
229
230	def _split_remote_ident(ident: str) -> tuple[str, str \| None]:	8✔
231	"""Split a `remote:` value into `(base_id, version_or_None)`.
232
233	`buf.build/foo/bar:v1.2` → (`buf.build/foo/bar`, `v1.2`).
234	`buf.build/foo/bar` → (`buf.build/foo/bar`, None).
235	"""
236	base, sep, suffix = ident.partition(":")	4✔
237	return (base, suffix) if sep else (base, None)	4✔
238
239
240	def _parse_pin_string(pin: str) -> tuple[str, int] \| None:	8✔
241	"""Parse `"vX.Y:N"` (Pants-internal pin format) into `(version, revision)`.
242	Returns `None` if the format is invalid."""
243	parts = pin.split(":")	1✔
244	if len(parts) != 2 or not parts[0]:	1✔
245	return None	1✔
246	try:	1✔
247	return parts[0], int(parts[1])	1✔
NEW 248	except ValueError:	×
NEW 249	return None	×
250
251
252	def synthesize_pinned_buf_gen_yaml(	8✔
253	content: bytes,
254	source_path: str,
255	*,
256	extra_pins: Mapping[str, str] \| None = None,
257	) -> bytes:
258	"""Return the user's `buf.gen.yaml` with `remote:` plugin pins resolved.
259
260	For each `remote:` entry, the buf-recognized pinned form is:
261
262	- remote: <id>:<version>
263	revision: <int>
264	out: ...
265
266	Pants requires both fields. If a plugin entry is missing either, it must be
267	in `DEFAULT_PLUGIN_PINS` (or the user's `extra_pins`) for Pants to fill in
268	defaults. `extra_pins` values are `"vX.Y:N"` strings, parsed here.
269	`protoc_builtin:` and `local:` plugins are not subject to pin enforcement.
270	"""
271	try:	4✔
272	data = yaml.safe_load(content)	4✔
NEW 273	except yaml.YAMLError:	×
NEW 274	return content	×
275	if not isinstance(data, dict):	4✔
NEW 276	return content	×
277	plugins = data.get("plugins")	4✔
278	if not isinstance(plugins, list):	4✔
NEW 279	return content	×
280
281	parsed_extra: dict[str, tuple[str, int]] = {}	4✔
282	for k, v in (extra_pins or {}).items():	4✔
283	parsed = _parse_pin_string(v)	1✔
284	if parsed is not None:	1✔
285	parsed_extra[k] = parsed	1✔
286	pins: Mapping[str, tuple[str, int]] = {DEFAULT_PLUGIN_PINS, parsed_extra}	4✔
287
288	unresolvable: list[str] = []	4✔
289	rewrote = False	4✔
290	for plugin in plugins:	4✔
291	if not isinstance(plugin, dict):	4✔
NEW 292	continue	×
293	val = plugin.get("remote")	4✔
294	if val is None:	4✔
295	continue	4✔
296	ident = " ".join(str(x) for x in val) if isinstance(val, list) else str(val)	4✔
297	base, version = _split_remote_ident(ident)	4✔
298	revision = plugin.get("revision")	4✔
299	if version is not None and isinstance(revision, int):	4✔
300	continue # already fully pinned	1✔
301	default = pins.get(base)	4✔
302	if default is None:	4✔
303	unresolvable.append(ident)	1✔
304	continue	1✔
305	default_version, default_revision = default	4✔
306	plugin["remote"] = f"{base}:{default_version}"	4✔
307	plugin["revision"] = default_revision	4✔
308	rewrote = True	4✔
309
310	if unresolvable:	4✔
311	bullets = "\n".join(f" - remote: {ident}" for ident in unresolvable)	1✔
312	known = ", ".join(sorted(DEFAULT_PLUGIN_PINS)) or "(none)"	1✔
313	raise UnpinnedBufPluginError(	1✔
314	f"`{source_path}` has `remote:` plugin entries that are missing a "
315	f"version or `revision:`:\n{bullets}\n\n"
316	"Pin both fields explicitly:\n\n"
317	" - remote: buf.build/owner/plugin:vX.Y\n"
318	" revision: N\n"
319	" out: ...\n\n"
320	"Alternatively, for a plugin in Pants's built-in registry, leave both "
321	"fields unset and Pants will fill in defaults. To extend the registry "
322	"for your own plugins, use `[buf].extra_plugin_pins`.\n\n"
323	f"Built-in registry: {known}."
324	)
325
326	if not rewrote:	4✔
327	return content	4✔
328	return yaml.safe_dump(data, sort_keys=False).encode("utf-8")	4✔
329
330
331	def check_pinned_remote_plugins(	8✔
332	content: bytes,
333	source_path: str,
334	*,
335	extra_pins: Mapping[str, str] \| None = None,
336	) -> None:
337	"""Raise if `remote:` plugin entries can't be resolved to a full pin, without
338	returning the synthesized content."""
NEW 339	synthesize_pinned_buf_gen_yaml(content, source_path, extra_pins=extra_pins)	×
340
341
342	# ---- Per-target template request resolvers --------------------------------
343
344
345	def gen_template_request_from_fields(	8✔
346	*,
347	spec_path: str,
348	address_str: str,
349	override: str \| None,
350	buf: BufSubsystem,
351	) -> ConfigFilesRequest:
352	"""Resolve the `buf.gen.yaml` request from already-extracted field values.
353
354	Precedence: per-target `buf_gen_template` (`override`) → `[buf].gen_template`
355	subsystem option → `[buf].gen_template_discovery`.
356	"""
357	if override is None:	4✔
358	return buf.gen_template_request	4✔
359	path = os.path.normpath(os.path.join(spec_path, override))	3✔
360	return ConfigFilesRequest(	3✔
361	specified=path,
362	specified_option_name=f"`{BufGenTemplateField.alias}` field on {address_str}",
363	discovery=False,
364	check_existence=(path,),
365	)
366
367
368	def gen_template_request_for_target(tgt: Target, buf: BufSubsystem) -> ConfigFilesRequest:	8✔
369	"""Convenience wrapper around `gen_template_request_from_fields` for a Target."""
370	return gen_template_request_from_fields(	3✔
371	spec_path=tgt.address.spec_path,
372	address_str=str(tgt.address),
373	override=tgt.get(BufGenTemplateField).value,
374	buf=buf,
375	)
376
377
378	def resolved_template_path(tgt: Target, buf: BufSubsystem) -> str \| None:	8✔
379	"""Path to pass to `buf generate --template`, or None to rely on discovery."""
380	override = tgt.get(BufGenTemplateField).value	3✔
381	if override is not None:	3✔
382	return os.path.normpath(os.path.join(tgt.address.spec_path, override))	3✔
383	return buf.gen_template	3✔
384
385
386	# ---- Async fetchers + result types ----------------------------------------
387
388
389	@dataclass(frozen=True)	8✔
390	class BufLayout:	8✔
391	"""Module layout derived from `buf.yaml`."""
392
393	buf_yaml_dir: str	8✔
394	module_paths: tuple[str, ...]	8✔
395	deps: tuple[str, ...] # BSR module ids (e.g. `buf.build/bufbuild/protovalidate`)	8✔
396
397	def root_for_proto(self, proto_path: str) -> str:	8✔
398	return resolve_buf_module_root(proto_path, self.buf_yaml_dir, self.module_paths)	1✔
399
400
401	async def fetch_buf_layout(buf: BufSubsystem) -> BufLayout:	8✔
402	"""Read `buf.yaml` and return the parsed module layout. Empty if not found.
403
404	`config_request` may also surface `buf.lock` (it's listed in `check_existence`
405	so codegen invalidates on lock changes), so we filter to `buf.yaml` here.
406	"""
407	files = await find_config_file(buf.config_request)	1✔
408	yaml_paths = [p for p in files.snapshot.files if os.path.basename(p) == "buf.yaml"]	1✔
409	if not yaml_paths:	1✔
410	return BufLayout("", (), ())	1✔
411	path = yaml_paths[0]	1✔
412	contents = await get_digest_contents(files.snapshot.digest)	1✔
413	content = next((dc.content for dc in contents if dc.path == path), b"")	1✔
414	return BufLayout(	1✔
415	os.path.dirname(path),
416	parse_buf_yaml_module_paths(content),
417	parse_buf_yaml_deps(content),
418	)
419
420
421	@dataclass(frozen=True)	8✔
422	class BufGenContent:	8✔
423	"""Per-target `buf.gen.yaml` resolution.
424
425	`template_path` is `None` when no template was found (callers should fall back
426	to source-root path arithmetic). When set, `content` is the raw yaml.
427	"""
428
429	target: Target	8✔
430	template_path: str \| None	8✔
431	content: bytes	8✔
432
433
434	async def fetch_buf_gen_contents(	8✔
435	targets: Sequence[Target], buf: BufSubsystem
436	) -> tuple[BufGenContent, ...]:
437	"""Resolve and read each target's effective `buf.gen.yaml`."""
438	if not targets:	1✔
NEW 439	return ()	×
440	template_files_per_target = await concurrently(	1✔
441	find_config_file(gen_template_request_for_target(t, buf)) for t in targets
442	)
443	contents_per_target = await concurrently(	1✔
444	get_digest_contents(tf.snapshot.digest) for tf in template_files_per_target
445	)
446	out: list[BufGenContent] = []	1✔
447	for tgt, files, dcs in zip(targets, template_files_per_target, contents_per_target):	1✔
448	if not files.snapshot.files:	1✔
449	out.append(BufGenContent(tgt, None, b""))	1✔
450	continue	1✔
451	path = files.snapshot.files[0]	1✔
452	content = next((dc.content for dc in dcs if dc.path == path), b"")	1✔
453	out.append(BufGenContent(tgt, path, content))	1✔
454	return tuple(out)	1✔

pantsbuild / pants / 26689585807

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