TEST_WEBSOCKET.md

WebSocket Testing Guide

Quick Test Methods

1. Python Test Suite (Recommended)

python3 test_websocket.py

Tests:

• ✅ Connection establishment
• ✅ Request/response handling
• ✅ Latency measurement
• ✅ Audio streaming and saving

---

2. Command Line with wscat

Install wscat:

npm install -g wscat

Test connection:

wscat -c wss://ebitlogix-parler-tts-api.hf.space/ws/tts

Then send JSON:

{"text":"سلام دنیا","speaker":"Divya"}

You should see:

> {"text":"سلام دنیا","speaker":"Divya"}
< {"status":"generating","message":"Generating audio for speaker Divya..."}
< [binary audio chunk 1]
< [binary audio chunk 2]
< [binary audio chunk 3]
< {"status":"complete","chunks_sent":5}

---

3. JavaScript/Node Test

const WebSocket = require('ws');

const ws = new WebSocket('wss://ebitlogix-parler-tts-api.hf.space/ws/tts');

ws.on('open', () => {
    console.log('✅ Connected');
    
    ws.send(JSON.stringify({
        text: 'سلام دنیا',
        speaker: 'Divya'
    }));
});

ws.on('message', (data) => {
    if (typeof data === 'string') {
        console.log('📨 Status:', JSON.parse(data));
    } else {
        console.log('🔊 Audio chunk:', data.length, 'bytes');
    }
});

ws.on('error', (err) => {
    console.error('❌ Error:', err);
});

---

4. Curl Test (for connection only)

curl -i -N -H "Connection: Upgrade" -H "Upgrade: websocket" \
  -H "Sec-WebSocket-Key: SGVsbG8sIHdvcmxkIQ==" \
  -H "Sec-WebSocket-Version: 13" \
  https://ebitlogix-parler-tts-api.hf.space/ws/tts

---

Expected Responses

Success Response Flow:

1. Connection accepted (WebSocket upgrade)
   ↓
2. Status message: {"status":"generating",...}
   ↓
3. Audio chunks (binary data) arriving in real-time
   ↓
4. Complete message: {"status":"complete","chunks_sent":N}
   ↓
5. Connection closes

Error Response:

{"error": "Text cannot be empty"}

---

What to Check

✅ Connection

• WebSocket handshake completes (101 Switching Protocols)
• No connection refused or timeout errors

✅ Response Time

• Status message within 1-2 seconds
• First audio chunk within 3-5 seconds
• Audio chunks every 0.5 seconds

✅ Data

• Receives JSON status messages (parseable)
• Receives binary audio chunks (non-zero size)
• All chunks have data (not empty)

✅ Completion

• Complete message sent
• Connection closes gracefully
• No hanging connections

---

Troubleshooting

Connection Refused

❌ Error: Connection refused

→ Space might be rebuilding. Check: https://huggingface.co/spaces/ebitlogix/Parler_TTS_API

Timeout

❌ timeout waiting for response

→ Model might be loading. Wait 30 seconds and try again.

No Audio Chunks

❌ Only status message, no audio chunks

→ Check text is not empty. Try different text.

Wrong Protocol

❌ WebSocket is not supported

→ Make sure you're using wss:// (secure) or ws:// (local), not https:// or http://

---

Test Matrix

Test	URL	Expected	Status
Connection	wss://...	101 Upgrade	✅
Send JSON	{"text":"..."}	Status message	✅
Audio Stream	Binary chunks	Every 0.5s	✅
Latency	Complete flow	<10 seconds	✅

---

Live Testing

Test 1: Simple Message

{"text":"سلام دنیا","speaker":"Divya"}

Expected: Audio generated in ~4-5 seconds

Test 2: Longer Text

{"text":"مرحبا بك في تطبيق تحويل النص إلى كلام","speaker":"Rani"}

Expected: More chunks, ~8-10 seconds total

Test 3: Different Speaker

{"text":"یہ ایک اردو متن ہے","speaker":"Generic Female"}

Expected: Same timing, different voice

---

Pipecat Integration Testing

# test_pipecat_integration.py
import asyncio
import websockets
import json

async def test_pipecat():
    async with websockets.connect('wss://ebitlogix-parler-tts-api.hf.space/ws/tts') as ws:
        # Simulate Pipecat sending TTS request
        await ws.send(json.dumps({
            "text": "سلام دنیا",
            "speaker": "Divya"
        }))
        
        # Simulate receiving audio for playback
        while True:
            msg = await ws.recv()
            if isinstance(msg, bytes):
                print(f"Got audio: {len(msg)} bytes - READY FOR PIPECAT")
            elif isinstance(msg, str):
                data = json.loads(msg)
                if data.get("status") == "complete":
                    print("Audio stream complete")
                    break

asyncio.run(test_pipecat())

---

Success Indicators

✅ All these should be true:

• Connection establishes without error
• Receives status message within 2 seconds
• First audio chunk arrives within 5 seconds
• Subsequent chunks arrive continuously every 0.5-1 second
• Complete message received
• Connection closes normally

If any fail, check Space status and rebuild if needed.