Coverage for tests / unit / tools / sqlfluff / test_execution.py: 100%

1"""Unit tests for sqlfluff plugin check and fix method execution.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6from unittest.mock import patch 

7 

8from assertpy import assert_that 

9 

10from lintro.tools.definitions.sqlfluff import SqlfluffPlugin 

11 

12# Tests for SqlfluffPlugin.check method 

13 

14 

15def test_check_with_mocked_subprocess_success( 

16 sqlfluff_plugin: SqlfluffPlugin, 

17 tmp_path: Path, 

18) -> None: 

19 """Check returns success when no issues found. 

20 

21 Args: 

22 sqlfluff_plugin: The SqlfluffPlugin instance to test. 

23 tmp_path: Temporary directory path for test files. 

24 """ 

25 test_file = tmp_path / "test_query.sql" 

26 test_file.write_text("SELECT * FROM users;\n") 

27 

28 # Note: verify_tool_version is already patched by the sqlfluff_plugin fixture 

29 with patch.object( 

30 sqlfluff_plugin, 

31 "_run_subprocess", 

32 return_value=(True, "[]"), 

33 ): 

34 result = sqlfluff_plugin.check([str(test_file)], {}) 

35 

36 assert_that(result.success).is_true() 

37 assert_that(result.issues_count).is_equal_to(0) 

38 

39 

40def test_check_with_mocked_subprocess_issues( 

41 sqlfluff_plugin: SqlfluffPlugin, 

42 tmp_path: Path, 

43) -> None: 

44 """Check returns issues when sqlfluff finds problems. 

45 

46 Args: 

47 sqlfluff_plugin: The SqlfluffPlugin instance to test. 

48 tmp_path: Temporary directory path for test files. 

49 """ 

50 test_file = tmp_path / "test_query.sql" 

51 test_file.write_text("select * from users;\n") 

52 

53 sqlfluff_output = """[ 

54 { 

55 "filepath": "test_query.sql", 

56 "violations": [ 

57 { 

58 "start_line_no": 1, 

59 "start_line_pos": 1, 

60 "end_line_no": 1, 

61 "end_line_pos": 6, 

62 "code": "L010", 

63 "description": "Keywords must be upper case.", 

64 "name": "capitalisation.keywords" 

65 } 

66 ] 

67 } 

68 ]""" 

69 

70 # Note: verify_tool_version is already patched by the sqlfluff_plugin fixture 

71 with patch.object( 

72 sqlfluff_plugin, 

73 "_run_subprocess", 

74 return_value=(False, sqlfluff_output), 

75 ): 

76 result = sqlfluff_plugin.check([str(test_file)], {}) 

77 

78 assert_that(result.success).is_false() 

79 assert_that(result.issues_count).is_greater_than(0) 

80 

81 

82def test_check_with_no_sql_files( 

83 sqlfluff_plugin: SqlfluffPlugin, 

84 tmp_path: Path, 

85) -> None: 

86 """Check returns success when no SQL files found. 

87 

88 Args: 

89 sqlfluff_plugin: The SqlfluffPlugin instance to test. 

90 tmp_path: Temporary directory path for test files. 

91 """ 

92 non_sql_file = tmp_path / "test.txt" 

93 non_sql_file.write_text("Not a SQL file") 

94 

95 # Note: verify_tool_version is already patched by the sqlfluff_plugin fixture 

96 result = sqlfluff_plugin.check([str(non_sql_file)], {}) 

97 

98 assert_that(result.success).is_true() 

99 assert_that(result.output).contains("No") 

100 

101 

102# Tests for SqlfluffPlugin.fix method 

103 

104 

105def test_fix_with_mocked_subprocess_success( 

106 sqlfluff_plugin: SqlfluffPlugin, 

107 tmp_path: Path, 

108) -> None: 

109 """Fix returns success when fixes applied. 

110 

111 Args: 

112 sqlfluff_plugin: The SqlfluffPlugin instance to test. 

113 tmp_path: Temporary directory path for test files. 

114 """ 

115 test_file = tmp_path / "test_query.sql" 

116 test_file.write_text("select * from users;\n") 

117 

118 # Note: verify_tool_version is already patched by the sqlfluff_plugin fixture 

119 with patch.object( 

120 sqlfluff_plugin, 

121 "_run_subprocess", 

122 return_value=(True, "Fixed 1 file(s)"), 

123 ): 

124 result = sqlfluff_plugin.fix([str(test_file)], {}) 

125 

126 assert_that(result.success).is_true() 

127 assert_that(result.issues_count).is_equal_to(0) 

128 

129 

130def test_fix_with_mocked_subprocess_no_changes( 

131 sqlfluff_plugin: SqlfluffPlugin, 

132 tmp_path: Path, 

133) -> None: 

134 """Fix returns success when no changes needed. 

135 

136 Args: 

137 sqlfluff_plugin: The SqlfluffPlugin instance to test. 

138 tmp_path: Temporary directory path for test files. 

139 """ 

140 test_file = tmp_path / "test_query.sql" 

141 test_file.write_text("SELECT * FROM users;\n") 

142 

143 # Note: verify_tool_version is already patched by the sqlfluff_plugin fixture 

144 with patch.object( 

145 sqlfluff_plugin, 

146 "_run_subprocess", 

147 return_value=(True, ""), 

148 ): 

149 result = sqlfluff_plugin.fix([str(test_file)], {}) 

150 

151 assert_that(result.success).is_true() 

152 

153 

154def test_fix_with_no_sql_files( 

155 sqlfluff_plugin: SqlfluffPlugin, 

156 tmp_path: Path, 

157) -> None: 

158 """Fix returns success when no SQL files found. 

159 

160 Args: 

161 sqlfluff_plugin: The SqlfluffPlugin instance to test. 

162 tmp_path: Temporary directory path for test files. 

163 """ 

164 non_sql_file = tmp_path / "test.txt" 

165 non_sql_file.write_text("Not a SQL file") 

166 

167 # Note: verify_tool_version is already patched by the sqlfluff_plugin fixture 

168 result = sqlfluff_plugin.fix([str(non_sql_file)], {}) 

169 

170 assert_that(result.success).is_true() 

171 assert_that(result.output).contains("No")