backend/tests/test_main.py

from dotenv import dotenv_values
from fastapi import FastAPI
from pymongo import MongoClient
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, app
from unittest.mock import patch

sio = socketio.AsyncServer(
    async_mode="asgi",
    cors_allowed_origins="*",
    # engineio_logger=logger,
)

config = dotenv_values(".env")

# Read connection string from environment vars
# uri = os.environ['MONGODB_URI']

# Read connection string from .env file
uri = config['MONGODB_URI']
app.mongodb_client = MongoClient(uri)
app.database = app.mongodb_client['IT-Cluster1'] #connect to interpretalk primary db
try:
    app.mongodb_client.admin.command('ping')
    print("MongoDB Connection Established...")
except Exception as e:
    print(e)
# shutdown logic
print("Closing MongoDB Connection...")

@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", "testusername", 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)
    app.mongodb_client.close()
    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"

@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