lovdata-chat/docker/scripts/test-structured-logging.py

#!/usr/bin/env python3
"""
Structured Logging Test Script

Tests the comprehensive logging infrastructure with structured logging,
request tracking, and proper log formatting.
"""

import os
import sys
import asyncio
import json
import time
import tempfile
from pathlib import Path

# Add session-manager to path for imports
sys.path.insert(0, str(Path(__file__).parent))

from logging_config import (
    setup_logging,
    get_logger,
    RequestContext,
    log_performance,
    log_request,
    log_session_operation,
    log_security_event,
    StructuredFormatter,
    HumanReadableFormatter,
)

# Set up logging
logger = get_logger(__name__)


async def test_log_formatters():
    """Test JSON and human-readable formatters."""
    print("🧪 Testing Log Formatters")
    print("=" * 50)

    # Test JSON formatter
    json_formatter = StructuredFormatter()
    log_record = logging.LogRecord(
        name="test.logger",
        level=logging.INFO,
        pathname="test.py",
        lineno=42,
        msg="Test message",
        args=(),
        exc_info=None,
    )
    log_record.request_id = "req-123"
    log_record.session_id = "ses-456"

    json_output = json_formatter.format(log_record)
    parsed = json.loads(json_output)

    # Verify JSON structure
    assert "timestamp" in parsed
    assert parsed["level"] == "INFO"
    assert parsed["message"] == "Test message"
    assert parsed["request_id"] == "req-123"
    assert parsed["session_id"] == "ses-456"

    print("✅ JSON formatter produces valid structured logs")

    # Test human-readable formatter
    human_formatter = HumanReadableFormatter()
    human_output = human_formatter.format(log_record)

    assert "Test message" in human_output
    assert "req-123" in human_output  # Request ID included
    print("✅ Human-readable formatter includes request context")

    return True


async def test_request_context():
    """Test request ID tracking and context management."""
    print("\n🆔 Testing Request Context")
    print("=" * 50)

    # Test context management
    with RequestContext("test-request-123") as ctx:
        assert RequestContext.get_current_request_id() == "test-request-123"
        print("✅ Request context properly set")

        # Nested context should override
        with RequestContext("nested-request-456"):
            assert RequestContext.get_current_request_id() == "nested-request-456"
            print("✅ Nested context overrides parent")

        # Should revert to parent
        assert RequestContext.get_current_request_id() == "test-request-123"
        print("✅ Context properly restored after nested context")

    # Should be cleared after context exit
    assert RequestContext.get_current_request_id() is None
    print("✅ Context properly cleared after exit")

    return True


async def test_log_levels_and_filtering():
    """Test different log levels and filtering."""
    print("\n📊 Testing Log Levels and Filtering")
    print("=" * 50)

    # Create a temporary log file
    with tempfile.NamedTemporaryFile(mode="w+", suffix=".log", delete=False) as f:
        log_file = f.name

    try:
        # Set up logging with file output
        test_logger = setup_logging(
            level="DEBUG",
            format_type="json",
            log_file=log_file,
            enable_console=False,
            enable_file=True,
        )

        # Log messages at different levels
        test_logger.debug("Debug message", extra={"test": "debug"})
        test_logger.info("Info message", extra={"test": "info"})
        test_logger.warning("Warning message", extra={"test": "warning"})
        test_logger.error("Error message", extra={"test": "error"})

        # Flush logs
        for handler in logging.getLogger().handlers:
            handler.flush()

        # Read and verify log file
        with open(log_file, "r") as f:
            lines = f.readlines()

        # Should have 4 log entries
        assert len(lines) == 4

        # Parse JSON and verify levels
        for line in lines:
            entry = json.loads(line)
            assert "level" in entry
            assert "message" in entry
            assert "timestamp" in entry

        # Verify specific messages
        messages = [json.loads(line)["message"] for line in lines]
        assert "Debug message" in messages
        assert "Info message" in messages
        assert "Warning message" in messages
        assert "Error message" in messages

        print("✅ All log levels captured correctly")
        print(f"✅ Log file contains {len(lines)} entries")

    finally:
        # Clean up
        if os.path.exists(log_file):
            os.unlink(log_file)

    return True


async def test_structured_logging_features():
    """Test structured logging features and extra fields."""
    print("\n🔧 Testing Structured Logging Features")
    print("=" * 50)

    # Test with temporary file
    with tempfile.NamedTemporaryFile(mode="w+", suffix=".log", delete=False) as f:
        log_file = f.name

    try:
        # Set up structured logging
        test_logger = setup_logging(
            level="INFO",
            format_type="json",
            log_file=log_file,
            enable_console=False,
            enable_file=True,
        )

        with RequestContext("req-abc-123"):
            # Test performance logging
            log_performance("test_operation", 150.5, user_id="user123", success=True)

            # Test request logging
            log_request("POST", "/api/test", 200, 45.2, user_agent="test-client")

            # Test session operation logging
            log_session_operation("session-xyz", "created", container_id="cont-123")

            # Test security event logging
            log_security_event(
                "authentication_success", "info", user_id="user123", ip="192.168.1.1"
            )

            # Flush logs
            for handler in logging.getLogger().handlers:
                handler.flush()

            # Read and analyze logs
            with open(log_file, "r") as f:
                lines = f.readlines()

            # Should have multiple entries
            assert len(lines) >= 4

            # Parse and verify structured data
            for line in lines:
                entry = json.loads(line)

                # All entries should have request_id
                assert "request_id" in entry
                assert entry["request_id"] == "req-abc-123"

                # Check for operation-specific fields
                if "operation" in entry:
                    if entry["operation"] == "test_operation":
                        assert "duration_ms" in entry
                        assert entry["duration_ms"] == 150.5
                        assert entry["user_id"] == "user123"
                    elif entry["operation"] == "http_request":
                        assert "method" in entry
                        assert "path" in entry
                        assert "status_code" in entry
                    elif entry["operation"] == "authentication_success":
                        assert "security_event" in entry
                        assert "severity" in entry

        print("✅ All structured logging features working correctly")
        print(f"✅ Generated {len(lines)} structured log entries")

    finally:
        if os.path.exists(log_file):
            os.unlink(log_file)

    return True


async def test_environment_configuration():
    """Test logging configuration from environment variables."""
    print("\n🌍 Testing Environment Configuration")
    print("=" * 50)

    # Save original environment
    original_env = {}
    test_vars = [
        ("LOG_LEVEL", "DEBUG"),
        ("LOG_FORMAT", "human"),
        ("LOG_FILE", "/tmp/test-env-log.log"),
        ("LOG_MAX_SIZE_MB", "5"),
        ("LOG_BACKUP_COUNT", "2"),
        ("LOG_CONSOLE", "false"),
        ("LOG_FILE_ENABLED", "true"),
    ]

    for var, _ in test_vars:
        original_env[var] = os.environ.get(var)

    try:
        # Set test environment
        for var, value in test_vars:
            os.environ[var] = value

        # Re-initialize logging
        from logging_config import init_logging

        init_logging()

        # Verify configuration applied
        root_logger = logging.getLogger()

        # Check level (DEBUG = 10)
        assert root_logger.level == 10

        # Check handlers
        file_handlers = [
            h
            for h in root_logger.handlers
            if isinstance(h, logging.handlers.RotatingFileHandler)
        ]
        console_handlers = [
            h for h in root_logger.handlers if isinstance(h, logging.StreamHandler)
        ]

        assert len(file_handlers) == 1, "Should have one file handler"
        assert len(console_handlers) == 0, "Should have no console handlers (disabled)"

        # Check file handler configuration
        file_handler = file_handlers[0]
        assert file_handler.baseFilename == "/tmp/test-env-log.log"

        print("✅ Environment-based logging configuration working")
        print("✅ Log level set to DEBUG")
        print("✅ File logging enabled, console disabled")
        print("✅ Log rotation configured correctly")

    finally:
        # Clean up test log file
        if os.path.exists("/tmp/test-env-log.log"):
            os.unlink("/tmp/test-env-log.log")

        # Restore original environment
        for var, original_value in original_env.items():
            if original_value is not None:
                os.environ[var] = original_value
            elif var in os.environ:
                del os.environ[var]


async def run_all_logging_tests():
    """Run all structured logging tests."""
    print("📝 Structured Logging Test Suite")
    print("=" * 70)

    tests = [
        ("Log Formatters", test_log_formatters),
        ("Request Context", test_request_context),
        ("Log Levels and Filtering", test_log_levels_and_filtering),
        ("Structured Logging Features", test_structured_logging_features),
        ("Environment Configuration", test_environment_configuration),
    ]

    results = []
    for test_name, test_func in tests:
        print(f"\n{'=' * 25} {test_name} {'=' * 25}")
        try:
            result = await test_func()
            results.append(result)
            status = "✅ PASSED" if result else "❌ FAILED"
            print(f"\n{status}: {test_name}")
        except Exception as e:
            print(f"\n❌ ERROR in {test_name}: {e}")
            results.append(False)

    # Summary
    print(f"\n{'=' * 70}")
    passed = sum(results)
    total = len(results)
    print(f"📊 Test Results: {passed}/{total} tests passed")

    if passed == total:
        print("🎉 All structured logging tests completed successfully!")
        print("🔍 Logging infrastructure provides comprehensive observability.")
    else:
        print("⚠️  Some tests failed. Check the output above for details.")
        print("💡 Ensure proper file permissions for log file creation.")

    return passed == total


if __name__ == "__main__":
    asyncio.run(run_all_logging_tests())