dnd-music-console/test/test_streamer.py

import io
import queue
import threading
from pathlib import Path
from unittest.mock import MagicMock

import pytest
import shout

from croaker import playlist, streamer


def get_stream_output(stream):
    return stream.read()


@pytest.fixture(scope="session")
def silence_bytes():
    return (Path(streamer.__file__).parent / "silence.mp3").read_bytes()


@pytest.fixture
def output_stream():
    return io.BytesIO()


@pytest.fixture
def mock_shout(output_stream, monkeypatch):
    def handle_send(buf):
        output_stream.write(buf)

    mm = MagicMock(spec=shout.Shout, **{"return_value.send.side_effect": handle_send})
    monkeypatch.setattr("shout.Shout", mm)
    return mm


@pytest.fixture
def input_queue():
    return queue.Queue()


@pytest.fixture
def skip_event():
    return threading.Event()


@pytest.fixture
def stop_event():
    return threading.Event()


@pytest.fixture
def load_event():
    return threading.Event()


@pytest.fixture
def audio_streamer(mock_shout, input_queue, skip_event, stop_event, load_event):
    return streamer.AudioStreamer(input_queue, skip_event, stop_event, load_event)


def test_streamer_stop(audio_streamer, stop_event, output_stream):
    stop_event.set()
    audio_streamer.stream_queued_audio()
    assert not stop_event.is_set()


def test_streamer_skip(audio_streamer, skip_event, output_stream):
    skip_event.set()
    audio_streamer.stream_queued_audio()
    assert not skip_event.is_set()


def test_streamer_load(audio_streamer, load_event, output_stream):
    load_event.set()
    audio_streamer.stream_queued_audio()
    assert not load_event.is_set()


def test_clear_queue(audio_streamer, input_queue):
    pl = playlist.Playlist(name="test_playlist")
    for track in pl.tracks:
        input_queue.put(bytes(track))
    assert input_queue.not_empty
    audio_streamer.clear_queue()
    assert input_queue.empty


@pytest.mark.skip
def test_streamer_defaults_to_silence(audio_streamer, input_queue, output_stream, silence_bytes):
    audio_streamer.stream_queued_audio()
    track = playlist.Playlist(name="test_playlist").tracks[0]
    input_queue.put(bytes(track))
    audio_streamer.stream_queued_audio()
    audio_streamer.stream_queued_audio()
    assert get_stream_output(output_stream) == silence_bytes + track.read_bytes() + silence_bytes


@pytest.mark.skip
def test_streamer_plays_silence_on_error(monkeypatch, audio_streamer, input_queue, output_stream, silence_bytes):
    monkeypatch.setattr(audio_streamer.queue, "get", MagicMock(side_effect=Exception))
    track = playlist.Playlist(name="test_playlist").tracks[0]
    input_queue.put(bytes(track))
    audio_streamer.stream_queued_audio()
    assert get_stream_output(output_stream) == silence_bytes


@pytest.mark.skip
def test_streamer_plays_from_queue(audio_streamer, input_queue, output_stream):
    pl = playlist.Playlist(name="test_playlist")
    expected = b""
    for track in pl.tracks:
        input_queue.put(bytes(track))
        expected += track.read_bytes()
    while not input_queue.empty():
        audio_streamer.stream_queued_audio()
    assert get_stream_output(output_stream) == expected


def test_streamer_handles_stop_interrupt(audio_streamer, output_stream, stop_event):
    stop_event.set()
    audio_streamer.stream_queued_audio()
    assert get_stream_output(output_stream) == b""


def test_streamer_handles_load_interrupt(audio_streamer, input_queue, output_stream, load_event):
    pl = playlist.Playlist(name="test_playlist")
    input_queue.put(bytes(pl.tracks[0]))
    load_event.set()
    audio_streamer.stream_queued_audio()
    assert get_stream_output(output_stream) == b""
    assert input_queue.empty