mirror of
https://github.com/varun-r-mallya/py-libp2p.git
synced 2025-12-31 20:36:24 +00:00
283 lines
8.4 KiB
Python
283 lines
8.4 KiB
Python
import logging
|
|
import logging.handlers
|
|
import os
|
|
from pathlib import (
|
|
Path,
|
|
)
|
|
import queue
|
|
import tempfile
|
|
import threading
|
|
from unittest.mock import (
|
|
patch,
|
|
)
|
|
|
|
import pytest
|
|
import trio
|
|
|
|
from libp2p.utils.logging import (
|
|
_current_handlers,
|
|
_current_listener,
|
|
_listener_ready,
|
|
log_queue,
|
|
setup_logging,
|
|
)
|
|
|
|
|
|
def _reset_logging():
|
|
"""Reset all logging state."""
|
|
global _current_listener, _listener_ready, _current_handlers
|
|
|
|
# Stop existing listener if any
|
|
if _current_listener is not None:
|
|
_current_listener.stop()
|
|
_current_listener = None
|
|
|
|
# Close all file handlers to ensure proper cleanup on Windows
|
|
for handler in _current_handlers:
|
|
if isinstance(handler, logging.FileHandler):
|
|
handler.close()
|
|
_current_handlers.clear()
|
|
|
|
# Reset the event
|
|
_listener_ready = threading.Event()
|
|
|
|
# Reset the root logger
|
|
root = logging.getLogger()
|
|
root.handlers.clear()
|
|
root.setLevel(logging.WARNING)
|
|
|
|
# Clear all loggers
|
|
for name in list(logging.Logger.manager.loggerDict.keys()):
|
|
if name.startswith("libp2p"):
|
|
del logging.Logger.manager.loggerDict[name] # Remove logger from registry
|
|
|
|
# Reset libp2p logger
|
|
logger = logging.getLogger("libp2p")
|
|
logger.handlers.clear()
|
|
logger.propagate = False # Don't propagate to Python's root logger
|
|
logger.setLevel(logging.WARNING) # Default level
|
|
|
|
# Clear the log queue
|
|
while not log_queue.empty():
|
|
try:
|
|
log_queue.get_nowait()
|
|
except queue.Empty:
|
|
break
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_env():
|
|
"""Remove relevant environment variables before each test."""
|
|
# Save original environment
|
|
original_env = {}
|
|
for var in ["LIBP2P_DEBUG", "LIBP2P_DEBUG_FILE"]:
|
|
if var in os.environ:
|
|
original_env[var] = os.environ[var]
|
|
del os.environ[var]
|
|
|
|
# Reset logging state
|
|
_reset_logging()
|
|
|
|
yield
|
|
|
|
# Reset logging state again
|
|
_reset_logging()
|
|
|
|
# Restore original environment
|
|
for var, value in original_env.items():
|
|
os.environ[var] = value
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_logger():
|
|
"""Reset the libp2p logger before each test."""
|
|
# Reset logging state
|
|
_reset_logging()
|
|
|
|
yield logging.getLogger("libp2p")
|
|
|
|
# Reset logging state again
|
|
_reset_logging()
|
|
|
|
|
|
def test_logging_disabled(clean_env):
|
|
"""Test that logging is disabled when LIBP2P_DEBUG is not set."""
|
|
setup_logging()
|
|
logger = logging.getLogger("libp2p")
|
|
assert logger.level == logging.WARNING
|
|
assert not logger.handlers
|
|
|
|
|
|
def test_logging_with_debug_env(clean_env):
|
|
"""Test that logging is properly configured when LIBP2P_DEBUG is set."""
|
|
os.environ["LIBP2P_DEBUG"] = "DEBUG"
|
|
setup_logging()
|
|
logger = logging.getLogger("libp2p")
|
|
|
|
assert logger.level == logging.DEBUG
|
|
assert len(logger.handlers) == 1 # Should have the QueueHandler
|
|
|
|
# The handler should be a QueueHandler
|
|
assert isinstance(logger.handlers[0], logging.handlers.QueueHandler)
|
|
|
|
|
|
def test_module_specific_logging(clean_env):
|
|
"""Test module-specific logging levels."""
|
|
os.environ["LIBP2P_DEBUG"] = "identity.identify:DEBUG,transport:INFO"
|
|
setup_logging()
|
|
|
|
# Check root logger (should be at INFO by default)
|
|
root_logger = logging.getLogger("libp2p")
|
|
assert root_logger.level == logging.INFO
|
|
|
|
# Check identify module (should be at DEBUG)
|
|
identify_logger = logging.getLogger("libp2p.identity.identify")
|
|
assert identify_logger.level == logging.DEBUG
|
|
|
|
# Check transport module (should be at INFO)
|
|
transport_logger = logging.getLogger("libp2p.transport")
|
|
assert transport_logger.level == logging.INFO
|
|
|
|
# Check unspecified module (should inherit from root)
|
|
other_logger = logging.getLogger("libp2p.other")
|
|
assert other_logger.getEffectiveLevel() == logging.INFO
|
|
|
|
|
|
def test_global_logging_level(clean_env):
|
|
"""Test setting a global logging level."""
|
|
os.environ["LIBP2P_DEBUG"] = "DEBUG"
|
|
setup_logging()
|
|
|
|
# Root logger should be at DEBUG
|
|
root_logger = logging.getLogger("libp2p")
|
|
assert root_logger.level == logging.DEBUG
|
|
|
|
# All modules should inherit the DEBUG level
|
|
for module in ["identity", "transport", "other"]:
|
|
logger = logging.getLogger(f"libp2p.{module}")
|
|
assert logger.getEffectiveLevel() == logging.DEBUG
|
|
|
|
|
|
@pytest.mark.trio
|
|
async def test_custom_log_file(clean_env):
|
|
"""Test logging to a custom file path."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
log_file = Path(temp_dir) / "test.log"
|
|
os.environ["LIBP2P_DEBUG"] = "INFO"
|
|
os.environ["LIBP2P_DEBUG_FILE"] = str(log_file)
|
|
|
|
setup_logging()
|
|
|
|
# Wait for the listener to be ready
|
|
_listener_ready.wait(timeout=1)
|
|
|
|
logger = logging.getLogger("libp2p")
|
|
logger.info("Test message")
|
|
|
|
# Give the listener time to process the message
|
|
await trio.sleep(0.1)
|
|
|
|
# Stop the listener to ensure all messages are written
|
|
if _current_listener is not None:
|
|
_current_listener.stop()
|
|
|
|
# Give a moment for the listener to fully stop
|
|
await trio.sleep(0.05)
|
|
|
|
# Close all file handlers to release the file
|
|
for handler in _current_handlers:
|
|
if isinstance(handler, logging.FileHandler):
|
|
handler.flush() # Ensure all writes are flushed
|
|
handler.close()
|
|
|
|
# Check if the file exists and contains our message
|
|
assert log_file.exists()
|
|
content = log_file.read_text()
|
|
assert "Test message" in content
|
|
|
|
|
|
@pytest.mark.trio
|
|
async def test_default_log_file(clean_env):
|
|
"""Test logging to the default file path."""
|
|
os.environ["LIBP2P_DEBUG"] = "INFO"
|
|
|
|
with patch("libp2p.utils.paths.create_temp_file") as mock_create_temp:
|
|
# Mock the temp file creation to return a predictable path
|
|
mock_temp_file = (
|
|
Path(tempfile.gettempdir()) / "test_py-libp2p_20240101_120000.log"
|
|
)
|
|
mock_create_temp.return_value = mock_temp_file
|
|
|
|
# Remove the log file if it exists
|
|
mock_temp_file.unlink(missing_ok=True)
|
|
|
|
setup_logging()
|
|
|
|
# Wait for the listener to be ready
|
|
_listener_ready.wait(timeout=1)
|
|
|
|
logger = logging.getLogger("libp2p")
|
|
logger.info("Test message")
|
|
|
|
# Give the listener time to process the message
|
|
await trio.sleep(0.1)
|
|
|
|
# Stop the listener to ensure all messages are written
|
|
if _current_listener is not None:
|
|
_current_listener.stop()
|
|
|
|
# Give a moment for the listener to fully stop
|
|
await trio.sleep(0.05)
|
|
|
|
# Close all file handlers to release the file
|
|
for handler in _current_handlers:
|
|
if isinstance(handler, logging.FileHandler):
|
|
handler.flush() # Ensure all writes are flushed
|
|
handler.close()
|
|
|
|
# Check the mocked temp file
|
|
if mock_temp_file.exists():
|
|
content = mock_temp_file.read_text()
|
|
assert "Test message" in content
|
|
|
|
|
|
def test_invalid_log_level(clean_env):
|
|
"""Test handling of invalid log level in LIBP2P_DEBUG."""
|
|
os.environ["LIBP2P_DEBUG"] = "INVALID_LEVEL"
|
|
setup_logging()
|
|
logger = logging.getLogger("libp2p")
|
|
|
|
# Should default to WARNING when invalid level is provided
|
|
assert logger.level == logging.WARNING
|
|
|
|
|
|
def test_invalid_module_format(clean_env):
|
|
"""Test handling of invalid module format in LIBP2P_DEBUG."""
|
|
os.environ["LIBP2P_DEBUG"] = "identity.identify:DEBUG,invalid_format"
|
|
setup_logging()
|
|
|
|
# The valid module should be configured
|
|
identify_logger = logging.getLogger("libp2p.identity.identify")
|
|
assert identify_logger.level == logging.DEBUG
|
|
|
|
|
|
def test_module_name_handling(clean_env):
|
|
"""Test various module name formats."""
|
|
# Test multiple formats in one go
|
|
os.environ["LIBP2P_DEBUG"] = (
|
|
"identity.identify:DEBUG," # No libp2p prefix
|
|
"libp2p.transport:INFO," # With libp2p prefix
|
|
"pubsub/gossipsub:WARN" # Using slash instead of dot
|
|
)
|
|
setup_logging()
|
|
|
|
# Check each logger
|
|
identify_logger = logging.getLogger("libp2p.identity.identify")
|
|
assert identify_logger.level == logging.DEBUG
|
|
|
|
transport_logger = logging.getLogger("libp2p.transport")
|
|
assert transport_logger.level == logging.INFO
|
|
|
|
gossipsub_logger = logging.getLogger("libp2p.pubsub.gossipsub")
|
|
assert gossipsub_logger.level == logging.WARNING
|