test_translation_service.py

#!/usr/bin/env python3
"""
Test script for the Translation Service

This script tests the translation processor functionality including:
- Single text translation
- Batch translation
- Language detection
- API endpoints integration
"""

import sys
import time
import requests
from pathlib import Path

# Add src to path for imports
sys.path.append(str(Path(__file__).parent / "src"))

from src.file_processors.translation_processor import TranslationProcessor
from src.config import Config


def test_processor_initialization():
    """Test if translation processor can be initialized."""
    print("🔧 Testing Translation Processor Initialization...")

    print(f"   OpenAI API Key configured: {bool(Config.OPENAI_API_KEY)}")
    print(f"   Gemini API Key configured: {bool(Config.GOOGLE_API_KEY)}")

    if not Config.OPENAI_API_KEY and not Config.GOOGLE_API_KEY:
        print("   ⚠️  No API keys configured - processor will have limited functionality")
        return None

    try:
        processor = TranslationProcessor()
        print("   ✅ Translation processor initialized successfully")

        # Test supported languages
        languages = processor.get_supported_languages()
        print(f"   📋 Supported languages: {len(languages)}")
        print(f"   🌍 Sample languages: {', '.join(list(languages.keys())[:5])}")

        return processor
    except Exception as e:
        print(f"   ❌ Error initializing processor: {e}")
        return None


def test_language_detection(processor):
    """Test language detection functionality."""
    if not processor:
        print("⏭️  Skipping language detection tests (no processor)")
        return

    print("\n🔍 Testing Language Detection...")

    test_texts = [
        ("Hello, this is English text.", "en"),
        ("Hola, esto es texto en español.", "es"),
        ("Bonjour, ceci est du texte français.", "fr"),
        ("Guten Tag, das ist deutscher Text.", "de"),
    ]

    for text, expected_lang in test_texts:
        print(f"   Text: '{text[:30]}...'")
        print(f"   Expected: {expected_lang}")

        try:
            if Config.OPENAI_API_KEY:
                result = processor.detect_language(text, use_openai=True)
                if result["success"]:
                    detected = result["detected_language"]
                    print(f"   📍 OpenAI detected: {detected} (expected: {expected_lang})")
                else:
                    print(f"   ❌ OpenAI detection failed: {result.get('error')}")

            if Config.GOOGLE_API_KEY:
                result = processor.detect_language(text, use_openai=False)
                if result["success"]:
                    detected = result["detected_language"]
                    print(f"   📍 Gemini detected: {detected} (expected: {expected_lang})")
                else:
                    print(f"   ❌ Gemini detection failed: {result.get('error')}")

        except Exception as e:
            print(f"   ❌ Error in detection: {e}")


def test_single_translation(processor):
    """Test single text translation."""
    if not processor:
        print("⏭️  Skipping single translation tests (no processor)")
        return

    print("\n🌐 Testing Single Text Translation...")

    test_cases = [
        ("Hello world!", "en", "es"),
        ("How are you today?", "en", "fr"),
        ("Thank you very much", "en", "de"),
    ]

    for text, source_lang, target_lang in test_cases:
        print(f"   Translating: '{text}' ({source_lang} → {target_lang})")

        try:
            if Config.OPENAI_API_KEY:
                result = processor.translate_text(text, source_lang, target_lang, use_openai=True)
                if result["success"]:
                    translation = result["translated_text"]
                    print(f"   ✅ OpenAI: '{translation}'")
                else:
                    print(f"   ❌ OpenAI failed: {result.get('error')}")

            if Config.GOOGLE_API_KEY:
                result = processor.translate_text(text, source_lang, target_lang, use_openai=False)
                if result["success"]:
                    translation = result["translated_text"]
                    print(f"   ✅ Gemini: '{translation}'")
                else:
                    print(f"   ❌ Gemini failed: {result.get('error')}")

        except Exception as e:
            print(f"   ❌ Error in translation: {e}")


def test_batch_translation(processor):
    """Test batch translation functionality."""
    if not processor:
        print("⏭️  Skipping batch translation tests (no processor)")
        return

    print("\n📦 Testing Batch Translation...")

    texts = [
        "Good morning!",
        "How can I help you?",
        "Thank you for your time.",
        "Have a great day!"
    ]

    print(f"   Translating {len(texts)} texts (en → es)")

    try:
        if Config.OPENAI_API_KEY:
            result = processor.translate_batch(texts, "en", "es", use_openai=True)
            if result["success"]:
                print(f"   ✅ OpenAI batch: {result['successful']}/{result['total_texts']} successful")
                for item in result["results"][:2]:  # Show first 2 results
                    if item["success"]:
                        print(f"      '{item['original_text']}' → '{item['translated_text']}'")
            else:
                print(f"   ❌ OpenAI batch failed")

        if Config.GOOGLE_API_KEY:
            result = processor.translate_batch(texts, "en", "es", use_openai=False)
            if result["success"]:
                print(f"   ✅ Gemini batch: {result['successful']}/{result['total_texts']} successful")
            else:
                print(f"   ❌ Gemini batch failed")

    except Exception as e:
        print(f"   ❌ Error in batch translation: {e}")


def test_api_endpoints():
    """Test API endpoints if server is running."""
    print("\n🌐 Testing API Endpoints...")

    base_url = "http://localhost:8000"

    # Check if server is running
    try:
        response = requests.get(f"{base_url}/health", timeout=5)
        if response.status_code != 200:
            print("   ⏭️  API server not running, skipping endpoint tests")
            return
    except requests.exceptions.RequestException:
        print("   ⏭️  API server not accessible, skipping endpoint tests")
        return

    print("   🟢 API server detected, testing endpoints...")

    # Test supported languages endpoint
    try:
        response = requests.get(f"{base_url}/supported-languages")
        if response.status_code == 200:
            data = response.json()
            print(f"   ✅ Supported languages: {data.get('total_languages', 0)} languages")
        else:
            print(f"   ❌ Supported languages endpoint failed: {response.status_code}")
    except Exception as e:
        print(f"   ❌ Error testing supported languages: {e}")

    # Test single translation endpoint
    try:
        payload = {
            "text": "Hello API test",
            "source_language": "en",
            "target_language": "es",
            "use_openai": True
        }
        response = requests.post(f"{base_url}/translate", json=payload)
        if response.status_code == 200:
            data = response.json()
            if data.get("success"):
                print(f"   ✅ Translation endpoint: '{data.get('translated_text')}'")
            else:
                print(f"   ❌ Translation failed: {data.get('error')}")
        else:
            print(f"   ❌ Translation endpoint failed: {response.status_code}")
    except Exception as e:
        print(f"   ❌ Error testing translation endpoint: {e}")

    # Test language detection endpoint
    try:
        payload = {
            "text": "Bonjour, comment ça va?",
            "use_openai": True
        }
        response = requests.post(f"{base_url}/detect-language", json=payload)
        if response.status_code == 200:
            data = response.json()
            if data.get("success"):
                print(f"   ✅ Detection endpoint: '{data.get('detected_language')}'")
            else:
                print(f"   ❌ Detection failed: {data.get('error')}")
        else:
            print(f"   ❌ Detection endpoint failed: {response.status_code}")
    except Exception as e:
        print(f"   ❌ Error testing detection endpoint: {e}")


def main():
    """Run all translation service tests."""
    print("🧪 AI Content Processing - Translation Service Tests")
    print("=" * 60)

    # Test processor initialization
    processor = test_processor_initialization()

    # Test core functionality
    test_language_detection(processor)
    test_single_translation(processor)
    test_batch_translation(processor)

    # Test API endpoints
    test_api_endpoints()

    print("\n" + "=" * 60)
    print("🎉 Translation service tests completed!")

    # Show configuration recommendations
    print("\n💡 Configuration Notes:")
    if not Config.OPENAI_API_KEY:
        print("   • Set OPENAI_API_KEY for OpenAI translations")
    if not Config.GOOGLE_API_KEY:
        print("   • Set GOOGLE_API_KEY for Gemini translations")
    if Config.OPENAI_API_KEY and Config.GOOGLE_API_KEY:
        print("   ✅ Both APIs configured - full functionality available")

    print("\n📖 Usage Examples:")
    print("   • python translation_examples.py")
    print("   • curl examples in examples/TRANSLATION_CURL_EXAMPLES.md")
    print("   • Start API: python api_server.py")


if __name__ == "__main__":
    main()