Spaces:
Paused
Paused
File size: 2,354 Bytes
275976f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 |
from fastapi import FastAPI
import pytest
from unittest.mock import AsyncMock, MagicMock, ANY
import socketio
import os
import sys
current_dir = os.path.dirname(os.path.abspath(__file__))
parent_dir = os.path.dirname(current_dir)
sys.path.append(parent_dir)
from Client import Client
from main import sio, connect, disconnect, target_language, call_user, answer_call, incoming_audio, clients, rooms
from unittest.mock import patch
sio = socketio.AsyncServer(
async_mode="asgi",
cors_allowed_origins="*",
# engineio_logger=logger,
)
# sio.logger.setLevel(logging.DEBUG)
socketio_app = socketio.ASGIApp(sio)
app = FastAPI()
app.mount("/", socketio_app)
@pytest.fixture(autouse=True)
def setup_clients_and_rooms():
global clients, rooms
clients.clear()
rooms.clear()
yield
@pytest.fixture
def mock_client():
client = Client("test_sid", "test_client_id", original_sr=44100)
return client
@pytest.mark.asyncio
async def test_connect(mock_client):
sid = mock_client.sid
environ = {'QUERY_STRING': 'client_id=test_client_id'}
await connect(sid, environ)
assert sid in clients
@pytest.mark.asyncio
async def test_disconnect(mock_client):
sid = mock_client.sid
clients[sid] = mock_client
await disconnect(sid)
assert sid not in clients
@pytest.mark.asyncio
async def test_target_language(mock_client):
sid = mock_client.sid
clients[sid] = mock_client
target_lang = "fr"
await target_language(sid, target_lang)
assert clients[sid].target_language == "fr"
# PM - issues with socketio enter_room in these tests
# @pytest.mark.asyncio
# async def test_call_user(mock_client):
# sid = mock_client.sid
# clients[sid] = mock_client
# call_id = "1234"
# await call_user(sid, call_id)
# assert call_id in rooms
# assert sid in rooms[call_id]
# @pytest.mark.asyncio
# async def test_answer_call(mock_client):
# sid = mock_client.sid
# clients[sid] = mock_client
# call_id = "1234"
# await answer_call(sid, call_id)
# assert call_id in rooms
# assert sid in rooms[call_id]
@pytest.mark.asyncio
async def test_incoming_audio(mock_client):
sid = mock_client.sid
clients[sid] = mock_client
data = b"\x01"
call_id = "1234"
await incoming_audio(sid, data, call_id)
assert clients[sid].get_length() != 0
|