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

56.84

/src/python/pants/backend/sql/lint/sqlfluff/rules_integration_test.py

# Copyright 2024 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

from pathlib import Path
from textwrap import dedent

import pytest

from pants.backend.sql.lint.sqlfluff import rules as sqlfluff_rules
from pants.backend.sql.lint.sqlfluff import skip_field
from pants.backend.sql.lint.sqlfluff import subsystem as sqlfluff_subsystem
from pants.backend.sql.lint.sqlfluff.rules import (
    SqlfluffFixRequest,
    SqlfluffFormatRequest,
    SqlfluffLintRequest,
)
from pants.backend.sql.lint.sqlfluff.skip_field import SkipSqlfluffField
from pants.backend.sql.lint.sqlfluff.subsystem import SqlfluffFieldSet
from pants.backend.sql.target_types import SqlSourcesGeneratorTarget
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 _EmptyMetadata
from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest
from pants.engine.addresses import Address
from pants.engine.target import Target
from pants.testutil.rule_runner import QueryRule, RuleRunner

GOOD_FILE = dedent(
    """\
    select
        e.id,
        e.name
    from employees as e
    """
)
BAD_FILE = dedent(
    """\
    select
        e.id,
        name
    from employees as e
    """
)
UNFORMATTED_FILE = dedent(
    """\
    select e.id, e.name
    from employees as e
    """
)
CONFIG_POSTGRES = dedent(
    """\
    [sqlfluff]
    dialect = postgres
    """
)


@pytest.fixture
def rule_runner() -> RuleRunner:
    return RuleRunner(
        rules=[
            *skip_field.rules(),
            *sqlfluff_subsystem.rules(),
            *config_files.rules(),
            *sqlfluff_rules.rules(),
            QueryRule(FixResult, [SqlfluffFixRequest.Batch]),
            QueryRule(LintResult, [SqlfluffLintRequest.Batch]),
            QueryRule(FmtResult, [SqlfluffFormatRequest.Batch]),
            QueryRule(SourceFiles, (SourceFilesRequest,)),
        ],
        target_types=[SqlSourcesGeneratorTarget],
    )


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

    field_sets = [
        SqlfluffFieldSet.create(tgt) for tgt in targets if SqlfluffFieldSet.is_applicable(tgt)
    ]
    source_reqs = [SourceFilesRequest(field_set.source for field_set in field_sets)]
    input_sources = rule_runner.request(SourceFiles, source_reqs)

    fix_result = rule_runner.request(
        FixResult,
        [
            SqlfluffFixRequest.Batch(
                "",
                tuple(field_sets),
                partition_metadata=_EmptyMetadata(),
                snapshot=input_sources.snapshot,
            ),
        ],
    )
    lint_result = rule_runner.request(
        LintResult,
        [
            SqlfluffLintRequest.Batch(
                "",
                tuple(field_sets),
                partition_metadata=_EmptyMetadata(),
            ),
        ],
    )
    fmt_result = rule_runner.request(
        FmtResult,
        [
            SqlfluffFormatRequest.Batch(
                "",
                tuple(field_sets),
                partition_metadata=_EmptyMetadata(),
                snapshot=input_sources.snapshot,
            )
        ],
    )

    return fix_result, lint_result, fmt_result


@pytest.mark.platform_specific_behavior
def test_passing(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "query.sql": GOOD_FILE,
            "BUILD": "sql_sources(name='t')",
            ".sqlfluff": CONFIG_POSTGRES,
        }
    )
    tgt = rule_runner.get_target(Address("", target_name="t", relative_file_path="query.sql"))
    fix_result, lint_result, fmt_result = run_sqlfluff(rule_runner, [tgt])
    assert fix_result.stdout == dedent(
        """\
        ==== finding fixable violations ====
        FORCE MODE: Attempting fixes...
        ==== no fixable linting violations found ====
        All Finished!
        """
    )
    assert fix_result.stderr == ""
    assert lint_result.exit_code == 0
    assert not fix_result.did_change
    assert fix_result.output == rule_runner.make_snapshot({"query.sql": GOOD_FILE})
    assert not fmt_result.did_change
    assert fmt_result.output == rule_runner.make_snapshot({"query.sql": GOOD_FILE})


def test_failing(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "query.sql": BAD_FILE,
            "BUILD": "sql_sources(name='t')",
            ".sqlfluff": CONFIG_POSTGRES,
        }
    )
    tgt = rule_runner.get_target(Address("", target_name="t", relative_file_path="query.sql"))
    fix_result, lint_result, fmt_result = run_sqlfluff(rule_runner, [tgt])
    assert fix_result.stdout == dedent(
        """\
        ==== finding fixable violations ====
        FORCE MODE: Attempting fixes...
        == [query.sql] FAIL
        L:   3 | P:   5 | RF03 | Unqualified reference 'name' found in single table
                               | select. [references.consistent]
        == [query.sql] FIXED
        1 fixable linting violations found
          [1 unfixable linting violations found]
        """
    )
    assert fix_result.stderr == ""
    assert lint_result.stdout == dedent(
        """\
         == [query.sql] FAIL
         L:   3 | P:   5 | RF03 | Unqualified reference 'name' found in single table
                                | select. [references.consistent]
         L:   3 | P:   5 | RF03 | Unqualified reference 'name' found in single table
                                | select which is inconsistent with previous references.
                                | [references.consistent]
         All Finished!
         """
    )
    assert lint_result.stderr == ""
    assert lint_result.exit_code == 1
    assert fix_result.did_change
    assert fix_result.output == rule_runner.make_snapshot({"query.sql": GOOD_FILE})
    assert not fmt_result.did_change
    assert fmt_result.output == rule_runner.make_snapshot({"query.sql": BAD_FILE})


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


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

    fix_result, lint_result, fmt_result = run_sqlfluff(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


@pytest.mark.parametrize(
    "file_path,config_path,extra_args,should_change",
    (
        [Path("query.sql"), Path("pyproject.toml"), [], False],
        [Path("query.sql"), Path(".sqlfluff"), [], False],
        [Path("custom/query.sql"), Path("custom/pyproject.toml"), [], False],
        [Path("custom/query.sql"), Path("custom/.sqlfluff"), [], False],
        [
            Path("query.sql"),
            Path("custom/config.sqlfluff"),
            ["--sqlfluff-config=custom/config.sqlfluff"],
            False,
        ],
        [
            Path("query.sql"),
            Path("custom/.sqlfluff"),
            ['--sqlfluff-args="--dialect=postgres"'],
            True,
        ],
    ),
)
def test_config_file(
    rule_runner: RuleRunner,
    file_path: Path,
    config_path: Path,
    extra_args: list[str],
    should_change: bool,
) -> None:
    if config_path.stem == "pyproject":
        config = dedent(
            """\
            [tool.sqlfluff.core]
            dialect = "postgres"
            exclude_rules = ["RF03"]
            """
        )
    else:
        config = dedent(
            """\
            [sqlfluff]
            dialect = postgres
            exclude_rules = RF03
            """
        )

    rule_runner.write_files(
        {
            file_path: BAD_FILE,
            file_path.parent / "BUILD": "sql_sources()",
            config_path: config,
        }
    )

    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, _ = run_sqlfluff(rule_runner, [tgt], extra_args=extra_args)
    assert lint_result.exit_code == (1 if should_change else 0)
    assert fix_result.did_change is should_change

1	# Copyright 2024 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	1✔
4
5	from pathlib import Path	1✔
6	from textwrap import dedent	1✔
7
8	import pytest	1✔
9
10	from pants.backend.sql.lint.sqlfluff import rules as sqlfluff_rules	1✔
11	from pants.backend.sql.lint.sqlfluff import skip_field	1✔
12	from pants.backend.sql.lint.sqlfluff import subsystem as sqlfluff_subsystem	1✔
13	from pants.backend.sql.lint.sqlfluff.rules import (	1✔
14	SqlfluffFixRequest,
15	SqlfluffFormatRequest,
16	SqlfluffLintRequest,
17	)
18	from pants.backend.sql.lint.sqlfluff.skip_field import SkipSqlfluffField	1✔
19	from pants.backend.sql.lint.sqlfluff.subsystem import SqlfluffFieldSet	1✔
20	from pants.backend.sql.target_types import SqlSourcesGeneratorTarget	1✔
21	from pants.core.goals.fix import FixResult	1✔
22	from pants.core.goals.fmt import FmtResult	1✔
23	from pants.core.goals.lint import LintResult	1✔
24	from pants.core.util_rules import config_files	1✔
25	from pants.core.util_rules.partitions import _EmptyMetadata	1✔
26	from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest	1✔
27	from pants.engine.addresses import Address	1✔
28	from pants.engine.target import Target	1✔
29	from pants.testutil.rule_runner import QueryRule, RuleRunner	1✔
30
31	GOOD_FILE = dedent(	1✔
32	"""\
33	select
34	e.id,
35	e.name
36	from employees as e
37	"""
38	)
39	BAD_FILE = dedent(	1✔
40	"""\
41	select
42	e.id,
43	name
44	from employees as e
45	"""
46	)
47	UNFORMATTED_FILE = dedent(	1✔
48	"""\
49	select e.id, e.name
50	from employees as e
51	"""
52	)
53	CONFIG_POSTGRES = dedent(	1✔
54	"""\
55	[sqlfluff]
56	dialect = postgres
57	"""
58	)
59
60
61	@pytest.fixture	1✔
62	def rule_runner() -> RuleRunner:	1✔
63	return RuleRunner(	1✔
64	rules=[
65	*skip_field.rules(),
66	*sqlfluff_subsystem.rules(),
67	*config_files.rules(),
68	*sqlfluff_rules.rules(),
69	QueryRule(FixResult, [SqlfluffFixRequest.Batch]),
70	QueryRule(LintResult, [SqlfluffLintRequest.Batch]),
71	QueryRule(FmtResult, [SqlfluffFormatRequest.Batch]),
72	QueryRule(SourceFiles, (SourceFilesRequest,)),
73	],
74	target_types=[SqlSourcesGeneratorTarget],
75	)
76
77
78	def run_sqlfluff(	1✔
79	rule_runner: RuleRunner,
80	targets: list[Target],
81	*,
82	extra_args: list[str] \| None = None,
83	) -> tuple[FixResult, LintResult, FmtResult]:
84	args = [	1✔
85	"--backend-packages=pants.backend.sql.lint.sqlfluff",
86	'--sqlfluff-fix-args="--force"',
87	*(extra_args or ()),
88	]
89	rule_runner.set_options(args, env_inherit={"PATH", "PYENV_ROOT", "HOME"})	1✔
90
91	field_sets = [	1✔
92	SqlfluffFieldSet.create(tgt) for tgt in targets if SqlfluffFieldSet.is_applicable(tgt)
93	]
94	source_reqs = [SourceFilesRequest(field_set.source for field_set in field_sets)]	1✔
95	input_sources = rule_runner.request(SourceFiles, source_reqs)	1✔
96
97	fix_result = rule_runner.request(	1✔
98	FixResult,
99	[
100	SqlfluffFixRequest.Batch(
101	"",
102	tuple(field_sets),
103	partition_metadata=_EmptyMetadata(),
104	snapshot=input_sources.snapshot,
105	),
106	],
107	)
108	lint_result = rule_runner.request(	1✔
109	LintResult,
110	[
111	SqlfluffLintRequest.Batch(
112	"",
113	tuple(field_sets),
114	partition_metadata=_EmptyMetadata(),
115	),
116	],
117	)
118	fmt_result = rule_runner.request(	1✔
119	FmtResult,
120	[
121	SqlfluffFormatRequest.Batch(
122	"",
123	tuple(field_sets),
124	partition_metadata=_EmptyMetadata(),
125	snapshot=input_sources.snapshot,
126	)
127	],
128	)
129
130	return fix_result, lint_result, fmt_result	1✔
131
132
133	@pytest.mark.platform_specific_behavior	1✔
134	def test_passing(rule_runner: RuleRunner) -> None:	1✔
135	rule_runner.write_files(	1✔
136	{
137	"query.sql": GOOD_FILE,
138	"BUILD": "sql_sources(name='t')",
139	".sqlfluff": CONFIG_POSTGRES,
140	}
141	)
142	tgt = rule_runner.get_target(Address("", target_name="t", relative_file_path="query.sql"))	1✔
143	fix_result, lint_result, fmt_result = run_sqlfluff(rule_runner, [tgt])	1✔
144	assert fix_result.stdout == dedent(	1✔
145	"""\
146	==== finding fixable violations ====
147	FORCE MODE: Attempting fixes...
148	==== no fixable linting violations found ====
149	All Finished!
150	"""
151	)
152	assert fix_result.stderr == ""	1✔
153	assert lint_result.exit_code == 0	1✔
154	assert not fix_result.did_change	1✔
155	assert fix_result.output == rule_runner.make_snapshot({"query.sql": GOOD_FILE})	1✔
156	assert not fmt_result.did_change	1✔
157	assert fmt_result.output == rule_runner.make_snapshot({"query.sql": GOOD_FILE})	1✔
158
159
160	def test_failing(rule_runner: RuleRunner) -> None:	1✔
UNCOV 161	rule_runner.write_files(	×
162	{
163	"query.sql": BAD_FILE,
164	"BUILD": "sql_sources(name='t')",
165	".sqlfluff": CONFIG_POSTGRES,
166	}
167	)
UNCOV 168	tgt = rule_runner.get_target(Address("", target_name="t", relative_file_path="query.sql"))	×
UNCOV 169	fix_result, lint_result, fmt_result = run_sqlfluff(rule_runner, [tgt])	×
UNCOV 170	assert fix_result.stdout == dedent(	×
171	"""\
172	==== finding fixable violations ====
173	FORCE MODE: Attempting fixes...
174	== [query.sql] FAIL
175	L: 3 \| P: 5 \| RF03 \| Unqualified reference 'name' found in single table
176	\| select. [references.consistent]
177	== [query.sql] FIXED
178	1 fixable linting violations found
179	[1 unfixable linting violations found]
180	"""
181	)
UNCOV 182	assert fix_result.stderr == ""	×
UNCOV 183	assert lint_result.stdout == dedent(	×
184	"""\
185	== [query.sql] FAIL
186	L: 3 \| P: 5 \| RF03 \| Unqualified reference 'name' found in single table
187	\| select. [references.consistent]
188	L: 3 \| P: 5 \| RF03 \| Unqualified reference 'name' found in single table
189	\| select which is inconsistent with previous references.
190	\| [references.consistent]
191	All Finished!
192	"""
193	)
UNCOV 194	assert lint_result.stderr == ""	×
UNCOV 195	assert lint_result.exit_code == 1	×
UNCOV 196	assert fix_result.did_change	×
UNCOV 197	assert fix_result.output == rule_runner.make_snapshot({"query.sql": GOOD_FILE})	×
UNCOV 198	assert not fmt_result.did_change	×
UNCOV 199	assert fmt_result.output == rule_runner.make_snapshot({"query.sql": BAD_FILE})	×
200
201
202	def test_multiple_targets(rule_runner: RuleRunner) -> None:	1✔
UNCOV 203	rule_runner.write_files(	×
204	{
205	"good.sql": GOOD_FILE,
206	"bad.sql": BAD_FILE,
207	"unformatted.sql": UNFORMATTED_FILE,
208	"BUILD": "sql_sources(name='t')",
209	".sqlfluff": CONFIG_POSTGRES,
210	}
211	)
UNCOV 212	tgts = [	×
213	rule_runner.get_target(Address("", target_name="t", relative_file_path="good.sql")),
214	rule_runner.get_target(Address("", target_name="t", relative_file_path="bad.sql")),
215	rule_runner.get_target(Address("", target_name="t", relative_file_path="unformatted.sql")),
216	]
UNCOV 217	fix_result, lint_result, fmt_result = run_sqlfluff(rule_runner, tgts)	×
UNCOV 218	assert lint_result.exit_code == 1	×
UNCOV 219	assert fix_result.output == rule_runner.make_snapshot(	×
220	{"good.sql": GOOD_FILE, "bad.sql": GOOD_FILE, "unformatted.sql": GOOD_FILE}
221	)
UNCOV 222	assert fix_result.did_change is True	×
UNCOV 223	assert fmt_result.output == rule_runner.make_snapshot(	×
224	{"good.sql": GOOD_FILE, "bad.sql": BAD_FILE, "unformatted.sql": GOOD_FILE}
225	)
UNCOV 226	assert fmt_result.did_change is True	×
227
228
229	def test_skip_field(rule_runner: RuleRunner) -> None:	1✔
UNCOV 230	rule_runner.write_files(	×
231	{
232	"good.sql": GOOD_FILE,
233	"bad.sql": BAD_FILE,
234	"unformatted.sql": UNFORMATTED_FILE,
235	"BUILD": "sql_sources(name='t', skip_sqlfluff=True)",
236	}
237	)
UNCOV 238	tgts = [	×
239	rule_runner.get_target(Address("", target_name="t", relative_file_path="good.sql")),
240	rule_runner.get_target(Address("", target_name="t", relative_file_path="bad.sql")),
241	rule_runner.get_target(Address("", target_name="t", relative_file_path="unformatted.sql")),
242	]
UNCOV 243	for tgt in tgts:	×
UNCOV 244	assert tgt.get(SkipSqlfluffField).value is True	×
245
UNCOV 246	fix_result, lint_result, fmt_result = run_sqlfluff(rule_runner, tgts)	×
247
UNCOV 248	assert lint_result.exit_code == 0	×
UNCOV 249	assert fix_result.output == rule_runner.make_snapshot({})	×
UNCOV 250	assert fix_result.did_change is False	×
UNCOV 251	assert fmt_result.output == rule_runner.make_snapshot({})	×
UNCOV 252	assert fmt_result.did_change is False	×
253
254
255	@pytest.mark.parametrize(	1✔
256	"file_path,config_path,extra_args,should_change",
257	(
258	[Path("query.sql"), Path("pyproject.toml"), [], False],
259	[Path("query.sql"), Path(".sqlfluff"), [], False],
260	[Path("custom/query.sql"), Path("custom/pyproject.toml"), [], False],
261	[Path("custom/query.sql"), Path("custom/.sqlfluff"), [], False],
262	[
263	Path("query.sql"),
264	Path("custom/config.sqlfluff"),
265	["--sqlfluff-config=custom/config.sqlfluff"],
266	False,
267	],
268	[
269	Path("query.sql"),
270	Path("custom/.sqlfluff"),
271	['--sqlfluff-args="--dialect=postgres"'],
272	True,
273	],
274	),
275	)
276	def test_config_file(	1✔
277	rule_runner: RuleRunner,
278	file_path: Path,
279	config_path: Path,
280	extra_args: list[str],
281	should_change: bool,
282	) -> None:
UNCOV 283	if config_path.stem == "pyproject":	×
UNCOV 284	config = dedent(	×
285	"""\
286	[tool.sqlfluff.core]
287	dialect = "postgres"
288	exclude_rules = ["RF03"]
289	"""
290	)
291	else:
UNCOV 292	config = dedent(	×
293	"""\
294	[sqlfluff]
295	dialect = postgres
296	exclude_rules = RF03
297	"""
298	)
299
UNCOV 300	rule_runner.write_files(	×
301	{
302	file_path: BAD_FILE,
303	file_path.parent / "BUILD": "sql_sources()",
304	config_path: config,
305	}
306	)
307
UNCOV 308	spec_path = str(file_path.parent).replace(".", "")	×
UNCOV 309	rel_file_path = file_path.relative_to(*file_path.parts[:1]) if spec_path else file_path	×
UNCOV 310	addr = Address(spec_path, relative_file_path=str(rel_file_path))	×
UNCOV 311	tgt = rule_runner.get_target(addr)	×
UNCOV 312	fix_result, lint_result, _ = run_sqlfluff(rule_runner, [tgt], extra_args=extra_args)	×
UNCOV 313	assert lint_result.exit_code == (1 if should_change else 0)	×
UNCOV 314	assert fix_result.did_change is should_change	×

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