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

1"""Unit tests for sqlfluff plugin error handling, timeouts, and edge cases.""" 

2 

3from __future__ import annotations 

4 

5import subprocess 

6from pathlib import Path 

7from unittest.mock import patch 

8 

9from assertpy import assert_that 

10 

11from lintro.tools.definitions.sqlfluff import ( 

12 SQLFLUFF_DEFAULT_TIMEOUT, 

13 SqlfluffPlugin, 

14) 

15 

16# Tests for check method error handling 

17 

18 

19def test_check_with_timeout( 

20 sqlfluff_plugin: SqlfluffPlugin, 

21 tmp_path: Path, 

22) -> None: 

23 """Check handles timeout correctly. 

24 

25 Args: 

26 sqlfluff_plugin: The SqlfluffPlugin instance to test. 

27 tmp_path: Temporary directory path for test files. 

28 """ 

29 test_file = tmp_path / "test_query.sql" 

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

31 

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

33 with patch.object( 

34 sqlfluff_plugin, 

35 "_run_subprocess", 

36 side_effect=subprocess.TimeoutExpired( 

37 cmd=["sqlfluff"], 

38 timeout=SQLFLUFF_DEFAULT_TIMEOUT, 

39 ), 

40 ): 

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

42 

43 assert_that(result.success).is_false() 

44 

45 

46def test_check_with_empty_output( 

47 sqlfluff_plugin: SqlfluffPlugin, 

48 tmp_path: Path, 

49) -> None: 

50 """Check handles empty output correctly. 

51 

52 Args: 

53 sqlfluff_plugin: The SqlfluffPlugin instance to test. 

54 tmp_path: Temporary directory path for test files. 

55 """ 

56 test_file = tmp_path / "test_query.sql" 

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

58 

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

60 with patch.object( 

61 sqlfluff_plugin, 

62 "_run_subprocess", 

63 return_value=(True, ""), 

64 ): 

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

66 

67 assert_that(result.success).is_true() 

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

69 

70 

71# Tests for fix method error handling 

72 

73 

74def test_fix_with_timeout( 

75 sqlfluff_plugin: SqlfluffPlugin, 

76 tmp_path: Path, 

77) -> None: 

78 """Fix handles timeout correctly. 

79 

80 Args: 

81 sqlfluff_plugin: The SqlfluffPlugin instance to test. 

82 tmp_path: Temporary directory path for test files. 

83 """ 

84 test_file = tmp_path / "test_query.sql" 

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

86 

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

88 with patch.object( 

89 sqlfluff_plugin, 

90 "_run_subprocess", 

91 side_effect=subprocess.TimeoutExpired( 

92 cmd=["sqlfluff"], 

93 timeout=SQLFLUFF_DEFAULT_TIMEOUT, 

94 ), 

95 ): 

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

97 

98 assert_that(result.success).is_false()