379 lines
16 KiB
Python
379 lines
16 KiB
Python
# Copyright 2022 (c) Anna Schumaker.
|
|
"""Tests our GObject audio player wrapping a GStreamer Playbin element."""
|
|
import io
|
|
import pathlib
|
|
import unittest
|
|
import unittest.mock
|
|
import emmental.audio
|
|
import tests.util
|
|
from gi.repository import GObject
|
|
from gi.repository import GLib
|
|
from gi.repository import Gst
|
|
|
|
|
|
class TestAudio(unittest.TestCase):
|
|
"""Our audio player test case."""
|
|
|
|
def setUp(self):
|
|
"""Set up common variables."""
|
|
self.player = emmental.audio.Player()
|
|
|
|
def tearDown(self):
|
|
"""Clean up the playbin."""
|
|
self.player.shutdown()
|
|
|
|
def main_loop(self) -> None:
|
|
"""Run a GLib main loop."""
|
|
while GLib.main_context_default().iteration():
|
|
pass
|
|
|
|
def test_constants(self):
|
|
"""Test audio player constants."""
|
|
self.assertEqual(emmental.audio.UPDATE_INTERVAL, 100)
|
|
self.assertEqual(emmental.audio.SEEK_FLAGS,
|
|
Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT)
|
|
|
|
def test_player(self):
|
|
"""Test that the audio player was set up correctly."""
|
|
self.assertIsInstance(self.player, GObject.GObject)
|
|
self.assertEqual(self.player.get_state(), Gst.State.READY)
|
|
|
|
self.assertIsNone(self.player.file)
|
|
self.assertFalse(self.player.have_track)
|
|
self.assertEqual(self.player.status, "Stopped")
|
|
|
|
def test_playbin(self):
|
|
"""Test that the playbin was configured correctly."""
|
|
self.assertIsInstance(self.player._playbin, Gst.Element)
|
|
self.assertIsInstance(self.player._playbin.get_property("video-sink"),
|
|
Gst.Element)
|
|
self.assertIsNone(self.player._timeout, None)
|
|
|
|
self.assertRegex(self.player._playbin.name, r"playbin\d+")
|
|
self.assertRegex(self.player._playbin.get_property("video-sink").name,
|
|
r"fakesink\d+")
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_filter(self, mock_stdout: io.StringIO):
|
|
"""Test the filter element added to the playbin."""
|
|
self.assertIsInstance(self.player._filter,
|
|
emmental.audio.filter.Filter)
|
|
self.assertEqual(self.player._playbin.get_property("audio-filter"),
|
|
self.player._filter)
|
|
|
|
self.assertFalse(self.player.bg_enabled)
|
|
self.assertEqual(self.player.bg_volume, 0.5)
|
|
|
|
self.player.bg_enabled = True
|
|
self.player.bg_volume = 0.75
|
|
self.assertTrue(self.player._filter.bg_enabled)
|
|
self.assertEqual(self.player._filter.bg_volume, 0.75)
|
|
|
|
self.player.bg_enabled = False
|
|
self.player.bg_volume = 0.5
|
|
self.assertFalse(self.player._filter.bg_enabled)
|
|
self.assertEqual(self.player.bg_volume, 0.5)
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_eos(self, mock_stdout: io.StringIO):
|
|
"""Test handling an EOS message."""
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.player.duration = 10
|
|
self.player.position = 8
|
|
self.player.playtime = 6
|
|
self.player.almost_done = True
|
|
self.player.artwork = pathlib.Path("/a/b/c.jpg")
|
|
self.player.stopwatch.reset = unittest.mock.Mock()
|
|
|
|
eos = Gst.Message.new_eos(self.player._playbin)
|
|
self.player._playbin.get_bus().post(eos)
|
|
self.main_loop()
|
|
|
|
self.assertRegex(mock_stdout.getvalue(), "audio: end of stream")
|
|
|
|
for prop in ["artist", "album-artist", "album", "title"]:
|
|
self.assertEqual(self.player.get_property(prop), "")
|
|
for prop in ["album-disc-number", "track-number",
|
|
"position", "duration", "playtime"]:
|
|
self.assertEqual(self.player.get_property(prop), 0)
|
|
self.assertIsNone(self.player.artwork)
|
|
self.player.stopwatch.reset.assert_called()
|
|
|
|
self.assertEqual(self.player.get_state(), Gst.State.READY)
|
|
self.assertEqual(self.player.status, "Stopped")
|
|
self.assertFalse(self.player.have_track)
|
|
self.assertFalse(self.player.almost_done)
|
|
self.assertIsNone(self.player.file)
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_almost_done(self, mock_stdout: io.StringIO):
|
|
"""Test notifying that the current track is almost done."""
|
|
self.assertFalse(self.player.almost_done)
|
|
|
|
self.player.almost_done = True
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.player.pause()
|
|
self.main_loop()
|
|
self.assertFalse(self.player.almost_done)
|
|
|
|
self.player.duration = (5 * Gst.SECOND) / Gst.USECOND
|
|
with unittest.mock.patch.object(self.player._playbin,
|
|
"query_position") as query_position:
|
|
query_position.return_value = (True, 2 * Gst.SECOND)
|
|
self.player._Player__update_position()
|
|
self.assertFalse(self.player.almost_done)
|
|
self.assertNotRegex(mock_stdout.getvalue(),
|
|
"audio: about to finish")
|
|
|
|
query_position.return_value = (True, 3 * Gst.SECOND)
|
|
self.player._Player__update_position()
|
|
self.assertTrue(self.player.almost_done)
|
|
self.assertRegex(mock_stdout.getvalue(), "audio: about to finish")
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_file(self, mock_stdout: io.StringIO):
|
|
"""Test that the file property works as expected."""
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.assertEqual(self.player._playbin.get_property("uri"),
|
|
tests.util.TRACK_OGG.as_uri())
|
|
self.assertEqual(self.player.title, "")
|
|
self.assertEqual(self.player.album, "")
|
|
|
|
started = unittest.mock.Mock()
|
|
self.player.connect("file-loaded", started)
|
|
self.player.pause()
|
|
self.main_loop()
|
|
started.assert_called_with(self.player, tests.util.TRACK_OGG)
|
|
self.assertTrue(self.player.have_track)
|
|
|
|
self.assertEqual(mock_stdout.getvalue(),
|
|
f"audio: loading {tests.util.TRACK_OGG.as_uri()}\n"
|
|
"audio: file loaded\n"
|
|
"audio: state changed to 'paused'\n")
|
|
|
|
self.player.playtime = 6
|
|
self.player.stopwatch.reset = unittest.mock.Mock()
|
|
self.player.emit("file-loaded", tests.util.TRACK_OGG)
|
|
for prop in ["artist", "album-artist", "album", "title"]:
|
|
self.assertEqual(self.player.get_property(prop), "")
|
|
for prop in ["album-disc-number", "track-number", "playtime"]:
|
|
self.assertEqual(self.player.get_property(prop), 0)
|
|
self.player.stopwatch.reset.assert_called()
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_play(self, mock_stdout: io.StringIO):
|
|
"""Test that the play() function works as expected."""
|
|
self.player.stopwatch.start = unittest.mock.Mock()
|
|
|
|
self.player.play()
|
|
self.main_loop()
|
|
self.assertFalse(self.player.playing)
|
|
self.assertEqual(self.player.status, "Stopped")
|
|
self.player.stopwatch.start.assert_not_called()
|
|
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.player.play()
|
|
self.main_loop()
|
|
self.assertTrue(self.player.playing)
|
|
self.assertEqual(self.player.status, "Playing")
|
|
self.assertEqual(self.player.get_state(), Gst.State.PLAYING)
|
|
self.assertRegex(mock_stdout.getvalue(),
|
|
"audio: state changed to 'playing'$")
|
|
self.player.stopwatch.start.assert_called()
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_pause(self, mock_stdout: io.StringIO):
|
|
"""Test that the pause() function works as expected."""
|
|
self.player.stopwatch.stop = unittest.mock.Mock()
|
|
|
|
self.player.pause()
|
|
self.main_loop()
|
|
self.assertEqual(self.player.status, "Stopped")
|
|
self.player.stopwatch.stop.assert_not_called()
|
|
|
|
self.player.playing = True
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.player.pause()
|
|
self.main_loop()
|
|
self.assertFalse(self.player.playing)
|
|
self.assertEqual(self.player.status, "Paused")
|
|
self.assertEqual(self.player.get_state(), Gst.State.PAUSED)
|
|
self.assertRegex(mock_stdout.getvalue(),
|
|
"audio: state changed to 'paused'$")
|
|
self.player.stopwatch.stop.assert_called()
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_play_pause(self, mock_stdout: io.StringIO):
|
|
"""Test that the play_pause() function works as expected."""
|
|
self.player.play_pause()
|
|
self.main_loop()
|
|
self.assertFalse(self.player.playing)
|
|
self.assertEqual(self.player.status, "Stopped")
|
|
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.player.play_pause()
|
|
self.main_loop()
|
|
self.assertTrue(self.player.playing)
|
|
self.assertEqual(self.player.status, "Playing")
|
|
self.assertEqual(self.player.get_state(), Gst.State.PLAYING)
|
|
self.assertRegex(mock_stdout.getvalue(),
|
|
"audio: state changed to 'playing'$")
|
|
|
|
self.player.play_pause()
|
|
self.main_loop()
|
|
self.assertFalse(self.player.playing)
|
|
self.assertEqual(self.player.status, "Paused")
|
|
self.assertEqual(self.player.get_state(), Gst.State.PAUSED)
|
|
self.assertRegex(mock_stdout.getvalue(),
|
|
"audio: state changed to 'playing'\n"
|
|
"audio: state changed to 'paused'\n$")
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_stop(self, mock_stdout: io.StringIO):
|
|
"""Test that the stop() function works as expected."""
|
|
self.player.stopwatch.stop = unittest.mock.Mock()
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.player.play()
|
|
self.main_loop()
|
|
|
|
self.player.stop()
|
|
self.main_loop()
|
|
self.assertFalse(self.player.playing)
|
|
self.assertEqual(self.player.status, "Stopped")
|
|
self.assertEqual(self.player.get_state(), Gst.State.READY)
|
|
self.assertRegex(mock_stdout.getvalue(),
|
|
"audio: state changed to 'playing'\n"
|
|
"audio: state changed to 'stopped'\n$")
|
|
self.player.stopwatch.stop.assert_called()
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_tags(self, mock_stdout: io.StringIO):
|
|
"""Test that the tag properties work as expected."""
|
|
for prop in ["artist", "album", "album-artist", "title"]:
|
|
self.assertEqual(self.player.get_property(prop), "")
|
|
for prop in ["album-disc-number", "track-number"]:
|
|
self.assertEqual(self.player.get_property(prop), 0)
|
|
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.player.pause()
|
|
self.main_loop()
|
|
self.assertEqual(self.player.artist, "Test Artist")
|
|
self.assertEqual(self.player.album, "Test Album")
|
|
self.assertEqual(self.player.album_artist, "Test Album Artist")
|
|
self.assertEqual(self.player.album_disc_number, 1)
|
|
self.assertEqual(self.player.title, "Test Title")
|
|
self.assertEqual(self.player.track_number, 1)
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_duration(self, mock_stdout: io.StringIO):
|
|
"""Test the duration property."""
|
|
self.assertEqual(self.player.duration, 0)
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.player.pause()
|
|
self.main_loop()
|
|
self.assertEqual(self.player.duration, 10 * Gst.SECOND / Gst.USECOND)
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_position(self, mock_stdout: io.StringIO):
|
|
"""Test the position property."""
|
|
self.assertEqual(self.player.position, 0)
|
|
|
|
query_position = unittest.mock.Mock(return_value=(True, Gst.SECOND))
|
|
self.player._playbin.query_position = query_position
|
|
self.player._Player__update_position()
|
|
self.assertEqual(self.player.position, Gst.SECOND / Gst.USECOND)
|
|
|
|
seek_simple = unittest.mock.Mock()
|
|
self.player._playbin.seek_simple = seek_simple
|
|
self.player.seek(5 * Gst.SECOND / Gst.USECOND)
|
|
seek_simple.assert_called_with(Gst.Format.TIME,
|
|
emmental.audio.SEEK_FLAGS,
|
|
5 * Gst.SECOND)
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_playtime(self, mock_stdout: io.StringIO):
|
|
"""Test the play time property."""
|
|
self.assertIsInstance(self.player.stopwatch,
|
|
emmental.audio.stopwatch.StopWatch)
|
|
self.assertEqual(self.player.playtime, 0.0)
|
|
|
|
with unittest.mock.patch.object(self.player.stopwatch,
|
|
"elapsed_time") as mock_elapsed:
|
|
mock_elapsed.return_value = 2.0
|
|
self.player._Player__update_position()
|
|
self.assertEqual(self.player.playtime, 2.0)
|
|
|
|
def test_volume(self):
|
|
"""Test that the volume property works as expected."""
|
|
self.assertEqual(self.player.volume, 1.0)
|
|
self.assertEqual(self.player._playbin.get_property("volume"), 1.0)
|
|
self.player.volume = 0.5
|
|
self.assertEqual(self.player.volume, 0.5)
|
|
self.assertEqual(self.player._playbin.get_property("volume"), 0.5)
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_replaygain(self, mock_stdout: io.StringIO):
|
|
"""Test that ReplayGain functions work as expected."""
|
|
self.assertEqual(self.player._filter.rg_mode, "disabled")
|
|
self.assertEqual(self.player.get_replaygain(), (False, None))
|
|
|
|
self.player.set_replaygain(True, "album")
|
|
self.assertEqual(self.player._filter.rg_mode, "album")
|
|
self.assertEqual(self.player.get_replaygain(), (True, "album"))
|
|
self.assertRegex(mock_stdout.getvalue(),
|
|
r"audio: setting ReplayGain mode to 'album'")
|
|
|
|
self.player.set_replaygain(False, "track")
|
|
self.assertEqual(self.player._filter.rg_mode, "disabled")
|
|
self.assertEqual(self.player.get_replaygain(), (False, None))
|
|
self.assertRegex(mock_stdout.getvalue(),
|
|
r"audio: setting ReplayGain mode to 'disabled'")
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_artwork(self, mock_stdout: io.StringIO):
|
|
"""Test that we handle album artwork."""
|
|
self.assertIsNone(self.player.artwork)
|
|
|
|
with unittest.mock.patch.object(pathlib.Path, "is_file",
|
|
return_value=False):
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.player.pause()
|
|
self.main_loop()
|
|
self.assertIsNone(self.player.artwork)
|
|
|
|
self.player.stop()
|
|
self.player.pause()
|
|
self.main_loop()
|
|
self.assertEqual(self.player.artwork, tests.util.COVER_JPG)
|
|
|
|
buffer = Gst.Buffer.new_wrapped_bytes(tests.util.COVER_JPG_BYTES)
|
|
taglist = Gst.TagList.new_empty()
|
|
taglist.add_value(Gst.TagMergeMode.APPEND, "image",
|
|
Gst.Sample.new(buffer))
|
|
tag = Gst.Message.new_tag(self.player._playbin, taglist)
|
|
self.player._playbin.get_bus().post(tag)
|
|
self.main_loop()
|
|
|
|
expected = emmental.tmpdir.cover_jpg()
|
|
self.assertEqual(self.player.artwork,
|
|
emmental.tmpdir.cover_jpg())
|
|
self.assertTrue(expected.is_file())
|
|
|
|
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
|
def test_pause_on_load(self, mock_stdout: io.StringIO):
|
|
"""Test pausing the next time a file is loaded."""
|
|
self.assertFalse(self.player.pause_on_load)
|
|
|
|
self.player.pause_on_load = True
|
|
self.player.file = tests.util.TRACK_OGG
|
|
self.player.play()
|
|
|
|
self.main_loop()
|
|
self.assertFalse(self.player.playing)
|
|
self.assertFalse(self.player.pause_on_load)
|
|
|
|
def test_shutdown(self):
|
|
"""Test that the shutdown function works as expected."""
|
|
self.player.shutdown()
|
|
self.assertEqual(self.player.get_state(), Gst.State.NULL)
|