20332790708

Committed 18 Dec 2025 09:48AM UTC coverage: 64.992% (-15.3%) from 80.295%

Build # 20332790708

Build Type

Pull #22949

github

Committed by

web-flow

Commit Message

Merge f730a56cd into 407284c67

Pull Request Pull Request #22949: Add experimental uv resolver for Python lockfiles

Run Details

54 of 97 new or added lines in 5 files covered. (55.67%)

8270 existing lines in 295 files now uncovered.

48990 of 75379 relevant lines covered (64.99%)

1.81 hits per line

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

57.33

/src/python/pants/backend/codegen/protobuf/lint/buf/lint_rules_integration_test.py

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

from __future__ import annotations

import json
from textwrap import dedent
from typing import Any

import pytest

from pants.backend.codegen.protobuf.lint.buf.lint_rules import BufFieldSet, BufLintRequest
from pants.backend.codegen.protobuf.lint.buf.lint_rules import rules as buf_rules
from pants.backend.codegen.protobuf.target_types import ProtobufSourcesGeneratorTarget
from pants.backend.codegen.protobuf.target_types import rules as target_types_rules
from pants.core.goals.lint import LintResult, Partitions
from pants.core.util_rules import config_files, external_tool, stripped_source_files
from pants.engine.addresses import Address
from pants.engine.target import Target
from pants.testutil.rule_runner import QueryRule, RuleRunner


@pytest.fixture
def rule_runner() -> RuleRunner:
    return RuleRunner(
        rules=[
            *buf_rules(),
            *config_files.rules(),
            *external_tool.rules(),
            *stripped_source_files.rules(),
            *target_types_rules(),
            QueryRule(Partitions, [BufLintRequest.PartitionRequest]),
            QueryRule(LintResult, [BufLintRequest.Batch]),
        ],
        target_types=[ProtobufSourcesGeneratorTarget],
    )


GOOD_FILE = 'syntax = "proto3";\npackage foo.v1;\nmessage Foo {\nstring snake_case = 1;\n}\n'
BAD_FILE = 'syntax = "proto3";\npackage foo.v1;\nmessage Bar {\nstring camelCase = 1;\n}\n'


def run_buf(
    rule_runner: RuleRunner,
    targets: list[Target],
    *,
    source_roots: list[str] | None = None,
    extra_args: list[str] | None = None,
) -> tuple[LintResult, ...]:
    rule_runner.set_options(
        [
            f"--source-root-patterns={repr(source_roots)}",
            "--backend-packages=pants.backend.codegen.protobuf.lint.buf",
            *(extra_args or ()),
        ],
        env_inherit={"PATH"},
    )
    partitions = rule_runner.request(
        Partitions[BufFieldSet, Any],
        [BufLintRequest.PartitionRequest(tuple(BufFieldSet.create(tgt) for tgt in targets))],
    )
    results = []
    for partition in partitions:
        result = rule_runner.request(
            LintResult,
            [BufLintRequest.Batch("", partition.elements, partition.metadata)],
        )
        results.append(result)
    return tuple(results)


def assert_success(
    rule_runner: RuleRunner,
    target: Target,
    *,
    source_roots: list[str] | None = None,
    extra_args: list[str] | None = None,
) -> None:
    result = run_buf(rule_runner, [target], source_roots=source_roots, extra_args=extra_args)
    assert len(result) == 1
    assert result[0].exit_code == 0
    assert not result[0].stdout
    assert not result[0].stderr


def test_passing(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {"foo/v1/f.proto": GOOD_FILE, "foo/v1/BUILD": "protobuf_sources(name='t')"}
    )
    tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))
    assert_success(rule_runner, tgt)


@pytest.mark.platform_specific_behavior
def test_failing(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {"foo/v1/f.proto": BAD_FILE, "foo/v1/BUILD": "protobuf_sources(name='t')"}
    )
    tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))
    result = run_buf(rule_runner, [tgt])
    assert len(result) == 1
    assert result[0].exit_code == 100
    assert "foo/v1/f.proto:" in result[0].stdout


def test_multiple_targets(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "foo/v1/good.proto": GOOD_FILE,
            "foo/v1/bad.proto": BAD_FILE,
            "foo/v1/BUILD": "protobuf_sources(name='t')",
        }
    )
    tgts = [
        rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="good.proto")),
        rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="bad.proto")),
    ]
    result = run_buf(rule_runner, tgts)
    assert len(result) == 1
    assert result[0].exit_code == 100
    assert "good.proto" not in result[0].stdout
    assert "foo/v1/bad.proto:" in result[0].stdout


def test_passthrough_args(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {"foo/v1/f.proto": BAD_FILE, "foo/v1/BUILD": "protobuf_sources(name='t')"}
    )
    tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))
    config = json.dumps(
        {
            "version": "v1",
            "lint": {
                "ignore_only": {
                    "FIELD_LOWER_SNAKE_CASE": [
                        "foo/v1/f.proto",
                    ],
                },
            },
        }
    )

    assert_success(
        rule_runner,
        tgt,
        extra_args=[f"--buf-lint-args='--config={config}'"],
    )


def test_skip(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {"foo/v1/f.proto": BAD_FILE, "foo/v1/BUILD": "protobuf_sources(name='t')"}
    )
    tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))
    result = run_buf(rule_runner, [tgt], extra_args=["--buf-lint-skip"])
    assert not result


def test_dependencies(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "src/protobuf/dir1/v1/f.proto": dedent(
                """\
                syntax = "proto3";
                package dir1.v1;
                message Person {
                  string name = 1;
                  int32 id = 2;
                  string email = 3;
                }
                """
            ),
            "src/protobuf/dir1/v1/BUILD": "protobuf_sources()",
            "src/protobuf/dir2/v1/f.proto": dedent(
                """\
                syntax = "proto3";
                package dir2.v1;
                import "dir1/v1/f.proto";

                message Person {
                  dir1.v1.Person person = 1;
                }
                """
            ),
            "src/protobuf/dir2/v1/BUILD": (
                "protobuf_sources(dependencies=['src/protobuf/dir1/v1'])"
            ),
            "tests/protobuf/test_protos/v1/f.proto": dedent(
                """\
                syntax = "proto3";
                package test_protos.v1;
                import "dir2/v1/f.proto";

                message Person {
                  dir2.v1.Person person = 1;
                }
                """
            ),
            "tests/protobuf/test_protos/v1/BUILD": (
                "protobuf_sources(dependencies=['src/protobuf/dir2/v1'])"
            ),
        }
    )

    tgt = rule_runner.get_target(
        Address("tests/protobuf/test_protos/v1", relative_file_path="f.proto")
    )
    assert_success(
        rule_runner, tgt, source_roots=["src/python", "/src/protobuf", "/tests/protobuf"]
    )


def test_config_discovery(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "foo/v1/f.proto": BAD_FILE,
            "foo/v1/BUILD": "protobuf_sources(name='t')",
            "buf.yaml": dedent(
                """\
             version: v1
             lint:
               ignore_only:
                 FIELD_LOWER_SNAKE_CASE:
                   - foo/v1/f.proto
             """
            ),
        }
    )

    tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))

    assert_success(
        rule_runner,
        tgt,
    )


def test_config_file_submitted(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "foo/v1/f.proto": BAD_FILE,
            "foo/v1/BUILD": "protobuf_sources(name='t')",
            # Intentionally placed somewhere config_discovery can't see.
            "foo/buf.yaml": dedent(
                """\
             version: v1
             lint:
               ignore_only:
                 FIELD_LOWER_SNAKE_CASE:
                   - foo/v1/f.proto
             """,
            ),
        }
    )

    tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))

    assert_success(
        rule_runner,
        tgt,
        extra_args=["--buf-config=foo/buf.yaml"],
    )

1	# Copyright 2022 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 json	1✔
7	from textwrap import dedent	1✔
8	from typing import Any	1✔
9
10	import pytest	1✔
11
12	from pants.backend.codegen.protobuf.lint.buf.lint_rules import BufFieldSet, BufLintRequest	1✔
13	from pants.backend.codegen.protobuf.lint.buf.lint_rules import rules as buf_rules	1✔
14	from pants.backend.codegen.protobuf.target_types import ProtobufSourcesGeneratorTarget	1✔
15	from pants.backend.codegen.protobuf.target_types import rules as target_types_rules	1✔
16	from pants.core.goals.lint import LintResult, Partitions	1✔
17	from pants.core.util_rules import config_files, external_tool, stripped_source_files	1✔
18	from pants.engine.addresses import Address	1✔
19	from pants.engine.target import Target	1✔
20	from pants.testutil.rule_runner import QueryRule, RuleRunner	1✔
21
22
23	@pytest.fixture	1✔
24	def rule_runner() -> RuleRunner:	1✔
25	return RuleRunner(	1✔
26	rules=[
27	*buf_rules(),
28	*config_files.rules(),
29	*external_tool.rules(),
30	*stripped_source_files.rules(),
31	*target_types_rules(),
32	QueryRule(Partitions, [BufLintRequest.PartitionRequest]),
33	QueryRule(LintResult, [BufLintRequest.Batch]),
34	],
35	target_types=[ProtobufSourcesGeneratorTarget],
36	)
37
38
39	GOOD_FILE = 'syntax = "proto3";\npackage foo.v1;\nmessage Foo {\nstring snake_case = 1;\n}\n'	1✔
40	BAD_FILE = 'syntax = "proto3";\npackage foo.v1;\nmessage Bar {\nstring camelCase = 1;\n}\n'	1✔
41
42
43	def run_buf(	1✔
44	rule_runner: RuleRunner,
45	targets: list[Target],
46	*,
47	source_roots: list[str] \| None = None,
48	extra_args: list[str] \| None = None,
49	) -> tuple[LintResult, ...]:
50	rule_runner.set_options(	1✔
51	[
52	f"--source-root-patterns={repr(source_roots)}",
53	"--backend-packages=pants.backend.codegen.protobuf.lint.buf",
54	*(extra_args or ()),
55	],
56	env_inherit={"PATH"},
57	)
58	partitions = rule_runner.request(	1✔
59	Partitions[BufFieldSet, Any],
60	[BufLintRequest.PartitionRequest(tuple(BufFieldSet.create(tgt) for tgt in targets))],
61	)
62	results = []	1✔
63	for partition in partitions:	1✔
64	result = rule_runner.request(	1✔
65	LintResult,
66	[BufLintRequest.Batch("", partition.elements, partition.metadata)],
67	)
68	results.append(result)	1✔
69	return tuple(results)	1✔
70
71
72	def assert_success(	1✔
73	rule_runner: RuleRunner,
74	target: Target,
75	*,
76	source_roots: list[str] \| None = None,
77	extra_args: list[str] \| None = None,
78	) -> None:
UNCOV 79	result = run_buf(rule_runner, [target], source_roots=source_roots, extra_args=extra_args)	×
UNCOV 80	assert len(result) == 1	×
UNCOV 81	assert result[0].exit_code == 0	×
UNCOV 82	assert not result[0].stdout	×
UNCOV 83	assert not result[0].stderr	×
84
85
86	def test_passing(rule_runner: RuleRunner) -> None:	1✔
UNCOV 87	rule_runner.write_files(	×
88	{"foo/v1/f.proto": GOOD_FILE, "foo/v1/BUILD": "protobuf_sources(name='t')"}
89	)
UNCOV 90	tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))	×
UNCOV 91	assert_success(rule_runner, tgt)	×
92
93
94	@pytest.mark.platform_specific_behavior	1✔
95	def test_failing(rule_runner: RuleRunner) -> None:	1✔
96	rule_runner.write_files(	1✔
97	{"foo/v1/f.proto": BAD_FILE, "foo/v1/BUILD": "protobuf_sources(name='t')"}
98	)
99	tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))	1✔
100	result = run_buf(rule_runner, [tgt])	1✔
101	assert len(result) == 1	1✔
102	assert result[0].exit_code == 100	1✔
103	assert "foo/v1/f.proto:" in result[0].stdout	1✔
104
105
106	def test_multiple_targets(rule_runner: RuleRunner) -> None:	1✔
UNCOV 107	rule_runner.write_files(	×
108	{
109	"foo/v1/good.proto": GOOD_FILE,
110	"foo/v1/bad.proto": BAD_FILE,
111	"foo/v1/BUILD": "protobuf_sources(name='t')",
112	}
113	)
UNCOV 114	tgts = [	×
115	rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="good.proto")),
116	rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="bad.proto")),
117	]
UNCOV 118	result = run_buf(rule_runner, tgts)	×
UNCOV 119	assert len(result) == 1	×
UNCOV 120	assert result[0].exit_code == 100	×
UNCOV 121	assert "good.proto" not in result[0].stdout	×
UNCOV 122	assert "foo/v1/bad.proto:" in result[0].stdout	×
123
124
125	def test_passthrough_args(rule_runner: RuleRunner) -> None:	1✔
UNCOV 126	rule_runner.write_files(	×
127	{"foo/v1/f.proto": BAD_FILE, "foo/v1/BUILD": "protobuf_sources(name='t')"}
128	)
UNCOV 129	tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))	×
UNCOV 130	config = json.dumps(	×
131	{
132	"version": "v1",
133	"lint": {
134	"ignore_only": {
135	"FIELD_LOWER_SNAKE_CASE": [
136	"foo/v1/f.proto",
137	],
138	},
139	},
140	}
141	)
142
UNCOV 143	assert_success(	×
144	rule_runner,
145	tgt,
146	extra_args=[f"--buf-lint-args='--config={config}'"],
147	)
148
149
150	def test_skip(rule_runner: RuleRunner) -> None:	1✔
UNCOV 151	rule_runner.write_files(	×
152	{"foo/v1/f.proto": BAD_FILE, "foo/v1/BUILD": "protobuf_sources(name='t')"}
153	)
UNCOV 154	tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))	×
UNCOV 155	result = run_buf(rule_runner, [tgt], extra_args=["--buf-lint-skip"])	×
UNCOV 156	assert not result	×
157
158
159	def test_dependencies(rule_runner: RuleRunner) -> None:	1✔
UNCOV 160	rule_runner.write_files(	×
161	{
162	"src/protobuf/dir1/v1/f.proto": dedent(
163	"""\
164	syntax = "proto3";
165	package dir1.v1;
166	message Person {
167	string name = 1;
168	int32 id = 2;
169	string email = 3;
170	}
171	"""
172	),
173	"src/protobuf/dir1/v1/BUILD": "protobuf_sources()",
174	"src/protobuf/dir2/v1/f.proto": dedent(
175	"""\
176	syntax = "proto3";
177	package dir2.v1;
178	import "dir1/v1/f.proto";
179
180	message Person {
181	dir1.v1.Person person = 1;
182	}
183	"""
184	),
185	"src/protobuf/dir2/v1/BUILD": (
186	"protobuf_sources(dependencies=['src/protobuf/dir1/v1'])"
187	),
188	"tests/protobuf/test_protos/v1/f.proto": dedent(
189	"""\
190	syntax = "proto3";
191	package test_protos.v1;
192	import "dir2/v1/f.proto";
193
194	message Person {
195	dir2.v1.Person person = 1;
196	}
197	"""
198	),
199	"tests/protobuf/test_protos/v1/BUILD": (
200	"protobuf_sources(dependencies=['src/protobuf/dir2/v1'])"
201	),
202	}
203	)
204
UNCOV 205	tgt = rule_runner.get_target(	×
206	Address("tests/protobuf/test_protos/v1", relative_file_path="f.proto")
207	)
UNCOV 208	assert_success(	×
209	rule_runner, tgt, source_roots=["src/python", "/src/protobuf", "/tests/protobuf"]
210	)
211
212
213	def test_config_discovery(rule_runner: RuleRunner) -> None:	1✔
UNCOV 214	rule_runner.write_files(	×
215	{
216	"foo/v1/f.proto": BAD_FILE,
217	"foo/v1/BUILD": "protobuf_sources(name='t')",
218	"buf.yaml": dedent(
219	"""\
220	version: v1
221	lint:
222	ignore_only:
223	FIELD_LOWER_SNAKE_CASE:
224	- foo/v1/f.proto
225	"""
226	),
227	}
228	)
229
UNCOV 230	tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))	×
231
UNCOV 232	assert_success(	×
233	rule_runner,
234	tgt,
235	)
236
237
238	def test_config_file_submitted(rule_runner: RuleRunner) -> None:	1✔
UNCOV 239	rule_runner.write_files(	×
240	{
241	"foo/v1/f.proto": BAD_FILE,
242	"foo/v1/BUILD": "protobuf_sources(name='t')",
243	# Intentionally placed somewhere config_discovery can't see.
244	"foo/buf.yaml": dedent(
245	"""\
246	version: v1
247	lint:
248	ignore_only:
249	FIELD_LOWER_SNAKE_CASE:
250	- foo/v1/f.proto
251	""",
252	),
253	}
254	)
255
UNCOV 256	tgt = rule_runner.get_target(Address("foo/v1", target_name="t", relative_file_path="f.proto"))	×
257
UNCOV 258	assert_success(	×
259	rule_runner,
260	tgt,
261	extra_args=["--buf-config=foo/buf.yaml"],
262	)

pantsbuild / pants / 20332790708

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