amq-adapter-python/tests/adapter/test_logging_utils.py

import os
from importlib.metadata import version
from unittest.mock import patch

from amqp.adapter.logging_utils import (
    get_context_info,
    logging_debug,
    logging_error,
    logging_info,
    logging_warning,
)

__version__ = version("amq_adapter")  # Name must match pyproject.toml [project] name
__verbose__ = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1"


class TestLoggingFunctions:
    """
    A class containing pytest unit tests for the logging functions.
    """

    @patch("logging.error")
    def test_logging_error_with_exception(self, mock_logger):
        """
        Test logging_error function when called within an exception handler.
        """
        try:
            raise ValueError("Test error")
        except ValueError as ve:
            logging_error("An error occurred: %s", ve)
        mock_logger.assert_called_once()
        # Assert that the traceback information is included in the log message
        assert len(mock_logger.call_args_list) == 1
        _val = mock_logger.call_args_list[0][0][0]
        assert "[E] 'An error occurred: %s'" in _val
        assert "test_logging_error_with_exception" in _val

    @patch("logging.error")
    def test_logging_error_without_exception(self, mock_logger):
        """
        Test logging_error function when called without an active exception.
        """
        logging_error("A regular error")
        mock_logger.assert_called_once()
        assert "A regular error" in mock_logger.call_args_list[0][0][0]

    @patch("logging.warning")
    def test_logging_warning_verbose_on(self, mock_logger):
        """
        Test logging_warning function when verbose logging is enabled.
        """
        with patch(
            "amqp.adapter.logging_utils.__verbose__", True
        ):  # Simulate verbose logging
            logging_warning("A warning message")
        mock_logger.assert_called_once()
        assert "A warning message" in mock_logger.call_args_list[0][0][0]

    @patch("logging.warning")
    def test_logging_warning_verbose_off(self, mock_logger):
        """
        Test logging_warning function when verbose logging is disabled.
        """
        with patch(
            "amqp.adapter.logging_utils.__verbose__", False
        ):  # Simulate non-verbose logging
            logging_warning("A warning message")
        mock_logger.assert_called_once()
        assert "A warning message" in mock_logger.call_args_list[0][0][0]

    @patch("logging.info")
    def test_logging_info_verbose_on(self, mock_logger):
        """
        Test logging_info function with verbose logging enabled.
        """
        with patch("amqp.adapter.logging_utils.__verbose__", True):
            logging_info("An info message")
        mock_logger.assert_called_once()
        assert "An info message" in mock_logger.call_args_list[0][0][0]

    @patch("logging.info")
    def test_logging_info_verbose_off(self, mock_logger):
        """
        Test logging_info function with verbose logging disabled.
        """
        with patch("amqp.adapter.logging_utils.__verbose__", False):
            logging_info("An info message")
        mock_logger.assert_called_once()
        assert "An info message" in mock_logger.call_args_list[0][0][0]

    @patch("logging.debug")
    def test_logging_debug(self, mock_logger):
        """
        Test logging_debug function.
        """
        logging_debug("A debug message")
        mock_logger.assert_called_once()
        assert "A debug message" in mock_logger.call_args_list[0][0][0]

    def test_get_context_info_normal(self):
        """
        Test get_context_info function in a normal function call.
        """

        def dummy_function():
            return get_context_info()

        module_name, line_number, function_name = dummy_function()
        assert (
            "test_get_context_info_normal" in function_name
        )  # changed from co_name to function name
        assert (
            "test_logging_utils.py" in module_name
        )  # changed from None to test_unit.py
        assert isinstance(line_number, int)

    def test_get_context_info_no_frame(self):
        """
        Test get_context_info function when no frame is available.
        """
        # Simulate a situation where inspect.currentframe() returns None (which is hard to do directly)
        with patch("inspect.currentframe") as mock_frame:
            mock_frame.return_value = None
            module_name, line_number, function_name = get_context_info()
            assert module_name is None
            assert line_number is None
            assert function_name is None

    def test_get_context_info_exception(self):
        """
        Test get_context_info function when an exception occurs.
        """
        with patch("inspect.currentframe") as mock_frame:
            mock_frame.side_effect = Exception("Simulated exception")
            module_name, line_number, function_name = get_context_info()
            assert module_name is None
            assert line_number is None
            assert function_name is None