import unittest from unittest.mock import patch, mock_open, MagicMock import os import logging from datetime import datetime, timedelta # Ensure tools/log_tool.py is accessible from tools.log_tool import LogTool class TestLogTool(unittest.TestCase): def setUp(self): self.test_log_file_path = "test_dummy_log.log" # Suppress logging output during tests unless explicitly testing for it self.logger = logging.getLogger('tools.log_tool') # Ensure only one NullHandler to prevent duplicate messages if tests run multiple times in a session if not any(isinstance(h, logging.NullHandler) for h in self.logger.handlers): self.logger.addHandler(logging.NullHandler()) self.logger.propagate = False # Prevent propagation to root logger if it has handlers def test_init_default_log_path(self): tool = LogTool(logger=self.logger) self.assertEqual(tool.configured_log_file_path, 'logs/output.log') def test_init_custom_log_path(self): tool = LogTool(log_file_path=self.test_log_file_path, logger=self.logger) self.assertEqual(tool.configured_log_file_path, self.test_log_file_path) def test_get_functions(self): tool = LogTool(logger=self.logger) functions = tool.get_functions() self.assertIsInstance(functions, list) self.assertEqual(len(functions), 1) self.assertEqual(functions[0]["function"]["name"], "get_log_contents") @patch("os.path.exists", return_value=False) def test_get_log_contents_file_not_exists(self, mock_exists): tool = LogTool(log_file_path=self.test_log_file_path, logger=self.logger) result = tool._get_log_contents() self.assertIn("Log file does not exist", result) mock_exists.assert_called_once_with(self.test_log_file_path) @patch("os.path.exists", return_value=True) @patch("builtins.open", new_callable=mock_open, read_data="line1\nline2\nline3\nline4\nline5") def test_get_log_contents_with_line_count(self, mock_file_open, mock_exists): tool = LogTool(log_file_path=self.test_log_file_path, logger=self.logger) result = tool._get_log_contents(line_count=3) self.assertEqual(result, "line3\nline4\nline5") mock_exists.assert_called_once_with(self.test_log_file_path) mock_file_open.assert_called_once_with(self.test_log_file_path, 'r', encoding='utf-8') @patch("os.path.exists", return_value=True) @patch("builtins.open", new_callable=mock_open, read_data="line1\nline2\n") def test_get_log_contents_line_count_more_than_available(self, mock_file_open, mock_exists): tool = LogTool(log_file_path=self.test_log_file_path, logger=self.logger) result = tool._get_log_contents(line_count=5) self.assertEqual(result, "line1\nline2\n") @patch("os.path.exists", return_value=True) @patch("builtins.open", new_callable=mock_open, read_data="line1\nline2\n") def test_get_log_contents_invalid_line_count_uses_default(self, mock_file_open, mock_exists): tool = LogTool(log_file_path=self.test_log_file_path, logger=self.logger) # Test with zero, negative, and non-integer line_count (though type hint is int) # The code defaults to 150 if invalid. Here, we only have 2 lines. with patch.object(tool.logger, 'warning') as mock_log_warning: result_zero = tool._get_log_contents(line_count=0) self.assertEqual(result_zero, "line1\nline2\n") mock_log_warning.assert_any_call("Invalid line_count '0' provided, defaulting to fetch last 150 lines.") mock_file_open.reset_mock() # Reset for next call result_neg = tool._get_log_contents(line_count=-5) self.assertEqual(result_neg, "line1\nline2\n") mock_log_warning.assert_any_call("Invalid line_count '-5' provided, defaulting to fetch last 150 lines.") @patch("os.path.exists", return_value=True) def test_get_log_contents_last_24_hours(self, mock_exists): tool = LogTool(log_file_path=self.test_log_file_path, logger=self.logger) now = datetime.now() one_hour_ago_dt = now - timedelta(hours=1) two_days_ago_dt = now - timedelta(days=2) one_hour_ago_str = one_hour_ago_dt.strftime(LogTool.EXPECTED_LOG_TIMESTAMP_FORMAT) two_days_ago_str = two_days_ago_dt.strftime(LogTool.EXPECTED_LOG_TIMESTAMP_FORMAT) log_data = ( f"{two_days_ago_str} - OLD - This is an old log entry.\n" f"No timestamp here - this line should be skipped by time filter.\n" f"{one_hour_ago_str} - RECENT - This is a recent log entry.\n" f"Malformed Date 2023-xx-01 - Another skipped line.\n" f"{now.strftime(LogTool.EXPECTED_LOG_TIMESTAMP_FORMAT)} - CURRENT - This is a current log entry.\n" ) expected_output = ( f"{one_hour_ago_str} - RECENT - This is a recent log entry.\n" f"{now.strftime(LogTool.EXPECTED_LOG_TIMESTAMP_FORMAT)} - CURRENT - This is a current log entry.\n" ) with patch("builtins.open", mock_open(read_data=log_data)): result = tool._get_log_contents(line_count=None) # Trigger 24-hour logic self.assertEqual(result, expected_output) @patch("os.path.exists", return_value=True) @patch("builtins.open", side_effect=IOError("File read error!")) def test_get_log_contents_file_read_exception(self, mock_file_open, mock_exists): tool = LogTool(log_file_path=self.test_log_file_path, logger=self.logger) result = tool._get_log_contents(line_count=10) self.assertIn("An error occurred while reading the log file: File read error!", result) def test_execute_get_log_contents(self): tool = LogTool(logger=self.logger) mock_return_value = "Mocked log content" with patch.object(tool, '_get_log_contents', return_value=mock_return_value) as mock_method: result = tool.execute(function_name="get_log_contents", line_count=50) mock_method.assert_called_once_with(line_count=50) self.assertEqual(result, mock_return_value) def test_execute_get_log_contents_no_line_count(self): tool = LogTool(logger=self.logger) mock_return_value = "Mocked log content for 24h" with patch.object(tool, '_get_log_contents', return_value=mock_return_value) as mock_method: result = tool.execute(function_name="get_log_contents") # No line_count mock_method.assert_called_once_with(line_count=None) # Expects None to trigger 24h self.assertEqual(result, mock_return_value) def test_execute_unknown_function(self): tool = LogTool(logger=self.logger) result = tool.execute(function_name="non_existent_log_function") self.assertIn("Unknown function: non_existent_log_function", result) def test_clear_method(self): tool = LogTool(logger=self.logger) # Set a specific level for the logger for this test if needed to capture debug original_level = tool.logger.level tool.logger.setLevel(logging.DEBUG) with self.assertLogs(tool.logger, level='DEBUG') as cm: tool.clear() self.assertTrue(any("LogTool clear called" in message for message in cm.output)) tool.logger.setLevel(original_level) # Reset level if __name__ == '__main__': unittest.main()