🔍 Code Extractor

Unit test class for testing logging utilities including InvoiceExtractionLogger, PerformanceLogger, and get_logger function.

Purpose

This test class validates the functionality of logging utilities used in an invoice extraction system. It tests logger initialization, configuration from environment variables, file logging, JSON formatted logging, request ID correlation, performance logging, and logger retrieval. Each test method sets up a clean logging environment, executes specific logging scenarios, and verifies expected behavior through assertions.

Source Code

class TestLoggingUtils(unittest.TestCase):
    """Test cases for the logging utilities."""
    
    def setUp(self):
        """Set up test environment before each test."""
        # Reset the logging configuration before each test
        logging.root.handlers = []
        logging.root.setLevel(logging.INFO)
    
    def tearDown(self):
        """Clean up after each test."""
        # Reset logger after each test
        logging.disable(logging.NOTSET)
        logging.root.handlers = []
    
    def test_invoice_extraction_logger_init(self):
        """Test initialization of InvoiceExtractionLogger."""
        config = {
            'log_level': 'DEBUG',
            'log_to_file': False
        }
        
        logger = InvoiceExtractionLogger(config)
        
        # Check log level
        self.assertEqual(logger.log_level, logging.DEBUG)
        self.assertFalse(logger.log_to_file)
        
        # Check that root logger was configured
        self.assertEqual(logging.getLogger().level, logging.DEBUG)
        self.assertEqual(len(logging.getLogger().handlers), 1)
    
    @patch.dict(os.environ, {'INVOICE_EXTRACTION_LOG_LEVEL': 'ERROR'})
    def test_log_level_from_env(self):
        """Test getting log level from environment variables."""
        config = {
            'log_level': 'DEBUG',  # This should be overridden by env var
            'log_to_file': False
        }
        
        logger = InvoiceExtractionLogger(config)
        
        # Log level should be from environment variable
        self.assertEqual(logger.log_level, logging.ERROR)
        self.assertEqual(logging.getLogger().level, logging.ERROR)
    
    def test_file_logging(self):
        """Test logging to file."""
        # Use a temporary directory for log files
        with tempfile.TemporaryDirectory() as temp_dir:
            config = {
                'log_level': 'INFO',
                'log_to_file': True,
                'log_dir': temp_dir,
                'log_file': 'test_log.log'
            }
            
            logger = InvoiceExtractionLogger(config)
            
            # Log a test message
            logging.info("Test log message")
            
            # Check if file exists and contains the message
            log_file_path = os.path.join(temp_dir, 'test_log.log')
            self.assertTrue(os.path.exists(log_file_path))
            
            with open(log_file_path, 'r') as f:
                log_content = f.read()
                self.assertIn("Test log message", log_content)
    
    def test_json_logging(self):
        """Test JSON formatted logs."""
        # Capture stdout to check JSON format
        captured_output = io.StringIO()
        sys.stdout = captured_output
        
        try:
            config = {
                'log_level': 'INFO',
                'log_to_file': False,
                'json_logs': True
            }
            
            logger = InvoiceExtractionLogger(config)
            
            # Log a test message
            logging.info("Test JSON log")
            
            # Get output
            log_output = captured_output.getvalue()
            
            # Try to parse as JSON
            try:
                log_data = json.loads(log_output.strip())
                self.assertEqual(log_data['level'], 'INFO')
                self.assertEqual(log_data['message'], 'Test JSON log')
            except json.JSONDecodeError:
                self.fail("Log output was not valid JSON")
        finally:
            sys.stdout = sys.__stdout__  # Restore stdout
    
    def test_request_id_correlation(self):
        """Test request ID correlation in logs."""
        # Capture stdout to check for correlation ID
        captured_output = io.StringIO()
        sys.stdout = captured_output
        
        try:
            config = {
                'log_level': 'INFO',
                'log_to_file': False,
                'json_logs': True
            }
            
            logger = InvoiceExtractionLogger(config)
            logger.set_request_id("test-request-123")
            
            # Log a test message
            logging.info("Test correlated log")
            
            # Get output
            log_output = captured_output.getvalue()
            
            # Try to parse as JSON
            try:
                log_data = json.loads(log_output.strip())
                self.assertEqual(log_data['correlation_id'], 'test-request-123')
            except json.JSONDecodeError:
                self.fail("Log output was not valid JSON")
        finally:
            sys.stdout = sys.__stdout__  # Restore stdout
    
    def test_performance_logger(self):
        """Test PerformanceLogger for measuring execution time."""
        # Reset the logging configuration
        logging.root.handlers = []
        
        # Capture log output
        log_capture = io.StringIO()
        handler = logging.StreamHandler(log_capture)
        logging.getLogger().addHandler(handler)
        logging.getLogger().setLevel(logging.INFO)
        
        # Use PerformanceLogger with a controlled delay
        with PerformanceLogger("test_operation") as perf:
            # Add some custom metrics
            perf.add_metric("test_count", 5)
            perf.add_metric("status", "success")
            
            # Simulate work
            datetime.now()  # Just to ensure some minimal time passes
        
        # Get log output
        log_output = log_capture.getvalue()
        
        # Check that log contains performance data
        self.assertIn("Performance: test_operation completed in", log_output)
        self.assertIn("test_count", log_output)
        self.assertIn("status", log_output)
    
    def test_get_logger(self):
        """Test get_logger utility function."""
        # Should create logging setup if none exists
        logger = get_logger("test_module")
        
        # Check logger name
        self.assertEqual(logger.name, "test_module")
        
        # Check that root logger has handlers
        self.assertGreater(len(logging.getLogger().handlers), 0)
        
        # Check that new loggers use the same configuration
        another_logger = get_logger("another_module")
        self.assertEqual(another_logger.level, logger.level)

Parameters

Name	Type	Default	Kind
bases	unittest.TestCase	-

Parameter Details

bases: Inherits from unittest.TestCase to provide testing framework functionality including setUp, tearDown, and assertion methods

Return Value

As a test class, it does not return values. Each test method performs assertions that either pass (no return) or fail (raises AssertionError). The unittest framework collects and reports test results.

Class Interface

Methods

Dependencies

• unittest
• logging
• json
• os
• sys
• io
• datetime
• tempfile
• requests

Required Imports

import unittest
from unittest.mock import patch, MagicMock, ANY
import logging
import json
import os
import sys
import io
from datetime import datetime, timedelta
import tempfile
import requests
from utils.llm_client import LLMClient
from utils.logging_utils import InvoiceExtractionLogger, PerformanceLogger, get_logger

Usage Example

import unittest
from test_logging_utils import TestLoggingUtils

# Run all tests in the class
suite = unittest.TestLoader().loadTestsFromTestCase(TestLoggingUtils)
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)

# Run a specific test
suite = unittest.TestSuite()
suite.addTest(TestLoggingUtils('test_invoice_extraction_logger_init'))
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)

# Run from command line
# python -m unittest test_logging_utils.TestLoggingUtils
# python -m unittest test_logging_utils.TestLoggingUtils.test_file_logging

Best Practices

• Each test method is independent and isolated through setUp and tearDown methods
• setUp resets logging configuration before each test to ensure clean state
• tearDown cleans up logging handlers and disables logging after each test
• Tests use temporary directories for file operations to avoid polluting the filesystem
• stdout is captured and restored properly in tests that check console output
• Environment variable mocking uses @patch.dict decorator for clean isolation
• Tests verify both positive cases (expected behavior) and edge cases (environment overrides)
• File logging tests use context managers (with statements) for proper resource cleanup
• JSON parsing tests include try-except blocks with explicit failure messages
• Tests should be run in isolation or as part of a test suite, not imported for production use

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class InvoiceExtractionLogger 73.7% similar

  A comprehensive logging configuration class for invoice extraction systems that provides console and file logging with optional JSON formatting, request tracking via correlation IDs, and configurable log levels.

  From: /tf/active/vicechatdev/invoice_extraction/utils/logging_utils.py

• class TestAUExtractor 65.1% similar

  Unit test class for testing the AUExtractor class, which extracts data from Australian invoices including ABN, GST, and payment details.

  From: /tf/active/vicechatdev/invoice_extraction/tests/test_extractors.py

• class TestUKExtractor 64.9% similar

  Unit test class for testing the UKExtractor class, which extracts structured data from UK invoices including VAT numbers, dates, amounts, and line items.

  From: /tf/active/vicechatdev/invoice_extraction/tests/test_extractors.py

• class TestBEExtractor 63.3% similar

  Unit test class for testing the BEExtractor class, which extracts structured data from Belgian invoices using LLM-based extraction.

  From: /tf/active/vicechatdev/invoice_extraction/tests/test_extractors.py

• class TestAUValidator 60.5% similar

  Unit test class for validating the AUValidator class, which validates Australian invoice extraction results including ABN, GST, banking details, and tax invoice requirements.

  From: /tf/active/vicechatdev/invoice_extraction/tests/test_validators.py

class TestLoggingUtils(unittest.TestCase):
    """Test cases for the logging utilities."""
    
    def setUp(self):
        """Set up test environment before each test."""
        # Reset the logging configuration before each test
        logging.root.handlers = []
        logging.root.setLevel(logging.INFO)
    
    def tearDown(self):
        """Clean up after each test."""
        # Reset logger after each test
        logging.disable(logging.NOTSET)
        logging.root.handlers = []
    
    def test_invoice_extraction_logger_init(self):
        """Test initialization of InvoiceExtractionLogger."""
        config = {
            'log_level': 'DEBUG',
            'log_to_file': False
        }
        
        logger = InvoiceExtractionLogger(config)
        
        # Check log level
        self.assertEqual(logger.log_level, logging.DEBUG)
        self.assertFalse(logger.log_to_file)
        
        # Check that root logger was configured
        self.assertEqual(logging.getLogger().level, logging.DEBUG)
        self.assertEqual(len(logging.getLogger().handlers), 1)
    
    @patch.dict(os.environ, {'INVOICE_EXTRACTION_LOG_LEVEL': 'ERROR'})
    def test_log_level_from_env(self):
        """Test getting log level from environment variables."""
        config = {
            'log_level': 'DEBUG',  # This should be overridden by env var
            'log_to_file': False
        }
        
        logger = InvoiceExtractionLogger(config)
        
        # Log level should be from environment variable
        self.assertEqual(logger.log_level, logging.ERROR)
        self.assertEqual(logging.getLogger().level, logging.ERROR)
    
    def test_file_logging(self):
        """Test logging to file."""
        # Use a temporary directory for log files
        with tempfile.TemporaryDirectory() as temp_dir:
            config = {
                'log_level': 'INFO',
                'log_to_file': True,
                'log_dir': temp_dir,
                'log_file': 'test_log.log'
            }
            
            logger = InvoiceExtractionLogger(config)
            
            # Log a test message
            logging.info("Test log message")
            
            # Check if file exists and contains the message
            log_file_path = os.path.join(temp_dir, 'test_log.log')
            self.assertTrue(os.path.exists(log_file_path))
            
            with open(log_file_path, 'r') as f:
                log_content = f.read()
                self.assertIn("Test log message", log_content)
    
    def test_json_logging(self):
        """Test JSON formatted logs."""
        # Capture stdout to check JSON format
        captured_output = io.StringIO()
        sys.stdout = captured_output
        
        try:
            config = {
                'log_level': 'INFO',
                'log_to_file': False,
                'json_logs': True
            }
            
            logger = InvoiceExtractionLogger(config)
            
            # Log a test message
            logging.info("Test JSON log")
            
            # Get output
            log_output = captured_output.getvalue()
            
            # Try to parse as JSON
            try:
                log_data = json.loads(log_output.strip())
                self.assertEqual(log_data['level'], 'INFO')
                self.assertEqual(log_data['message'], 'Test JSON log')
            except json.JSONDecodeError:
                self.fail("Log output was not valid JSON")
        finally:
            sys.stdout = sys.__stdout__  # Restore stdout
    
    def test_request_id_correlation(self):
        """Test request ID correlation in logs."""
        # Capture stdout to check for correlation ID
        captured_output = io.StringIO()
        sys.stdout = captured_output
        
        try:
            config = {
                'log_level': 'INFO',
                'log_to_file': False,
                'json_logs': True
            }
            
            logger = InvoiceExtractionLogger(config)
            logger.set_request_id("test-request-123")
            
            # Log a test message
            logging.info("Test correlated log")
            
            # Get output
            log_output = captured_output.getvalue()
            
            # Try to parse as JSON
            try:
                log_data = json.loads(log_output.strip())
                self.assertEqual(log_data['correlation_id'], 'test-request-123')
            except json.JSONDecodeError:
                self.fail("Log output was not valid JSON")
        finally:
            sys.stdout = sys.__stdout__  # Restore stdout
    
    def test_performance_logger(self):
        """Test PerformanceLogger for measuring execution time."""
        # Reset the logging configuration
        logging.root.handlers = []
        
        # Capture log output
        log_capture = io.StringIO()
        handler = logging.StreamHandler(log_capture)
        logging.getLogger().addHandler(handler)
        logging.getLogger().setLevel(logging.INFO)
        
        # Use PerformanceLogger with a controlled delay
        with PerformanceLogger("test_operation") as perf:
            # Add some custom metrics
            perf.add_metric("test_count", 5)
            perf.add_metric("status", "success")
            
            # Simulate work
            datetime.now()  # Just to ensure some minimal time passes
        
        # Get log output
        log_output = log_capture.getvalue()
        
        # Check that log contains performance data
        self.assertIn("Performance: test_operation completed in", log_output)
        self.assertIn("test_count", log_output)
        self.assertIn("status", log_output)
    
    def test_get_logger(self):
        """Test get_logger utility function."""
        # Should create logging setup if none exists
        logger = get_logger("test_module")
        
        # Check logger name
        self.assertEqual(logger.name, "test_module")
        
        # Check that root logger has handlers
        self.assertGreater(len(logging.getLogger().handlers), 0)
        
        # Check that new loggers use the same configuration
        another_logger = get_logger("another_module")
        self.assertEqual(another_logger.level, logger.level)