anna/emmental
anna
/
emmental
1
0
Fork 0
Code Issues 13 Pull Requests Projects 1 Releases 12 Wiki Activity

rind: Gst unit test cleanups 

I noticed that I was getting some unexpected behavior due to the global
Audio class still running in parallel with my testing class. This patch
fixes that by adding an easy disconnect function for signals, and also
implements some other improvements to waiting for state changes.

Additionally, I added a check to see if we have already set the
playbin's "uri" property before picking the next track when calling
next().

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
Anna Schumaker 2019-03-24 17:24:46 -04:00
parent 3afffc54d4
commit e433097309
2 changed files with 75 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
rind/gst.py
View File

@ -24,21 +24,32 @@ class EmmentalAudio:
self.bus = self.playbin.get_bus()
self.bus.add_signal_watch()
self.bus.connect("message", self.on_message)
PauseButton.connect("clicked", self.pause)
PlayButton.connect( "clicked", self.play)
NextButton.connect( "clicked", self.next)
ProgScale.connect("change-value", self.seek)
self.bus_cb = self.bus.connect("message", self.on_message)
self.pause_cb = PauseButton.connect("clicked", self.pause)
self.play_cb = PlayButton.connect( "clicked", self.play)
self.next_cb = NextButton.connect( "clicked", self.next)
self.seek_cb = ProgScale.connect("change-value", self.seek)
self.playbin.connect("about-to-finish", self.about_to_finish)
self.timeout = GLib.timeout_add(100, self.update_progress)
self.finish_cb = self.playbin.connect("about-to-finish",
self.about_to_finish)
self.timeout_cb = GLib.timeout_add(100, self.update_progress)
def about_to_finish(self, playbin):
track = curds.PlaylistManager.next()
if track:
self.playbin.set_property("uri", f"file://{track.path}")
def disconnect(self):
self.playbin.disconnect(self.finish_cb)
self.bus.disconnect(self.bus_cb)
PauseButton.disconnect(self.pause_cb)
PlayButton.disconnect(self.play_cb)
NextButton.disconnect(self.next_cb)
ProgScale.disconnect(self.seek_cb)
GLib.source_remove(self.timeout_cb)
self.playbin.set_state(Gst.State.NULL)
def duration(self):
(res, cur) = self.playbin.query_duration(Gst.Format.TIME)
return cur / Gst.SECOND if res == True else 0.0
@ -50,19 +61,18 @@ class EmmentalAudio:
self.playbin.set_state(Gst.State.PLAYING)
def next(self, *args):
self.load(curds.PlaylistManager.next())
if self.playbin.get_property("uri") == None:
self.load(curds.PlaylistManager.next())
else:
self.load(curds.PlaylistManager.track)
def on_message(self, bus, message):
if message.type == Gst.MessageType.ASYNC_DONE:
state = message.src.get_state(Gst.Format.TIME)
self.state_changed(message.src, state[1])
if message.type == Gst.MessageType.STATE_CHANGED:
(old, new, pending) = message.parse_state_changed()
PlayButton.set_visible( new != Gst.State.PLAYING)
PauseButton.set_visible(new == Gst.State.PLAYING)
ProgScale.set_show_fill_level(new == Gst.State.PLAYING or
new == Gst.State.PAUSED)
if new == Gst.State.READY:
Title.set_markup("<big>Emmental</big>")
Artist.set_markup("<big>The Cheesy Music Player</big>")
curds.Notify.notify("state-changed", new)
self.state_changed(message.src, new)
elif message.type == Gst.MessageType.TAG:
taglist = message.parse_tag()
for i in range(taglist.n_tags()):
@ -73,7 +83,6 @@ class EmmentalAudio:
elif name == "title":
title = GLib.markup_escape_text(taglist.get_string(name)[1])
Title.set_markup(f"<big>{title}</big>")
return True
def pause(self, *args):
self.playbin.set_state(Gst.State.PAUSED)
@ -96,6 +105,17 @@ class EmmentalAudio:
pos = (dur * value) / 100.0
self.playbin.seek_simple(Gst.Format.TIME, Gst.SeekFlags.FLUSH, pos)
def state_changed(self, src, state):
if src == self.playbin:
PlayButton.set_visible( state != Gst.State.PLAYING)
PauseButton.set_visible(state == Gst.State.PLAYING)
ProgScale.set_show_fill_level(state == Gst.State.PLAYING or
state == Gst.State.PAUSED)
if state == Gst.State.READY:
Title.set_markup("<big>Emmental</big>")
Artist.set_markup("<big>The Cheesy Music Player</big>")
curds.Notify.notify("state-changed", state)
def update_progress(self):
(p_m, p_s) = divmod(int(self.position()), 60)
(d_m, d_s) = divmod(int(self.duration()) - int(self.position()), 60)

67
rind/test_gst.py
View File

@ -20,6 +20,7 @@ class TestGst(unittest.TestCase):
curds.Notify.notify_me("state-changed", self.on_state_changed)
def setUpClass():
gst.Audio.disconnect()
if not curds.playlist.library.library_thread.is_alive():
curds.playlist.library.library_thread = curds.ThreadQueue()
@ -28,15 +29,19 @@ class TestGst(unittest.TestCase):
self.audio.seek(value=0)
time.sleep(0.1)
self.main_loop()
GLib.source_remove(self.audio.timeout)
self.audio.disconnect()
def tearDownClass():
curds.playlist.library.library_thread.stop()
def main_loop(self):
events = GLib.main_context_default().iteration(may_block=True)
while events == True:
events = GLib.main_context_default().iteration(may_block=False)
def main_loop(self, delay=0.0):
time.sleep(delay)
while Gtk.events_pending():
Gtk.main_iteration_do(True)
def main_loop_until(self, state):
while self.state != state:
Gtk.main_iteration_do(True)
def on_state_changed(self, state):
self.state = state
@ -89,15 +94,12 @@ class TestGst(unittest.TestCase):
def test_gst_load(self):
track = curds.Track.lookup(os.path.join(test_album, "10 - Test Track 10.ogg"))
self.audio.load(track)
while self.state != Gst.State.PLAYING:
time.sleep(0.1)
self.main_loop()
self.assertEqual(self.state, Gst.State.PLAYING)
self.main_loop_until(Gst.State.PLAYING)
self.assertTrue( gst.ProgScale.get_show_fill_level())
self.main_loop(delay=0.1)
self.assertGreater(self.audio.progress(), 0)
self.assertGreater(self.audio.position(), 0)
self.assertEqual( self.audio.duration(), 10.0)
@ -116,35 +118,42 @@ class TestGst(unittest.TestCase):
curds.playlist.library.library_thread.join()
gst.NextButton.clicked()
self.assertIsNotNone(curds.PlaylistManager.track)
while self.state != Gst.State.PLAYING:
time.sleep(0.1)
self.main_loop()
track1 = curds.PlaylistManager.track
self.assertIsNotNone(track1)
gst.NextButton.clicked()
self.assertNotEqual(track1, curds.PlaylistManager.track)
track2 = curds.PlaylistManager.track
self.main_loop_until(Gst.State.PLAYING)
self.assertTrue( gst.PauseButton.is_visible())
self.assertFalse(gst.PlayButton.is_visible())
self.assertEqual(curds.PlaylistManager.track, track2)
gst.PauseButton.clicked()
while self.state != Gst.State.PAUSED:
self.main_loop()
self.main_loop_until(Gst.State.PAUSED)
self.assertFalse(gst.PauseButton.is_visible())
self.assertTrue( gst.PlayButton.is_visible())
self.assertEqual(curds.PlaylistManager.track, track2)
self.audio.seek(value=50.0)
time.sleep(0.1)
self.main_loop()
self.assertEqual(self.audio.duration(), 3.0)
self.assertEqual(self.audio.position(), 1.5)
self.main_loop(delay=0.1)
self.assertEqual(self.audio.duration(), 2.0)
self.assertEqual(self.audio.position(), 1.0)
self.assertEqual(gst.Position.get_text(), "00:01")
self.assertEqual(gst.Duration.get_text(), "-00:02")
self.assertEqual(gst.Duration.get_text(), "-00:01")
gst.PlayButton.clicked()
while self.state != Gst.State.PLAYING:
self.main_loop()
self.main_loop_until(Gst.State.PLAYING)
self.assertTrue( gst.PauseButton.is_visible())
self.assertFalse(gst.PlayButton.is_visible())
track = curds.PlaylistManager.track
self.audio.seek(value=99)
time.sleep(0.1)
self.main_loop()
self.assertNotEqual(curds.PlaylistManager.track, track)
gst.NextButton.clicked()
self.main_loop_until(Gst.State.PLAYING)
self.assertNotEqual(curds.PlaylistManager.track, track2)
track3 = curds.PlaylistManager.track
self.audio.seek(value=90)
self.main_loop_until(Gst.State.PLAYING)
self.main_loop(delay=1)
self.assertNotEqual(curds.PlaylistManager.track, track3)