andpp/mopidy
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Switched playback test over to testing via MpdSession slightly higher level testing

Browse Source
This commit is contained in:
Thomas Adamcik 2011-07-19 22:41:58 +02:00
parent 6cf5deb216
commit da9b6470ba
3 changed files with 203 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
mopidy/frontends/mpd/__init__.py
View File

@ -49,8 +49,8 @@ class MpdSession(network.LineProtocol):
terminator = protocol.LINE_TERMINATOR
encoding = protocol.ENCODING
def __init__(self, client):
super(MpdSession, self).__init__(client)
def __init__(self, connection):
super(MpdSession, self).__init__(connection)
self.dispatcher = dispatcher.MpdDispatcher(self)
def on_start(self):
@ -67,7 +67,7 @@ class MpdSession(network.LineProtocol):
logger.debug(u'Response to [%s]:%s from %s: %s', self.host, self.port,
self.actor_urn, log.indent(self.terminator.join(response)))
self.send_lines(response)
def close(self):

43
tests/frontends/mpd/protocol/__init__.py Normal file
View File

@ -0,0 +1,43 @@
import unittest
import mock
from mopidy.backends import dummy as backend
from mopidy.frontends import mpd
from mopidy.frontends.mpd import dispatcher
from mopidy.mixers import dummy as mixer
from mopidy.utils import network
class MockConnetion(mock.Mock):
def __init__(self, *args, **kwargs):
super(MockConnetion, self).__init__(*args, **kwargs)
self.host = mock.sentinel.host
self.port = mock.sentinel.port
self.response = []
def send(self, data):
self.response.extend(data.split('\n'))
class BaseTestCase(unittest.TestCase):
def setUp(self):
self.backend = backend.DummyBackend.start().proxy()
self.mixer = mixer.DummyMixer.start().proxy()
self.dispatcher = dispatcher.MpdDispatcher()
self.connection = MockConnetion()
self.session = mpd.MpdSession(self.connection)
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
def sendRequest(self, request, clear=False):
self.connection.response = []
self.session.on_line_received(request)
def assertResponse(self, value, index=None):
if index is not None:
self.assertEqual(value, self.connection.response[index])
else:
self.assert_(value in self.connection.response)

311
tests/frontends/mpd/playback_test.py → tests/frontends/mpd/protocol/playback_test.py
View File

@ -1,163 +1,150 @@
import unittest
from mopidy.backends.base import PlaybackController
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
from mopidy.mixers.dummy import DummyMixer
from mopidy.models import Track
from tests import SkipTest
from tests.frontends.mpd import protocol
PAUSED = PlaybackController.PAUSED
PLAYING = PlaybackController.PLAYING
STOPPED = PlaybackController.STOPPED
class PlaybackOptionsHandlerTest(unittest.TestCase):
def setUp(self):
self.backend = DummyBackend.start().proxy()
self.mixer = DummyMixer.start().proxy()
self.dispatcher = MpdDispatcher()
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
class PlaybackOptionsHandlerTest(protocol.BaseTestCase):
def test_consume_off(self):
result = self.dispatcher.handle_request(u'consume "0"')
self.sendRequest(u'consume "0"')
self.assertFalse(self.backend.playback.consume.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_consume_off_without_quotes(self):
result = self.dispatcher.handle_request(u'consume 0')
self.sendRequest(u'consume 0')
self.assertFalse(self.backend.playback.consume.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_consume_on(self):
result = self.dispatcher.handle_request(u'consume "1"')
self.sendRequest(u'consume "1"')
self.assertTrue(self.backend.playback.consume.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_consume_on_without_quotes(self):
result = self.dispatcher.handle_request(u'consume 1')
self.sendRequest(u'consume 1')
self.assertTrue(self.backend.playback.consume.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_crossfade(self):
result = self.dispatcher.handle_request(u'crossfade "10"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
self.sendRequest(u'crossfade "10"')
self.assertResponse(u'ACK [0@0] {} Not implemented')
def test_random_off(self):
result = self.dispatcher.handle_request(u'random "0"')
self.sendRequest(u'random "0"')
self.assertFalse(self.backend.playback.random.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_random_off_without_quotes(self):
result = self.dispatcher.handle_request(u'random 0')
self.sendRequest(u'random 0')
self.assertFalse(self.backend.playback.random.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_random_on(self):
result = self.dispatcher.handle_request(u'random "1"')
self.sendRequest(u'random "1"')
self.assertTrue(self.backend.playback.random.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_random_on_without_quotes(self):
result = self.dispatcher.handle_request(u'random 1')
self.sendRequest(u'random 1')
self.assertTrue(self.backend.playback.random.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_repeat_off(self):
result = self.dispatcher.handle_request(u'repeat "0"')
self.sendRequest(u'repeat "0"')
self.assertFalse(self.backend.playback.repeat.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_repeat_off_without_quotes(self):
result = self.dispatcher.handle_request(u'repeat 0')
self.sendRequest(u'repeat 0')
self.assertFalse(self.backend.playback.repeat.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_repeat_on(self):
result = self.dispatcher.handle_request(u'repeat "1"')
self.sendRequest(u'repeat "1"')
self.assertTrue(self.backend.playback.repeat.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_repeat_on_without_quotes(self):
result = self.dispatcher.handle_request(u'repeat 1')
self.sendRequest(u'repeat 1')
self.assertTrue(self.backend.playback.repeat.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_setvol_below_min(self):
result = self.dispatcher.handle_request(u'setvol "-10"')
self.assert_(u'OK' in result)
self.sendRequest(u'setvol "-10"')
self.assertEqual(0, self.mixer.volume.get())
self.assertResponse(u'OK')
def test_setvol_min(self):
result = self.dispatcher.handle_request(u'setvol "0"')
self.assert_(u'OK' in result)
self.sendRequest(u'setvol "0"')
self.assertEqual(0, self.mixer.volume.get())
self.assertResponse(u'OK')
def test_setvol_middle(self):
result = self.dispatcher.handle_request(u'setvol "50"')
self.assert_(u'OK' in result)
self.sendRequest(u'setvol "50"')
self.assertEqual(50, self.mixer.volume.get())
self.assertResponse(u'OK')
def test_setvol_max(self):
result = self.dispatcher.handle_request(u'setvol "100"')
self.assert_(u'OK' in result)
self.sendRequest(u'setvol "100"')
self.assertEqual(100, self.mixer.volume.get())
self.assertResponse(u'OK')
def test_setvol_above_max(self):
result = self.dispatcher.handle_request(u'setvol "110"')
self.assert_(u'OK' in result)
self.sendRequest(u'setvol "110"')
self.assertEqual(100, self.mixer.volume.get())
self.assertResponse(u'OK')
def test_setvol_plus_is_ignored(self):
result = self.dispatcher.handle_request(u'setvol "+10"')
self.assert_(u'OK' in result)
self.sendRequest(u'setvol "+10"')
self.assertEqual(10, self.mixer.volume.get())
self.assertResponse(u'OK')
def test_setvol_without_quotes(self):
result = self.dispatcher.handle_request(u'setvol 50')
self.assert_(u'OK' in result)
self.sendRequest(u'setvol 50')
self.assertEqual(50, self.mixer.volume.get())
self.assertResponse(u'OK')
def test_single_off(self):
result = self.dispatcher.handle_request(u'single "0"')
self.sendRequest(u'single "0"')
self.assertFalse(self.backend.playback.single.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_single_off_without_quotes(self):
result = self.dispatcher.handle_request(u'single 0')
self.sendRequest(u'single 0')
self.assertFalse(self.backend.playback.single.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_single_on(self):
result = self.dispatcher.handle_request(u'single "1"')
self.sendRequest(u'single "1"')
self.assertTrue(self.backend.playback.single.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_single_on_without_quotes(self):
result = self.dispatcher.handle_request(u'single 1')
self.sendRequest(u'single 1')
self.assertTrue(self.backend.playback.single.get())
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
def test_replay_gain_mode_off(self):
result = self.dispatcher.handle_request(u'replay_gain_mode "off"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
self.sendRequest(u'replay_gain_mode "off"')
self.assertResponse(u'ACK [0@0] {} Not implemented')
def test_replay_gain_mode_track(self):
result = self.dispatcher.handle_request(u'replay_gain_mode "track"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
self.sendRequest(u'replay_gain_mode "track"')
self.assertResponse(u'ACK [0@0] {} Not implemented')
def test_replay_gain_mode_album(self):
result = self.dispatcher.handle_request(u'replay_gain_mode "album"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
self.sendRequest(u'replay_gain_mode "album"')
self.assertResponse(u'ACK [0@0] {} Not implemented')
def test_replay_gain_status_default(self):
expected = u'off'
result = self.dispatcher.handle_request(u'replay_gain_status')
self.assert_(u'OK' in result)
self.assert_(expected in result)
self.sendRequest(u'replay_gain_status')
self.assertResponse(u'OK')
self.assertResponse(u'off')
def test_replay_gain_status_off(self):
raise SkipTest # TODO
@ -169,79 +156,80 @@ class PlaybackOptionsHandlerTest(unittest.TestCase):
raise SkipTest # TODO
class PlaybackControlHandlerTest(unittest.TestCase):
def setUp(self):
self.backend = DummyBackend.start().proxy()
self.mixer = DummyMixer.start().proxy()
self.dispatcher = MpdDispatcher()
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
class PlaybackControlHandlerTest(protocol.BaseTestCase):
def test_next(self):
result = self.dispatcher.handle_request(u'next')
self.assert_(u'OK' in result)
self.sendRequest(u'next')
self.assertResponse(u'OK')
def test_pause_off(self):
self.backend.current_playlist.append([Track()])
self.dispatcher.handle_request(u'play "0"')
self.dispatcher.handle_request(u'pause "1"')
result = self.dispatcher.handle_request(u'pause "0"')
self.assert_(u'OK' in result)
self.sendRequest(u'play "0"')
self.sendRequest(u'pause "1"')
self.sendRequest(u'pause "0"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertResponse(u'OK')
def test_pause_on(self):
self.backend.current_playlist.append([Track()])
self.dispatcher.handle_request(u'play "0"')
result = self.dispatcher.handle_request(u'pause "1"')
self.assert_(u'OK' in result)
self.sendRequest(u'play "0"')
self.sendRequest(u'pause "1"')
self.assertEqual(PAUSED, self.backend.playback.state.get())
self.assertResponse(u'OK')
def test_pause_toggle(self):
self.backend.current_playlist.append([Track()])
result = self.dispatcher.handle_request(u'play "0"')
self.assert_(u'OK' in result)
self.sendRequest(u'play "0"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
result = self.dispatcher.handle_request(u'pause')
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
self.sendRequest(u'pause')
self.assertEqual(PAUSED, self.backend.playback.state.get())
result = self.dispatcher.handle_request(u'pause')
self.assert_(u'OK' in result)
self.assertResponse(u'OK')
self.sendRequest(u'pause')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertResponse(u'OK')
def test_play_without_pos(self):
self.backend.current_playlist.append([Track()])
self.backend.playback.state = PAUSED
result = self.dispatcher.handle_request(u'play')
self.assert_(u'OK' in result)
self.sendRequest(u'play')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertResponse(u'OK')
def test_play_with_pos(self):
self.backend.current_playlist.append([Track()])
result = self.dispatcher.handle_request(u'play "0"')
self.assert_(u'OK' in result)
self.sendRequest(u'play "0"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertResponse(u'OK')
def test_play_with_pos_without_quotes(self):
self.backend.current_playlist.append([Track()])
result = self.dispatcher.handle_request(u'play 0')
self.assert_(u'OK' in result)
self.sendRequest(u'play 0')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertResponse(u'OK')
def test_play_with_pos_out_of_bounds(self):
self.backend.current_playlist.append([])
result = self.dispatcher.handle_request(u'play "0"')
self.assertEqual(result[0], u'ACK [2@0] {play} Bad song index')
self.sendRequest(u'play "0"')
self.assertEqual(STOPPED, self.backend.playback.state.get())
self.assertResponse(u'ACK [2@0] {play} Bad song index', index=0)
def test_play_minus_one_plays_first_in_playlist_if_no_current_track(self):
self.assertEqual(self.backend.playback.current_track.get(), None)
self.backend.current_playlist.append([Track(uri='a'), Track(uri='b')])
result = self.dispatcher.handle_request(u'play "-1"')
self.assert_(u'OK' in result)
self.sendRequest(u'play "-1"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertEqual(self.backend.playback.current_track.get().uri, 'a')
self.assertEqual('a', self.backend.playback.current_track.get().uri)
self.assertResponse(u'OK')
def test_play_minus_one_plays_current_track_if_current_track_is_set(self):
self.backend.current_playlist.append([Track(uri='a'), Track(uri='b')])
@ -250,27 +238,30 @@ class PlaybackControlHandlerTest(unittest.TestCase):
self.backend.playback.next()
self.backend.playback.stop()
self.assertNotEqual(self.backend.playback.current_track.get(), None)
result = self.dispatcher.handle_request(u'play "-1"')
self.assert_(u'OK' in result)
self.sendRequest(u'play "-1"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertEqual(self.backend.playback.current_track.get().uri, 'b')
self.assertEqual('b', self.backend.playback.current_track.get().uri)
self.assertResponse(u'OK')
def test_play_minus_one_on_empty_playlist_does_not_ack(self):
self.backend.current_playlist.clear()
result = self.dispatcher.handle_request(u'play "-1"')
self.assert_(u'OK' in result)
self.sendRequest(u'play "-1"')
self.assertEqual(STOPPED, self.backend.playback.state.get())
self.assertEqual(self.backend.playback.current_track.get(), None)
self.assertEqual(None, self.backend.playback.current_track.get())
self.assertResponse(u'OK')
def test_play_minus_is_ignored_if_playing(self):
self.backend.current_playlist.append([Track(length=40000)])
self.backend.playback.seek(30000)
self.assert_(self.backend.playback.time_position.get() >= 30000)
self.assertEquals(PLAYING, self.backend.playback.state.get())
result = self.dispatcher.handle_request(u'play "-1"')
self.assert_(u'OK' in result)
self.sendRequest(u'play "-1"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assert_(self.backend.playback.time_position.get() >= 30000)
self.assertResponse(u'OK')
def test_play_minus_one_resumes_if_paused(self):
self.backend.current_playlist.append([Track(length=40000)])
@ -279,24 +270,27 @@ class PlaybackControlHandlerTest(unittest.TestCase):
self.assertEquals(PLAYING, self.backend.playback.state.get())
self.backend.playback.pause()
self.assertEquals(PAUSED, self.backend.playback.state.get())
result = self.dispatcher.handle_request(u'play "-1"')
self.assert_(u'OK' in result)
self.sendRequest(u'play "-1"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assert_(self.backend.playback.time_position.get() >= 30000)
self.assertResponse(u'OK')
def test_playid(self):
self.backend.current_playlist.append([Track()])
result = self.dispatcher.handle_request(u'playid "0"')
self.assert_(u'OK' in result)
self.sendRequest(u'playid "0"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertResponse(u'OK')
def test_playid_minus_one_plays_first_in_playlist_if_no_current_track(self):
self.assertEqual(self.backend.playback.current_track.get(), None)
self.backend.current_playlist.append([Track(uri='a'), Track(uri='b')])
result = self.dispatcher.handle_request(u'playid "-1"')
self.assert_(u'OK' in result)
self.sendRequest(u'playid "-1"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertEqual(self.backend.playback.current_track.get().uri, 'a')
self.assertEqual('a', self.backend.playback.current_track.get().uri)
self.assertResponse(u'OK')
def test_playid_minus_one_plays_current_track_if_current_track_is_set(self):
self.backend.current_playlist.append([Track(uri='a'), Track(uri='b')])
@ -304,28 +298,31 @@ class PlaybackControlHandlerTest(unittest.TestCase):
self.backend.playback.play()
self.backend.playback.next()
self.backend.playback.stop()
self.assertNotEqual(self.backend.playback.current_track.get(), None)
result = self.dispatcher.handle_request(u'playid "-1"')
self.assert_(u'OK' in result)
self.assertNotEqual(None, self.backend.playback.current_track.get())
self.sendRequest(u'playid "-1"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertEqual(self.backend.playback.current_track.get().uri, 'b')
self.assertEqual('b', self.backend.playback.current_track.get().uri)
self.assertResponse(u'OK')
def test_playid_minus_one_on_empty_playlist_does_not_ack(self):
self.backend.current_playlist.clear()
result = self.dispatcher.handle_request(u'playid "-1"')
self.assert_(u'OK' in result)
self.sendRequest(u'playid "-1"')
self.assertEqual(STOPPED, self.backend.playback.state.get())
self.assertEqual(self.backend.playback.current_track.get(), None)
self.assertEqual(None, self.backend.playback.current_track.get())
self.assertResponse(u'OK')
def test_playid_minus_is_ignored_if_playing(self):
self.backend.current_playlist.append([Track(length=40000)])
self.backend.playback.seek(30000)
self.assert_(self.backend.playback.time_position.get() >= 30000)
self.assertEquals(PLAYING, self.backend.playback.state.get())
result = self.dispatcher.handle_request(u'playid "-1"')
self.assert_(u'OK' in result)
self.sendRequest(u'playid "-1"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assert_(self.backend.playback.time_position.get() >= 30000)
self.assertResponse(u'OK')
def test_playid_minus_one_resumes_if_paused(self):
self.backend.current_playlist.append([Track(length=40000)])
@ -334,58 +331,64 @@ class PlaybackControlHandlerTest(unittest.TestCase):
self.assertEquals(PLAYING, self.backend.playback.state.get())
self.backend.playback.pause()
self.assertEquals(PAUSED, self.backend.playback.state.get())
result = self.dispatcher.handle_request(u'playid "-1"')
self.assert_(u'OK' in result)
self.sendRequest(u'playid "-1"')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assert_(self.backend.playback.time_position.get() >= 30000)
self.assertResponse(u'OK')
def test_playid_which_does_not_exist(self):
self.backend.current_playlist.append([Track()])
result = self.dispatcher.handle_request(u'playid "12345"')
self.assertEqual(result[0], u'ACK [50@0] {playid} No such song')
self.sendRequest(u'playid "12345"')
self.assertResponse(u'ACK [50@0] {playid} No such song', index=0)
def test_previous(self):
result = self.dispatcher.handle_request(u'previous')
self.assert_(u'OK' in result)
self.sendRequest(u'previous')
self.assertResponse(u'OK')
def test_seek(self):
self.backend.current_playlist.append([Track(length=40000)])
self.dispatcher.handle_request(u'seek "0"')
result = self.dispatcher.handle_request(u'seek "0" "30"')
self.assert_(u'OK' in result)
self.sendRequest(u'seek "0"')
self.sendRequest(u'seek "0" "30"')
self.assert_(self.backend.playback.time_position >= 30000)
self.assertResponse(u'OK')
def test_seek_with_songpos(self):
seek_track = Track(uri='2', length=40000)
self.backend.current_playlist.append(
[Track(uri='1', length=40000), seek_track])
result = self.dispatcher.handle_request(u'seek "1" "30"')
self.assert_(u'OK' in result)
self.sendRequest(u'seek "1" "30"')
self.assertEqual(self.backend.playback.current_track.get(), seek_track)
self.assertResponse(u'OK')
def test_seek_without_quotes(self):
self.backend.current_playlist.append([Track(length=40000)])
self.dispatcher.handle_request(u'seek 0')
result = self.dispatcher.handle_request(u'seek 0 30')
self.assert_(u'OK' in result)
self.sendRequest(u'seek 0')
self.sendRequest(u'seek 0 30')
self.assert_(self.backend.playback.time_position.get() >= 30000)
self.assertResponse(u'OK')
def test_seekid(self):
self.backend.current_playlist.append([Track(length=40000)])
result = self.dispatcher.handle_request(u'seekid "0" "30"')
self.assert_(u'OK' in result)
self.sendRequest(u'seekid "0" "30"')
self.assert_(self.backend.playback.time_position.get() >= 30000)
self.assertResponse(u'OK')
def test_seekid_with_cpid(self):
seek_track = Track(uri='2', length=40000)
self.backend.current_playlist.append(
[Track(length=40000), seek_track])
result = self.dispatcher.handle_request(u'seekid "1" "30"')
self.assert_(u'OK' in result)
self.assertEqual(self.backend.playback.current_cpid.get(), 1)
self.assertEqual(self.backend.playback.current_track.get(), seek_track)
self.sendRequest(u'seekid "1" "30"')
self.assertEqual(1, self.backend.playback.current_cpid.get())
self.assertEqual(seek_track, self.backend.playback.current_track.get())
self.assertResponse(u'OK')
def test_stop(self):
result = self.dispatcher.handle_request(u'stop')
self.assert_(u'OK' in result)
self.sendRequest(u'stop')
self.assertEqual(STOPPED, self.backend.playback.state.get())
self.assertResponse(u'OK')