test_logging_fix.py

#!/usr/bin/env python3
"""
Test script to verify the STT service logging fixes work properly.
This simulates the ZeroGPU environment conditions that cause logging conflicts.
"""

import os
import sys
import io
import tempfile
import subprocess

def test_stream_protection():
    """Test that the SafeStream class prevents I/O errors"""
    print("🧪 Testing SafeStream protection...")
    
    # Simulate closed file descriptor scenario
    try:
        # Import the SafeStream from our app
        sys.path.insert(0, os.path.dirname(__file__))
        
        # Mock a closed stream scenario
        original_stdout = sys.stdout
        
        # Create a StringIO that we'll "close"
        closed_stream = io.StringIO()
        closed_stream.close()
        
        # Test if our SafeStream handles this gracefully
        from app import SafeStream
        safe_stream = SafeStream('stdout')
        
        # This should not raise an exception
        safe_stream.write("Test message")
        safe_stream.flush()
        
        # Test isatty() always returns False to prevent tty errors
        assert safe_stream.isatty() == False
        
        print("✅ SafeStream protection working correctly")
        return True
        
    except Exception as e:
        print(f"❌ SafeStream test failed: {e}")
        return False
    finally:
        sys.stdout = original_stdout

def test_uvicorn_patching():
    """Test that uvicorn logging configuration is properly disabled"""
    print("🧪 Testing uvicorn patching...")
    
    try:
        # Import our app which should patch uvicorn
        import app
        
        # Try to import uvicorn to see if it's patched
        try:
            import uvicorn.config
            
            # Create a config instance and check if configure_logging is patched
            config = uvicorn.config.Config("dummy:app")
            
            # This should not cause any logging setup
            config.configure_logging()
            
            print("✅ uvicorn.Config.configure_logging() patched successfully")
            return True
            
        except ImportError:
            print("⚠️ uvicorn not available, but that's okay for basic testing")
            return True
            
    except Exception as e:
        print(f"❌ uvicorn patching test failed: {e}")
        return False

def test_logging_module_disabled():
    """Test that the logging module is properly disabled"""
    print("🧪 Testing logging module disablement...")
    
    try:
        import logging
        
        # Check if logging is disabled
        if logging.getLogger().disabled:
            print("✅ Root logger properly disabled")
        else:
            print("⚠️ Root logger not disabled, but may still work")
        
        # Try to log something - should be silent
        logger = logging.getLogger("test")
        logger.error("This should be silent")
        
        print("✅ Logging module configuration working")
        return True
        
    except Exception as e:
        print(f"❌ Logging module test failed: {e}")
        return False

def test_import_safety():
    """Test that importing our app doesn't cause stream conflicts"""
    print("🧪 Testing safe import process...")
    
    try:
        # This should not raise any I/O errors
        import app
        
        # Test that safe_log function works
        app.safe_log("info", "Test log message")
        
        print("✅ App import and logging working correctly")
        return True
        
    except Exception as e:
        print(f"❌ Import safety test failed: {e}")
        return False

def test_environment_variables():
    """Test that required environment variables are set correctly"""
    print("🧪 Testing environment variables...")
    
    expected_vars = {
        "PYTHONWARNINGS": "ignore",
        "TRANSFORMERS_VERBOSITY": "error",
        "TOKENIZERS_PARALLELISM": "false",
        "GRADIO_ANALYTICS_ENABLED": "False",
        "GRADIO_ALLOW_FLAGGING": "never"
    }
    
    all_correct = True
    for var, expected_value in expected_vars.items():
        actual_value = os.environ.get(var)
        if actual_value == expected_value:
            print(f"✅ {var} = {actual_value}")
        else:
            print(f"⚠️ {var} = {actual_value} (expected: {expected_value})")
            all_correct = False
    
    return all_correct

def main():
    """Run all logging fix tests"""
    print("🔒 STT Service Logging Fix Validation")
    print("=" * 50)
    
    tests = [
        ("Stream Protection", test_stream_protection),
        ("uvicorn Patching", test_uvicorn_patching), 
        ("Logging Module Disabled", test_logging_module_disabled),
        ("Safe Import Process", test_import_safety),
        ("Environment Variables", test_environment_variables)
    ]
    
    passed = 0
    total = len(tests)
    
    for test_name, test_func in tests:
        print(f"\n📋 {test_name}")
        print("-" * 30)
        if test_func():
            passed += 1
        else:
            print(f"❌ {test_name} failed")
    
    print("\n" + "=" * 50)
    print("📊 Test Results")
    print("=" * 50)
    print(f"✅ Passed: {passed}/{total}")
    print(f"❌ Failed: {total - passed}/{total}")
    
    if passed == total:
        print("\n🎉 All logging fix tests passed!")
        print("🚀 STT service should work without stream conflicts")
        return True
    else:
        print(f"\n⚠️ {total - passed} tests failed")
        print("🔧 Some logging fixes may need adjustment")
        return False

if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)