Coverage for tests / unit / utils / test_console_output_writer.py: 100%

1"""Unit tests for ThreadSafeConsoleLogger class message tracking.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6from unittest.mock import patch 

7 

8import pytest 

9from assertpy import assert_that 

10 

11from lintro.utils.console import ThreadSafeConsoleLogger 

12 

13 

14@pytest.fixture 

15def logger() -> ThreadSafeConsoleLogger: 

16 """Create a ThreadSafeConsoleLogger instance for testing. 

17 

18 Returns: 

19 ThreadSafeConsoleLogger instance for testing. 

20 """ 

21 return ThreadSafeConsoleLogger() 

22 

23 

24def test_initialization(logger: ThreadSafeConsoleLogger) -> None: 

25 """Test that logger initializes with empty messages list. 

26 

27 Args: 

28 logger: ThreadSafeConsoleLogger instance for testing. 

29 """ 

30 assert_that(logger._messages).is_empty() 

31 

32 

33def test_console_output_no_color(logger: ThreadSafeConsoleLogger) -> None: 

34 """Test console_output without color. 

35 

36 Args: 

37 logger: ThreadSafeConsoleLogger instance for testing. 

38 """ 

39 with patch("lintro.utils.console.logger.click.echo") as mock_echo: 

40 logger.console_output("Test message") 

41 

42 mock_echo.assert_called_once_with("Test message") 

43 assert_that(logger._messages).contains("Test message") 

44 

45 

46def test_console_output_with_color(logger: ThreadSafeConsoleLogger) -> None: 

47 """Test console_output with color. 

48 

49 Args: 

50 logger: ThreadSafeConsoleLogger instance for testing. 

51 """ 

52 with ( 

53 patch("lintro.utils.console.logger.click.echo") as mock_echo, 

54 patch("lintro.utils.console.logger.click.style") as mock_style, 

55 ): 

56 mock_style.return_value = "styled text" 

57 logger.console_output("Test message", color="green") 

58 

59 mock_style.assert_called_once_with("Test message", fg="green") 

60 mock_echo.assert_called_once_with("styled text") 

61 assert_that(logger._messages).contains("Test message") 

62 

63 

64def test_info(logger: ThreadSafeConsoleLogger) -> None: 

65 """Test info method logs message. 

66 

67 Args: 

68 logger: ThreadSafeConsoleLogger instance for testing. 

69 """ 

70 with ( 

71 patch("lintro.utils.console.logger.click.echo") as mock_echo, 

72 patch("lintro.utils.console.logger.logger") as mock_logger, 

73 ): 

74 logger.info("Info message") 

75 

76 mock_echo.assert_called_once_with("Info message") 

77 mock_logger.info.assert_called_once() 

78 assert_that(logger._messages).contains("Info message") 

79 

80 

81def test_info_blue(logger: ThreadSafeConsoleLogger) -> None: 

82 """Test info_blue method logs in cyan. 

83 

84 Args: 

85 logger: ThreadSafeConsoleLogger instance for testing. 

86 """ 

87 with ( 

88 patch("lintro.utils.console.logger.click.echo") as mock_echo, 

89 patch("lintro.utils.console.logger.click.style") as mock_style, 

90 patch("lintro.utils.console.logger.logger") as mock_logger, 

91 ): 

92 mock_style.return_value = "styled" 

93 logger.info_blue("Blue message") 

94 

95 mock_style.assert_called_once_with("Blue message", fg="cyan") 

96 mock_echo.assert_called_once_with("styled") 

97 mock_logger.info.assert_called_once() 

98 assert_that(logger._messages).contains("Blue message") 

99 

100 

101def test_success(logger: ThreadSafeConsoleLogger) -> None: 

102 """Test success method logs in green with emoji. 

103 

104 Args: 

105 logger: ThreadSafeConsoleLogger instance for testing. 

106 """ 

107 with ( 

108 patch("lintro.utils.console.logger.click.echo"), 

109 patch("lintro.utils.console.logger.click.style") as mock_style, 

110 patch("lintro.utils.console.logger.logger") as mock_logger, 

111 ): 

112 mock_style.return_value = "styled" 

113 logger.success("Success message") 

114 

115 mock_style.assert_called_once_with("✅ Success message", fg="green") 

116 mock_logger.info.assert_called_once() 

117 assert_that(logger._messages).contains("✅ Success message") 

118 

119 

120def test_warning(logger: ThreadSafeConsoleLogger) -> None: 

121 """Test warning method prefixes with WARNING. 

122 

123 Args: 

124 logger: ThreadSafeConsoleLogger instance for testing. 

125 """ 

126 with ( 

127 patch("lintro.utils.console.logger.click.echo"), 

128 patch("lintro.utils.console.logger.click.style") as mock_style, 

129 patch("lintro.utils.console.logger.logger") as mock_logger, 

130 ): 

131 mock_style.return_value = "styled" 

132 logger.warning("Warning message") 

133 

134 mock_style.assert_called_once_with("WARNING: Warning message", fg="yellow") 

135 mock_logger.warning.assert_called_once() 

136 assert_that(logger._messages).contains("WARNING: Warning message") 

137 

138 

139def test_error(logger: ThreadSafeConsoleLogger) -> None: 

140 """Test error method prefixes with ERROR. 

141 

142 Args: 

143 logger: ThreadSafeConsoleLogger instance for testing. 

144 """ 

145 with ( 

146 patch("lintro.utils.console.logger.click.echo"), 

147 patch("lintro.utils.console.logger.click.style") as mock_style, 

148 patch("lintro.utils.console.logger.logger") as mock_logger, 

149 ): 

150 mock_style.return_value = "styled" 

151 logger.error("Error message") 

152 

153 mock_style.assert_called_once_with("ERROR: Error message", fg="red", bold=True) 

154 mock_logger.error.assert_called_once() 

155 assert_that(logger._messages).contains("ERROR: Error message") 

156 

157 

158def test_save_console_log(logger: ThreadSafeConsoleLogger, tmp_path: Path) -> None: 

159 """Test saving console messages to file. 

160 

161 Args: 

162 logger: ThreadSafeConsoleLogger instance for testing. 

163 tmp_path: Temporary directory path for testing. 

164 """ 

165 logger._messages = ["Message 1", "Message 2", "Message 3"] 

166 

167 with patch("lintro.utils.console.logger.logger"): 

168 logger.save_console_log(tmp_path) 

169 

170 log_path = tmp_path / "console.log" 

171 assert_that(log_path.exists()).is_true() 

172 content = log_path.read_text() 

173 assert_that(content).contains("Message 1") 

174 assert_that(content).contains("Message 2") 

175 assert_that(content).contains("Message 3") 

176 

177 

178def test_save_console_log_handles_error(logger: ThreadSafeConsoleLogger) -> None: 

179 """Test save_console_log handles write errors gracefully. 

180 

181 Args: 

182 logger: ThreadSafeConsoleLogger instance for testing. 

183 """ 

184 logger._messages = ["Message"] 

185 

186 with ( 

187 patch("builtins.open", side_effect=OSError("Permission denied")), 

188 patch("lintro.utils.console.logger.logger") as mock_logger, 

189 ): 

190 logger.save_console_log("/invalid/path") 

191 mock_logger.error.assert_called_once() 

192 

193 

194def test_multiple_messages_tracked(logger: ThreadSafeConsoleLogger) -> None: 

195 """Test that multiple messages are tracked in order. 

196 

197 Args: 

198 logger: ThreadSafeConsoleLogger instance for testing. 

199 """ 

200 with ( 

201 patch("lintro.utils.console.logger.click.echo"), 

202 patch("lintro.utils.console.logger.click.style", return_value="s"), 

203 patch("lintro.utils.console.logger.logger"), 

204 ): 

205 logger.info("First") 

206 logger.success("Second") 

207 logger.warning("Third") 

208 logger.error("Fourth") 

209 

210 assert_that(len(logger._messages)).is_equal_to(4) 

211 assert_that(logger._messages[0]).is_equal_to("First") 

212 assert_that(logger._messages[1]).is_equal_to("✅ Second") 

213 assert_that(logger._messages[2]).is_equal_to("WARNING: Third") 

214 assert_that(logger._messages[3]).is_equal_to("ERROR: Fourth") 

215 

216 

217def test_thread_safe_message_tracking(logger: ThreadSafeConsoleLogger) -> None: 

218 """Test that message tracking is thread-safe. 

219 

220 Args: 

221 logger: ThreadSafeConsoleLogger instance for testing. 

222 """ 

223 import threading 

224 

225 messages_to_add = 100 

226 threads = [] 

227 

228 with ( 

229 patch("lintro.utils.console.logger.click.echo"), 

230 patch("lintro.utils.console.logger.click.style", return_value="s"), 

231 ): 

232 

233 def add_message(i: int) -> None: 

234 logger.console_output(f"Message {i}") 

235 

236 for i in range(messages_to_add): 

237 t = threading.Thread(target=add_message, args=(i,)) 

238 threads.append(t) 

239 t.start() 

240 

241 for t in threads: 

242 t.join() 

243 

244 # All messages should be tracked without loss 

245 assert_that(len(logger._messages)).is_equal_to(messages_to_add)