test_services_detailed.py

#!/usr/bin/env python3
"""
Script de test détaillé pour diagnostiquer les services backend
"""

import asyncio
import aiohttp
import json
import sys
import tempfile
import wave
import audioop

async def test_stt_service():
    """Teste le service STT avec un fichier audio réel"""
    print("🔍 Test du service STT...")
    
    try:
        # Créer un fichier audio de test
        with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f:
            # Créer un fichier WAV de 2 secondes de silence
            with wave.open(f.name, 'w') as wav_file:
                wav_file.setnchannels(1)
                wav_file.setsampwidth(2)
                wav_file.setframerate(16000)
                wav_file.writeframes(b'\x00' * 64000)  # 2 secondes de silence
            
            # Lire le fichier audio
            with open(f.name, 'rb') as audio_file:
                audio_data = audio_file.read()
            
            print(f"✅ Fichier audio créé: {len(audio_data)} bytes")
            
            # Test du service STT
            form = aiohttp.FormData()
            form.add_field("file", audio_data, filename="audio.wav", content_type="audio/wav")
            form.add_field("language", "fr")
            
            async with aiohttp.ClientSession() as session:
                async with session.post("http://localhost:5001/transcribe", data=form) as resp:
                    print(f"✅ STT Status: {resp.status}")
                    if resp.status == 200:
                        result = await resp.json()
                        print(f"✅ STT Response: {result}")
                        return True
                    else:
                        text = await resp.text()
                        print(f"❌ STT Error: {text}")
                        return False
                        
    except Exception as e:
        print(f"❌ STT Exception: {e}")
        import traceback
        print(f"Traceback: {traceback.format_exc()}")
        return False

async def test_llm_service():
    """Teste le service LLM"""
    print("🔍 Test du service LLM...")
    
    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(
                "http://localhost:5002/generate",
                json={
                    "text": "Bonjour, comment allez-vous ?",
                    "user_id": "test_user",
                    "history": []
                }
            ) as resp:
                print(f"✅ LLM Status: {resp.status}")
                if resp.status == 200:
                    result = await resp.json()
                    print(f"✅ LLM Response: {result}")
                    return True
                else:
                    text = await resp.text()
                    print(f"❌ LLM Error: {text}")
                    return False
                    
    except Exception as e:
        print(f"❌ LLM Exception: {e}")
        import traceback
        print(f"Traceback: {traceback.format_exc()}")
        return False

async def test_tts_service():
    """Teste le service TTS"""
    print("🔍 Test du service TTS...")
    
    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(
                "http://localhost:5000/generate-tts",
                json={
                    "text": "Bonjour, je suis un test de synthèse vocale.",
                    "lang": "fr"
                }
            ) as resp:
                print(f"✅ TTS Status: {resp.status}")
                if resp.status == 200:
                    result = await resp.json()
                    print(f"✅ TTS Response: {result}")
                    
                    # Tester le téléchargement du fichier audio
                    audio_url = result.get("url")
                    if audio_url:
                        # Construire l'URL complète si c'est une URL relative
                        if audio_url.startswith("/"):
                            audio_url = f"http://localhost:5000{audio_url}"
                        
                        async with session.get(audio_url) as audio_resp:
                            if audio_resp.status == 200:
                                audio_data = await audio_resp.read()
                                print(f"✅ Audio téléchargé: {len(audio_data)} bytes")
                                return True
                            else:
                                print(f"❌ Erreur téléchargement audio: {audio_resp.status}")
                                return False
                    else:
                        print("❌ Pas d'URL audio dans la réponse")
                        return False
                else:
                    text = await resp.text()
                    print(f"❌ TTS Error: {text}")
                    return False
                    
    except Exception as e:
        print(f"❌ TTS Exception: {e}")
        import traceback
        print(f"Traceback: {traceback.format_exc()}")
        return False

async def main():
    print("🚀 Démarrage des tests détaillés des services...")
    
    # Test de chaque service
    stt_ok = await test_stt_service()
    llm_ok = await test_llm_service()
    tts_ok = await test_tts_service()
    
    print("\n📊 Résultats des tests:")
    print(f"STT: {'✅ OK' if stt_ok else '❌ ERREUR'}")
    print(f"LLM: {'✅ OK' if llm_ok else '❌ ERREUR'}")
    print(f"TTS: {'✅ OK' if tts_ok else '❌ ERREUR'}")
    
    if all([stt_ok, llm_ok, tts_ok]):
        print("🎉 Tous les services fonctionnent correctement !")
    else:
        print("⚠️ Certains services ont des problèmes")

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