147 lines
7.3 KiB
Python
147 lines
7.3 KiB
Python
|
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()
|