test_scheduler_visibility.py

#!/usr/bin/env python3
"""
Test script for APScheduler visibility and functionality.
This script tests whether APScheduler is working and properly configured for logging.
"""

import sys
import os
import logging
from pathlib import Path
from datetime import datetime

# Add the backend directory to the Python path
backend_dir = Path(__file__).parent / "backend"
sys.path.insert(0, str(backend_dir))

def setup_logging():
    """Setup logging for the test script."""
    logging.basicConfig(
        level=logging.DEBUG,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    # Configure APScheduler logger specifically
    logging.getLogger('apscheduler').setLevel(logging.DEBUG)
    print("Logging configured for APScheduler")

def test_apscheduler_import():
    """Test that APScheduler can be imported."""
    try:
        from backend.scheduler.apscheduler_service import APSchedulerService
        print("SUCCESS: APSchedulerService imported successfully")
        return True
    except Exception as e:
        print(f"ERROR: Failed to import APSchedulerService: {e}")
        return False

def test_scheduler_initialization():
    """Test APScheduler initialization with mock app."""
    try:
        from backend.scheduler.apscheduler_service import APSchedulerService
        
        # Create a mock app object
        class MockApp:
            def __init__(self):
                self.config = {
                    'SUPABASE_URL': 'https://test.supabase.co',
                    'SUPABASE_KEY': 'test_key',
                    'SCHEDULER_ENABLED': True
                }
        
        # Initialize the scheduler service
        app = MockApp()
        scheduler_service = APSchedulerService()
        
        # Mock the Supabase client initialization
        class MockSupabaseClient:
            def table(self, table_name):
                return self
                
            def select(self, columns):
                return self
                
            def execute(self):
                # Return empty schedule data for testing
                return type('obj', (object,), {'data': []})()
        
        scheduler_service.supabase_client = MockSupabaseClient()
        
        # Test initialization
        scheduler_service.init_app(app)
        
        if scheduler_service.scheduler is not None:
            print("SUCCESS: APScheduler initialized successfully")
            print(f"INFO: Current jobs: {len(scheduler_service.scheduler.get_jobs())}")
            return True
        else:
            print("ERROR: APScheduler initialization failed")
            return False
            
    except Exception as e:
        print(f"ERROR: Error testing APScheduler initialization: {e}")
        import traceback
        traceback.print_exc()
        return False

def test_schedule_loading():
    """Test the schedule loading functionality."""
    try:
        from backend.scheduler.apscheduler_service import APSchedulerService
        
        # Create scheduler service
        scheduler_service = APSchedulerService()
        
        # Mock the Supabase client
        class MockSupabaseClient:
            def table(self, table_name):
                return self
                
            def select(self, columns):
                return self
                
            def execute(self):
                # Return mock schedule data
                mock_data = [
                    {
                        'id': 'test_schedule_1',
                        'schedule_time': 'Monday 09:00',
                        'adjusted_time': 'Monday 08:55',
                        'Social_network': {
                            'id_utilisateur': 'test_user_1',
                            'token': 'test_token',
                            'sub': 'test_sub'
                        }
                    }
                ]
                return type('obj', (object,), {'data': mock_data})()
        
        scheduler_service.supabase_client = MockSupabaseClient()
        
        # Test schedule loading
        scheduler_service.load_schedules()
        
        if scheduler_service.scheduler is not None:
            jobs = scheduler_service.scheduler.get_jobs()
            print(f"SUCCESS: Schedule loading test completed")
            print(f"INFO: Total jobs: {len(jobs)}")
            
            # Check for specific job types
            loader_jobs = [job for job in jobs if job.id == 'load_schedules']
            content_jobs = [job for job in jobs if job.id.startswith('gen_')]
            publish_jobs = [job for job in jobs if job.id.startswith('pub_')]
            
            print(f"INFO: Loader jobs: {len(loader_jobs)}")
            print(f"INFO: Content generation jobs: {len(content_jobs)}")
            print(f"INFO: Publishing jobs: {len(publish_jobs)}")
            
            return len(jobs) > 0
        else:
            print("ERROR: Scheduler not initialized for schedule loading test")
            return False
            
    except Exception as e:
        print(f"ERROR: Error testing schedule loading: {e}")
        import traceback
        traceback.print_exc()
        return False

def main():
    """Main test function."""
    print("Testing APScheduler visibility and functionality...")
    print("=" * 60)
    
    setup_logging()
    
    tests = [
        ("APScheduler Import", test_apscheduler_import),
        ("Scheduler Initialization", test_scheduler_initialization),
        ("Schedule Loading", test_schedule_loading),
    ]
    
    passed = 0
    total = len(tests)
    
    for test_name, test_func in tests:
        print(f"\nRunning test: {test_name}")
        print("-" * 40)
        
        if test_func():
            passed += 1
            print(f"SUCCESS: {test_name} PASSED")
        else:
            print(f"FAILED: {test_name} FAILED")
    
    print("\n" + "=" * 60)
    print(f"Test Results: {passed}/{total} tests passed")
    
    if passed == total:
        print("SUCCESS: All tests passed! APScheduler is working correctly.")
        return True
    else:
        print("WARNING: Some tests failed. Please check the error messages above.")
        return False

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