26260209689

Committed 21 May 2026 11:59PM UTC coverage: 75.453% (-15.7%) from 91.156%

Build # 26260209689

Build Type

Pull #23365

github

Committed by

web-flow

Commit Message

Merge 5fe873b58 into 7ea655ba0

Pull Request Pull Request #23365: uv.lock -> pex optimization

Coverage Stats

5 of 16 new or added lines in 1 file covered. (31.25%)

10118 existing lines in 378 files now uncovered.

54669 of 72454 relevant lines covered (75.45%)

2.31 hits per line

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

45.75

/src/python/pants/backend/python/lint/ruff/rules_integration_test.py

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

from __future__ import annotations

from pathlib import Path

import pytest

from pants.backend.python import target_types_rules
from pants.backend.python.lint.ruff import skip_field
from pants.backend.python.lint.ruff.check import skip_field as ruff_check_skip_field
from pants.backend.python.lint.ruff.check.rules import (
    RuffCheckFieldSet,
    RuffFixRequest,
    RuffLintRequest,
)
from pants.backend.python.lint.ruff.check.rules import rules as ruff_check_rules
from pants.backend.python.lint.ruff.check.skip_field import SkipRuffCheckField
from pants.backend.python.lint.ruff.format import skip_field as ruff_format_skip_field
from pants.backend.python.lint.ruff.format.rules import RuffFormatFieldSet, RuffFormatRequest
from pants.backend.python.lint.ruff.format.rules import rules as ruff_fmt_rules
from pants.backend.python.lint.ruff.format.skip_field import SkipRuffFormatField
from pants.backend.python.lint.ruff.skip_field import SkipRuffField
from pants.backend.python.lint.ruff.subsystem import rules as ruff_subsystem_rules
from pants.backend.python.target_types import PythonSourcesGeneratorTarget
from pants.core.goals.fix import FixResult
from pants.core.goals.fmt import FmtResult
from pants.core.goals.lint import LintResult
from pants.core.util_rules import config_files
from pants.core.util_rules.partitions import Partitions, _EmptyMetadata
from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest
from pants.engine.addresses import Address
from pants.engine.fs import DigestContents
from pants.engine.target import Target
from pants.testutil.python_interpreter_selection import all_major_minor_python_versions
from pants.testutil.rule_runner import QueryRule, RuleRunner

GOOD_FILE = 'a = "string without any placeholders"\n'
BAD_FILE = 'a = f"string without any placeholders"\n'
UNFORMATTED_FILE = 'a ="string without any placeholders"\n'
IMPORTS_SHADOWED_BY_LOCAL_PACKAGE = (
    "from opentelemetry import trace\n"
    "\n"
    "from pants.backend.observability.opentelemetry.config import Config\n"
)
IMPORTS_FIXED_FOR_LOCAL_PACKAGE = (
    "from opentelemetry import trace\n"
    "from pants.backend.observability.opentelemetry.config import Config\n"
)


@pytest.fixture
def rule_runner() -> RuleRunner:
    return RuleRunner(
        rules=[
            *ruff_check_rules(),
            *ruff_fmt_rules(),
            *skip_field.rules(),
            *ruff_check_skip_field.rules(),
            *ruff_format_skip_field.rules(),
            *ruff_subsystem_rules(),
            *config_files.rules(),
            *target_types_rules.rules(),
            QueryRule(FixResult, [RuffFixRequest.Batch]),
            QueryRule(LintResult, [RuffLintRequest.Batch]),
            QueryRule(FmtResult, [RuffFormatRequest.Batch]),
            QueryRule(Partitions, [RuffFixRequest.PartitionRequest]),
            QueryRule(SourceFiles, (SourceFilesRequest,)),
        ],
        target_types=[PythonSourcesGeneratorTarget],
    )


def run_ruff(
    rule_runner: RuleRunner,
    targets: list[Target],
    *,
    extra_args: list[str] | None = None,
) -> tuple[FixResult, LintResult, FmtResult]:
    args = ["--backend-packages=pants.backend.python.lint.ruff", *(extra_args or ())]
    rule_runner.set_options(args, env_inherit={"PATH", "PYENV_ROOT", "HOME"})

    check_field_sets = [
        RuffCheckFieldSet.create(tgt) for tgt in targets if RuffCheckFieldSet.is_applicable(tgt)
    ]
    check_source_reqs = [SourceFilesRequest(field_set.source for field_set in check_field_sets)]
    check_input_sources = rule_runner.request(SourceFiles, check_source_reqs)

    format_field_sets = [
        RuffFormatFieldSet.create(tgt) for tgt in targets if RuffFormatFieldSet.is_applicable(tgt)
    ]
    format_source_reqs = [SourceFilesRequest(field_set.source for field_set in format_field_sets)]
    format_input_sources = rule_runner.request(SourceFiles, format_source_reqs)

    fix_result = rule_runner.request(
        FixResult,
        [
            RuffFixRequest.Batch(
                "",
                check_input_sources.snapshot.files,
                partition_metadata=_EmptyMetadata(),
                snapshot=check_input_sources.snapshot,
            ),
        ],
    )
    lint_result = rule_runner.request(
        LintResult,
        [
            RuffLintRequest.Batch(
                "",
                tuple(check_field_sets),
                partition_metadata=_EmptyMetadata(),
            ),
        ],
    )
    fmt_result = rule_runner.request(
        FmtResult,
        [
            RuffFormatRequest.Batch(
                "",
                format_input_sources.snapshot.files,
                partition_metadata=_EmptyMetadata(),
                snapshot=format_input_sources.snapshot,
            )
        ],
    )

    return fix_result, lint_result, fmt_result


@pytest.mark.platform_specific_behavior
@pytest.mark.parametrize(
    "major_minor_interpreter",
    all_major_minor_python_versions(["CPython>=3.9,<3.15"]),
)
def test_passing(rule_runner: RuleRunner, major_minor_interpreter: str) -> None:
    rule_runner.write_files({"f.py": GOOD_FILE, "BUILD": "python_sources(name='t')"})
    tgt = rule_runner.get_target(Address("", target_name="t", relative_file_path="f.py"))
    fix_result, lint_result, fmt_result = run_ruff(
        rule_runner,
        [tgt],
        extra_args=[f"--python-interpreter-constraints=['=={major_minor_interpreter}.*']"],
    )
    assert lint_result.exit_code == 0
    assert fix_result.stderr == ""
    assert fix_result.stdout == "All checks passed!\n"
    assert not fix_result.did_change
    assert fix_result.output == rule_runner.make_snapshot({"f.py": GOOD_FILE})
    assert not fmt_result.did_change
    assert fmt_result.output == rule_runner.make_snapshot({"f.py": GOOD_FILE})


def test_failing(rule_runner: RuleRunner) -> None:
    rule_runner.write_files({"f.py": BAD_FILE, "BUILD": "python_sources(name='t')"})
    tgt = rule_runner.get_target(Address("", target_name="t", relative_file_path="f.py"))
    fix_result, lint_result, fmt_result = run_ruff(rule_runner, [tgt])
    assert lint_result.exit_code == 1
    assert fix_result.stdout == "Found 1 error (1 fixed, 0 remaining).\n"
    assert fix_result.stderr == ""
    assert fix_result.did_change
    assert fix_result.output == rule_runner.make_snapshot({"f.py": GOOD_FILE})
    assert not fmt_result.did_change
    assert fmt_result.output == rule_runner.make_snapshot({"f.py": BAD_FILE})


def test_multiple_targets(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "good.py": GOOD_FILE,
            "bad.py": BAD_FILE,
            "unformatted.py": UNFORMATTED_FILE,
            "BUILD": "python_sources(name='t')",
        }
    )
    tgts = [
        rule_runner.get_target(Address("", target_name="t", relative_file_path="good.py")),
        rule_runner.get_target(Address("", target_name="t", relative_file_path="bad.py")),
        rule_runner.get_target(Address("", target_name="t", relative_file_path="unformatted.py")),
    ]
    fix_result, lint_result, fmt_result = run_ruff(rule_runner, tgts)
    assert lint_result.exit_code == 1
    assert fix_result.output == rule_runner.make_snapshot(
        {"good.py": GOOD_FILE, "bad.py": GOOD_FILE, "unformatted.py": UNFORMATTED_FILE}
    )
    assert fix_result.did_change is True
    assert fmt_result.output == rule_runner.make_snapshot(
        {"good.py": GOOD_FILE, "bad.py": BAD_FILE, "unformatted.py": GOOD_FILE}
    )
    assert fmt_result.did_change is True


def test_fix_preserves_init_py_package_context_for_import_sorting(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "pyproject.toml": "\n".join(
                [
                    "[tool.ruff.lint]",
                    'select = ["I"]',
                    "",
                    "[tool.ruff.lint.isort]",
                    'known-first-party = ["pants"]',
                    "",
                ]
            ),
            "src/python/pants/backend/observability/opentelemetry/BUILD": "python_sources()",
            "src/python/pants/backend/observability/opentelemetry/__init__.py": "",
            "src/python/pants/backend/observability/opentelemetry/processor.py": (
                IMPORTS_SHADOWED_BY_LOCAL_PACKAGE
            ),
        }
    )
    rule_runner.set_options(
        ["--backend-packages=pants.backend.python.lint.ruff"],
        env_inherit={"PATH", "PYENV_ROOT", "HOME"},
    )

    init_tgt = rule_runner.get_target(
        Address(
            "src/python/pants/backend/observability/opentelemetry",
            relative_file_path="__init__.py",
        )
    )
    processor_tgt = rule_runner.get_target(
        Address(
            "src/python/pants/backend/observability/opentelemetry",
            relative_file_path="processor.py",
        )
    )
    init_field_set = RuffCheckFieldSet.create(init_tgt)
    processor_field_set = RuffCheckFieldSet.create(processor_tgt)

    lint_result = rule_runner.request(
        LintResult,
        [
            RuffLintRequest.Batch(
                "",
                (init_field_set, processor_field_set),
                partition_metadata=_EmptyMetadata(),
            ),
        ],
    )
    assert lint_result.exit_code == 1
    assert "I001" in lint_result.stdout

    partitions = rule_runner.request(
        Partitions,
        [RuffFixRequest.PartitionRequest((init_field_set, processor_field_set))],
    )
    assert len(partitions) == 1
    partition = partitions[0]
    assert partition.elements == (
        "src/python/pants/backend/observability/opentelemetry/__init__.py",
        "src/python/pants/backend/observability/opentelemetry/processor.py",
    )
    assert partition.metadata.init_files == (
        "src/python/pants/backend/observability/opentelemetry/__init__.py",
    )

    fix_input_sources = rule_runner.request(
        SourceFiles, [SourceFilesRequest([processor_field_set.source])]
    )
    fix_result = rule_runner.request(
        FixResult,
        [
            RuffFixRequest.Batch(
                "",
                ("src/python/pants/backend/observability/opentelemetry/processor.py",),
                partition_metadata=partition.metadata,
                snapshot=fix_input_sources.snapshot,
            ),
        ],
    )

    assert fix_result.stdout == "Found 1 error (1 fixed, 0 remaining).\n"
    assert fix_result.output == rule_runner.make_snapshot(
        {
            "src/python/pants/backend/observability/opentelemetry/processor.py": (
                IMPORTS_FIXED_FOR_LOCAL_PACKAGE
            )
        }
    )


def test_skip_field(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "good.py": GOOD_FILE,
            "bad.py": BAD_FILE,
            "unformatted.py": UNFORMATTED_FILE,
            "BUILD": "python_sources(name='t', skip_ruff=True)",
        }
    )
    tgts = [
        rule_runner.get_target(Address("", target_name="t", relative_file_path="good.py")),
        rule_runner.get_target(Address("", target_name="t", relative_file_path="bad.py")),
        rule_runner.get_target(Address("", target_name="t", relative_file_path="unformatted.py")),
    ]
    for tgt in tgts:
        assert tgt.get(SkipRuffField).value is True

    fix_result, lint_result, fmt_result = run_ruff(rule_runner, tgts)

    assert lint_result.exit_code == 0
    assert fix_result.output == rule_runner.make_snapshot({})
    assert fix_result.did_change is False
    assert fmt_result.output == rule_runner.make_snapshot({})
    assert fmt_result.did_change is False


def test_skip_check_field(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "good.py": GOOD_FILE,
            "bad.py": BAD_FILE,
            "unformatted.py": UNFORMATTED_FILE,
            "BUILD": "python_sources(name='t', skip_ruff_check=True)",
        }
    )
    tgts = [
        rule_runner.get_target(Address("", target_name="t", relative_file_path="good.py")),
        rule_runner.get_target(Address("", target_name="t", relative_file_path="bad.py")),
        rule_runner.get_target(Address("", target_name="t", relative_file_path="unformatted.py")),
    ]
    for tgt in tgts:
        assert tgt.get(SkipRuffCheckField).value is True

    fix_result, lint_result, fmt_result = run_ruff(rule_runner, tgts)

    assert lint_result.exit_code == 0
    assert fix_result.output == rule_runner.make_snapshot({})
    assert fix_result.did_change is False
    assert fmt_result.output == rule_runner.make_snapshot(
        {"good.py": GOOD_FILE, "bad.py": BAD_FILE, "unformatted.py": GOOD_FILE}
    )
    assert fmt_result.did_change is True


def test_skip_format_field(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "good.py": GOOD_FILE,
            "bad.py": BAD_FILE,
            "unformatted.py": UNFORMATTED_FILE,
            "BUILD": "python_sources(name='t', skip_ruff_format=True)",
        }
    )
    tgts = [
        rule_runner.get_target(Address("", target_name="t", relative_file_path="good.py")),
        rule_runner.get_target(Address("", target_name="t", relative_file_path="bad.py")),
        rule_runner.get_target(Address("", target_name="t", relative_file_path="unformatted.py")),
    ]
    for tgt in tgts:
        assert tgt.get(SkipRuffFormatField).value is True

    fix_result, lint_result, fmt_result = run_ruff(rule_runner, tgts)

    assert lint_result.exit_code == 1
    assert fix_result.output == rule_runner.make_snapshot(
        {"good.py": GOOD_FILE, "bad.py": GOOD_FILE, "unformatted.py": UNFORMATTED_FILE}
    )
    assert fix_result.did_change is True
    assert fmt_result.output == rule_runner.make_snapshot({})
    assert fmt_result.did_change is False


@pytest.mark.parametrize(
    "file_path,config_path,extra_args,should_change",
    (
        [Path("f.py"), Path("pyproject.toml"), [], False],
        [Path("f.py"), Path("ruff.toml"), [], False],
        [Path("custom/f.py"), Path("custom/ruff.toml"), [], False],
        [Path("custom/f.py"), Path("custom/pyproject.toml"), [], False],
        [Path("f.py"), Path("custom/ruff.toml"), ["--ruff-config=custom/ruff.toml"], False],
        [Path("f.py"), Path("custom/ruff.toml"), [], True],
    ),
)
def test_config_file(
    rule_runner: RuleRunner,
    file_path: Path,
    config_path: Path,
    extra_args: list[str],
    should_change: bool,
) -> None:
    hierarchy = "[tool.ruff]\n" if config_path.stem == "pyproject" else ""
    rule_runner.write_files(
        {
            file_path: BAD_FILE,
            file_path.parent / "BUILD": "python_sources()",
            config_path: f'{hierarchy}ignore = ["F541"]',
        }
    )
    spec_path = str(file_path.parent).replace(".", "")
    rel_file_path = file_path.relative_to(*file_path.parts[:1]) if spec_path else file_path
    addr = Address(spec_path, relative_file_path=str(rel_file_path))
    tgt = rule_runner.get_target(addr)
    fix_result, lint_result, fmt_result = run_ruff(rule_runner, [tgt], extra_args=extra_args)
    assert lint_result.exit_code == bool(should_change)
    assert fix_result.did_change is should_change


def test_report_file(rule_runner: RuleRunner) -> None:
    rule_runner.write_files({"f.py": BAD_FILE, "BUILD": "python_sources(name='t')"})
    tgt = rule_runner.get_target(Address("", target_name="t", relative_file_path="f.py"))
    fix_result, lint_result, fmt_result = run_ruff(
        rule_runner,
        [tgt],
        extra_args=["--ruff-args='--output-file=reports/foo.txt'"],
    )
    assert lint_result.exit_code == 1
    report_files = rule_runner.request(DigestContents, [lint_result.report])
    assert len(report_files) == 1
    assert "f.py:1:5" in report_files[0].content.decode()
    assert "F541" in report_files[0].content.decode()

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	2✔
5
6	from pathlib import Path	2✔
7
8	import pytest	2✔
9
10	from pants.backend.python import target_types_rules	2✔
11	from pants.backend.python.lint.ruff import skip_field	2✔
12	from pants.backend.python.lint.ruff.check import skip_field as ruff_check_skip_field	2✔
13	from pants.backend.python.lint.ruff.check.rules import (	2✔
14	RuffCheckFieldSet,
15	RuffFixRequest,
16	RuffLintRequest,
17	)
18	from pants.backend.python.lint.ruff.check.rules import rules as ruff_check_rules	2✔
19	from pants.backend.python.lint.ruff.check.skip_field import SkipRuffCheckField	2✔
20	from pants.backend.python.lint.ruff.format import skip_field as ruff_format_skip_field	2✔
21	from pants.backend.python.lint.ruff.format.rules import RuffFormatFieldSet, RuffFormatRequest	2✔
22	from pants.backend.python.lint.ruff.format.rules import rules as ruff_fmt_rules	2✔
23	from pants.backend.python.lint.ruff.format.skip_field import SkipRuffFormatField	2✔
24	from pants.backend.python.lint.ruff.skip_field import SkipRuffField	2✔
25	from pants.backend.python.lint.ruff.subsystem import rules as ruff_subsystem_rules	2✔
26	from pants.backend.python.target_types import PythonSourcesGeneratorTarget	2✔
27	from pants.core.goals.fix import FixResult	2✔
28	from pants.core.goals.fmt import FmtResult	2✔
29	from pants.core.goals.lint import LintResult	2✔
30	from pants.core.util_rules import config_files	2✔
31	from pants.core.util_rules.partitions import Partitions, _EmptyMetadata	2✔
32	from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest	2✔
33	from pants.engine.addresses import Address	2✔
34	from pants.engine.fs import DigestContents	2✔
35	from pants.engine.target import Target	2✔
36	from pants.testutil.python_interpreter_selection import all_major_minor_python_versions	2✔
37	from pants.testutil.rule_runner import QueryRule, RuleRunner	2✔
38
39	GOOD_FILE = 'a = "string without any placeholders"\n'	2✔
40	BAD_FILE = 'a = f"string without any placeholders"\n'	2✔
41	UNFORMATTED_FILE = 'a ="string without any placeholders"\n'	2✔
42	IMPORTS_SHADOWED_BY_LOCAL_PACKAGE = (	2✔
43	"from opentelemetry import trace\n"
44	"\n"
45	"from pants.backend.observability.opentelemetry.config import Config\n"
46	)
47	IMPORTS_FIXED_FOR_LOCAL_PACKAGE = (	2✔
48	"from opentelemetry import trace\n"
49	"from pants.backend.observability.opentelemetry.config import Config\n"
50	)
51
52
53	@pytest.fixture	2✔
54	def rule_runner() -> RuleRunner:	2✔
55	return RuleRunner(	2✔
56	rules=[
57	*ruff_check_rules(),
58	*ruff_fmt_rules(),
59	*skip_field.rules(),
60	*ruff_check_skip_field.rules(),
61	*ruff_format_skip_field.rules(),
62	*ruff_subsystem_rules(),
63	*config_files.rules(),
64	*target_types_rules.rules(),
65	QueryRule(FixResult, [RuffFixRequest.Batch]),
66	QueryRule(LintResult, [RuffLintRequest.Batch]),
67	QueryRule(FmtResult, [RuffFormatRequest.Batch]),
68	QueryRule(Partitions, [RuffFixRequest.PartitionRequest]),
69	QueryRule(SourceFiles, (SourceFilesRequest,)),
70	],
71	target_types=[PythonSourcesGeneratorTarget],
72	)
73
74
75	def run_ruff(	2✔
76	rule_runner: RuleRunner,
77	targets: list[Target],
78	*,
79	extra_args: list[str] \| None = None,
80	) -> tuple[FixResult, LintResult, FmtResult]:
81	args = ["--backend-packages=pants.backend.python.lint.ruff", *(extra_args or ())]	2✔
82	rule_runner.set_options(args, env_inherit={"PATH", "PYENV_ROOT", "HOME"})	2✔
83
84	check_field_sets = [	2✔
85	RuffCheckFieldSet.create(tgt) for tgt in targets if RuffCheckFieldSet.is_applicable(tgt)
86	]
87	check_source_reqs = [SourceFilesRequest(field_set.source for field_set in check_field_sets)]	2✔
88	check_input_sources = rule_runner.request(SourceFiles, check_source_reqs)	2✔
89
90	format_field_sets = [	2✔
91	RuffFormatFieldSet.create(tgt) for tgt in targets if RuffFormatFieldSet.is_applicable(tgt)
92	]
93	format_source_reqs = [SourceFilesRequest(field_set.source for field_set in format_field_sets)]	2✔
94	format_input_sources = rule_runner.request(SourceFiles, format_source_reqs)	2✔
95
96	fix_result = rule_runner.request(	2✔
97	FixResult,
98	[
99	RuffFixRequest.Batch(
100	"",
101	check_input_sources.snapshot.files,
102	partition_metadata=_EmptyMetadata(),
103	snapshot=check_input_sources.snapshot,
104	),
105	],
106	)
107	lint_result = rule_runner.request(	2✔
108	LintResult,
109	[
110	RuffLintRequest.Batch(
111	"",
112	tuple(check_field_sets),
113	partition_metadata=_EmptyMetadata(),
114	),
115	],
116	)
117	fmt_result = rule_runner.request(	2✔
118	FmtResult,
119	[
120	RuffFormatRequest.Batch(
121	"",
122	format_input_sources.snapshot.files,
123	partition_metadata=_EmptyMetadata(),
124	snapshot=format_input_sources.snapshot,
125	)
126	],
127	)
128
129	return fix_result, lint_result, fmt_result	2✔
130
131
132	@pytest.mark.platform_specific_behavior	2✔
133	@pytest.mark.parametrize(	2✔
134	"major_minor_interpreter",
135	all_major_minor_python_versions(["CPython>=3.9,<3.15"]),
136	)
137	def test_passing(rule_runner: RuleRunner, major_minor_interpreter: str) -> None:	2✔
138	rule_runner.write_files({"f.py": GOOD_FILE, "BUILD": "python_sources(name='t')"})	2✔
139	tgt = rule_runner.get_target(Address("", target_name="t", relative_file_path="f.py"))	2✔
140	fix_result, lint_result, fmt_result = run_ruff(	2✔
141	rule_runner,
142	[tgt],
143	extra_args=[f"--python-interpreter-constraints=['=={major_minor_interpreter}.*']"],
144	)
145	assert lint_result.exit_code == 0	2✔
146	assert fix_result.stderr == ""	2✔
147	assert fix_result.stdout == "All checks passed!\n"	2✔
148	assert not fix_result.did_change	2✔
149	assert fix_result.output == rule_runner.make_snapshot({"f.py": GOOD_FILE})	2✔
150	assert not fmt_result.did_change	2✔
151	assert fmt_result.output == rule_runner.make_snapshot({"f.py": GOOD_FILE})	2✔
152
153
154	def test_failing(rule_runner: RuleRunner) -> None:	2✔
UNCOV 155	rule_runner.write_files({"f.py": BAD_FILE, "BUILD": "python_sources(name='t')"})	×
UNCOV 156	tgt = rule_runner.get_target(Address("", target_name="t", relative_file_path="f.py"))	×
UNCOV 157	fix_result, lint_result, fmt_result = run_ruff(rule_runner, [tgt])	×
UNCOV 158	assert lint_result.exit_code == 1	×
UNCOV 159	assert fix_result.stdout == "Found 1 error (1 fixed, 0 remaining).\n"	×
UNCOV 160	assert fix_result.stderr == ""	×
UNCOV 161	assert fix_result.did_change	×
UNCOV 162	assert fix_result.output == rule_runner.make_snapshot({"f.py": GOOD_FILE})	×
UNCOV 163	assert not fmt_result.did_change	×
UNCOV 164	assert fmt_result.output == rule_runner.make_snapshot({"f.py": BAD_FILE})	×
165
166
167	def test_multiple_targets(rule_runner: RuleRunner) -> None:	2✔
UNCOV 168	rule_runner.write_files(	×
169	{
170	"good.py": GOOD_FILE,
171	"bad.py": BAD_FILE,
172	"unformatted.py": UNFORMATTED_FILE,
173	"BUILD": "python_sources(name='t')",
174	}
175	)
UNCOV 176	tgts = [	×
177	rule_runner.get_target(Address("", target_name="t", relative_file_path="good.py")),
178	rule_runner.get_target(Address("", target_name="t", relative_file_path="bad.py")),
179	rule_runner.get_target(Address("", target_name="t", relative_file_path="unformatted.py")),
180	]
UNCOV 181	fix_result, lint_result, fmt_result = run_ruff(rule_runner, tgts)	×
UNCOV 182	assert lint_result.exit_code == 1	×
UNCOV 183	assert fix_result.output == rule_runner.make_snapshot(	×
184	{"good.py": GOOD_FILE, "bad.py": GOOD_FILE, "unformatted.py": UNFORMATTED_FILE}
185	)
UNCOV 186	assert fix_result.did_change is True	×
UNCOV 187	assert fmt_result.output == rule_runner.make_snapshot(	×
188	{"good.py": GOOD_FILE, "bad.py": BAD_FILE, "unformatted.py": GOOD_FILE}
189	)
UNCOV 190	assert fmt_result.did_change is True	×
191
192
193	def test_fix_preserves_init_py_package_context_for_import_sorting(rule_runner: RuleRunner) -> None:	2✔
UNCOV 194	rule_runner.write_files(	×
195	{
196	"pyproject.toml": "\n".join(
197	[
198	"[tool.ruff.lint]",
199	'select = ["I"]',
200	"",
201	"[tool.ruff.lint.isort]",
202	'known-first-party = ["pants"]',
203	"",
204	]
205	),
206	"src/python/pants/backend/observability/opentelemetry/BUILD": "python_sources()",
207	"src/python/pants/backend/observability/opentelemetry/__init__.py": "",
208	"src/python/pants/backend/observability/opentelemetry/processor.py": (
209	IMPORTS_SHADOWED_BY_LOCAL_PACKAGE
210	),
211	}
212	)
UNCOV 213	rule_runner.set_options(	×
214	["--backend-packages=pants.backend.python.lint.ruff"],
215	env_inherit={"PATH", "PYENV_ROOT", "HOME"},
216	)
217
UNCOV 218	init_tgt = rule_runner.get_target(	×
219	Address(
220	"src/python/pants/backend/observability/opentelemetry",
221	relative_file_path="__init__.py",
222	)
223	)
UNCOV 224	processor_tgt = rule_runner.get_target(	×
225	Address(
226	"src/python/pants/backend/observability/opentelemetry",
227	relative_file_path="processor.py",
228	)
229	)
UNCOV 230	init_field_set = RuffCheckFieldSet.create(init_tgt)	×
UNCOV 231	processor_field_set = RuffCheckFieldSet.create(processor_tgt)	×
232
UNCOV 233	lint_result = rule_runner.request(	×
234	LintResult,
235	[
236	RuffLintRequest.Batch(
237	"",
238	(init_field_set, processor_field_set),
239	partition_metadata=_EmptyMetadata(),
240	),
241	],
242	)
UNCOV 243	assert lint_result.exit_code == 1	×
UNCOV 244	assert "I001" in lint_result.stdout	×
245
UNCOV 246	partitions = rule_runner.request(	×
247	Partitions,
248	[RuffFixRequest.PartitionRequest((init_field_set, processor_field_set))],
249	)
UNCOV 250	assert len(partitions) == 1	×
UNCOV 251	partition = partitions[0]	×
UNCOV 252	assert partition.elements == (	×
253	"src/python/pants/backend/observability/opentelemetry/__init__.py",
254	"src/python/pants/backend/observability/opentelemetry/processor.py",
255	)
UNCOV 256	assert partition.metadata.init_files == (	×
257	"src/python/pants/backend/observability/opentelemetry/__init__.py",
258	)
259
UNCOV 260	fix_input_sources = rule_runner.request(	×
261	SourceFiles, [SourceFilesRequest([processor_field_set.source])]
262	)
UNCOV 263	fix_result = rule_runner.request(	×
264	FixResult,
265	[
266	RuffFixRequest.Batch(
267	"",
268	("src/python/pants/backend/observability/opentelemetry/processor.py",),
269	partition_metadata=partition.metadata,
270	snapshot=fix_input_sources.snapshot,
271	),
272	],
273	)
274
UNCOV 275	assert fix_result.stdout == "Found 1 error (1 fixed, 0 remaining).\n"	×
UNCOV 276	assert fix_result.output == rule_runner.make_snapshot(	×
277	{
278	"src/python/pants/backend/observability/opentelemetry/processor.py": (
279	IMPORTS_FIXED_FOR_LOCAL_PACKAGE
280	)
281	}
282	)
283
284
285	def test_skip_field(rule_runner: RuleRunner) -> None:	2✔
UNCOV 286	rule_runner.write_files(	×
287	{
288	"good.py": GOOD_FILE,
289	"bad.py": BAD_FILE,
290	"unformatted.py": UNFORMATTED_FILE,
291	"BUILD": "python_sources(name='t', skip_ruff=True)",
292	}
293	)
UNCOV 294	tgts = [	×
295	rule_runner.get_target(Address("", target_name="t", relative_file_path="good.py")),
296	rule_runner.get_target(Address("", target_name="t", relative_file_path="bad.py")),
297	rule_runner.get_target(Address("", target_name="t", relative_file_path="unformatted.py")),
298	]
UNCOV 299	for tgt in tgts:	×
UNCOV 300	assert tgt.get(SkipRuffField).value is True	×
301
UNCOV 302	fix_result, lint_result, fmt_result = run_ruff(rule_runner, tgts)	×
303
UNCOV 304	assert lint_result.exit_code == 0	×
UNCOV 305	assert fix_result.output == rule_runner.make_snapshot({})	×
UNCOV 306	assert fix_result.did_change is False	×
UNCOV 307	assert fmt_result.output == rule_runner.make_snapshot({})	×
UNCOV 308	assert fmt_result.did_change is False	×
309
310
311	def test_skip_check_field(rule_runner: RuleRunner) -> None:	2✔
UNCOV 312	rule_runner.write_files(	×
313	{
314	"good.py": GOOD_FILE,
315	"bad.py": BAD_FILE,
316	"unformatted.py": UNFORMATTED_FILE,
317	"BUILD": "python_sources(name='t', skip_ruff_check=True)",
318	}
319	)
UNCOV 320	tgts = [	×
321	rule_runner.get_target(Address("", target_name="t", relative_file_path="good.py")),
322	rule_runner.get_target(Address("", target_name="t", relative_file_path="bad.py")),
323	rule_runner.get_target(Address("", target_name="t", relative_file_path="unformatted.py")),
324	]
UNCOV 325	for tgt in tgts:	×
UNCOV 326	assert tgt.get(SkipRuffCheckField).value is True	×
327
UNCOV 328	fix_result, lint_result, fmt_result = run_ruff(rule_runner, tgts)	×
329
UNCOV 330	assert lint_result.exit_code == 0	×
UNCOV 331	assert fix_result.output == rule_runner.make_snapshot({})	×
UNCOV 332	assert fix_result.did_change is False	×
UNCOV 333	assert fmt_result.output == rule_runner.make_snapshot(	×
334	{"good.py": GOOD_FILE, "bad.py": BAD_FILE, "unformatted.py": GOOD_FILE}
335	)
UNCOV 336	assert fmt_result.did_change is True	×
337
338
339	def test_skip_format_field(rule_runner: RuleRunner) -> None:	2✔
UNCOV 340	rule_runner.write_files(	×
341	{
342	"good.py": GOOD_FILE,
343	"bad.py": BAD_FILE,
344	"unformatted.py": UNFORMATTED_FILE,
345	"BUILD": "python_sources(name='t', skip_ruff_format=True)",
346	}
347	)
UNCOV 348	tgts = [	×
349	rule_runner.get_target(Address("", target_name="t", relative_file_path="good.py")),
350	rule_runner.get_target(Address("", target_name="t", relative_file_path="bad.py")),
351	rule_runner.get_target(Address("", target_name="t", relative_file_path="unformatted.py")),
352	]
UNCOV 353	for tgt in tgts:	×
UNCOV 354	assert tgt.get(SkipRuffFormatField).value is True	×
355
UNCOV 356	fix_result, lint_result, fmt_result = run_ruff(rule_runner, tgts)	×
357
UNCOV 358	assert lint_result.exit_code == 1	×
UNCOV 359	assert fix_result.output == rule_runner.make_snapshot(	×
360	{"good.py": GOOD_FILE, "bad.py": GOOD_FILE, "unformatted.py": UNFORMATTED_FILE}
361	)
UNCOV 362	assert fix_result.did_change is True	×
UNCOV 363	assert fmt_result.output == rule_runner.make_snapshot({})	×
UNCOV 364	assert fmt_result.did_change is False	×
365
366
367	@pytest.mark.parametrize(	2✔
368	"file_path,config_path,extra_args,should_change",
369	(
370	[Path("f.py"), Path("pyproject.toml"), [], False],
371	[Path("f.py"), Path("ruff.toml"), [], False],
372	[Path("custom/f.py"), Path("custom/ruff.toml"), [], False],
373	[Path("custom/f.py"), Path("custom/pyproject.toml"), [], False],
374	[Path("f.py"), Path("custom/ruff.toml"), ["--ruff-config=custom/ruff.toml"], False],
375	[Path("f.py"), Path("custom/ruff.toml"), [], True],
376	),
377	)
378	def test_config_file(	2✔
379	rule_runner: RuleRunner,
380	file_path: Path,
381	config_path: Path,
382	extra_args: list[str],
383	should_change: bool,
384	) -> None:
UNCOV 385	hierarchy = "[tool.ruff]\n" if config_path.stem == "pyproject" else ""	×
UNCOV 386	rule_runner.write_files(	×
387	{
388	file_path: BAD_FILE,
389	file_path.parent / "BUILD": "python_sources()",
390	config_path: f'{hierarchy}ignore = ["F541"]',
391	}
392	)
UNCOV 393	spec_path = str(file_path.parent).replace(".", "")	×
UNCOV 394	rel_file_path = file_path.relative_to(*file_path.parts[:1]) if spec_path else file_path	×
UNCOV 395	addr = Address(spec_path, relative_file_path=str(rel_file_path))	×
UNCOV 396	tgt = rule_runner.get_target(addr)	×
UNCOV 397	fix_result, lint_result, fmt_result = run_ruff(rule_runner, [tgt], extra_args=extra_args)	×
UNCOV 398	assert lint_result.exit_code == bool(should_change)	×
UNCOV 399	assert fix_result.did_change is should_change	×
400
401
402	def test_report_file(rule_runner: RuleRunner) -> None:	2✔
UNCOV 403	rule_runner.write_files({"f.py": BAD_FILE, "BUILD": "python_sources(name='t')"})	×
UNCOV 404	tgt = rule_runner.get_target(Address("", target_name="t", relative_file_path="f.py"))	×
UNCOV 405	fix_result, lint_result, fmt_result = run_ruff(	×
406	rule_runner,
407	[tgt],
408	extra_args=["--ruff-args='--output-file=reports/foo.txt'"],
409	)
UNCOV 410	assert lint_result.exit_code == 1	×
UNCOV 411	report_files = rule_runner.request(DigestContents, [lint_result.report])	×
UNCOV 412	assert len(report_files) == 1	×
UNCOV 413	assert "f.py:1:5" in report_files[0].content.decode()	×
UNCOV 414	assert "F541" in report_files[0].content.decode()	×

pantsbuild / pants / 26260209689

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