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

Merge branch 'dz0ny/tornado_http' into develop

Browse Source
This commit is contained in:
Stein Magnus Jodal 2014-05-14 23:07:19 +02:00
commit cf22fc98ad
10 changed files with 498 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
docs/ext/http.rst
View File

@ -46,17 +46,30 @@ See :ref:`http-api` for details on how to integrate with Mopidy over HTTP. If
you're looking for a web based client for Mopidy, go check out
:ref:`http-clients`.
Extending server
================
If you wish to extend server additional service side functionality you must
create class that implements the :class:`mopidy.http.Router` interface and
install it in the extension registry under ``http:router``.
The default implementation already supports serving static files. If you just
want to serve static files you only need to define class variable :attr:`mopidy
.http.Router.name` and :attr:`mopidy.http.Router.path`, for example::
class WebClient(http.Router):
name = 'webclient'
path = os.path.join(os.path.dirname(__file__), 'public_html')
If you wish to extend server with custom methods you can override class method
``mopidy.http.Router.setup_routes`` and define custom routes.
Dependencies
============
In addition to Mopidy's dependencies, Mopidy-HTTP requires the following:
- cherrypy >= 3.2.2. Available as python-cherrypy3 in Debian/Ubuntu.
- ws4py >= 0.2.3. Available as python-ws4py in newer Debian/Ubuntu and from
`apt.mopidy.com <http://apt.mopidy.com/>`__ for older releases of
Debian/Ubuntu.
- tornado >= 3.1.1 Available as python-tornado in Debian/Ubuntu.
If you're installing Mopidy with pip, you can run the following command to
install Mopidy with the extra dependencies for required for Mopidy-HTTP::

95
mopidy/http/__init__.py
View File

@ -1,40 +1,103 @@
from __future__ import unicode_literals
import logging
import os
import mopidy
from mopidy import config, exceptions, ext
import tornado.web
from mopidy import __version__, config as config_lib, exceptions, ext
logger = logging.getLogger(__name__)
class Extension(ext.Extension):
dist_name = 'Mopidy-HTTP'
ext_name = 'http'
version = mopidy.__version__
version = __version__
def get_default_config(self):
conf_file = os.path.join(os.path.dirname(__file__), 'ext.conf')
return config.read(conf_file)
return config_lib.read(conf_file)
def get_config_schema(self):
schema = super(Extension, self).get_config_schema()
schema['hostname'] = config.Hostname()
schema['port'] = config.Port()
schema['static_dir'] = config.Path(optional=True)
schema['zeroconf'] = config.String(optional=True)
schema['hostname'] = config_lib.Hostname()
schema['port'] = config_lib.Port()
schema['static_dir'] = config_lib.Path(optional=True)
schema['zeroconf'] = config_lib.String(optional=True)
return schema
def validate_environment(self):
try:
import cherrypy # noqa
import tornado.web # noqa
except ImportError as e:
raise exceptions.ExtensionError('cherrypy library not found', e)
try:
import ws4py # noqa
except ImportError as e:
raise exceptions.ExtensionError('ws4py library not found', e)
raise exceptions.ExtensionError('tornado library not found', e)
def setup(self, registry):
from .actor import HttpFrontend
HttpFrontend.routers = registry['http:routers']
registry.add('frontend', HttpFrontend)
class StaticFileHandler(tornado.web.StaticFileHandler):
def set_extra_headers(self, path):
self.set_header("Cache-Control", "no-cache")
self.set_header("X-Mopidy-Version", __version__.encode('utf-8'))
class Router(object):
"""
HTTP router interface.
Extensions that wish to add custom routes to HTTP server
need to sub-class this class and install and configure it with an
extension.
:param config:dict Config dictionary
"""
#: Name of the HTTP router implementation, must be overridden.
name = None
#: Path to location of static files.
path = None
def __init__(self, config):
self.config = config
self.hostname = config['http']['hostname']
self.port = config['http']['port']
if not self.name:
raise ValueError('Undefined name in %s' % self)
def linkify(self):
"""
Absolute URL to the root of this router.
:return string: URI
"""
return 'http://%s:%s/%s/' % (self.hostname, self.port, self.name)
def setup_routes(self):
"""
Configure routes for this interface
Implementation must return list of routes, compatible with
:`class:tornado.web.Application`.
:return list: List of tornado routes
"""
if not self.path:
raise ValueError('Undefined path in %s' % self)
logger.info(
'Serving HTTP extension %s at %s' %
(type(self), self.linkify())
)
return [
(r"/%s/(.*)" % self.name, StaticFileHandler, {
'path': self.path,
'default_filename': 'index.html'
})
]

184
mopidy/http/actor.py
View File

@ -3,23 +3,25 @@ from __future__ import unicode_literals
import json
import logging
import os
import cherrypy
import threading
import pykka
from ws4py.messaging import TextMessage
from ws4py.server.cherrypyserver import WebSocketPlugin, WebSocketTool
import tornado.ioloop
import tornado.web
import tornado.websocket
from mopidy import models, zeroconf
from mopidy.core import CoreListener
from mopidy.http import ws
from mopidy.http import StaticFileHandler, handlers
logger = logging.getLogger(__name__)
class HttpFrontend(pykka.ThreadingActor, CoreListener):
routers = []
def __init__(self, config, core):
super(HttpFrontend, self).__init__()
self.config = config
@ -29,123 +31,85 @@ class HttpFrontend(pykka.ThreadingActor, CoreListener):
self.port = config['http']['port']
self.zeroconf_name = config['http']['zeroconf']
self.zeroconf_service = None
self.app = None
self.websocket_clients = set()
self._setup_server()
self._setup_websocket_plugin()
app = self._create_app()
self._setup_logging(app)
def _load_extensions(self):
routes = []
for extension in self.routers:
extension = extension(self.config)
if callable(getattr(extension, "setup_routes", None)):
routes.extend(extension.setup_routes())
logger.info('Loaded HTTP extension: %s',
extension.__class__.__name__)
else:
logger.info(
'Disabled HTTP router %s: missing setup_routes method',
extension.__class__.__name__)
def _setup_server(self):
cherrypy.config.update({
'engine.autoreload_on': False,
'server.socket_host': self.hostname,
'server.socket_port': self.port,
return routes
def _create_routes(self):
mopidy_dir = os.path.join(os.path.dirname(__file__), 'data')
static_dir = self.config['http']['static_dir']
# either default mopidy or user defined path to files
primary_dir = (r"/(.*)", StaticFileHandler, {
'path': static_dir if static_dir else mopidy_dir,
'default_filename': 'index.html'
})
def _setup_websocket_plugin(self):
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
routes = self._load_extensions()
logger.debug(
'HTTP routes from extensions: %s',
list((l[0], l[1]) for l in routes)
)
def _create_app(self):
root = RootResource()
root.mopidy = MopidyResource()
root.mopidy.ws = ws.WebSocketResource(self.core)
if self.config['http']['static_dir']:
static_dir = self.config['http']['static_dir']
else:
static_dir = os.path.join(os.path.dirname(__file__), 'data')
logger.debug('HTTP server will serve "%s" at /', static_dir)
mopidy_dir = os.path.join(os.path.dirname(__file__), 'data')
favicon = os.path.join(mopidy_dir, 'favicon.png')
config = {
b'/': {
'tools.staticdir.on': True,
'tools.staticdir.index': 'index.html',
'tools.staticdir.dir': static_dir,
},
b'/favicon.ico': {
'tools.staticfile.on': True,
'tools.staticfile.filename': favicon,
},
b'/mopidy': {
'tools.staticdir.on': True,
'tools.staticdir.index': 'mopidy.html',
'tools.staticdir.dir': mopidy_dir,
},
b'/mopidy/ws': {
'tools.websocket.on': True,
'tools.websocket.handler_cls': ws.WebSocketHandler,
},
}
return cherrypy.tree.mount(root, '/', config)
def _setup_logging(self, app):
cherrypy.log.access_log.setLevel(logging.NOTSET)
cherrypy.log.error_log.setLevel(logging.NOTSET)
cherrypy.log.screen = False
app.log.access_log.setLevel(logging.NOTSET)
app.log.error_log.setLevel(logging.NOTSET)
# TODO: Dynamically define all endpoints
routes.extend([
(r"/mopidy/ws/?", handlers.WebSocketHandler, {'actor': self}),
(r"/mopidy/rpc", handlers.JsonRpcHandler, {'actor': self}),
(r"/mopidy/(.*)", StaticFileHandler, {
'path': mopidy_dir, 'default_filename': 'mopidy.html'
}),
primary_dir,
])
return routes
def on_start(self):
if self.zeroconf_name:
self.zeroconf_service = zeroconf.Zeroconf(
stype='_http._tcp', name=self.zeroconf_name,
host=self.hostname, port=self.port)
if self.zeroconf_service.publish():
logger.debug(
'Registered HTTP with Zeroconf as "%s"',
self.zeroconf_service.name)
else:
logger.debug('Registering HTTP with Zeroconf failed.')
threading.Thread(target=self._startup).start()
def _startup(self):
logger.debug('Starting HTTP server')
cherrypy.engine.start()
logger.info('HTTP server running at %s', cherrypy.server.base())
self._publish_zeroconf()
self.app = tornado.web.Application(self._create_routes())
self.app.listen(self.port, self.hostname)
logger.info('HTTP server running at http://%s:%s', self.hostname,
self.port)
tornado.ioloop.IOLoop.instance().start()
def _shutdown(self):
logger.debug('Stopping HTTP server')
tornado.ioloop.IOLoop.instance().stop()
logger.info('Stopped HTTP server')
def on_stop(self):
logger.debug('Stopping HTTP server')
self._unpublish_zeroconf()
cherrypy.engine.exit()
logger.info('Stopped HTTP server')
tornado.ioloop.IOLoop.instance().add_callback(self._shutdown)
if self.zeroconf_service:
self.zeroconf_service.unpublish()
def on_event(self, name, **data):
event = data
event['event'] = name
message = json.dumps(event, cls=models.ModelJSONEncoder)
cherrypy.engine.publish('websocket-broadcast', TextMessage(message))
def _publish_zeroconf(self):
if not self.zeroconf_name:
return
self.zeroconf_http_service = zeroconf.Zeroconf(
stype='_http._tcp', name=self.zeroconf_name,
host=self.hostname, port=self.port)
if self.zeroconf_http_service.publish():
logger.debug(
'Registered HTTP with Zeroconf as "%s"',
self.zeroconf_http_service.name)
else:
logger.debug('Registering HTTP with Zeroconf failed.')
self.zeroconf_mopidy_http_service = zeroconf.Zeroconf(
stype='_mopidy-http._tcp', name=self.zeroconf_name,
host=self.hostname, port=self.port)
if self.zeroconf_mopidy_http_service.publish():
logger.debug(
'Registered Mopidy-HTTP with Zeroconf as "%s"',
self.zeroconf_mopidy_http_service.name)
else:
logger.debug('Registering Mopidy-HTTP with Zeroconf failed.')
def _unpublish_zeroconf(self):
if self.zeroconf_http_service:
self.zeroconf_http_service.unpublish()
if self.zeroconf_mopidy_http_service:
self.zeroconf_mopidy_http_service.unpublish()
class RootResource(object):
pass
class MopidyResource(object):
pass
handlers.WebSocketHandler.broadcast(self.websocket_clients, message)

0
mopidy/http/data/favicon.png → mopidy/http/data/favicon.ico
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 5.9 KiB

After

Width:  |  Height:  |  Size: 5.9 KiB

1
mopidy/http/data/mopidy.html
View File

@ -38,6 +38,7 @@
<p>For more information, please refer to the Mopidy documentation at
<a href="http://docs.mopidy.com/">docs.mopidy.com</a>.</p>
</div>
<script src="mopidy.js" type="text/javascript"></script>
<script type="text/javascript">
var ws = new WebSocket("ws://" + document.location.host + "/mopidy/ws/");
ws.onmessage = function (message) {

115
mopidy/http/handlers.py Normal file
View File

@ -0,0 +1,115 @@
from __future__ import unicode_literals
import logging
import tornado.escape
import tornado.web
import tornado.websocket
from mopidy import __version__, core, models
from mopidy.utils import jsonrpc
logger = logging.getLogger(__name__)
def construct_rpc(actor):
inspector = jsonrpc.JsonRpcInspector(
objects={
'core.get_uri_schemes': core.Core.get_uri_schemes,
'core.get_version': core.Core.get_version,
'core.library': core.LibraryController,
'core.playback': core.PlaybackController,
'core.playlists': core.PlaylistsController,
'core.tracklist': core.TracklistController,
})
return jsonrpc.JsonRpcWrapper(
objects={
'core.describe': inspector.describe,
'core.get_uri_schemes': actor.core.get_uri_schemes,
'core.get_version': actor.core.get_version,
'core.library': actor.core.library,
'core.playback': actor.core.playback,
'core.playlists': actor.core.playlists,
'core.tracklist': actor.core.tracklist,
},
decoders=[models.model_json_decoder],
encoders=[models.ModelJSONEncoder]
)
class WebSocketHandler(tornado.websocket.WebSocketHandler):
def initialize(self, actor):
self.actor = actor
self.jsonrpc = construct_rpc(actor)
@classmethod
def broadcast(cls, clients, msg):
for client in clients:
client.write_message(msg)
def open(self):
self.set_nodelay(True)
self.actor.websocket_clients.add(self)
logger.debug(
'New WebSocket connection from %s', self.request.remote_ip)
def on_close(self):
self.actor.websocket_clients.discard(self)
logger.debug(
'Closed WebSocket connection from %s',
self.request.remote_ip)
def on_message(self, message):
if not message:
return
logger.debug(
'Received WebSocket message from %s: %r',
self.request.remote_ip, message)
try:
response = self.jsonrpc.handle_json(
tornado.escape.native_str(message)
)
if response and self.write_message(response):
logger.debug(
'Sent WebSocket message to %s: %r',
self.request.remote_ip, response)
except Exception as e:
logger.error('WebSocket request error:', e)
self.close()
class JsonRpcHandler(tornado.web.RequestHandler):
def initialize(self, actor):
self.jsonrpc = construct_rpc(actor)
def head(self):
self.set_extra_headers()
self.finish()
def post(self):
data = self.request.body
if not data:
return
logger.debug('Received RPC message from %s: %r',
self.request.remote_ip, data)
try:
self.set_extra_headers()
response = self.jsonrpc.handle_json(
tornado.escape.native_str(data))
if response and self.write(response):
logger.debug('Sent RPC message to %s: %r',
self.request.remote_ip, response)
except Exception as e:
logger.error('HTTP JSON-RPC request error:', e)
self.write_error(500)
def set_extra_headers(self):
self.set_header('Accept', 'application/json')
self.set_header('Cache-Control', 'no-cache')
self.set_header('X-Mopidy-Version', __version__.encode(
'utf-8'))
self.set_header('Content-Type', 'application/json; utf-8')

94
mopidy/http/ws.py
View File

@ -1,94 +0,0 @@
from __future__ import unicode_literals
import logging
import socket
import cherrypy
import ws4py.websocket
from mopidy import core, models
from mopidy.utils import jsonrpc
logger = logging.getLogger(__name__)
class WebSocketResource(object):
def __init__(self, core_proxy):
self._core = core_proxy
inspector = jsonrpc.JsonRpcInspector(
objects={
'core.get_uri_schemes': core.Core.get_uri_schemes,
'core.get_version': core.Core.get_version,
'core.library': core.LibraryController,
'core.playback': core.PlaybackController,
'core.playlists': core.PlaylistsController,
'core.tracklist': core.TracklistController,
})
self.jsonrpc = jsonrpc.JsonRpcWrapper(
objects={
'core.describe': inspector.describe,
'core.get_uri_schemes': self._core.get_uri_schemes,
'core.get_version': self._core.get_version,
'core.library': self._core.library,
'core.playback': self._core.playback,
'core.playlists': self._core.playlists,
'core.tracklist': self._core.tracklist,
},
decoders=[models.model_json_decoder],
encoders=[models.ModelJSONEncoder])
@cherrypy.expose
def index(self):
logger.debug('WebSocket handler created')
cherrypy.request.ws_handler.jsonrpc = self.jsonrpc
class _WebSocket(ws4py.websocket.WebSocket):
"""Sub-class ws4py WebSocket with better error handling."""
def send(self, *args, **kwargs):
try:
super(_WebSocket, self).send(*args, **kwargs)
return True
except socket.error as e:
logger.warning('Send message failed: %s', e)
# This isn't really correct, but its the only way to break of out
# the loop in run and trick ws4py into cleaning up.
self.client_terminated = self.server_terminated = True
return False
def close(self, *args, **kwargs):
try:
super(_WebSocket, self).close(*args, **kwargs)
except socket.error as e:
logger.warning('Closing WebSocket failed: %s', e)
class WebSocketHandler(_WebSocket):
def opened(self):
remote = cherrypy.request.remote
logger.debug(
'New WebSocket connection from %s:%d',
remote.ip, remote.port)
def closed(self, code, reason=None):
remote = cherrypy.request.remote
logger.debug(
'Closed WebSocket connection from %s:%d '
'with code %s and reason %r',
remote.ip, remote.port, code, reason)
def received_message(self, request):
remote = cherrypy.request.remote
request = str(request)
logger.debug(
'Received WebSocket message from %s:%d: %r',
remote.ip, remote.port, request)
response = self.jsonrpc.handle_json(request)
if response and self.send(response):
logger.debug(
'Sent WebSocket message to %s:%d: %r',
remote.ip, remote.port, response)

5
setup.py
View File

@ -26,10 +26,9 @@ setup(
install_requires=[
'setuptools',
'Pykka >= 1.1',
'tornado >= 3.1.1',
],
extras_require={
'http': ['CherryPy >= 3.2.2', 'ws4py >= 0.2.3'],
},
extras_requires={'http': []},
test_suite='nose.collector',
tests_require=[
'nose',

31
tests/http/test_events.py
View File

@ -5,23 +5,18 @@ import unittest
import mock
try:
import cherrypy
except ImportError:
cherrypy = False
try:
import ws4py
import tornado
except ImportError:
ws4py = False
tornado = False
if cherrypy and ws4py:
if tornado:
from mopidy.http import actor
@unittest.skipUnless(cherrypy, 'cherrypy not found')
@unittest.skipUnless(ws4py, 'ws4py not found')
@mock.patch('cherrypy.engine.publish')
@unittest.skipUnless(tornado, 'tornado is missing')
@mock.patch('mopidy.http.handlers.WebSocketHandler.broadcast')
class HttpEventsTest(unittest.TestCase):
def setUp(self):
config = {
@ -34,22 +29,22 @@ class HttpEventsTest(unittest.TestCase):
}
self.http = actor.HttpFrontend(config=config, core=mock.Mock())
def test_track_playback_paused_is_broadcasted(self, publish):
publish.reset_mock()
def test_track_playback_paused_is_broadcasted(self, broadcast):
broadcast.reset_mock()
self.http.on_event('track_playback_paused', foo='bar')
self.assertEqual(publish.call_args[0][0], 'websocket-broadcast')
self.assertEqual(broadcast.call_args[0][0], set([]))
self.assertDictEqual(
json.loads(str(publish.call_args[0][1])), {
json.loads(str(broadcast.call_args[0][1])), {
'event': 'track_playback_paused',
'foo': 'bar',
})
def test_track_playback_resumed_is_broadcasted(self, publish):
publish.reset_mock()
def test_track_playback_resumed_is_broadcasted(self, broadcast):
broadcast.reset_mock()
self.http.on_event('track_playback_resumed', foo='bar')
self.assertEqual(publish.call_args[0][0], 'websocket-broadcast')
self.assertEqual(broadcast.call_args[0][0], set([]))
self.assertDictEqual(
json.loads(str(publish.call_args[0][1])), {
json.loads(str(broadcast.call_args[0][1])), {
'event': 'track_playback_resumed',
'foo': 'bar',
})

196
tests/http/test_router.py Normal file
View File

@ -0,0 +1,196 @@
from __future__ import unicode_literals
import os
import unittest
import mock
from tornado.escape import json_decode, json_encode, to_unicode
from tornado.testing import AsyncHTTPTestCase
from tornado.web import Application
from mopidy import __version__, http
try:
import tornado
except ImportError:
tornado = False
if tornado:
from mopidy.http import actor
class TestRouter(http.Router):
name = 'test'
path = os.path.join(os.path.dirname(__file__), 'static')
class TestRouterMissingPath(http.Router):
name = 'test'
class TestRouterMissingName(http.Router):
path = os.path.join(os.path.dirname(__file__), 'static')
@unittest.skipUnless(tornado, 'tornado is missing')
class HttpRouterTest(unittest.TestCase):
def setUp(self):
self.config = {
'http': {
'hostname': '127.0.0.1',
'port': 6680,
'static_dir': None,
'zeroconf': '',
}
}
self.http = actor.HttpFrontend(config=self.config, core=mock.Mock())
def test_default_router(self):
router = TestRouter(self.config)
self.assertEqual(router.setup_routes()[0][0], r'/test/(.*)')
self.assertIs(router.setup_routes()[0][1], http.StaticFileHandler)
self.assertEqual(router.setup_routes()[0][2]['path'],
os.path.join(os.path.dirname(__file__), 'static'))
def test_default_router_missing_name(self):
with self.assertRaises(ValueError):
TestRouterMissingName(self.config)
def test_default_router_missing_path(self):
with self.assertRaises(ValueError):
TestRouterMissingPath(self.config).setup_routes()
def test_default_uri_helper(self):
router = TestRouter(self.config)
self.assertEqual('http://127.0.0.1:6680/test/', router.linkify())
class StaticFileHandlerTest(AsyncHTTPTestCase):
def get_app(self):
app = Application([(r'/(.*)', http.StaticFileHandler, {
'path': os.path.dirname(__file__),
'default_filename': 'test_router.py'
})])
return app
def test_static_handler(self):
response = self.fetch('/test_router.py', method='GET')
self.assertEqual(response.headers['X-Mopidy-Version'],
__version__)
self.assertEqual(response.headers['Cache-Control'],
'no-cache')
def test_static_default_filename(self):
response = self.fetch('/', method='GET')
self.assertEqual(response.headers['X-Mopidy-Version'],
__version__)
self.assertEqual(response.headers['Cache-Control'],
'no-cache')
class DefaultHTTPServerTest(AsyncHTTPTestCase):
def get_app(self):
config = {
'http': {
'hostname': '127.0.0.1',
'port': 6680,
'static_dir': None,
'zeroconf': '',
}
}
core = mock.Mock()
core.get_version = mock.MagicMock(name='get_version')
core.get_version.return_value = __version__
actor_http = actor.HttpFrontend(config=config, core=core)
return Application(actor_http._create_routes())
def test_root_should_return_index(self):
response = self.fetch('/', method='GET')
self.assertIn(
'Static content serving',
to_unicode(response.body)
)
self.assertEqual(response.headers['X-Mopidy-Version'],
__version__)
self.assertEqual(response.headers['Cache-Control'],
'no-cache')
def test_mopidy_should_return_index(self):
response = self.fetch('/mopidy/', method='GET')
self.assertIn(
'Here you can see events arriving from Mopidy in real time:',
to_unicode(response.body)
)
self.assertEqual(response.headers['X-Mopidy-Version'],
__version__)
self.assertEqual(response.headers['Cache-Control'],
'no-cache')
def test_should_return_js(self):
response = self.fetch('/mopidy/mopidy.js', method='GET')
self.assertIn(
'function Mopidy',
to_unicode(response.body)
)
self.assertEqual(response.headers['X-Mopidy-Version'],
__version__)
self.assertEqual(response.headers['Cache-Control'],
'no-cache')
def test_should_return_ws(self):
response = self.fetch('/mopidy/ws', method='GET')
self.assertEqual(
'Can "Upgrade" only to "WebSocket".',
to_unicode(response.body)
)
def test_should_return_ws_old(self):
response = self.fetch('/mopidy/ws/', method='GET')
self.assertEqual(
'Can "Upgrade" only to "WebSocket".',
to_unicode(response.body)
)
def test_should_return_rpc_error(self):
cmd = json_encode({
'action': 'get_version'
})
response = self.fetch('/mopidy/rpc', method='POST', body=cmd)
self.assertEqual(
{'jsonrpc': '2.0', 'id': None, 'error':
{'message': 'Invalid Request', 'code': -32600,
'data': '"jsonrpc" member must be included'}},
json_decode(response.body)
)
def test_should_return_parse_error(self):
cmd = '{[[[]}'
response = self.fetch('/mopidy/rpc', method='POST', body=cmd)
self.assertEqual(
{'jsonrpc': '2.0', 'id': None, 'error':
{'message': 'Parse error', 'code': -32700}},
json_decode(response.body)
)
def test_should_return_mopidy_version(self):
cmd = json_encode({
'method': 'core.get_version',
'params': [],
'jsonrpc': '2.0',
'id': 1
})
response = self.fetch('/mopidy/rpc', method='POST', body=cmd)
self.assertEqual(
{'jsonrpc': '2.0', 'id': 1, 'result': __version__},
json_decode(response.body)
)
def test_should_return_extra_headers(self):
response = self.fetch('/mopidy/rpc', method='HEAD')
self.assertIn('Accept', response.headers)
self.assertIn('X-Mopidy-Version', response.headers)
self.assertIn('Cache-Control', response.headers)
self.assertIn('Content-Type', response.headers)