pstruebi
dc8b6cc66e
- Implent support to play precoded files - Implement a basic client -server architecture Reviewed-on: https://gitea.pstruebi.xyz/auracaster/bumble-auracast/pulls/2
74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
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 |