backend/tests/test_main.py

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