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

Merge pull request #124 from adamcik/feature/higher-level-protocol-testing

Browse Source
This commit is contained in:
Stein Magnus Jodal 2011-07-20 15:55:59 -07:00
commit 54b226eda4
23 changed files with 1176 additions and 1277 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):

2
mopidy/frontends/mpd/protocol/empty.py
View File

@ -1,6 +1,6 @@
from mopidy.frontends.mpd.protocol import handle_request
@handle_request(r'^$')
@handle_request(r'^\s*$')
def empty(context):
"""The original MPD server returns ``OK`` on an empty request."""
pass

30
tests/frontends/mpd/audio_output_test.py
View File

@ -1,30 +0,0 @@
import unittest
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
from mopidy.mixers.dummy import DummyMixer
class AudioOutputHandlerTest(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()
def test_enableoutput(self):
result = self.dispatcher.handle_request(u'enableoutput "0"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_disableoutput(self):
result = self.dispatcher.handle_request(u'disableoutput "0"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_outputs(self):
result = self.dispatcher.handle_request(u'outputs')
self.assert_(u'outputid: 0' in result)
self.assert_(u'outputname: None' in result)
self.assert_(u'outputenabled: 1' in result)
self.assert_(u'OK' in result)

63
tests/frontends/mpd/command_list_test.py
View File

@ -1,63 +0,0 @@
import unittest
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd import dispatcher
from mopidy.mixers.dummy import DummyMixer
class CommandListsTest(unittest.TestCase):
def setUp(self):
self.b = DummyBackend.start().proxy()
self.mixer = DummyMixer.start().proxy()
self.dispatcher = dispatcher.MpdDispatcher()
def tearDown(self):
self.b.stop().get()
self.mixer.stop().get()
def test_command_list_begin(self):
result = self.dispatcher.handle_request(u'command_list_begin')
self.assertEquals(result, [])
def test_command_list_end(self):
self.dispatcher.handle_request(u'command_list_begin')
result = self.dispatcher.handle_request(u'command_list_end')
self.assert_(u'OK' in result)
def test_command_list_end_without_start_first_is_an_unknown_command(self):
result = self.dispatcher.handle_request(u'command_list_end')
self.assertEquals(result[0],
u'ACK [5@0] {} unknown command "command_list_end"')
def test_command_list_with_ping(self):
self.dispatcher.handle_request(u'command_list_begin')
self.assertEqual([], self.dispatcher.command_list)
self.assertEqual(False, self.dispatcher.command_list_ok)
self.dispatcher.handle_request(u'ping')
self.assert_(u'ping' in self.dispatcher.command_list)
result = self.dispatcher.handle_request(u'command_list_end')
self.assert_(u'OK' in result)
self.assertEqual(False, self.dispatcher.command_list)
def test_command_list_with_error_returns_ack_with_correct_index(self):
self.dispatcher.handle_request(u'command_list_begin')
self.dispatcher.handle_request(u'play') # Known command
self.dispatcher.handle_request(u'paly') # Unknown command
result = self.dispatcher.handle_request(u'command_list_end')
self.assertEqual(len(result), 1, result)
self.assertEqual(result[0], u'ACK [5@1] {} unknown command "paly"')
def test_command_list_ok_begin(self):
result = self.dispatcher.handle_request(u'command_list_ok_begin')
self.assertEquals(result, [])
def test_command_list_ok_with_ping(self):
self.dispatcher.handle_request(u'command_list_ok_begin')
self.assertEqual([], self.dispatcher.command_list)
self.assertEqual(True, self.dispatcher.command_list_ok)
self.dispatcher.handle_request(u'ping')
self.assert_(u'ping' in self.dispatcher.command_list)
result = self.dispatcher.handle_request(u'command_list_end')
self.assert_(u'list_OK' in result)
self.assert_(u'OK' in result)
self.assertEqual(False, self.dispatcher.command_list)
self.assertEqual(False, self.dispatcher.command_list_ok)

53
tests/frontends/mpd/connection_test.py
View File

@ -1,53 +0,0 @@
import mock
import unittest
from mopidy import settings
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd import MpdSession
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
from mopidy.mixers.dummy import DummyMixer
class ConnectionHandlerTest(unittest.TestCase):
def setUp(self):
self.backend = DummyBackend.start().proxy()
self.mixer = DummyMixer.start().proxy()
self.session = mock.Mock(spec=MpdSession)
self.dispatcher = MpdDispatcher(session=self.session)
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
settings.runtime.clear()
def test_close_closes_the_client_connection(self):
result = self.dispatcher.handle_request(u'close')
self.assert_(self.session.close.called,
u'Should call close() on MpdSession')
self.assert_(u'OK' in result)
def test_empty_request(self):
result = self.dispatcher.handle_request(u'')
self.assert_(u'OK' in result)
def test_kill(self):
result = self.dispatcher.handle_request(u'kill')
self.assert_(u'ACK [4@0] {kill} you don\'t have permission for "kill"' in result)
def test_valid_password_is_accepted(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
result = self.dispatcher.handle_request(u'password "topsecret"')
self.assert_(u'OK' in result)
def test_invalid_password_is_not_accepted(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
result = self.dispatcher.handle_request(u'password "secret"')
self.assert_(u'ACK [3@0] {password} incorrect password' in result)
def test_any_password_is_not_accepted_when_password_check_turned_off(self):
settings.MPD_SERVER_PASSWORD = None
result = self.dispatcher.handle_request(u'password "secret"')
self.assert_(u'ACK [3@0] {password} incorrect password' in result)
def test_ping(self):
result = self.dispatcher.handle_request(u'ping')
self.assert_(u'OK' in result)

412
tests/frontends/mpd/music_db_test.py
View File

@ -1,412 +0,0 @@
import unittest
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
from mopidy.mixers.dummy import DummyMixer
class MusicDatabaseHandlerTest(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()
def test_count(self):
result = self.dispatcher.handle_request(u'count "tag" "needle"')
self.assert_(u'songs: 0' in result)
self.assert_(u'playtime: 0' in result)
self.assert_(u'OK' in result)
def test_findadd(self):
result = self.dispatcher.handle_request(u'findadd "album" "what"')
self.assert_(u'OK' in result)
def test_listall(self):
result = self.dispatcher.handle_request(
u'listall "file:///dev/urandom"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_listallinfo(self):
result = self.dispatcher.handle_request(
u'listallinfo "file:///dev/urandom"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_lsinfo_without_path_returns_same_as_listplaylists(self):
lsinfo_result = self.dispatcher.handle_request(u'lsinfo')
listplaylists_result = self.dispatcher.handle_request(u'listplaylists')
self.assertEqual(lsinfo_result, listplaylists_result)
def test_lsinfo_with_empty_path_returns_same_as_listplaylists(self):
lsinfo_result = self.dispatcher.handle_request(u'lsinfo ""')
listplaylists_result = self.dispatcher.handle_request(u'listplaylists')
self.assertEqual(lsinfo_result, listplaylists_result)
def test_lsinfo_for_root_returns_same_as_listplaylists(self):
lsinfo_result = self.dispatcher.handle_request(u'lsinfo "/"')
listplaylists_result = self.dispatcher.handle_request(u'listplaylists')
self.assertEqual(lsinfo_result, listplaylists_result)
def test_update_without_uri(self):
result = self.dispatcher.handle_request(u'update')
self.assert_(u'OK' in result)
self.assert_(u'updating_db: 0' in result)
def test_update_with_uri(self):
result = self.dispatcher.handle_request(u'update "file:///dev/urandom"')
self.assert_(u'OK' in result)
self.assert_(u'updating_db: 0' in result)
def test_rescan_without_uri(self):
result = self.dispatcher.handle_request(u'rescan')
self.assert_(u'OK' in result)
self.assert_(u'updating_db: 0' in result)
def test_rescan_with_uri(self):
result = self.dispatcher.handle_request(u'rescan "file:///dev/urandom"')
self.assert_(u'OK' in result)
self.assert_(u'updating_db: 0' in result)
class MusicDatabaseFindTest(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()
def test_find_album(self):
result = self.dispatcher.handle_request(u'find "album" "what"')
self.assert_(u'OK' in result)
def test_find_album_without_quotes(self):
result = self.dispatcher.handle_request(u'find album "what"')
self.assert_(u'OK' in result)
def test_find_artist(self):
result = self.dispatcher.handle_request(u'find "artist" "what"')
self.assert_(u'OK' in result)
def test_find_artist_without_quotes(self):
result = self.dispatcher.handle_request(u'find artist "what"')
self.assert_(u'OK' in result)
def test_find_title(self):
result = self.dispatcher.handle_request(u'find "title" "what"')
self.assert_(u'OK' in result)
def test_find_title_without_quotes(self):
result = self.dispatcher.handle_request(u'find title "what"')
self.assert_(u'OK' in result)
def test_find_date(self):
result = self.dispatcher.handle_request(u'find "date" "2002-01-01"')
self.assert_(u'OK' in result)
def test_find_date_without_quotes(self):
result = self.dispatcher.handle_request(u'find date "2002-01-01"')
self.assert_(u'OK' in result)
def test_find_date_with_capital_d_and_incomplete_date(self):
result = self.dispatcher.handle_request(u'find Date "2005"')
self.assert_(u'OK' in result)
def test_find_else_should_fail(self):
result = self.dispatcher.handle_request(u'find "somethingelse" "what"')
self.assertEqual(result[0], u'ACK [2@0] {find} incorrect arguments')
def test_find_album_and_artist(self):
result = self.dispatcher.handle_request(
u'find album "album_what" artist "artist_what"')
self.assert_(u'OK' in result)
class MusicDatabaseListTest(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()
def test_list_foo_returns_ack(self):
result = self.dispatcher.handle_request(u'list "foo"')
self.assertEqual(result[0],
u'ACK [2@0] {list} incorrect arguments')
### Artist
def test_list_artist_with_quotes(self):
result = self.dispatcher.handle_request(u'list "artist"')
self.assert_(u'OK' in result)
def test_list_artist_without_quotes(self):
result = self.dispatcher.handle_request(u'list artist')
self.assert_(u'OK' in result)
def test_list_artist_without_quotes_and_capitalized(self):
result = self.dispatcher.handle_request(u'list Artist')
self.assert_(u'OK' in result)
def test_list_artist_with_query_of_one_token(self):
result = self.dispatcher.handle_request(u'list "artist" "anartist"')
self.assertEqual(result[0],
u'ACK [2@0] {list} should be "Album" for 3 arguments')
def test_list_artist_with_unknown_field_in_query_returns_ack(self):
result = self.dispatcher.handle_request(u'list "artist" "foo" "bar"')
self.assertEqual(result[0],
u'ACK [2@0] {list} not able to parse args')
def test_list_artist_by_artist(self):
result = self.dispatcher.handle_request(
u'list "artist" "artist" "anartist"')
self.assert_(u'OK' in result)
def test_list_artist_by_album(self):
result = self.dispatcher.handle_request(
u'list "artist" "album" "analbum"')
self.assert_(u'OK' in result)
def test_list_artist_by_full_date(self):
result = self.dispatcher.handle_request(
u'list "artist" "date" "2001-01-01"')
self.assert_(u'OK' in result)
def test_list_artist_by_year(self):
result = self.dispatcher.handle_request(
u'list "artist" "date" "2001"')
self.assert_(u'OK' in result)
def test_list_artist_by_genre(self):
result = self.dispatcher.handle_request(
u'list "artist" "genre" "agenre"')
self.assert_(u'OK' in result)
def test_list_artist_by_artist_and_album(self):
result = self.dispatcher.handle_request(
u'list "artist" "artist" "anartist" "album" "analbum"')
self.assert_(u'OK' in result)
### Album
def test_list_album_with_quotes(self):
result = self.dispatcher.handle_request(u'list "album"')
self.assert_(u'OK' in result)
def test_list_album_without_quotes(self):
result = self.dispatcher.handle_request(u'list album')
self.assert_(u'OK' in result)
def test_list_album_without_quotes_and_capitalized(self):
result = self.dispatcher.handle_request(u'list Album')
self.assert_(u'OK' in result)
def test_list_album_with_artist_name(self):
result = self.dispatcher.handle_request(u'list "album" "anartist"')
self.assert_(u'OK' in result)
def test_list_album_by_artist(self):
result = self.dispatcher.handle_request(
u'list "album" "artist" "anartist"')
self.assert_(u'OK' in result)
def test_list_album_by_album(self):
result = self.dispatcher.handle_request(
u'list "album" "album" "analbum"')
self.assert_(u'OK' in result)
def test_list_album_by_full_date(self):
result = self.dispatcher.handle_request(
u'list "album" "date" "2001-01-01"')
self.assert_(u'OK' in result)
def test_list_album_by_year(self):
result = self.dispatcher.handle_request(
u'list "album" "date" "2001"')
self.assert_(u'OK' in result)
def test_list_album_by_genre(self):
result = self.dispatcher.handle_request(
u'list "album" "genre" "agenre"')
self.assert_(u'OK' in result)
def test_list_album_by_artist_and_album(self):
result = self.dispatcher.handle_request(
u'list "album" "artist" "anartist" "album" "analbum"')
self.assert_(u'OK' in result)
### Date
def test_list_date_with_quotes(self):
result = self.dispatcher.handle_request(u'list "date"')
self.assert_(u'OK' in result)
def test_list_date_without_quotes(self):
result = self.dispatcher.handle_request(u'list date')
self.assert_(u'OK' in result)
def test_list_date_without_quotes_and_capitalized(self):
result = self.dispatcher.handle_request(u'list Date')
self.assert_(u'OK' in result)
def test_list_date_with_query_of_one_token(self):
result = self.dispatcher.handle_request(u'list "date" "anartist"')
self.assertEqual(result[0],
u'ACK [2@0] {list} should be "Album" for 3 arguments')
def test_list_date_by_artist(self):
result = self.dispatcher.handle_request(
u'list "date" "artist" "anartist"')
self.assert_(u'OK' in result)
def test_list_date_by_album(self):
result = self.dispatcher.handle_request(
u'list "date" "album" "analbum"')
self.assert_(u'OK' in result)
def test_list_date_by_full_date(self):
result = self.dispatcher.handle_request(
u'list "date" "date" "2001-01-01"')
self.assert_(u'OK' in result)
def test_list_date_by_year(self):
result = self.dispatcher.handle_request(u'list "date" "date" "2001"')
self.assert_(u'OK' in result)
def test_list_date_by_genre(self):
result = self.dispatcher.handle_request(u'list "date" "genre" "agenre"')
self.assert_(u'OK' in result)
def test_list_date_by_artist_and_album(self):
result = self.dispatcher.handle_request(
u'list "date" "artist" "anartist" "album" "analbum"')
self.assert_(u'OK' in result)
### Genre
def test_list_genre_with_quotes(self):
result = self.dispatcher.handle_request(u'list "genre"')
self.assert_(u'OK' in result)
def test_list_genre_without_quotes(self):
result = self.dispatcher.handle_request(u'list genre')
self.assert_(u'OK' in result)
def test_list_genre_without_quotes_and_capitalized(self):
result = self.dispatcher.handle_request(u'list Genre')
self.assert_(u'OK' in result)
def test_list_genre_with_query_of_one_token(self):
result = self.dispatcher.handle_request(u'list "genre" "anartist"')
self.assertEqual(result[0],
u'ACK [2@0] {list} should be "Album" for 3 arguments')
def test_list_genre_by_artist(self):
result = self.dispatcher.handle_request(
u'list "genre" "artist" "anartist"')
self.assert_(u'OK' in result)
def test_list_genre_by_album(self):
result = self.dispatcher.handle_request(
u'list "genre" "album" "analbum"')
self.assert_(u'OK' in result)
def test_list_genre_by_full_date(self):
result = self.dispatcher.handle_request(
u'list "genre" "date" "2001-01-01"')
self.assert_(u'OK' in result)
def test_list_genre_by_year(self):
result = self.dispatcher.handle_request(
u'list "genre" "date" "2001"')
self.assert_(u'OK' in result)
def test_list_genre_by_genre(self):
result = self.dispatcher.handle_request(
u'list "genre" "genre" "agenre"')
self.assert_(u'OK' in result)
def test_list_genre_by_artist_and_album(self):
result = self.dispatcher.handle_request(
u'list "genre" "artist" "anartist" "album" "analbum"')
self.assert_(u'OK' in result)
class MusicDatabaseSearchTest(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()
def test_search_album(self):
result = self.dispatcher.handle_request(u'search "album" "analbum"')
self.assert_(u'OK' in result)
def test_search_album_without_quotes(self):
result = self.dispatcher.handle_request(u'search album "analbum"')
self.assert_(u'OK' in result)
def test_search_artist(self):
result = self.dispatcher.handle_request(u'search "artist" "anartist"')
self.assert_(u'OK' in result)
def test_search_artist_without_quotes(self):
result = self.dispatcher.handle_request(u'search artist "anartist"')
self.assert_(u'OK' in result)
def test_search_filename(self):
result = self.dispatcher.handle_request(
u'search "filename" "afilename"')
self.assert_(u'OK' in result)
def test_search_filename_without_quotes(self):
result = self.dispatcher.handle_request(u'search filename "afilename"')
self.assert_(u'OK' in result)
def test_search_title(self):
result = self.dispatcher.handle_request(u'search "title" "atitle"')
self.assert_(u'OK' in result)
def test_search_title_without_quotes(self):
result = self.dispatcher.handle_request(u'search title "atitle"')
self.assert_(u'OK' in result)
def test_search_any(self):
result = self.dispatcher.handle_request(u'search "any" "anything"')
self.assert_(u'OK' in result)
def test_search_any_without_quotes(self):
result = self.dispatcher.handle_request(u'search any "anything"')
self.assert_(u'OK' in result)
def test_search_date(self):
result = self.dispatcher.handle_request(u'search "date" "2002-01-01"')
self.assert_(u'OK' in result)
def test_search_date_without_quotes(self):
result = self.dispatcher.handle_request(u'search date "2002-01-01"')
self.assert_(u'OK' in result)
def test_search_date_with_capital_d_and_incomplete_date(self):
result = self.dispatcher.handle_request(u'search Date "2005"')
self.assert_(u'OK' in result)
def test_search_else_should_fail(self):
result = self.dispatcher.handle_request(
u'search "sometype" "something"')
self.assertEqual(result[0], u'ACK [2@0] {search} incorrect arguments')

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

@ -0,0 +1,51 @@
import unittest
import mock
from mopidy import settings
from mopidy.backends import dummy as backend
from mopidy.frontends import mpd
from mopidy.mixers import dummy as mixer
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):
lines = (line for line in data.split('\n') if line)
self.response.extend(lines)
class BaseTestCase(unittest.TestCase):
def setUp(self):
self.backend = backend.DummyBackend.start().proxy()
self.mixer = mixer.DummyMixer.start().proxy()
self.connection = MockConnetion()
self.session = mpd.MpdSession(self.connection)
self.dispatcher = self.session.dispatcher
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
settings.runtime.clear()
def sendRequest(self, request, clear=False):
self.connection.response = []
self.session.on_line_received(request)
return self.connection.response
def assertInResponse(self, value):
self.assert_(value in self.connection.response, u'Did not find %s '
'in %s' % (repr(value), repr(self.connection.response)))
def assertNotInResponse(self, value):
self.assert_(value not in self.connection.response, u'Found %s in %s' %
(repr(value), repr(self.connection.response)))
def assertEqualResponse(self, value):
self.assertEqual(1, len(self.connection.response))
self.assertEqual(value, self.connection.response[0])

17
tests/frontends/mpd/protocol/audio_output_test.py Normal file
View File

@ -0,0 +1,17 @@
from tests.frontends.mpd import protocol
class AudioOutputHandlerTest(protocol.BaseTestCase):
def test_enableoutput(self):
self.sendRequest(u'enableoutput "0"')
self.assertInResponse(u'ACK [0@0] {} Not implemented')
def test_disableoutput(self):
self.sendRequest(u'disableoutput "0"')
self.assertInResponse(u'ACK [0@0] {} Not implemented')
def test_outputs(self):
self.sendRequest(u'outputs')
self.assertInResponse(u'outputid: 0')
self.assertInResponse(u'outputname: None')
self.assertInResponse(u'outputenabled: 1')
self.assertInResponse(u'OK')

56
tests/frontends/mpd/authentication_test.py → tests/frontends/mpd/protocol/authentication_test.py
View File

@ -1,63 +1,61 @@
import mock
import unittest
from mopidy import settings
from mopidy.frontends.mpd import MpdSession
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
class AuthenticationTest(unittest.TestCase):
def setUp(self):
self.session = mock.Mock(spec=MpdSession)
self.dispatcher = MpdDispatcher(session=self.session)
def tearDown(self):
settings.runtime.clear()
from tests.frontends.mpd import protocol
class AuthenticationTest(protocol.BaseTestCase):
def test_authentication_with_valid_password_is_accepted(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
response = self.dispatcher.handle_request(u'password "topsecret"')
self.sendRequest(u'password "topsecret"')
self.assertTrue(self.dispatcher.authenticated)
self.assert_(u'OK' in response)
self.assertInResponse(u'OK')
def test_authentication_with_invalid_password_is_not_accepted(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
response = self.dispatcher.handle_request(u'password "secret"')
self.sendRequest(u'password "secret"')
self.assertFalse(self.dispatcher.authenticated)
self.assert_(u'ACK [3@0] {password} incorrect password' in response)
self.assertEqualResponse(u'ACK [3@0] {password} incorrect password')
def test_authentication_with_anything_when_password_check_turned_off(self):
settings.MPD_SERVER_PASSWORD = None
response = self.dispatcher.handle_request(u'any request at all')
self.sendRequest(u'any request at all')
self.assertTrue(self.dispatcher.authenticated)
self.assert_('ACK [5@0] {} unknown command "any"' in response)
self.assertEqualResponse('ACK [5@0] {} unknown command "any"')
def test_anything_when_not_authenticated_should_fail(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
response = self.dispatcher.handle_request(u'any request at all')
self.sendRequest(u'any request at all')
self.assertFalse(self.dispatcher.authenticated)
self.assert_(
u'ACK [4@0] {any} you don\'t have permission for "any"' in response)
self.assertEqualResponse(
u'ACK [4@0] {any} you don\'t have permission for "any"')
def test_close_is_allowed_without_authentication(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
response = self.dispatcher.handle_request(u'close')
self.sendRequest(u'close')
self.assertFalse(self.dispatcher.authenticated)
self.assert_(u'OK' in response)
self.assertInResponse(u'OK')
def test_commands_is_allowed_without_authentication(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
response = self.dispatcher.handle_request(u'commands')
self.sendRequest(u'commands')
self.assertFalse(self.dispatcher.authenticated)
self.assert_(u'OK' in response)
self.assertInResponse(u'OK')
def test_notcommands_is_allowed_without_authentication(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
response = self.dispatcher.handle_request(u'notcommands')
self.sendRequest(u'notcommands')
self.assertFalse(self.dispatcher.authenticated)
self.assert_(u'OK' in response)
self.assertInResponse(u'OK')
def test_ping_is_allowed_without_authentication(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
response = self.dispatcher.handle_request(u'ping')
self.sendRequest(u'ping')
self.assertFalse(self.dispatcher.authenticated)
self.assert_(u'OK' in response)
self.assertInResponse(u'OK')

53
tests/frontends/mpd/protocol/command_list_test.py Normal file
View File

@ -0,0 +1,53 @@
from tests.frontends.mpd import protocol
class CommandListsTest(protocol.BaseTestCase):
def test_command_list_begin(self):
response = self.sendRequest(u'command_list_begin')
self.assertEquals([], response)
def test_command_list_end(self):
self.sendRequest(u'command_list_begin')
self.sendRequest(u'command_list_end')
self.assertInResponse(u'OK')
def test_command_list_end_without_start_first_is_an_unknown_command(self):
self.sendRequest(u'command_list_end')
self.assertEqualResponse(
u'ACK [5@0] {} unknown command "command_list_end"')
def test_command_list_with_ping(self):
self.sendRequest(u'command_list_begin')
self.assertEqual([], self.dispatcher.command_list)
self.assertEqual(False, self.dispatcher.command_list_ok)
self.sendRequest(u'ping')
self.assert_(u'ping' in self.dispatcher.command_list)
self.sendRequest(u'command_list_end')
self.assertInResponse(u'OK')
self.assertEqual(False, self.dispatcher.command_list)
def test_command_list_with_error_returns_ack_with_correct_index(self):
self.sendRequest(u'command_list_begin')
self.sendRequest(u'play') # Known command
self.sendRequest(u'paly') # Unknown command
self.sendRequest(u'command_list_end')
self.assertEqualResponse(u'ACK [5@1] {} unknown command "paly"')
def test_command_list_ok_begin(self):
response = self.sendRequest(u'command_list_ok_begin')
self.assertEquals([], response)
def test_command_list_ok_with_ping(self):
self.sendRequest(u'command_list_ok_begin')
self.assertEqual([], self.dispatcher.command_list)
self.assertEqual(True, self.dispatcher.command_list_ok)
self.sendRequest(u'ping')
self.assert_(u'ping' in self.dispatcher.command_list)
self.sendRequest(u'command_list_end')
self.assertInResponse(u'list_OK')
self.assertInResponse(u'OK')
self.assertEqual(False, self.dispatcher.command_list)
self.assertEqual(False, self.dispatcher.command_list_ok)
# FIXME this should also include the special handling of idle within a
# command list. That is that once a idle/noidle command is found inside a
# commad list, the rest of the list seems to be ignored.

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

@ -0,0 +1,43 @@
from mock import patch
from mopidy import settings
from tests.frontends.mpd import protocol
class ConnectionHandlerTest(protocol.BaseTestCase):
def test_close_closes_the_client_connection(self):
with patch.object(self.session, 'close') as close_mock:
response = self.sendRequest(u'close')
close_mock.assertEqualResponsecalled_once_with()
self.assertEqualResponse(u'OK')
def test_empty_request(self):
self.sendRequest(u'')
self.assertEqualResponse(u'OK')
self.sendRequest(u' ')
self.assertEqualResponse(u'OK')
def test_kill(self):
self.sendRequest(u'kill')
self.assertEqualResponse(
u'ACK [4@0] {kill} you don\'t have permission for "kill"')
def test_valid_password_is_accepted(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
self.sendRequest(u'password "topsecret"')
self.assertEqualResponse(u'OK')
def test_invalid_password_is_not_accepted(self):
settings.MPD_SERVER_PASSWORD = u'topsecret'
self.sendRequest(u'password "secret"')
self.assertEqualResponse(u'ACK [3@0] {password} incorrect password')
def test_any_password_is_not_accepted_when_password_check_turned_off(self):
settings.MPD_SERVER_PASSWORD = None
self.sendRequest(u'password "secret"')
self.assertEqualResponse(u'ACK [3@0] {password} incorrect password')
def test_ping(self):
self.sendRequest(u'ping')
self.assertEqualResponse(u'OK')

351
tests/frontends/mpd/current_playlist_test.py → tests/frontends/mpd/protocol/current_playlist_test.py
View File

@ -1,20 +1,8 @@
import unittest
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
from mopidy.mixers.dummy import DummyMixer
from mopidy.models import Track
class CurrentPlaylistHandlerTest(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()
from tests.frontends.mpd import protocol
class CurrentPlaylistHandlerTest(protocol.BaseTestCase):
def test_add(self):
needle = Track(uri='dummy://foo')
self.backend.library.provider.dummy_library = [
@ -22,21 +10,21 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
self.backend.current_playlist.append(
[Track(), Track(), Track(), Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
result = self.dispatcher.handle_request(u'add "dummy://foo"')
self.assertEqual(len(result), 1)
self.assertEqual(result[0], u'OK')
self.sendRequest(u'add "dummy://foo"')
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 6)
self.assertEqual(self.backend.current_playlist.tracks.get()[5], needle)
self.assertEqualResponse(u'OK')
def test_add_with_uri_not_found_in_library_should_ack(self):
result = self.dispatcher.handle_request(u'add "dummy://foo"')
self.assertEqual(result[0],
self.sendRequest(u'add "dummy://foo"')
self.assertEqualResponse(
u'ACK [50@0] {add} directory or file not found')
def test_add_with_empty_uri_should_add_all_known_tracks_and_ok(self):
result = self.dispatcher.handle_request(u'add ""')
self.sendRequest(u'add ""')
# TODO check that we add all tracks (we currently don't)
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_addid_without_songpos(self):
needle = Track(uri='dummy://foo')
@ -45,16 +33,17 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
self.backend.current_playlist.append(
[Track(), Track(), Track(), Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
result = self.dispatcher.handle_request(u'addid "dummy://foo"')
self.sendRequest(u'addid "dummy://foo"')
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 6)
self.assertEqual(self.backend.current_playlist.tracks.get()[5], needle)
self.assert_(u'Id: %d' %
self.backend.current_playlist.cp_tracks.get()[5][0] in result)
self.assert_(u'OK' in result)
self.assertInResponse(u'Id: %d' %
self.backend.current_playlist.cp_tracks.get()[5][0])
self.assertInResponse(u'OK')
def test_addid_with_empty_uri_acks(self):
result = self.dispatcher.handle_request(u'addid ""')
self.assertEqual(result[0], u'ACK [50@0] {addid} No such song')
self.sendRequest(u'addid ""')
self.assertEqualResponse(u'ACK [50@0] {addid} No such song')
def test_addid_with_songpos(self):
needle = Track(uri='dummy://foo')
@ -63,12 +52,13 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
self.backend.current_playlist.append(
[Track(), Track(), Track(), Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
result = self.dispatcher.handle_request(u'addid "dummy://foo" "3"')
self.sendRequest(u'addid "dummy://foo" "3"')
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 6)
self.assertEqual(self.backend.current_playlist.tracks.get()[3], needle)
self.assert_(u'Id: %d' %
self.backend.current_playlist.cp_tracks.get()[3][0] in result)
self.assert_(u'OK' in result)
self.assertInResponse(u'Id: %d' %
self.backend.current_playlist.cp_tracks.get()[3][0])
self.assertInResponse(u'OK')
def test_addid_with_songpos_out_of_bounds_should_ack(self):
needle = Track(uri='dummy://foo')
@ -77,83 +67,93 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
self.backend.current_playlist.append(
[Track(), Track(), Track(), Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
result = self.dispatcher.handle_request(u'addid "dummy://foo" "6"')
self.assertEqual(result[0], u'ACK [2@0] {addid} Bad song index')
self.sendRequest(u'addid "dummy://foo" "6"')
self.assertEqualResponse(u'ACK [2@0] {addid} Bad song index')
def test_addid_with_uri_not_found_in_library_should_ack(self):
result = self.dispatcher.handle_request(u'addid "dummy://foo"')
self.assertEqual(result[0], u'ACK [50@0] {addid} No such song')
self.sendRequest(u'addid "dummy://foo"')
self.assertEqualResponse(u'ACK [50@0] {addid} No such song')
def test_clear(self):
self.backend.current_playlist.append(
[Track(), Track(), Track(), Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
result = self.dispatcher.handle_request(u'clear')
self.sendRequest(u'clear')
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 0)
self.assertEqual(self.backend.playback.current_track.get(), None)
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_delete_songpos(self):
self.backend.current_playlist.append(
[Track(), Track(), Track(), Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
result = self.dispatcher.handle_request(u'delete "%d"' %
self.sendRequest(u'delete "%d"' %
self.backend.current_playlist.cp_tracks.get()[2][0])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 4)
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_delete_songpos_out_of_bounds(self):
self.backend.current_playlist.append(
[Track(), Track(), Track(), Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
result = self.dispatcher.handle_request(u'delete "5"')
self.sendRequest(u'delete "5"')
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
self.assertEqual(result[0], u'ACK [2@0] {delete} Bad song index')
self.assertEqualResponse(u'ACK [2@0] {delete} Bad song index')
def test_delete_open_range(self):
self.backend.current_playlist.append(
[Track(), Track(), Track(), Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
result = self.dispatcher.handle_request(u'delete "1:"')
self.sendRequest(u'delete "1:"')
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 1)
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_delete_closed_range(self):
self.backend.current_playlist.append(
[Track(), Track(), Track(), Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
result = self.dispatcher.handle_request(u'delete "1:3"')
self.sendRequest(u'delete "1:3"')
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 3)
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_delete_range_out_of_bounds(self):
self.backend.current_playlist.append(
[Track(), Track(), Track(), Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
result = self.dispatcher.handle_request(u'delete "5:7"')
self.sendRequest(u'delete "5:7"')
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 5)
self.assertEqual(result[0], u'ACK [2@0] {delete} Bad song index')
self.assertEqualResponse(u'ACK [2@0] {delete} Bad song index')
def test_deleteid(self):
self.backend.current_playlist.append([Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 2)
result = self.dispatcher.handle_request(u'deleteid "1"')
self.sendRequest(u'deleteid "1"')
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 1)
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_deleteid_does_not_exist(self):
self.backend.current_playlist.append([Track(), Track()])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 2)
result = self.dispatcher.handle_request(u'deleteid "12345"')
self.sendRequest(u'deleteid "12345"')
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 2)
self.assertEqual(result[0], u'ACK [50@0] {deleteid} No such song')
self.assertEqualResponse(u'ACK [50@0] {deleteid} No such song')
def test_move_songpos(self):
self.backend.current_playlist.append([
Track(name='a'), Track(name='b'), Track(name='c'),
Track(name='d'), Track(name='e'), Track(name='f'),
])
result = self.dispatcher.handle_request(u'move "1" "0"')
self.sendRequest(u'move "1" "0"')
tracks = self.backend.current_playlist.tracks.get()
self.assertEqual(tracks[0].name, 'b')
self.assertEqual(tracks[1].name, 'a')
@ -161,14 +161,15 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
self.assertEqual(tracks[3].name, 'd')
self.assertEqual(tracks[4].name, 'e')
self.assertEqual(tracks[5].name, 'f')
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_move_open_range(self):
self.backend.current_playlist.append([
Track(name='a'), Track(name='b'), Track(name='c'),
Track(name='d'), Track(name='e'), Track(name='f'),
])
result = self.dispatcher.handle_request(u'move "2:" "0"')
self.sendRequest(u'move "2:" "0"')
tracks = self.backend.current_playlist.tracks.get()
self.assertEqual(tracks[0].name, 'c')
self.assertEqual(tracks[1].name, 'd')
@ -176,14 +177,15 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
self.assertEqual(tracks[3].name, 'f')
self.assertEqual(tracks[4].name, 'a')
self.assertEqual(tracks[5].name, 'b')
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_move_closed_range(self):
self.backend.current_playlist.append([
Track(name='a'), Track(name='b'), Track(name='c'),
Track(name='d'), Track(name='e'), Track(name='f'),
])
result = self.dispatcher.handle_request(u'move "1:3" "0"')
self.sendRequest(u'move "1:3" "0"')
tracks = self.backend.current_playlist.tracks.get()
self.assertEqual(tracks[0].name, 'b')
self.assertEqual(tracks[1].name, 'c')
@ -191,14 +193,15 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
self.assertEqual(tracks[3].name, 'd')
self.assertEqual(tracks[4].name, 'e')
self.assertEqual(tracks[5].name, 'f')
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_moveid(self):
self.backend.current_playlist.append([
Track(name='a'), Track(name='b'), Track(name='c'),
Track(name='d'), Track(name='e'), Track(name='f'),
])
result = self.dispatcher.handle_request(u'moveid "4" "2"')
self.sendRequest(u'moveid "4" "2"')
tracks = self.backend.current_playlist.tracks.get()
self.assertEqual(tracks[0].name, 'a')
self.assertEqual(tracks[1].name, 'b')
@ -206,179 +209,182 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
self.assertEqual(tracks[3].name, 'c')
self.assertEqual(tracks[4].name, 'd')
self.assertEqual(tracks[5].name, 'f')
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_playlist_returns_same_as_playlistinfo(self):
playlist_result = self.dispatcher.handle_request(u'playlist')
playlistinfo_result = self.dispatcher.handle_request(u'playlistinfo')
self.assertEqual(playlist_result, playlistinfo_result)
playlist_response = self.sendRequest(u'playlist')
playlistinfo_response = self.sendRequest(u'playlistinfo')
self.assertEqual(playlist_response, playlistinfo_response)
def test_playlistfind(self):
result = self.dispatcher.handle_request(u'playlistfind "tag" "needle"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
self.sendRequest(u'playlistfind "tag" "needle"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_playlistfind_by_filename_not_in_current_playlist(self):
result = self.dispatcher.handle_request(
u'playlistfind "filename" "file:///dev/null"')
self.assertEqual(len(result), 1)
self.assert_(u'OK' in result)
self.sendRequest(u'playlistfind "filename" "file:///dev/null"')
self.assertEqualResponse(u'OK')
def test_playlistfind_by_filename_without_quotes(self):
result = self.dispatcher.handle_request(
u'playlistfind filename "file:///dev/null"')
self.assertEqual(len(result), 1)
self.assert_(u'OK' in result)
self.sendRequest(u'playlistfind filename "file:///dev/null"')
self.assertEqualResponse(u'OK')
def test_playlistfind_by_filename_in_current_playlist(self):
self.backend.current_playlist.append([
Track(uri='file:///exists')])
result = self.dispatcher.handle_request(
u'playlistfind filename "file:///exists"')
self.assert_(u'file: file:///exists' in result)
self.assert_(u'Id: 0' in result)
self.assert_(u'Pos: 0' in result)
self.assert_(u'OK' in result)
self.sendRequest( u'playlistfind filename "file:///exists"')
self.assertInResponse(u'file: file:///exists')
self.assertInResponse(u'Id: 0')
self.assertInResponse(u'Pos: 0')
self.assertInResponse(u'OK')
def test_playlistid_without_songid(self):
self.backend.current_playlist.append([Track(name='a'), Track(name='b')])
result = self.dispatcher.handle_request(u'playlistid')
self.assert_(u'Title: a' in result)
self.assert_(u'Title: b' in result)
self.assert_(u'OK' in result)
self.sendRequest(u'playlistid')
self.assertInResponse(u'Title: a')
self.assertInResponse(u'Title: b')
self.assertInResponse(u'OK')
def test_playlistid_with_songid(self):
self.backend.current_playlist.append([Track(name='a'), Track(name='b')])
result = self.dispatcher.handle_request(u'playlistid "1"')
self.assert_(u'Title: a' not in result)
self.assert_(u'Id: 0' not in result)
self.assert_(u'Title: b' in result)
self.assert_(u'Id: 1' in result)
self.assert_(u'OK' in result)
self.sendRequest(u'playlistid "1"')
self.assertNotInResponse(u'Title: a')
self.assertNotInResponse(u'Id: 0')
self.assertInResponse(u'Title: b')
self.assertInResponse(u'Id: 1')
self.assertInResponse(u'OK')
def test_playlistid_with_not_existing_songid_fails(self):
self.backend.current_playlist.append([Track(name='a'), Track(name='b')])
result = self.dispatcher.handle_request(u'playlistid "25"')
self.assertEqual(result[0], u'ACK [50@0] {playlistid} No such song')
self.sendRequest(u'playlistid "25"')
self.assertEqualResponse(u'ACK [50@0] {playlistid} No such song')
def test_playlistinfo_without_songpos_or_range(self):
self.backend.current_playlist.append([
Track(name='a'), Track(name='b'), Track(name='c'),
Track(name='d'), Track(name='e'), Track(name='f'),
])
result = self.dispatcher.handle_request(u'playlistinfo')
self.assert_(u'Title: a' in result)
self.assert_(u'Title: b' in result)
self.assert_(u'Title: c' in result)
self.assert_(u'Title: d' in result)
self.assert_(u'Title: e' in result)
self.assert_(u'Title: f' in result)
self.assert_(u'OK' in result)
self.sendRequest(u'playlistinfo')
self.assertInResponse(u'Title: a')
self.assertInResponse(u'Title: b')
self.assertInResponse(u'Title: c')
self.assertInResponse(u'Title: d')
self.assertInResponse(u'Title: e')
self.assertInResponse(u'Title: f')
self.assertInResponse(u'OK')
def test_playlistinfo_with_songpos(self):
self.backend.current_playlist.append([
Track(name='a'), Track(name='b'), Track(name='c'),
Track(name='d'), Track(name='e'), Track(name='f'),
])
result = self.dispatcher.handle_request(u'playlistinfo "4"')
self.assert_(u'Title: a' not in result)
self.assert_(u'Title: b' not in result)
self.assert_(u'Title: c' not in result)
self.assert_(u'Title: d' not in result)
self.assert_(u'Title: e' in result)
self.assert_(u'Title: f' not in result)
self.assert_(u'OK' in result)
self.sendRequest(u'playlistinfo "4"')
self.assertNotInResponse(u'Title: a')
self.assertNotInResponse(u'Title: b')
self.assertNotInResponse(u'Title: c')
self.assertNotInResponse(u'Title: d')
self.assertInResponse(u'Title: e')
self.assertNotInResponse(u'Title: f')
self.assertInResponse(u'OK')
def test_playlistinfo_with_negative_songpos_same_as_playlistinfo(self):
result1 = self.dispatcher.handle_request(u'playlistinfo "-1"')
result2 = self.dispatcher.handle_request(u'playlistinfo')
self.assertEqual(result1, result2)
response1 = self.sendRequest(u'playlistinfo "-1"')
response2 = self.sendRequest(u'playlistinfo')
self.assertEqual(response1, response2)
def test_playlistinfo_with_open_range(self):
self.backend.current_playlist.append([
Track(name='a'), Track(name='b'), Track(name='c'),
Track(name='d'), Track(name='e'), Track(name='f'),
])
result = self.dispatcher.handle_request(u'playlistinfo "2:"')
self.assert_(u'Title: a' not in result)
self.assert_(u'Title: b' not in result)
self.assert_(u'Title: c' in result)
self.assert_(u'Title: d' in result)
self.assert_(u'Title: e' in result)
self.assert_(u'Title: f' in result)
self.assert_(u'OK' in result)
self.sendRequest(u'playlistinfo "2:"')
self.assertNotInResponse(u'Title: a')
self.assertNotInResponse(u'Title: b')
self.assertInResponse(u'Title: c')
self.assertInResponse(u'Title: d')
self.assertInResponse(u'Title: e')
self.assertInResponse(u'Title: f')
self.assertInResponse(u'OK')
def test_playlistinfo_with_closed_range(self):
self.backend.current_playlist.append([
Track(name='a'), Track(name='b'), Track(name='c'),
Track(name='d'), Track(name='e'), Track(name='f'),
])
result = self.dispatcher.handle_request(u'playlistinfo "2:4"')
self.assert_(u'Title: a' not in result)
self.assert_(u'Title: b' not in result)
self.assert_(u'Title: c' in result)
self.assert_(u'Title: d' in result)
self.assert_(u'Title: e' not in result)
self.assert_(u'Title: f' not in result)
self.assert_(u'OK' in result)
self.sendRequest(u'playlistinfo "2:4"')
self.assertNotInResponse(u'Title: a')
self.assertNotInResponse(u'Title: b')
self.assertInResponse(u'Title: c')
self.assertInResponse(u'Title: d')
self.assertNotInResponse(u'Title: e')
self.assertNotInResponse(u'Title: f')
self.assertInResponse(u'OK')
def test_playlistinfo_with_too_high_start_of_range_returns_arg_error(self):
result = self.dispatcher.handle_request(u'playlistinfo "10:20"')
self.assert_(u'ACK [2@0] {playlistinfo} Bad song index' in result)
self.sendRequest(u'playlistinfo "10:20"')
self.assertEqualResponse(u'ACK [2@0] {playlistinfo} Bad song index')
def test_playlistinfo_with_too_high_end_of_range_returns_ok(self):
result = self.dispatcher.handle_request(u'playlistinfo "0:20"')
self.assert_(u'OK' in result)
self.sendRequest(u'playlistinfo "0:20"')
self.assertInResponse(u'OK')
def test_playlistsearch(self):
result = self.dispatcher.handle_request(
u'playlistsearch "any" "needle"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
self.sendRequest( u'playlistsearch "any" "needle"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_playlistsearch_without_quotes(self):
result = self.dispatcher.handle_request(u'playlistsearch any "needle"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
self.sendRequest(u'playlistsearch any "needle"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_plchanges(self):
self.backend.current_playlist.append(
[Track(name='a'), Track(name='b'), Track(name='c')])
result = self.dispatcher.handle_request(u'plchanges "0"')
self.assert_(u'Title: a' in result)
self.assert_(u'Title: b' in result)
self.assert_(u'Title: c' in result)
self.assert_(u'OK' in result)
self.sendRequest(u'plchanges "0"')
self.assertInResponse(u'Title: a')
self.assertInResponse(u'Title: b')
self.assertInResponse(u'Title: c')
self.assertInResponse(u'OK')
def test_plchanges_with_minus_one_returns_entire_playlist(self):
self.backend.current_playlist.append(
[Track(name='a'), Track(name='b'), Track(name='c')])
result = self.dispatcher.handle_request(u'plchanges "-1"')
self.assert_(u'Title: a' in result)
self.assert_(u'Title: b' in result)
self.assert_(u'Title: c' in result)
self.assert_(u'OK' in result)
self.sendRequest(u'plchanges "-1"')
self.assertInResponse(u'Title: a')
self.assertInResponse(u'Title: b')
self.assertInResponse(u'Title: c')
self.assertInResponse(u'OK')
def test_plchanges_without_quotes_works(self):
self.backend.current_playlist.append(
[Track(name='a'), Track(name='b'), Track(name='c')])
result = self.dispatcher.handle_request(u'plchanges 0')
self.assert_(u'Title: a' in result)
self.assert_(u'Title: b' in result)
self.assert_(u'Title: c' in result)
self.assert_(u'OK' in result)
self.sendRequest(u'plchanges 0')
self.assertInResponse(u'Title: a')
self.assertInResponse(u'Title: b')
self.assertInResponse(u'Title: c')
self.assertInResponse(u'OK')
def test_plchangesposid(self):
self.backend.current_playlist.append([Track(), Track(), Track()])
result = self.dispatcher.handle_request(u'plchangesposid "0"')
self.sendRequest(u'plchangesposid "0"')
cp_tracks = self.backend.current_playlist.cp_tracks.get()
self.assert_(u'cpos: 0' in result)
self.assert_(u'Id: %d' % cp_tracks[0][0]
in result)
self.assert_(u'cpos: 2' in result)
self.assert_(u'Id: %d' % cp_tracks[1][0]
in result)
self.assert_(u'cpos: 2' in result)
self.assert_(u'Id: %d' % cp_tracks[2][0]
in result)
self.assert_(u'OK' in result)
self.assertInResponse(u'cpos: 0')
self.assertInResponse(u'Id: %d' % cp_tracks[0][0])
self.assertInResponse(u'cpos: 2')
self.assertInResponse(u'Id: %d' % cp_tracks[1][0])
self.assertInResponse(u'cpos: 2')
self.assertInResponse(u'Id: %d' % cp_tracks[2][0])
self.assertInResponse(u'OK')
def test_shuffle_without_range(self):
self.backend.current_playlist.append([
@ -386,9 +392,10 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
Track(name='d'), Track(name='e'), Track(name='f'),
])
version = self.backend.current_playlist.version.get()
result = self.dispatcher.handle_request(u'shuffle')
self.sendRequest(u'shuffle')
self.assert_(version < self.backend.current_playlist.version.get())
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_shuffle_with_open_range(self):
self.backend.current_playlist.append([
@ -396,14 +403,15 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
Track(name='d'), Track(name='e'), Track(name='f'),
])
version = self.backend.current_playlist.version.get()
result = self.dispatcher.handle_request(u'shuffle "4:"')
self.sendRequest(u'shuffle "4:"')
self.assert_(version < self.backend.current_playlist.version.get())
tracks = self.backend.current_playlist.tracks.get()
self.assertEqual(tracks[0].name, 'a')
self.assertEqual(tracks[1].name, 'b')
self.assertEqual(tracks[2].name, 'c')
self.assertEqual(tracks[3].name, 'd')
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_shuffle_with_closed_range(self):
self.backend.current_playlist.append([
@ -411,21 +419,23 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
Track(name='d'), Track(name='e'), Track(name='f'),
])
version = self.backend.current_playlist.version.get()
result = self.dispatcher.handle_request(u'shuffle "1:3"')
self.sendRequest(u'shuffle "1:3"')
self.assert_(version < self.backend.current_playlist.version.get())
tracks = self.backend.current_playlist.tracks.get()
self.assertEqual(tracks[0].name, 'a')
self.assertEqual(tracks[3].name, 'd')
self.assertEqual(tracks[4].name, 'e')
self.assertEqual(tracks[5].name, 'f')
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_swap(self):
self.backend.current_playlist.append([
Track(name='a'), Track(name='b'), Track(name='c'),
Track(name='d'), Track(name='e'), Track(name='f'),
])
result = self.dispatcher.handle_request(u'swap "1" "4"')
self.sendRequest(u'swap "1" "4"')
tracks = self.backend.current_playlist.tracks.get()
self.assertEqual(tracks[0].name, 'a')
self.assertEqual(tracks[1].name, 'e')
@ -433,14 +443,15 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
self.assertEqual(tracks[3].name, 'd')
self.assertEqual(tracks[4].name, 'b')
self.assertEqual(tracks[5].name, 'f')
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')
def test_swapid(self):
self.backend.current_playlist.append([
Track(name='a'), Track(name='b'), Track(name='c'),
Track(name='d'), Track(name='e'), Track(name='f'),
])
result = self.dispatcher.handle_request(u'swapid "1" "4"')
self.sendRequest(u'swapid "1" "4"')
tracks = self.backend.current_playlist.tracks.get()
self.assertEqual(tracks[0].name, 'a')
self.assertEqual(tracks[1].name, 'e')
@ -448,4 +459,4 @@ class CurrentPlaylistHandlerTest(unittest.TestCase):
self.assertEqual(tracks[3].name, 'd')
self.assertEqual(tracks[4].name, 'b')
self.assertEqual(tracks[5].name, 'f')
self.assert_(u'OK' in result)
self.assertInResponse(u'OK')

343
tests/frontends/mpd/protocol/music_db_test.py Normal file
View File

@ -0,0 +1,343 @@
from tests.frontends.mpd import protocol
class MusicDatabaseHandlerTest(protocol.BaseTestCase):
def test_count(self):
self.sendRequest(u'count "tag" "needle"')
self.assertInResponse(u'songs: 0')
self.assertInResponse(u'playtime: 0')
self.assertInResponse(u'OK')
def test_findadd(self):
self.sendRequest(u'findadd "album" "what"')
self.assertInResponse(u'OK')
def test_listall(self):
self.sendRequest(u'listall "file:///dev/urandom"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_listallinfo(self):
self.sendRequest(u'listallinfo "file:///dev/urandom"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_lsinfo_without_path_returns_same_as_listplaylists(self):
lsinfo_response = self.sendRequest(u'lsinfo')
listplaylists_response = self.sendRequest(u'listplaylists')
self.assertEqual(lsinfo_response, listplaylists_response)
def test_lsinfo_with_empty_path_returns_same_as_listplaylists(self):
lsinfo_response = self.sendRequest(u'lsinfo ""')
listplaylists_response = self.sendRequest(u'listplaylists')
self.assertEqual(lsinfo_response, listplaylists_response)
def test_lsinfo_for_root_returns_same_as_listplaylists(self):
lsinfo_response = self.sendRequest(u'lsinfo "/"')
listplaylists_response = self.sendRequest(u'listplaylists')
self.assertEqual(lsinfo_response, listplaylists_response)
def test_update_without_uri(self):
self.sendRequest(u'update')
self.assertInResponse(u'updating_db: 0')
self.assertInResponse(u'OK')
def test_update_with_uri(self):
self.sendRequest(u'update "file:///dev/urandom"')
self.assertInResponse(u'updating_db: 0')
self.assertInResponse(u'OK')
def test_rescan_without_uri(self):
self.sendRequest(u'rescan')
self.assertInResponse(u'updating_db: 0')
self.assertInResponse(u'OK')
def test_rescan_with_uri(self):
self.sendRequest(u'rescan "file:///dev/urandom"')
self.assertInResponse(u'updating_db: 0')
self.assertInResponse(u'OK')
class MusicDatabaseFindTest(protocol.BaseTestCase):
def test_find_album(self):
self.sendRequest(u'find "album" "what"')
self.assertInResponse(u'OK')
def test_find_album_without_quotes(self):
self.sendRequest(u'find album "what"')
self.assertInResponse(u'OK')
def test_find_artist(self):
self.sendRequest(u'find "artist" "what"')
self.assertInResponse(u'OK')
def test_find_artist_without_quotes(self):
self.sendRequest(u'find artist "what"')
self.assertInResponse(u'OK')
def test_find_title(self):
self.sendRequest(u'find "title" "what"')
self.assertInResponse(u'OK')
def test_find_title_without_quotes(self):
self.sendRequest(u'find title "what"')
self.assertInResponse(u'OK')
def test_find_date(self):
self.sendRequest(u'find "date" "2002-01-01"')
self.assertInResponse(u'OK')
def test_find_date_without_quotes(self):
self.sendRequest(u'find date "2002-01-01"')
self.assertInResponse(u'OK')
def test_find_date_with_capital_d_and_incomplete_date(self):
self.sendRequest(u'find Date "2005"')
self.assertInResponse(u'OK')
def test_find_else_should_fail(self):
self.sendRequest(u'find "somethingelse" "what"')
self.assertEqualResponse(u'ACK [2@0] {find} incorrect arguments')
def test_find_album_and_artist(self):
self.sendRequest(u'find album "album_what" artist "artist_what"')
self.assertInResponse(u'OK')
class MusicDatabaseListTest(protocol.BaseTestCase):
def test_list_foo_returns_ack(self):
self.sendRequest(u'list "foo"')
self.assertEqualResponse(u'ACK [2@0] {list} incorrect arguments')
### Artist
def test_list_artist_with_quotes(self):
self.sendRequest(u'list "artist"')
self.assertInResponse(u'OK')
def test_list_artist_without_quotes(self):
self.sendRequest(u'list artist')
self.assertInResponse(u'OK')
def test_list_artist_without_quotes_and_capitalized(self):
self.sendRequest(u'list Artist')
self.assertInResponse(u'OK')
def test_list_artist_with_query_of_one_token(self):
self.sendRequest(u'list "artist" "anartist"')
self.assertEqualResponse(
u'ACK [2@0] {list} should be "Album" for 3 arguments')
def test_list_artist_with_unknown_field_in_query_returns_ack(self):
self.sendRequest(u'list "artist" "foo" "bar"')
self.assertEqualResponse(u'ACK [2@0] {list} not able to parse args')
def test_list_artist_by_artist(self):
self.sendRequest(u'list "artist" "artist" "anartist"')
self.assertInResponse(u'OK')
def test_list_artist_by_album(self):
self.sendRequest(u'list "artist" "album" "analbum"')
self.assertInResponse(u'OK')
def test_list_artist_by_full_date(self):
self.sendRequest(u'list "artist" "date" "2001-01-01"')
self.assertInResponse(u'OK')
def test_list_artist_by_year(self):
self.sendRequest(u'list "artist" "date" "2001"')
self.assertInResponse(u'OK')
def test_list_artist_by_genre(self):
self.sendRequest(u'list "artist" "genre" "agenre"')
self.assertInResponse(u'OK')
def test_list_artist_by_artist_and_album(self):
self.sendRequest(
u'list "artist" "artist" "anartist" "album" "analbum"')
self.assertInResponse(u'OK')
### Album
def test_list_album_with_quotes(self):
self.sendRequest(u'list "album"')
self.assertInResponse(u'OK')
def test_list_album_without_quotes(self):
self.sendRequest(u'list album')
self.assertInResponse(u'OK')
def test_list_album_without_quotes_and_capitalized(self):
self.sendRequest(u'list Album')
self.assertInResponse(u'OK')
def test_list_album_with_artist_name(self):
self.sendRequest(u'list "album" "anartist"')
self.assertInResponse(u'OK')
def test_list_album_by_artist(self):
self.sendRequest(u'list "album" "artist" "anartist"')
self.assertInResponse(u'OK')
def test_list_album_by_album(self):
self.sendRequest(u'list "album" "album" "analbum"')
self.assertInResponse(u'OK')
def test_list_album_by_full_date(self):
self.sendRequest(u'list "album" "date" "2001-01-01"')
self.assertInResponse(u'OK')
def test_list_album_by_year(self):
self.sendRequest(u'list "album" "date" "2001"')
self.assertInResponse(u'OK')
def test_list_album_by_genre(self):
self.sendRequest(u'list "album" "genre" "agenre"')
self.assertInResponse(u'OK')
def test_list_album_by_artist_and_album(self):
self.sendRequest(
u'list "album" "artist" "anartist" "album" "analbum"')
self.assertInResponse(u'OK')
### Date
def test_list_date_with_quotes(self):
self.sendRequest(u'list "date"')
self.assertInResponse(u'OK')
def test_list_date_without_quotes(self):
self.sendRequest(u'list date')
self.assertInResponse(u'OK')
def test_list_date_without_quotes_and_capitalized(self):
self.sendRequest(u'list Date')
self.assertInResponse(u'OK')
def test_list_date_with_query_of_one_token(self):
self.sendRequest(u'list "date" "anartist"')
self.assertEqualResponse(
u'ACK [2@0] {list} should be "Album" for 3 arguments')
def test_list_date_by_artist(self):
self.sendRequest(u'list "date" "artist" "anartist"')
self.assertInResponse(u'OK')
def test_list_date_by_album(self):
self.sendRequest(u'list "date" "album" "analbum"')
self.assertInResponse(u'OK')
def test_list_date_by_full_date(self):
self.sendRequest(u'list "date" "date" "2001-01-01"')
self.assertInResponse(u'OK')
def test_list_date_by_year(self):
self.sendRequest(u'list "date" "date" "2001"')
self.assertInResponse(u'OK')
def test_list_date_by_genre(self):
self.sendRequest(u'list "date" "genre" "agenre"')
self.assertInResponse(u'OK')
def test_list_date_by_artist_and_album(self):
self.sendRequest(u'list "date" "artist" "anartist" "album" "analbum"')
self.assertInResponse(u'OK')
### Genre
def test_list_genre_with_quotes(self):
self.sendRequest(u'list "genre"')
self.assertInResponse(u'OK')
def test_list_genre_without_quotes(self):
self.sendRequest(u'list genre')
self.assertInResponse(u'OK')
def test_list_genre_without_quotes_and_capitalized(self):
self.sendRequest(u'list Genre')
self.assertInResponse(u'OK')
def test_list_genre_with_query_of_one_token(self):
self.sendRequest(u'list "genre" "anartist"')
self.assertEqualResponse(
u'ACK [2@0] {list} should be "Album" for 3 arguments')
def test_list_genre_by_artist(self):
self.sendRequest(u'list "genre" "artist" "anartist"')
self.assertInResponse(u'OK')
def test_list_genre_by_album(self):
self.sendRequest(u'list "genre" "album" "analbum"')
self.assertInResponse(u'OK')
def test_list_genre_by_full_date(self):
self.sendRequest(u'list "genre" "date" "2001-01-01"')
self.assertInResponse(u'OK')
def test_list_genre_by_year(self):
self.sendRequest(u'list "genre" "date" "2001"')
self.assertInResponse(u'OK')
def test_list_genre_by_genre(self):
self.sendRequest(u'list "genre" "genre" "agenre"')
self.assertInResponse(u'OK')
def test_list_genre_by_artist_and_album(self):
self.sendRequest(
u'list "genre" "artist" "anartist" "album" "analbum"')
self.assertInResponse(u'OK')
class MusicDatabaseSearchTest(protocol.BaseTestCase):
def test_search_album(self):
self.sendRequest(u'search "album" "analbum"')
self.assertInResponse(u'OK')
def test_search_album_without_quotes(self):
self.sendRequest(u'search album "analbum"')
self.assertInResponse(u'OK')
def test_search_artist(self):
self.sendRequest(u'search "artist" "anartist"')
self.assertInResponse(u'OK')
def test_search_artist_without_quotes(self):
self.sendRequest(u'search artist "anartist"')
self.assertInResponse(u'OK')
def test_search_filename(self):
self.sendRequest(u'search "filename" "afilename"')
self.assertInResponse(u'OK')
def test_search_filename_without_quotes(self):
self.sendRequest(u'search filename "afilename"')
self.assertInResponse(u'OK')
def test_search_title(self):
self.sendRequest(u'search "title" "atitle"')
self.assertInResponse(u'OK')
def test_search_title_without_quotes(self):
self.sendRequest(u'search title "atitle"')
self.assertInResponse(u'OK')
def test_search_any(self):
self.sendRequest(u'search "any" "anything"')
self.assertInResponse(u'OK')
def test_search_any_without_quotes(self):
self.sendRequest(u'search any "anything"')
self.assertInResponse(u'OK')
def test_search_date(self):
self.sendRequest(u'search "date" "2002-01-01"')
self.assertInResponse(u'OK')
def test_search_date_without_quotes(self):
self.sendRequest(u'search date "2002-01-01"')
self.assertInResponse(u'OK')
def test_search_date_with_capital_d_and_incomplete_date(self):
self.sendRequest(u'search Date "2005"')
self.assertInResponse(u'OK')
def test_search_else_should_fail(self):
self.sendRequest(u'search "sometype" "something"')
self.assertEqualResponse(u'ACK [2@0] {search} incorrect arguments')

319
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.backends import base as backend
from mopidy.models import Track
from tests import SkipTest
from tests.frontends.mpd import protocol
PAUSED = PlaybackController.PAUSED
PLAYING = PlaybackController.PLAYING
STOPPED = PlaybackController.STOPPED
PAUSED = backend.PlaybackController.PAUSED
PLAYING = backend.PlaybackController.PLAYING
STOPPED = backend.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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(u'OK')
self.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(u'OK')
self.sendRequest(u'pause')
self.assertEqual(PLAYING, self.backend.playback.state.get())
self.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(u'ACK [2@0] {play} Bad song index')
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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(u'ACK [50@0] {playid} No such song')
def test_previous(self):
result = self.dispatcher.handle_request(u'previous')
self.assert_(u'OK' in result)
self.sendRequest(u'previous')
self.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(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.assertInResponse(u'OK')

66
tests/frontends/mpd/protocol/reflection_test.py Normal file
View File

@ -0,0 +1,66 @@
from mopidy import settings
from tests.frontends.mpd import protocol
class ReflectionHandlerTest(protocol.BaseTestCase):
def test_commands_returns_list_of_all_commands(self):
self.sendRequest(u'commands')
# Check if some random commands are included
self.assertInResponse(u'command: commands')
self.assertInResponse(u'command: play')
self.assertInResponse(u'command: status')
# Check if commands you do not have access to are not present
self.assertNotInResponse(u'command: kill')
# Check if the blacklisted commands are not present
self.assertNotInResponse(u'command: command_list_begin')
self.assertNotInResponse(u'command: command_list_ok_begin')
self.assertNotInResponse(u'command: command_list_end')
self.assertNotInResponse(u'command: idle')
self.assertNotInResponse(u'command: noidle')
self.assertNotInResponse(u'command: sticker')
self.assertInResponse(u'OK')
def test_commands_show_less_if_auth_required_and_not_authed(self):
settings.MPD_SERVER_PASSWORD = u'secret'
self.sendRequest(u'commands')
# Not requiring auth
self.assertInResponse(u'command: close')
self.assertInResponse(u'command: commands')
self.assertInResponse(u'command: notcommands')
self.assertInResponse(u'command: password')
self.assertInResponse(u'command: ping')
# Requiring auth
self.assertNotInResponse(u'command: play')
self.assertNotInResponse(u'command: status')
def test_decoders(self):
self.sendRequest(u'decoders')
self.assertInResponse(u'ACK [0@0] {} Not implemented')
def test_notcommands_returns_only_kill_and_ok(self):
response = self.sendRequest(u'notcommands')
self.assertEqual(2, len(response))
self.assertInResponse(u'command: kill')
self.assertInResponse(u'OK')
def test_notcommands_returns_more_if_auth_required_and_not_authed(self):
settings.MPD_SERVER_PASSWORD = u'secret'
self.sendRequest(u'notcommands')
# Not requiring auth
self.assertNotInResponse(u'command: close')
self.assertNotInResponse(u'command: commands')
self.assertNotInResponse(u'command: notcommands')
self.assertNotInResponse(u'command: password')
self.assertNotInResponse(u'command: ping')
# Requiring auth
self.assertInResponse(u'command: play')
self.assertInResponse(u'command: status')
def test_tagtypes(self):
self.sendRequest(u'tagtypes')
self.assertInResponse(u'OK')
def test_urlhandlers(self):
self.sendRequest(u'urlhandlers')
self.assertInResponse(u'OK')
self.assertInResponse(u'handler: dummy')

120
tests/frontends/mpd/regression_test.py → tests/frontends/mpd/protocol/regression_test.py
View File

@ -1,12 +1,10 @@
import random
import unittest
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd import dispatcher
from mopidy.mixers.dummy import DummyMixer
from mopidy.models import Track
class IssueGH17RegressionTest(unittest.TestCase):
from tests.frontends.mpd import protocol
class IssueGH17RegressionTest(protocol.BaseTestCase):
"""
The issue: http://github.com/mopidy/mopidy/issues#issue/17
@ -16,36 +14,27 @@ class IssueGH17RegressionTest(unittest.TestCase):
- Turn on random mode
- Press next until you get to the unplayable track
"""
def setUp(self):
self.backend = DummyBackend.start().proxy()
def test(self):
self.backend.current_playlist.append([
Track(uri='a'), Track(uri='b'), None,
Track(uri='d'), Track(uri='e'), Track(uri='f')])
self.mixer = DummyMixer.start().proxy()
self.mpd = dispatcher.MpdDispatcher()
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
def test(self):
random.seed(1) # Playlist order: abcfde
self.mpd.handle_request(u'play')
self.sendRequest(u'play')
self.assertEquals('a', self.backend.playback.current_track.get().uri)
self.mpd.handle_request(u'random "1"')
self.mpd.handle_request(u'next')
self.sendRequest(u'random "1"')
self.sendRequest(u'next')
self.assertEquals('b', self.backend.playback.current_track.get().uri)
self.mpd.handle_request(u'next')
self.sendRequest(u'next')
# Should now be at track 'c', but playback fails and it skips ahead
self.assertEquals('f', self.backend.playback.current_track.get().uri)
self.mpd.handle_request(u'next')
self.sendRequest(u'next')
self.assertEquals('d', self.backend.playback.current_track.get().uri)
self.mpd.handle_request(u'next')
self.sendRequest(u'next')
self.assertEquals('e', self.backend.playback.current_track.get().uri)
class IssueGH18RegressionTest(unittest.TestCase):
class IssueGH18RegressionTest(protocol.BaseTestCase):
"""
The issue: http://github.com/mopidy/mopidy/issues#issue/18
@ -56,38 +45,30 @@ class IssueGH18RegressionTest(unittest.TestCase):
At this point it gives the same song over and over.
"""
def setUp(self):
self.backend = DummyBackend.start().proxy()
def test(self):
self.backend.current_playlist.append([
Track(uri='a'), Track(uri='b'), Track(uri='c'),
Track(uri='d'), Track(uri='e'), Track(uri='f')])
self.mixer = DummyMixer.start().proxy()
self.mpd = dispatcher.MpdDispatcher()
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
def test(self):
random.seed(1)
self.mpd.handle_request(u'play')
self.mpd.handle_request(u'random "1"')
self.mpd.handle_request(u'next')
self.mpd.handle_request(u'random "0"')
self.mpd.handle_request(u'next')
self.mpd.handle_request(u'next')
self.sendRequest(u'play')
self.sendRequest(u'random "1"')
self.sendRequest(u'next')
self.sendRequest(u'random "0"')
self.sendRequest(u'next')
self.sendRequest(u'next')
cp_track_1 = self.backend.playback.current_cp_track.get()
self.mpd.handle_request(u'next')
self.sendRequest(u'next')
cp_track_2 = self.backend.playback.current_cp_track.get()
self.mpd.handle_request(u'next')
self.sendRequest(u'next')
cp_track_3 = self.backend.playback.current_cp_track.get()
self.assertNotEqual(cp_track_1, cp_track_2)
self.assertNotEqual(cp_track_2, cp_track_3)
class IssueGH22RegressionTest(unittest.TestCase):
class IssueGH22RegressionTest(protocol.BaseTestCase):
"""
The issue: http://github.com/mopidy/mopidy/issues/#issue/22
@ -100,32 +81,24 @@ class IssueGH22RegressionTest(unittest.TestCase):
playlist, press next until it crashes.
"""
def setUp(self):
self.backend = DummyBackend.start().proxy()
def test(self):
self.backend.current_playlist.append([
Track(uri='a'), Track(uri='b'), Track(uri='c'),
Track(uri='d'), Track(uri='e'), Track(uri='f')])
self.mixer = DummyMixer.start().proxy()
self.mpd = dispatcher.MpdDispatcher()
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
def test(self):
random.seed(1)
self.mpd.handle_request(u'play')
self.mpd.handle_request(u'random "1"')
self.mpd.handle_request(u'deleteid "1"')
self.mpd.handle_request(u'deleteid "2"')
self.mpd.handle_request(u'deleteid "3"')
self.mpd.handle_request(u'deleteid "4"')
self.mpd.handle_request(u'deleteid "5"')
self.mpd.handle_request(u'deleteid "6"')
self.mpd.handle_request(u'status')
self.sendRequest(u'play')
self.sendRequest(u'random "1"')
self.sendRequest(u'deleteid "1"')
self.sendRequest(u'deleteid "2"')
self.sendRequest(u'deleteid "3"')
self.sendRequest(u'deleteid "4"')
self.sendRequest(u'deleteid "5"')
self.sendRequest(u'deleteid "6"')
self.sendRequest(u'status')
class IssueGH69RegressionTest(unittest.TestCase):
class IssueGH69RegressionTest(protocol.BaseTestCase):
"""
The issue: https://github.com/mopidy/mopidy/issues#issue/69
@ -136,23 +109,14 @@ class IssueGH69RegressionTest(unittest.TestCase):
The status response now contains "song: None".
"""
def setUp(self):
self.backend = DummyBackend.start().proxy()
def test(self):
self.backend.stored_playlists.create('foo')
self.backend.current_playlist.append([
Track(uri='a'), Track(uri='b'), Track(uri='c'),
Track(uri='d'), Track(uri='e'), Track(uri='f')])
self.backend.stored_playlists.create('foo')
self.mixer = DummyMixer.start().proxy()
self.mpd = dispatcher.MpdDispatcher()
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
def test(self):
self.mpd.handle_request(u'play')
self.mpd.handle_request(u'stop')
self.mpd.handle_request(u'clear')
self.mpd.handle_request(u'load "foo"')
response = self.mpd.handle_request(u'status')
self.assert_('song: None' not in response)
self.sendRequest(u'play')
self.sendRequest(u'stop')
self.sendRequest(u'clear')
self.sendRequest(u'load "foo"')
self.assertNotInResponse('song: None')

51
tests/frontends/mpd/protocol/status_test.py Normal file
View File

@ -0,0 +1,51 @@
from mopidy.models import Track
from tests.frontends.mpd import protocol
class StatusHandlerTest(protocol.BaseTestCase):
def test_clearerror(self):
self.sendRequest(u'clearerror')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_currentsong(self):
track = Track()
self.backend.current_playlist.append([track])
self.backend.playback.play()
self.sendRequest(u'currentsong')
self.assertInResponse(u'file: ')
self.assertInResponse(u'Time: 0')
self.assertInResponse(u'Artist: ')
self.assertInResponse(u'Title: ')
self.assertInResponse(u'Album: ')
self.assertInResponse(u'Track: 0')
self.assertInResponse(u'Date: ')
self.assertInResponse(u'Pos: 0')
self.assertInResponse(u'Id: 0')
self.assertInResponse(u'OK')
def test_currentsong_without_song(self):
self.sendRequest(u'currentsong')
self.assertInResponse(u'OK')
def test_idle_without_subsystems(self):
# FIXME this is not the correct behaviour for idle...
self.sendRequest(u'idle')
self.assertInResponse(u'OK')
def test_idle_with_subsystems(self):
# FIXME this is not the correct behaviour for idle...
self.sendRequest(u'idle database playlist')
self.assertInResponse(u'OK')
def test_noidle(self):
# FIXME this is not the correct behaviour for idle...
self.sendRequest(u'noidle')
self.assertInResponse(u'OK')
def test_stats_command(self):
self.sendRequest(u'stats')
self.assertInResponse(u'OK')
def test_status_command(self):
self.sendRequest(u'status')
self.assertInResponse(u'OK')

32
tests/frontends/mpd/protocol/stickers_test.py Normal file
View File

@ -0,0 +1,32 @@
from tests.frontends.mpd import protocol
class StickersHandlerTest(protocol.BaseTestCase):
def test_sticker_get(self):
self.sendRequest(
u'sticker get "song" "file:///dev/urandom" "a_name"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_sticker_set(self):
self.sendRequest(
u'sticker set "song" "file:///dev/urandom" "a_name" "a_value"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_sticker_delete_with_name(self):
self.sendRequest(
u'sticker delete "song" "file:///dev/urandom" "a_name"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_sticker_delete_without_name(self):
self.sendRequest(
u'sticker delete "song" "file:///dev/urandom"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_sticker_list(self):
self.sendRequest(
u'sticker list "song" "file:///dev/urandom"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_sticker_find(self):
self.sendRequest(
u'sticker find "song" "file:///dev/urandom" "a_name"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')

93
tests/frontends/mpd/protocol/stored_playlists_test.py Normal file
View File

@ -0,0 +1,93 @@
import datetime as dt
from mopidy.models import Track, Playlist
from tests.frontends.mpd import protocol
class StoredPlaylistsHandlerTest(protocol.BaseTestCase):
def test_listplaylist(self):
self.backend.stored_playlists.playlists = [
Playlist(name='name', tracks=[Track(uri='file:///dev/urandom')])]
self.sendRequest(u'listplaylist "name"')
self.assertInResponse(u'file: file:///dev/urandom')
self.assertInResponse(u'OK')
def test_listplaylist_fails_if_no_playlist_is_found(self):
self.sendRequest(u'listplaylist "name"')
self.assertEqualResponse(u'ACK [50@0] {listplaylist} No such playlist')
def test_listplaylistinfo(self):
self.backend.stored_playlists.playlists = [
Playlist(name='name', tracks=[Track(uri='file:///dev/urandom')])]
self.sendRequest(u'listplaylistinfo "name"')
self.assertInResponse(u'file: file:///dev/urandom')
self.assertInResponse(u'Track: 0')
self.assertNotInResponse(u'Pos: 0')
self.assertInResponse(u'OK')
def test_listplaylistinfo_fails_if_no_playlist_is_found(self):
self.sendRequest(u'listplaylistinfo "name"')
self.assertEqualResponse(
u'ACK [50@0] {listplaylistinfo} No such playlist')
def test_listplaylists(self):
last_modified = dt.datetime(2001, 3, 17, 13, 41, 17, 12345)
self.backend.stored_playlists.playlists = [Playlist(name='a',
last_modified=last_modified)]
self.sendRequest(u'listplaylists')
self.assertInResponse(u'playlist: a')
# Date without microseconds and with time zone information
self.assertInResponse(u'Last-Modified: 2001-03-17T13:41:17Z')
self.assertInResponse(u'OK')
def test_load_known_playlist_appends_to_current_playlist(self):
self.backend.current_playlist.append([Track(uri='a'), Track(uri='b')])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 2)
self.backend.stored_playlists.playlists = [Playlist(name='A-list',
tracks=[Track(uri='c'), Track(uri='d'), Track(uri='e')])]
self.sendRequest(u'load "A-list"')
tracks = self.backend.current_playlist.tracks.get()
self.assertEqual(5, len(tracks))
self.assertEqual('a', tracks[0].uri)
self.assertEqual('b', tracks[1].uri)
self.assertEqual('c', tracks[2].uri)
self.assertEqual('d', tracks[3].uri)
self.assertEqual('e', tracks[4].uri)
self.assertInResponse(u'OK')
def test_load_unknown_playlist_acks(self):
self.sendRequest(u'load "unknown playlist"')
self.assertEqual(0, len(self.backend.current_playlist.tracks.get()))
self.assertEqualResponse(u'ACK [50@0] {load} No such playlist')
def test_playlistadd(self):
self.sendRequest(u'playlistadd "name" "file:///dev/urandom"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_playlistclear(self):
self.sendRequest(u'playlistclear "name"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_playlistdelete(self):
self.sendRequest(u'playlistdelete "name" "5"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_playlistmove(self):
self.sendRequest(u'playlistmove "name" "5" "10"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_rename(self):
self.sendRequest(u'rename "old_name" "new_name"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_rm(self):
self.sendRequest(u'rm "name"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')
def test_save(self):
self.sendRequest(u'save "name"')
self.assertEqualResponse(u'ACK [0@0] {} Not implemented')

79
tests/frontends/mpd/reflection_test.py
View File

@ -1,79 +0,0 @@
import unittest
from mopidy import settings
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
from mopidy.mixers.dummy import DummyMixer
class ReflectionHandlerTest(unittest.TestCase):
def setUp(self):
self.backend = DummyBackend.start().proxy()
self.mixer = DummyMixer.start().proxy()
self.dispatcher = MpdDispatcher()
def tearDown(self):
settings.runtime.clear()
self.backend.stop().get()
self.mixer.stop().get()
def test_commands_returns_list_of_all_commands(self):
result = self.dispatcher.handle_request(u'commands')
# Check if some random commands are included
self.assert_(u'command: commands' in result)
self.assert_(u'command: play' in result)
self.assert_(u'command: status' in result)
# Check if commands you do not have access to are not present
self.assert_(u'command: kill' not in result)
# Check if the blacklisted commands are not present
self.assert_(u'command: command_list_begin' not in result)
self.assert_(u'command: command_list_ok_begin' not in result)
self.assert_(u'command: command_list_end' not in result)
self.assert_(u'command: idle' not in result)
self.assert_(u'command: noidle' not in result)
self.assert_(u'command: sticker' not in result)
self.assert_(u'OK' in result)
def test_commands_show_less_if_auth_required_and_not_authed(self):
settings.MPD_SERVER_PASSWORD = u'secret'
result = self.dispatcher.handle_request(u'commands')
# Not requiring auth
self.assert_(u'command: close' in result, result)
self.assert_(u'command: commands' in result, result)
self.assert_(u'command: notcommands' in result, result)
self.assert_(u'command: password' in result, result)
self.assert_(u'command: ping' in result, result)
# Requiring auth
self.assert_(u'command: play' not in result, result)
self.assert_(u'command: status' not in result, result)
def test_decoders(self):
result = self.dispatcher.handle_request(u'decoders')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_notcommands_returns_only_kill_and_ok(self):
result = self.dispatcher.handle_request(u'notcommands')
self.assertEqual(2, len(result))
self.assert_(u'command: kill' in result)
self.assert_(u'OK' in result)
def test_notcommands_returns_more_if_auth_required_and_not_authed(self):
settings.MPD_SERVER_PASSWORD = u'secret'
result = self.dispatcher.handle_request(u'notcommands')
# Not requiring auth
self.assert_(u'command: close' not in result, result)
self.assert_(u'command: commands' not in result, result)
self.assert_(u'command: notcommands' not in result, result)
self.assert_(u'command: password' not in result, result)
self.assert_(u'command: ping' not in result, result)
# Requiring auth
self.assert_(u'command: play' in result, result)
self.assert_(u'command: status' in result, result)
def test_tagtypes(self):
result = self.dispatcher.handle_request(u'tagtypes')
self.assert_(u'OK' in result)
def test_urlhandlers(self):
result = self.dispatcher.handle_request(u'urlhandlers')
self.assert_(u'OK' in result)
self.assert_(u'handler: dummy' in result)

66
tests/frontends/mpd/status_test.py
View File

@ -1,67 +1,29 @@
import unittest
from mopidy.backends.base import PlaybackController
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
from mopidy.backends import dummy as backend
from mopidy.frontends.mpd import dispatcher
from mopidy.frontends.mpd.protocol import status
from mopidy.mixers.dummy import DummyMixer
from mopidy.mixers import dummy as mixer
from mopidy.models import Track
PAUSED = PlaybackController.PAUSED
PLAYING = PlaybackController.PLAYING
STOPPED = PlaybackController.STOPPED
PAUSED = backend.PlaybackController.PAUSED
PLAYING = backend.PlaybackController.PLAYING
STOPPED = backend.PlaybackController.STOPPED
# FIXME migrate to using protocol.BaseTestCase instead of status.stats
# directly?
class StatusHandlerTest(unittest.TestCase):
def setUp(self):
self.backend = DummyBackend.start().proxy()
self.mixer = DummyMixer.start().proxy()
self.dispatcher = MpdDispatcher()
self.backend = backend.DummyBackend.start().proxy()
self.mixer = mixer.DummyMixer.start().proxy()
self.dispatcher = dispatcher.MpdDispatcher()
self.context = self.dispatcher.context
def tearDown(self):
self.backend.stop().get()
self.mixer.stop().get()
def test_clearerror(self):
result = self.dispatcher.handle_request(u'clearerror')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_currentsong(self):
track = Track()
self.backend.current_playlist.append([track])
self.backend.playback.play()
result = self.dispatcher.handle_request(u'currentsong')
self.assert_(u'file: ' in result)
self.assert_(u'Time: 0' in result)
self.assert_(u'Artist: ' in result)
self.assert_(u'Title: ' in result)
self.assert_(u'Album: ' in result)
self.assert_(u'Track: 0' in result)
self.assert_(u'Date: ' in result)
self.assert_(u'Pos: 0' in result)
self.assert_(u'Id: 0' in result)
self.assert_(u'OK' in result)
def test_currentsong_without_song(self):
result = self.dispatcher.handle_request(u'currentsong')
self.assert_(u'OK' in result)
def test_idle_without_subsystems(self):
result = self.dispatcher.handle_request(u'idle')
self.assert_(u'OK' in result)
def test_idle_with_subsystems(self):
result = self.dispatcher.handle_request(u'idle database playlist')
self.assert_(u'OK' in result)
def test_noidle(self):
result = self.dispatcher.handle_request(u'noidle')
self.assert_(u'OK' in result)
def test_stats_command(self):
result = self.dispatcher.handle_request(u'stats')
self.assert_(u'OK' in result)
def test_stats_method(self):
result = status.stats(self.context)
self.assert_('artists' in result)
@ -79,10 +41,6 @@ class StatusHandlerTest(unittest.TestCase):
self.assert_('playtime' in result)
self.assert_(int(result['playtime']) >= 0)
def test_status_command(self):
result = self.dispatcher.handle_request(u'status')
self.assert_(u'OK' in result)
def test_status_method_contains_volume_which_defaults_to_0(self):
result = dict(status.status(self.context))
self.assert_('volume' in result)

45
tests/frontends/mpd/stickers_test.py
View File

@ -1,45 +0,0 @@
import unittest
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
from mopidy.mixers.dummy import DummyMixer
class StickersHandlerTest(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()
def test_sticker_get(self):
result = self.dispatcher.handle_request(
u'sticker get "song" "file:///dev/urandom" "a_name"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_sticker_set(self):
result = self.dispatcher.handle_request(
u'sticker set "song" "file:///dev/urandom" "a_name" "a_value"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_sticker_delete_with_name(self):
result = self.dispatcher.handle_request(
u'sticker delete "song" "file:///dev/urandom" "a_name"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_sticker_delete_without_name(self):
result = self.dispatcher.handle_request(
u'sticker delete "song" "file:///dev/urandom"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_sticker_list(self):
result = self.dispatcher.handle_request(
u'sticker list "song" "file:///dev/urandom"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_sticker_find(self):
result = self.dispatcher.handle_request(
u'sticker find "song" "file:///dev/urandom" "a_name"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)

102
tests/frontends/mpd/stored_playlists_test.py
View File

@ -1,102 +0,0 @@
import datetime as dt
import unittest
from mopidy.backends.dummy import DummyBackend
from mopidy.frontends.mpd.dispatcher import MpdDispatcher
from mopidy.mixers.dummy import DummyMixer
from mopidy.models import Track, Playlist
class StoredPlaylistsHandlerTest(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()
def test_listplaylist(self):
self.backend.stored_playlists.playlists = [
Playlist(name='name', tracks=[Track(uri='file:///dev/urandom')])]
result = self.dispatcher.handle_request(u'listplaylist "name"')
self.assert_(u'file: file:///dev/urandom' in result)
self.assert_(u'OK' in result)
def test_listplaylist_fails_if_no_playlist_is_found(self):
result = self.dispatcher.handle_request(u'listplaylist "name"')
self.assertEqual(result[0],
u'ACK [50@0] {listplaylist} No such playlist')
def test_listplaylistinfo(self):
self.backend.stored_playlists.playlists = [
Playlist(name='name', tracks=[Track(uri='file:///dev/urandom')])]
result = self.dispatcher.handle_request(u'listplaylistinfo "name"')
self.assert_(u'file: file:///dev/urandom' in result)
self.assert_(u'Track: 0' in result)
self.assert_(u'Pos: 0' not in result)
self.assert_(u'OK' in result)
def test_listplaylistinfo_fails_if_no_playlist_is_found(self):
result = self.dispatcher.handle_request(u'listplaylistinfo "name"')
self.assertEqual(result[0],
u'ACK [50@0] {listplaylistinfo} No such playlist')
def test_listplaylists(self):
last_modified = dt.datetime(2001, 3, 17, 13, 41, 17, 12345)
self.backend.stored_playlists.playlists = [Playlist(name='a',
last_modified=last_modified)]
result = self.dispatcher.handle_request(u'listplaylists')
self.assert_(u'playlist: a' in result)
# Date without microseconds and with time zone information
self.assert_(u'Last-Modified: 2001-03-17T13:41:17Z' in result)
self.assert_(u'OK' in result)
def test_load_known_playlist_appends_to_current_playlist(self):
self.backend.current_playlist.append([Track(uri='a'), Track(uri='b')])
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 2)
self.backend.stored_playlists.playlists = [Playlist(name='A-list',
tracks=[Track(uri='c'), Track(uri='d'), Track(uri='e')])]
result = self.dispatcher.handle_request(u'load "A-list"')
self.assert_(u'OK' in result)
tracks = self.backend.current_playlist.tracks.get()
self.assertEqual(len(tracks), 5)
self.assertEqual(tracks[0].uri, 'a')
self.assertEqual(tracks[1].uri, 'b')
self.assertEqual(tracks[2].uri, 'c')
self.assertEqual(tracks[3].uri, 'd')
self.assertEqual(tracks[4].uri, 'e')
def test_load_unknown_playlist_acks(self):
result = self.dispatcher.handle_request(u'load "unknown playlist"')
self.assert_(u'ACK [50@0] {load} No such playlist' in result)
self.assertEqual(len(self.backend.current_playlist.tracks.get()), 0)
def test_playlistadd(self):
result = self.dispatcher.handle_request(
u'playlistadd "name" "file:///dev/urandom"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_playlistclear(self):
result = self.dispatcher.handle_request(u'playlistclear "name"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_playlistdelete(self):
result = self.dispatcher.handle_request(u'playlistdelete "name" "5"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_playlistmove(self):
result = self.dispatcher.handle_request(u'playlistmove "name" "5" "10"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_rename(self):
result = self.dispatcher.handle_request(u'rename "old_name" "new_name"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_rm(self):
result = self.dispatcher.handle_request(u'rm "name"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)
def test_save(self):
result = self.dispatcher.handle_request(u'save "name"')
self.assert_(u'ACK [0@0] {} Not implemented' in result)