Coverage for tests / unit / plugins / base / test_subprocess_streaming.py: 100%

1"""Unit tests for run_subprocess_streaming function.""" 

2 

3from __future__ import annotations 

4 

5import subprocess 

6from unittest.mock import MagicMock, patch 

7 

8import pytest 

9from assertpy import assert_that 

10 

11from lintro.plugins.subprocess_executor import run_subprocess_streaming 

12 

13# ============================================================================= 

14# run_subprocess_streaming - Success Cases 

15# ============================================================================= 

16 

17 

18def test_streaming_successful_command() -> None: 

19 """Verify successful streaming command returns True and captured output.""" 

20 with patch("lintro.plugins.subprocess_executor.subprocess.Popen") as mock_popen: 

21 mock_process = MagicMock() 

22 mock_process.stdout = iter(["line1\n", "line2\n"]) 

23 mock_process.wait.return_value = 0 

24 mock_popen.return_value = mock_process 

25 

26 success, output = run_subprocess_streaming(["echo", "hello"], timeout=30) 

27 

28 assert_that(success).is_true() 

29 assert_that(output).contains("line1") 

30 assert_that(output).contains("line2") 

31 

32 

33def test_streaming_failed_command_nonzero_exit() -> None: 

34 """Verify failed streaming command returns False and logs output.""" 

35 with patch("lintro.plugins.subprocess_executor.subprocess.Popen") as mock_popen: 

36 mock_process = MagicMock() 

37 mock_process.stdout = iter(["error output\n"]) 

38 mock_process.wait.return_value = 1 

39 mock_popen.return_value = mock_process 

40 

41 success, output = run_subprocess_streaming(["false"], timeout=30) 

42 

43 assert_that(success).is_false() 

44 assert_that(output).contains("error output") 

45 

46 

47def test_streaming_with_line_handler() -> None: 

48 """Verify line handler is called for each output line.""" 

49 lines_received: list[str] = [] 

50 

51 def handler(line: str) -> None: 

52 lines_received.append(line) 

53 

54 with patch("lintro.plugins.subprocess_executor.subprocess.Popen") as mock_popen: 

55 mock_process = MagicMock() 

56 mock_process.stdout = iter(["first\n", "second\n", "third\n"]) 

57 mock_process.wait.return_value = 0 

58 mock_popen.return_value = mock_process 

59 

60 run_subprocess_streaming(["echo"], timeout=30, line_handler=handler) 

61 

62 assert_that(lines_received).is_length(3) 

63 assert_that(lines_received).contains("first") 

64 assert_that(lines_received).contains("second") 

65 assert_that(lines_received).contains("third") 

66 

67 

68# ============================================================================= 

69# run_subprocess_streaming - Timeout Cases 

70# ============================================================================= 

71 

72 

73def test_streaming_timeout_during_read() -> None: 

74 """Verify TimeoutExpired is raised when reading times out.""" 

75 with ( 

76 patch("lintro.plugins.subprocess_executor.subprocess.Popen") as mock_popen, 

77 patch("lintro.plugins.subprocess_executor.threading.Thread") as mock_thread, 

78 ): 

79 mock_process = MagicMock() 

80 mock_process.stdout = iter([]) 

81 mock_popen.return_value = mock_process 

82 

83 # Simulate thread still alive after join (timeout occurred) 

84 mock_thread_instance = MagicMock() 

85 mock_thread_instance.is_alive.return_value = True 

86 mock_thread.return_value = mock_thread_instance 

87 

88 with pytest.raises(subprocess.TimeoutExpired): 

89 run_subprocess_streaming(["long", "cmd"], timeout=1) 

90 

91 mock_process.kill.assert_called_once() 

92 

93 

94def test_streaming_timeout_during_wait() -> None: 

95 """Verify TimeoutExpired is raised when process.wait times out.""" 

96 with patch("lintro.plugins.subprocess_executor.subprocess.Popen") as mock_popen: 

97 mock_process = MagicMock() 

98 mock_process.stdout = iter(["partial\n"]) 

99 mock_process.wait.side_effect = subprocess.TimeoutExpired( 

100 cmd=["slow"], 

101 timeout=1, 

102 ) 

103 mock_popen.return_value = mock_process 

104 

105 with pytest.raises(subprocess.TimeoutExpired): 

106 run_subprocess_streaming(["slow", "cmd"], timeout=1) 

107 

108 mock_process.kill.assert_called_once() 

109 

110 

111# ============================================================================= 

112# run_subprocess_streaming - FileNotFoundError Cases 

113# ============================================================================= 

114 

115 

116def test_streaming_file_not_found() -> None: 

117 """Verify FileNotFoundError is raised when command is not found.""" 

118 with patch("lintro.plugins.subprocess_executor.subprocess.Popen") as mock_popen: 

119 mock_popen.side_effect = FileNotFoundError("not found") 

120 

121 with pytest.raises(FileNotFoundError, match="Command not found"): 

122 run_subprocess_streaming(["nonexistent"], timeout=30) 

123 

124 

125# ============================================================================= 

126# run_subprocess_streaming - Edge Cases 

127# ============================================================================= 

128 

129 

130def test_streaming_empty_output() -> None: 

131 """Verify empty output is handled correctly.""" 

132 with patch("lintro.plugins.subprocess_executor.subprocess.Popen") as mock_popen: 

133 mock_process = MagicMock() 

134 mock_process.stdout = iter([]) 

135 mock_process.wait.return_value = 0 

136 mock_popen.return_value = mock_process 

137 

138 success, output = run_subprocess_streaming(["true"], timeout=30) 

139 

140 assert_that(success).is_true() 

141 assert_that(output).is_equal_to("") 

142 

143 

144def test_streaming_with_cwd_and_env() -> None: 

145 """Verify cwd and env are passed to Popen.""" 

146 with patch("lintro.plugins.subprocess_executor.subprocess.Popen") as mock_popen: 

147 mock_process = MagicMock() 

148 mock_process.stdout = iter([]) 

149 mock_process.wait.return_value = 0 

150 mock_popen.return_value = mock_process 

151 

152 custom_env = {"MY_VAR": "value"} 

153 run_subprocess_streaming( 

154 ["cmd"], 

155 timeout=30, 

156 cwd="/custom/path", 

157 env=custom_env, 

158 ) 

159 

160 mock_popen.assert_called_once() 

161 call_kwargs = mock_popen.call_args[1] 

162 assert_that(call_kwargs["cwd"]).is_equal_to("/custom/path") 

163 # Custom env is merged with os.environ to preserve PATH 

164 assert_that(call_kwargs["env"]["MY_VAR"]).is_equal_to("value") 

165 assert_that(call_kwargs["env"]).contains_key("PATH")