bytelyst-devops-tools/supabase monitor/test.py

"""
Test script for the YouTube Processing Workflow.
"""
import unittest
import os
import tempfile
from unittest.mock import patch, MagicMock
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

class TestWorkflowComponents(unittest.TestCase):
    """Test cases for workflow components."""
    
    def setUp(self):
        """Set up test fixtures."""
        self.sample_youtube_url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
        self.sample_transcript = """Welcome to this educational video about machine learning.
        Today we'll cover supervised learning, including algorithms like linear regression."""
        
    def test_configuration(self):
        """Test configuration loading."""
        from config import Config
        
        config = Config()
        self.assertIsNotNone(config)
        
    @patch('utils.speech_processing.YouTubeTranscriber')
    def test_transcriber_agent(self, mock_transcriber):
        """Test transcriber agent."""
        from agents.transcriber_agent import TranscriberAgent
        from openai import OpenAI
        
        # Mock the transcriber
        mock_transcriber_instance = MagicMock()
        mock_transcriber_instance.transcribe_youtube_video.return_value = self.sample_transcript
        mock_transcriber.return_value = mock_transcriber_instance
        
        # Mock OpenAI
        with patch('openai.OpenAI'):
            transcriber = TranscriberAgent(MagicMock())
            result = transcriber.transcribe(self.sample_youtube_url)
            
            # Note: This will now return an error string because we're mocking
            self.assertIsInstance(result, str)
    
    def test_translator_agent(self):
        """Test translator agent."""
        from agents.translator_agent import TranslatorAgent
        
        translator = TranslatorAgent(MagicMock())
        
        # Test task creation
        task = translator.create_translation_task(self.sample_transcript, "Spanish")
        self.assertIsNotNone(task)
        self.assertIn("Spanish", task.description)
    
    def test_summarizer_agent(self):
        """Test summarizer agent."""
        from agents.summarizer_agent import SummarizerAgent
        
        summarizer = SummarizerAgent(MagicMock())
        sample_translated = "Bienvenidos a este video educativo..."
        sample_prompt = "Summarize in 3 bullet points"
        
        # Test task creation
        task = summarizer.create_summarization_task(sample_translated, sample_prompt)
        self.assertIsNotNone(task)
        self.assertIn("summarization_prompt", expected_output=str)

def test_api_endpoints():
    """Test API endpoints."""
    import json
    from api import app
    
    # Create test client
    client = app.test_client()
    
    # Test health endpoint
    response = client.get('/health')
    assert response.status_code == 200
    
    data = json.loads(response.data)
    assert 'status' in data

def test_individual_functions():
    """Test individual utility functions."""
    
    # Test YouTube URL validation
    def is_valid_youtube_url(url):
        return "youtube.com" in url and "/watch" in url
    
    assert is_valid_youtube_url("https://www.youtube.com/watch?v=example")
    assert not is_valid_youtube_url("https://example.com")
    
    # Test language name validation
    def is_valid_language(language):
        valid_languages = ["English", "Spanish", "French", "German", "Italian"]
        return language in valid_languages
    
    assert is_valid_language("Spanish")
    assert is_valid_language("French")
    assert not is_valid_language("Klingon")

def test_error_handling():
    """Test error handling scenarios."""
    
    # Test transcription error
    error_result = "Error transcribing video: Network timeout"
    assert error_result.startswith("Error")
    
    # Test translation error  
    error_result = "Error translating text: Invalid language"
    assert error_result.startswith("Error")

def run_quick_tests():
    """Run quick tests without requiring API keys."""
    
    print("🧪 Running Quick Tests...")
    print("=" * 40)
    
    try:
        # Test individual functions
        test_individual_functions()
        print("✅ Individual function tests passed")
        
        # Test error handling
        test_error_handling() 
        print("✅ Error handling tests passed")
        
        # Test workflow components (basic)
        test_workflow_components()
        print("✅ Workflow component tests passed")
        
        print("\n🎉 All quick tests passed!")
        return True
        
    except Exception as e:
        print(f"❌ Test failed: {str(e)}")
        return False

def test_workflow_components():
    """Test workflow components without external dependencies."""
    
    # Test configuration
    test_configuration()
    
    # Test agents (basic initialization)
    from agents.transcriber_agent import TranscriberAgent
    from agents.translator_agent import TranslatorAgent  
    from agents.summarizer_agent import SummarizerAgent
    
    # Mock LLM for testing
    mock_llm = MagicMock()
    
    try:
        transcriber = TranscriberAgent(mock_llm)
        print("✅ Transcriber agent initialized")
        
        translator = TranslatorAgent(mock_llm)
        print("✅ Translator agent initialized")
        
        summarizer = SummarizerAgent(mock_llm)
        print("✅ Summarizer agent initialized")
        
    except Exception as e:
        print(f"❌ Agent initialization failed: {str(e)}")

if __name__ == "__main__":
    print("🚀 YouTube Processing Workflow - Test Suite")
    print("=" * 50)
    
    # Check for API keys
    api_keys_available = os.getenv("PERPLEXITY_API_KEY") or os.getenv("OPENAI_API_KEY")
    
    if not api_keys_available:
        print("⚠️  No API keys found. Running quick tests only...")
        success = run_quick_tests()
        
        if success:
            print(f"\n💡 To run full tests:")
            print(f"1. Add API keys to .env file")
            print(f"2. Run: python test.py --full")
        else:
            print(f"\n❌ Some tests failed")
    else:
        print("✅ API keys found. Running full test suite...")
        
        # Run full tests
        unittest.main(verbosity=2)