feature/precode (#2)
- Implent support to play precoded files - Implement a basic client -server architecture Reviewed-on: https://gitea.pstruebi.xyz/auracaster/bumble-auracast/pulls/2
This commit was merged in pull request #2.
This commit is contained in:
74
tests/test_server.py
Normal file
74
tests/test_server.py
Normal file
@@ -0,0 +1,74 @@
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
import asyncio
|
||||
from quart import jsonify
|
||||
from auracast.utils.read_lc3_file import read_lc3_file
|
||||
from auracast import multicast_server
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
"""Fixture to create and return a Quart test client."""
|
||||
yield multicast_server.app.test_client()
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def init_broadcast(client):
|
||||
response = await client.post('/init')
|
||||
json_data = await response.get_json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert json_data["status"] == "initialized"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shutdown(client, init_broadcast):
|
||||
"""Tests the /shutdown endpoint."""
|
||||
|
||||
response = await client.post('/shutdown')
|
||||
json_data = await response.get_json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert json_data["status"] == "stopped"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_audio(client, init_broadcast):
|
||||
"""Tests the /stop_audio endpoint."""
|
||||
|
||||
response = await client.post('/stop_audio')
|
||||
json_data = await response.get_json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert json_data["status"] == "stopped"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_audio(client, init_broadcast):
|
||||
"""Tests the /stream_lc3 endpoint."""
|
||||
test_audio_data = { # TODO: investigate further whats the best way to actually transfer the data
|
||||
"broadcast_de": read_lc3_file('src/auracast/testdata/announcement_de_10_16_32.lc3').decode('latin-1'),
|
||||
"broadcast_en": read_lc3_file('src/auracast/testdata/announcement_en_10_16_32.lc3').decode('latin-1')
|
||||
}
|
||||
|
||||
response = await client.post('/stream_lc3', json=test_audio_data)
|
||||
json_data = await response.get_json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert json_data["status"] == "audio_sent"
|
||||
|
||||
await asyncio.wait([multicast_server.multicaster.streamer.task], timeout=10)
|
||||
# Ensure the audio data is correctly assigned
|
||||
for key, val in multicast_server.big_conf.items():
|
||||
if key in test_audio_data:
|
||||
assert val.audio_source == test_audio_data[key].encode('latin-1')
|
||||
else:
|
||||
assert val.audio_source == b""
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_status(client):
|
||||
"""Tests the /status endpoint."""
|
||||
response = await client.get('/status')
|
||||
json_data = await response.get_json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "status" in json_data
|
||||
Reference in New Issue
Block a user