test: Add comprehensive STT service connection testing

- Created direct STT service test to verify audio processing
- Added WebRTC handler connection test with endpoint discovery
- Identified STT service internal exception requiring verbose errors
- Test scripts help diagnose WebRTC to STT pipeline issues

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

test_stt_direct.py

@@ -0,0 +1,149 @@
+"""
+Direct STT Service Test
+Test the STT service directly with a real audio file to verify it's receiving and processing data
+"""
+
+import asyncio
+import tempfile
+import wave
+import numpy as np
+import requests
+import logging
+from datetime import datetime
+from mcp_voice_service import mcp_create_test_voice
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class STTDirectTest:
+    """Test STT service directly with audio files"""
+    
+    def __init__(self):
+        self.stt_service_url = "https://pgits-stt-gpu-service.hf.space"
+    
+    async def test_stt_service_direct(self):
+        """Test STT service with direct audio file upload"""
+        logger.info("🎤 Starting direct STT service test")
+        
+        try:
+            # Step 1: Create test audio file
+            logger.info("📁 Creating test audio file...")
+            test_voice_file = await mcp_create_test_voice()
+            logger.info(f"✅ Test audio file created: {test_voice_file}")
+            
+            # Step 2: Test Gradio API endpoint
+            logger.info("🔗 Testing Gradio API endpoint...")
+            await self.test_gradio_api(test_voice_file)
+            
+            # Step 3: Test HTTP API endpoint  
+            logger.info("🌐 Testing HTTP API endpoint...")
+            await self.test_http_api(test_voice_file)
+            
+            # Step 4: Verify service is receiving requests
+            logger.info("📊 Checking service status...")
+            await self.check_service_status()
+            
+        except Exception as e:
+            logger.error(f"❌ Direct STT test failed: {e}")
+    
+    async def test_gradio_api(self, audio_file_path):
+        """Test STT service via Gradio Client API"""
+        try:
+            from gradio_client import Client
+            
+            logger.info(f"📡 Connecting to Gradio API at {self.stt_service_url}")
+            client = Client(self.stt_service_url)
+            
+            logger.info(f"🎤 Sending audio file to Gradio API...")
+            result = client.predict(
+                audio_file_path,  # audio file path
+                "en",            # language
+                "base",          # model_size_param
+                api_name="/gradio_transcribe_wrapper"
+            )
+            
+            logger.info(f"✅ Gradio API response: {result}")
+            
+            # Parse response
+            transcription_text, timing_json, status_text = result
+            logger.info(f"📝 Transcription: {transcription_text}")
+            logger.info(f"⏱️ Timing: {timing_json}")
+            logger.info(f"📊 Status: {status_text}")
+            
+            return result
+            
+        except Exception as e:
+            logger.error(f"❌ Gradio API test failed: {e}")
+            return None
+    
+    async def test_http_api(self, audio_file_path):
+        """Test STT service via HTTP API endpoint"""
+        try:
+            logger.info(f"🌐 Testing HTTP API at {self.stt_service_url}/api/transcribe")
+            
+            # Prepare file for upload
+            with open(audio_file_path, 'rb') as audio_file:
+                files = {
+                    'file': ('test_audio.wav', audio_file, 'audio/wav')
+                }
+                data = {
+                    'language': 'en',
+                    'model_size_param': 'base'
+                }
+                
+                logger.info("📤 Uploading audio file to HTTP API...")
+                response = requests.post(
+                    f"{self.stt_service_url}/api/transcribe",
+                    files=files,
+                    data=data,
+                    timeout=30
+                )
+            
+            logger.info(f"📥 HTTP API response status: {response.status_code}")
+            
+            if response.status_code == 200:
+                result = response.json()
+                logger.info(f"✅ HTTP API response: {result}")
+                return result
+            else:
+                logger.error(f"❌ HTTP API failed with status {response.status_code}: {response.text}")
+                return None
+                
+        except Exception as e:
+            logger.error(f"❌ HTTP API test failed: {e}")
+            return None
+    
+    async def check_service_status(self):
+        """Check if STT service is responding"""
+        try:
+            # Test basic connectivity
+            logger.info(f"🔍 Checking service connectivity...")
+            response = requests.get(f"{self.stt_service_url}/", timeout=10)
+            logger.info(f"📡 Service status: {response.status_code}")
+            
+            if response.status_code == 200:
+                logger.info("✅ STT service is responding")
+            else:
+                logger.warning(f"⚠️ STT service returned status {response.status_code}")
+            
+            # Test API health endpoint if it exists
+            try:
+                health_response = requests.get(f"{self.stt_service_url}/api/health", timeout=5)
+                if health_response.status_code == 200:
+                    health_data = health_response.json()
+                    logger.info(f"💚 Health check: {health_data}")
+                else:
+                    logger.info(f"ℹ️ No health endpoint available")
+            except:
+                logger.info("ℹ️ No health endpoint available")
+                
+        except Exception as e:
+            logger.error(f"❌ Service status check failed: {e}")
+
+async def run_stt_direct_test():
+    """Run the direct STT service test"""
+    test = STTDirectTest()
+    await test.test_stt_service_direct()
+
+if __name__ == "__main__":
+    asyncio.run(run_stt_direct_test())

test_webrtc_connection.py

@@ -0,0 +1,103 @@
+"""
+Test WebRTC Handler Connection to STT Service
+Verify if the WebRTC handler can actually connect and send audio to STT service
+"""
+
+import asyncio
+import tempfile
+import wave
+import numpy as np
+import logging
+from datetime import datetime
+from mcp_voice_service import mcp_create_test_voice
+from webrtc_streamlit import StreamlitWebRTCHandler
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+async def test_webrtc_stt_connection():
+    """Test WebRTC handler connection to STT service"""
+    logger.info("🔗 Testing WebRTC handler connection to STT service")
+    
+    try:
+        # Step 1: Create test audio
+        logger.info("📁 Creating test audio file...")
+        test_voice_file = await mcp_create_test_voice()
+        logger.info(f"✅ Test audio created: {test_voice_file}")
+        
+        # Step 2: Initialize WebRTC handler
+        logger.info("🎤 Initializing WebRTC handler...")
+        handler = StreamlitWebRTCHandler()
+        
+        # Step 3: Test client connection
+        logger.info("🔗 Testing STT client connection...")
+        client = handler.client
+        if client:
+            logger.info("✅ WebRTC handler client initialized successfully")
+        else:
+            logger.error("❌ WebRTC handler client initialization failed")
+            return
+        
+        # Step 4: Test transcription directly via handler
+        logger.info("🎤 Testing transcription via WebRTC handler...")
+        transcription = await handler.transcribe_audio_file(test_voice_file)
+        
+        if transcription and not transcription.startswith("ERROR"):
+            logger.info(f"✅ WebRTC transcription successful: '{transcription}'")
+        else:
+            logger.error(f"❌ WebRTC transcription failed: {transcription}")
+        
+        # Step 5: Test with HTTP fallback if available
+        logger.info("🌐 Testing HTTP fallback...")
+        await test_http_fallback(test_voice_file)
+        
+    except Exception as e:
+        logger.error(f"❌ WebRTC connection test failed: {e}")
+
+async def test_http_fallback(audio_file_path):
+    """Test HTTP fallback for WebRTC connection"""
+    try:
+        import requests
+        
+        stt_url = "https://pgits-stt-gpu-service.hf.space"
+        
+        # Try the Gradio HTTP endpoint directly
+        logger.info("📤 Testing Gradio HTTP upload...")
+        
+        with open(audio_file_path, 'rb') as f:
+            files = {'file': ('test_audio.wav', f, 'audio/wav')}
+            
+            # Try Gradio's file upload endpoint
+            try:
+                response = requests.post(
+                    f"{stt_url}/upload",
+                    files=files,
+                    timeout=10
+                )
+                logger.info(f"📥 Upload response: {response.status_code}")
+                if response.status_code != 200:
+                    logger.info(f"📄 Response text: {response.text[:200]}")
+            except Exception as e:
+                logger.info(f"ℹ️ Upload endpoint not available: {e}")
+        
+        # Check available endpoints
+        logger.info("🔍 Checking available endpoints...")
+        endpoints_to_try = [
+            "/queue/join",
+            "/run/gradio_transcribe_wrapper", 
+            "/api/gradio_transcribe_wrapper",
+            "/gradio_api/run/gradio_transcribe_wrapper"
+        ]
+        
+        for endpoint in endpoints_to_try:
+            try:
+                response = requests.get(f"{stt_url}{endpoint}", timeout=5)
+                logger.info(f"📡 {endpoint}: {response.status_code}")
+            except Exception as e:
+                logger.info(f"❌ {endpoint}: Not available")
+        
+    except Exception as e:
+        logger.error(f"❌ HTTP fallback test failed: {e}")
+
+if __name__ == "__main__":
+    asyncio.run(test_webrtc_stt_connection())