V5.0.1

Author: ddc

tests/context_management/test_context_managers.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 import logging
 import os
 import sys
@@ -20,7 +19,10 @@
 )
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Windows file locking issues with TemporaryDirectory - see equivalent Windows-specific test file")
+@pytest.mark.skipif(
+    sys.platform == "win32",
+    reason="Windows file locking issues with TemporaryDirectory - see equivalent Windows-specific test file",
+)
 class TestContextManagers:
     """Test context manager functionality for resource management."""
 
@@ -35,7 +37,7 @@ def setup_temp_dir(self):
             self.temp_dir = temp_dir
             self.log_file = "test.log"
             yield
-        
+
         # Clear registry after each test
         clear_logger_registry()
 
@@ -64,7 +66,7 @@ def test_size_rotating_context_manager(self):
             directory=self.temp_dir,
             filenames=[self.log_file],
             maxmbytes=1,
-            daystokeep=2
+            daystokeep=2,
         ) as logger:
             assert isinstance(logger, logging.Logger)
             assert logger.name == logger_name
@@ -91,7 +93,7 @@ def test_timed_rotating_context_manager(self):
             directory=self.temp_dir,
             filenames=[self.log_file],
             when=RotateWhen.HOURLY.value,
-            daystokeep=3
+            daystokeep=3,
         ) as logger:
             assert isinstance(logger, logging.Logger)
             assert logger.name == logger_name
@@ -180,7 +182,10 @@ def test_stream_handler_cleanup(self):
         logger_name = "test_stream_cleanup"
 
         with SizeRotatingLog(
-            name=logger_name, directory=self.temp_dir, filenames=[self.log_file], streamhandler=True
+            name=logger_name,
+            directory=self.temp_dir,
+            filenames=[self.log_file],
+            streamhandler=True,
             # Enable stream handler
         ) as logger:
             # Should have both file and stream handlers
@@ -217,26 +222,26 @@ def test_nested_context_managers(self):
     def test_shutdown_logger(self):
         """Test shutdown_logger functionality."""
         logger_name = "test_shutdown_logger"
-        
+
         # Create a logger using factory (with registry caching)
         logger = LoggerFactory.get_or_create_logger("basic", name=logger_name, level=LogLevel.INFO.value)
-        
+
         # Verify logger is in registry
         assert logger_name in LoggerFactory._logger_registry
-        
+
         # Add some handlers
         handler = logging.StreamHandler()
         logger.addHandler(handler)
-        
+
         # Verify handler is attached
         assert len(logger.handlers) > 0
-        
+
         # Shutdown the specific logger
         LoggerFactory.shutdown_logger(logger_name)
-        
+
         # Verify logger handlers are cleaned up
         assert len(logger.handlers) == 0
-        
+
         # Verify logger is removed from registry
         assert logger_name not in LoggerFactory._logger_registry
 

tests/context_management/test_resource_management.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 import concurrent.futures
 import gc
 import logging
@@ -14,17 +13,21 @@
 
 import pytest
 from pythonLogs import (
-    LoggerFactory, basic_logger,
+    LoggerFactory,
+    basic_logger,
     size_rotating_logger,
     clear_logger_registry,
     shutdown_logger,
     get_registered_loggers,
-    LogLevel
+    LogLevel,
 )
 from pythonLogs.basic_log import BasicLog
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Windows file locking issues with TemporaryDirectory - see equivalent Windows-specific test file")
+@pytest.mark.skipif(
+    sys.platform == "win32",
+    reason="Windows file locking issues with TemporaryDirectory - see equivalent Windows-specific test file",
+)
 class TestResourceManagement:
     """Test resource management functionality."""
 
@@ -39,7 +42,7 @@ def setup_temp_dir(self):
             self.temp_dir = temp_dir
             self.log_file = "resource_test.log"
             yield
-        
+
         # Clear registry after each test
         clear_logger_registry()
 
@@ -52,7 +55,7 @@ def test_factory_registry_cleanup(self):
             name=logger_name,
             directory=self.temp_dir,
             filenames=[self.log_file],
-            maxmbytes=1
+            maxmbytes=1,
         )
 
         # Add to registry
@@ -156,7 +159,7 @@ def test_registry_clear_with_file_handlers(self):
             directory=self.temp_dir,
             filenames=[self.log_file, "second.log"],
             maxmbytes=1,
-            streamhandler=True  # Add stream handler too
+            streamhandler=True,  # Add stream handler too
         )
 
         # Add to registry
@@ -191,7 +194,7 @@ def test_resource_cleanup_performance(self):
                 name=name,
                 directory=self.temp_dir,
                 filenames=[f"{name}.log"],
-                maxmbytes=1
+                maxmbytes=1,
             )
             LoggerFactory._logger_registry[name] = (logger, time.time())
 
@@ -223,7 +226,7 @@ def test_memory_usage_after_cleanup(self):
             name=logger_name,
             directory=self.temp_dir,
             filenames=[self.log_file],
-            maxmbytes=1
+            maxmbytes=1,
         )
 
         # Add to registry
@@ -259,31 +262,31 @@ def create_and_cleanup_logger(index):
             """Create a logger and immediately clean it up."""
             logger_name = f"concurrent_test_{index}"
             logger = basic_logger(name=logger_name, level=LogLevel.INFO.value)
-            
+
             # Add to registry
             LoggerFactory._logger_registry[logger_name] = (logger, time.time())
-            
+
             # Small delay to increase chance of concurrent access
             time.sleep(0.01)
-            
+
             # Shutdown this specific logger
             return shutdown_logger(logger_name)
-        
+
         # Create multiple threads doing concurrent operations
         num_threads = 5
         with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
             futures = []
             for i in range(num_threads):
                 futures.append(executor.submit(create_and_cleanup_logger, i))
-            
+
             # Wait for all to complete
             results = []
             for future in concurrent.futures.as_completed(futures):
                 results.append(future.result())
-        
+
         # All operations should succeed
         assert all(results)
-        
+
         # The Registry should be empty
         assert len(get_registered_loggers()) == 0
 

tests/core/test_basic_log.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 import logging
 import os
 import sys
@@ -43,7 +42,7 @@ def test_basic_log_initialization_with_params(self):
             encoding="utf-8",
             datefmt="%Y-%m-%d %H:%M:%S",
             timezone="UTC",
-            showlocation=True
+            showlocation=True,
         )
         assert basic_log.level == logging.DEBUG
         assert basic_log.appname == "test_app"
@@ -56,7 +55,7 @@ def test_basic_log_init_method(self):
         """Test BasicLog init method creates logger."""
         basic_log = BasicLog(name="test_init", level=LogLevel.INFO.value)
         logger = basic_log.init()
-        
+
         assert isinstance(logger, logging.Logger)
         assert logger.name == "test_init"
         assert logger.level == logging.INFO
@@ -67,19 +66,19 @@ def test_basic_log_logger_functionality(self):
         """Test that BasicLog logger can log messages."""
         basic_log = BasicLog(name="test_logging", level=LogLevel.INFO.value)
         logger = basic_log.init()
-        
+
         # Test logging at different levels
         logger.info("Test info message")
         logger.warning("Test warning message")
         logger.error("Test error message")
-        
+
         # Should not raise any exceptions
 
     def test_basic_log_level_filtering(self):
         """Test that BasicLog respects log level filtering."""
         basic_log = BasicLog(name="test_level", level=LogLevel.WARNING.value)
         logger = basic_log.init()
-        
+
         assert logger.level == logging.WARNING
         assert logger.isEnabledFor(logging.WARNING)
         assert logger.isEnabledFor(logging.ERROR)
@@ -90,12 +89,12 @@ def test_basic_log_cleanup_static_method(self):
         """Test BasicLog static cleanup method."""
         basic_log = BasicLog(name="test_cleanup", level=LogLevel.INFO.value)
         logger = basic_log.init()
-        
+
         # Add a handler
         handler = logging.StreamHandler()
         logger.addHandler(handler)
         assert len(logger.handlers) > 0
-        
+
         # Clean up using static method
         BasicLog.cleanup_logger(logger)
         assert len(logger.handlers) == 0
@@ -104,12 +103,12 @@ def test_basic_log_instance_cleanup_method(self):
         """Test BasicLog instance cleanup method."""
         basic_log = BasicLog(name="test_instance_cleanup", level=LogLevel.INFO.value)
         logger = basic_log.init()
-        
+
         # Add a handler
         handler = logging.StreamHandler()
         logger.addHandler(handler)
         assert len(logger.handlers) > 0
-        
+
         # Clean up using instance method
         basic_log._cleanup_logger(logger)
         assert len(logger.handlers) == 0
@@ -118,11 +117,11 @@ def test_basic_log_thread_safety(self):
         """Test BasicLog thread safety with concurrent operations."""
         import threading
         import time
-        
+
         basic_log = BasicLog(name="test_thread_safety", level=LogLevel.INFO.value)
         logger = basic_log.init()
         results = []
-        
+
         def log_messages(thread_id):
             try:
                 for i in range(10):
@@ -131,18 +130,18 @@ def log_messages(thread_id):
                 results.append(f"thread_{thread_id}_success")
             except Exception as e:
                 results.append(f"thread_{thread_id}_error_{e}")
-        
+
         # Create multiple threads
         threads = []
         for i in range(5):
             thread = threading.Thread(target=log_messages, args=(i,))
             threads.append(thread)
             thread.start()
-        
+
         # Wait for all threads to complete
         for thread in threads:
             thread.join()
-        
+
         # All threads should complete successfully
         assert len(results) == 5
         assert all("success" in result for result in results)
@@ -156,27 +155,27 @@ def test_basic_log_different_log_levels(self):
             (LogLevel.ERROR.value, logging.ERROR),
             (LogLevel.CRITICAL.value, logging.CRITICAL),
         ]
-        
+
         for level_str, level_int in log_levels:
             basic_log = BasicLog(name=f"test_{level_str}", level=level_str)
             logger = basic_log.init()
             assert logger.level == level_int
-            
+
             # Clean up
             BasicLog.cleanup_logger(logger)
 
     def test_basic_log_multiple_instances(self):
         """Test creating multiple BasicLog instances."""
         loggers = []
-        
+
         for i in range(5):
             basic_log = BasicLog(name=f"test_multi_{i}", level=LogLevel.INFO.value)
             logger = basic_log.init()
             loggers.append((basic_log, logger))
-            
+
             assert logger.name == f"test_multi_{i}"
             assert logger.level == logging.INFO
-        
+
         # Clean up all loggers
         for basic_log, logger in loggers:
             BasicLog.cleanup_logger(logger)