Coverage for tests / unit / tools / sqlfluff / test_fix_method.py: 100%
47 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-04-05 16:51 +0000
1"""Tests for SqlfluffPlugin.fix method initial_issues population."""
3from __future__ import annotations
5from pathlib import Path
6from unittest.mock import patch
8from assertpy import assert_that
10from lintro.tools.definitions.sqlfluff import SqlfluffPlugin
12SQLFLUFF_LINT_OUTPUT_WITH_ISSUES = """[
13 {
14 "filepath": "test_query.sql",
15 "violations": [
16 {
17 "start_line_no": 1,
18 "start_line_pos": 1,
19 "end_line_no": 1,
20 "end_line_pos": 6,
21 "code": "L010",
22 "description": "Keywords must be upper case.",
23 "name": "capitalisation.keywords"
24 }
25 ]
26 }
27]"""
30def test_fix_populates_initial_issues(
31 sqlfluff_plugin: SqlfluffPlugin,
32 tmp_path: Path,
33) -> None:
34 """Fix populates initial_issues when issues are found and fixed.
36 Args:
37 sqlfluff_plugin: The SqlfluffPlugin instance to test.
38 tmp_path: Temporary directory path for test files.
39 """
40 test_file = tmp_path / "test_query.sql"
41 test_file.write_text("select * from users;\n")
43 with patch.object(
44 sqlfluff_plugin,
45 "_run_subprocess",
46 side_effect=[
47 (False, SQLFLUFF_LINT_OUTPUT_WITH_ISSUES), # Initial lint check
48 (True, "Fixed 1 file(s)"), # Fix command
49 (True, "[]"), # Verification lint - no issues
50 ],
51 ):
52 result = sqlfluff_plugin.fix([str(test_file)], {})
54 assert_that(result.success).is_true()
55 assert_that(result.initial_issues).is_not_none()
56 assert_that(result.initial_issues).is_length(1)
57 assert_that(result.initial_issues_count).is_equal_to(1)
58 assert_that(result.fixed_issues_count).is_equal_to(1)
59 assert_that(result.remaining_issues_count).is_equal_to(0)
62def test_fix_initial_issues_none_when_no_issues(
63 sqlfluff_plugin: SqlfluffPlugin,
64 tmp_path: Path,
65) -> None:
66 """Fix sets initial_issues to None when no issues detected.
68 Args:
69 sqlfluff_plugin: The SqlfluffPlugin instance to test.
70 tmp_path: Temporary directory path for test files.
71 """
72 test_file = tmp_path / "test_query.sql"
73 test_file.write_text("SELECT * FROM users;\n")
75 with patch.object(
76 sqlfluff_plugin,
77 "_run_subprocess",
78 return_value=(True, "[]"),
79 ):
80 result = sqlfluff_plugin.fix([str(test_file)], {})
82 assert_that(result.success).is_true()
83 assert_that(result.initial_issues).is_none()
86def test_fix_partial_fix_preserves_initial_issues(
87 sqlfluff_plugin: SqlfluffPlugin,
88 tmp_path: Path,
89) -> None:
90 """Fix preserves initial_issues when some issues remain after fix.
92 Args:
93 sqlfluff_plugin: The SqlfluffPlugin instance to test.
94 tmp_path: Temporary directory path for test files.
95 """
96 test_file = tmp_path / "test_query.sql"
97 test_file.write_text("select * from users;\n")
99 remaining_output = """[
100 {
101 "filepath": "test_query.sql",
102 "violations": [
103 {
104 "start_line_no": 1,
105 "start_line_pos": 1,
106 "end_line_no": 1,
107 "end_line_pos": 6,
108 "code": "L010",
109 "description": "Keywords must be upper case.",
110 "name": "capitalisation.keywords"
111 }
112 ]
113 }
114 ]"""
116 with patch.object(
117 sqlfluff_plugin,
118 "_run_subprocess",
119 side_effect=[
120 (False, SQLFLUFF_LINT_OUTPUT_WITH_ISSUES), # Initial lint check
121 (True, "Fixed 1 file(s)"), # Fix command
122 (False, remaining_output), # Verification - issue remains
123 ],
124 ):
125 result = sqlfluff_plugin.fix([str(test_file)], {})
127 assert_that(result.success).is_false()
128 assert_that(result.initial_issues).is_not_none()
129 assert_that(result.initial_issues).is_length(1)
130 assert_that(result.initial_issues_count).is_equal_to(1)
131 assert_that(result.fixed_issues_count).is_equal_to(0)
132 assert_that(result.remaining_issues_count).is_equal_to(1)
135def test_fix_runs_verify_even_when_fix_exits_nonzero(
136 sqlfluff_plugin: SqlfluffPlugin,
137 tmp_path: Path,
138) -> None:
139 """Partial fixes are counted when sqlfluff fix exits non-zero.
141 sqlfluff's fix command can apply some fixes and still exit non-zero
142 if other rules are unfixable. The verify lint must still run to get
143 the true remaining count.
145 Args:
146 sqlfluff_plugin: The SqlfluffPlugin instance to test.
147 tmp_path: Temporary directory path for test files.
148 """
149 test_file = tmp_path / "test_query.sql"
150 test_file.write_text("select * from users;\n")
152 two_issues_output = """[
153 {
154 "filepath": "test_query.sql",
155 "violations": [
156 {
157 "start_line_no": 1, "start_line_pos": 1,
158 "end_line_no": 1, "end_line_pos": 6,
159 "code": "L010", "description": "kw",
160 "name": "capitalisation.keywords"
161 },
162 {
163 "start_line_no": 1, "start_line_pos": 10,
164 "end_line_no": 1, "end_line_pos": 11,
165 "code": "L030", "description": "fn",
166 "name": "capitalisation.functions"
167 }
168 ]
169 }
170 ]"""
171 one_remaining_output = """[
172 {
173 "filepath": "test_query.sql",
174 "violations": [
175 {
176 "start_line_no": 1, "start_line_pos": 10,
177 "end_line_no": 1, "end_line_pos": 11,
178 "code": "L030", "description": "fn",
179 "name": "capitalisation.functions"
180 }
181 ]
182 }
183 ]"""
185 with patch.object(
186 sqlfluff_plugin,
187 "_run_subprocess",
188 side_effect=[
189 (False, two_issues_output), # Initial lint: 2 issues
190 (False, "Unfixable rule"), # Fix: exits non-zero
191 (False, one_remaining_output), # Verify: 1 remaining (1 was fixed)
192 ],
193 ):
194 result = sqlfluff_plugin.fix([str(test_file)], {})
196 # Fix reports success=False because issues remain
197 assert_that(result.success).is_false()
198 # But the partial fix was counted
199 assert_that(result.initial_issues_count).is_equal_to(2)
200 assert_that(result.fixed_issues_count).is_equal_to(1)
201 assert_that(result.remaining_issues_count).is_equal_to(1)