secstate/Iris
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Merge pull request #477 from kingosticks/feature/python3-py37

Browse Source 
Feature/python3 py37
This commit is contained in:
James Barnsley
2020-01-17 16:41:39 +13:00
committed by GitHub
parent d0048a4f8b 985f7bb542
commit be9602e72c
8 changed files with 280 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
LICENSE.md → LICENSE
View File

49
mopidy_iris/__init__.py
View File

@ -1,18 +1,13 @@
import logging, json, pathlib
from __future__ import unicode_literals
import logging, os, json, pathlib
import tornado.web
import tornado.websocket
import pkg_resources
from mopidy import config, ext
from .frontend import IrisFrontend
from .handlers import WebsocketHandler, HttpHandler
from .core import IrisCore
from .mem import iris
__version__ = pkg_resources.get_distribution("Mopidy-Iris").version
logger = logging.getLogger(__name__)
##
# Core extension class
#
@ -22,10 +17,10 @@ class Extension( ext.Extension ):
dist_name = 'Mopidy-Iris'
ext_name = 'iris'
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.read(pathlib.Path(__file__).parent / "ext.conf")
def get_config_schema(self):
schema = config.ConfigSchema(self.ext_name)
@ -39,7 +34,7 @@ class Extension( ext.Extension ):
return schema
def setup(self, registry):
from .frontend import IrisFrontend
# Add web extension
registry.add('http:app', {
'name': self.ext_name,
@ -49,32 +44,19 @@ class Extension( ext.Extension ):
# Add our frontend
registry.add('frontend', IrisFrontend)
##
# Customised handler for react router URLS
#
# This routes all URLs to the same path, so that React can handle the path etc
##
class ReactRouterHandler(tornado.web.StaticFileHandler):
def initialize(self, path):
self.path = path
self.absolute_path = path
self.dirname, self.filename = os.path.split(path)
super(ReactRouterHandler, self).initialize(self.dirname)
def get(self, path=None, include_body=True):
return super(ReactRouterHandler, self).get(self.path, include_body)
##
# Frontend factory
##
def iris_factory(config, core):
from tornado.web import StaticFileHandler
from .handlers import HttpHandler, ReactRouterHandler, WebsocketHandler
path = pathlib.Path(__file__).parent / 'static'
return [
(
r'/http/([^/]*)',
handlers.HttpHandler,
HttpHandler,
{
'core': core,
'config': config
@ -82,7 +64,7 @@ def iris_factory(config, core):
),
(
r'/ws/?',
handlers.WebsocketHandler,
WebsocketHandler,
{
'core': core,
'config': config
@ -90,21 +72,22 @@ def iris_factory(config, core):
),
(
r'/assets/(.*)',
tornado.web.StaticFileHandler,
StaticFileHandler,
{
'path': path / 'assets'
}
),
(
r'/((.*)(?:css|js|json|map)$)',
tornado.web.StaticFileHandler,
StaticFileHandler,
{
'path': path
}
),
(
r'/(.*)',
ReactRouterHandler, {
ReactRouterHandler,
{
'path': path / 'index.html'
}
),

34
mopidy_iris/core.py
View File

@ -1,7 +1,4 @@
from __future__ import unicode_literals
import random, string, logging, json, pykka, urllib, os, sys, mopidy_iris, subprocess
import random, string, logging, json, pathlib, pykka, urllib, os, sys, mopidy_iris, subprocess
import tornado.web
import tornado.ioloop
import tornado.httpclient
@ -14,6 +11,7 @@ from pkg_resources import parse_version
from tornado.escape import json_encode, json_decode
from tornado.httpclient import AsyncHTTPClient
from . import Extension
from .system import IrisSystemThread
if sys.platform == 'win32':
@ -37,6 +35,11 @@ class IrisCore(pykka.ThreadingActor):
"results": []
}
@classmethod
async def do_fetch(cls, client, request):
# This wrapper function exists to ease mocking.
return await client.fetch(request)
def setup(self, config, core):
self.config = config
self.core = core
@ -65,15 +68,11 @@ class IrisCore(pykka.ThreadingActor):
# @return void
##
def save_to_file(self, dict, name):
path = self.config['iris'].get('data_dir')
# Create the folder if it doesn't yet exist
if not os.path.exists(path):
os.makedirs(path)
file_path = Extension.get_data_dir(self.config) / ('%s.pkl' % name)
# And now open the file, and drop in our dict
try:
with open(path + '/' + name + '.pkl', 'wb') as f:
with file_path.open('wb') as f:
pickle.dump(dict, f, pickle.HIGHEST_PROTOCOL)
except Exception:
return False
@ -85,10 +84,10 @@ class IrisCore(pykka.ThreadingActor):
# @return Dict
##
def load_from_file(self, name):
path = self.config['iris'].get('data_dir')
file_path = Extension.get_data_dir(self.config) / ('%s.pkl' % name)
try:
with open(path + '/' + name + '.pkl', 'rb') as f:
with file_path.open('wb') as f:
return pickle.load(f)
except Exception:
return {}
@ -99,10 +98,9 @@ class IrisCore(pykka.ThreadingActor):
# @return String
##
def load_version(self):
filepath = os.path.join(os.path.dirname(__file__), '..', 'IRIS_VERSION')
file_path = pathlib.Path(__file__).parent.parent / 'IRIS_VERSION'
try:
with open(filepath, 'r') as f:
return f.read()
return file_path.read_text()
except Exception:
return "Unknown"
@ -975,7 +973,7 @@ class IrisCore(pykka.ThreadingActor):
else:
return response
def refresh_spotify_token(self, *args, **kwargs):
async def refresh_spotify_token(self, *args, **kwargs):
callback = kwargs.get('callback', None)
# Use client_id and client_secret from config
@ -988,9 +986,9 @@ class IrisCore(pykka.ThreadingActor):
}
try:
http_client = tornado.httpclient.HTTPClient()
http_client = tornado.httpclient.AsyncHTTPClient()
request = tornado.httpclient.HTTPRequest(url, method='POST', body=urllib.parse.urlencode(data))
response = http_client.fetch(request)
response = await self.do_fetch(http_client, request)
token = json.loads(response.body)
token['expires_at'] = time.time() + token['expires_in']

4
mopidy_iris/frontend.py
View File

@ -1,5 +1,3 @@
from __future__ import unicode_literals
from mopidy.core import CoreListener
from .core import IrisCore
@ -13,7 +11,7 @@ logger = logging.getLogger(__name__)
class IrisFrontend(pykka.ThreadingActor, CoreListener):
def __init__(self, config, core):
super(IrisFrontend, self).__init__()
super().__init__()
# Pass our Mopidy config and core to the IrisCore instance
iris.config = config

17
mopidy_iris/handlers.py
View File

@ -1,5 +1,3 @@
from __future__ import unicode_literals
from datetime import datetime
from tornado.escape import json_encode, json_decode
import tornado.ioloop, tornado.web, tornado.websocket, tornado.template
@ -235,4 +233,19 @@ class HttpHandler(tornado.web.RequestHandler):
self.finish()
##
# Customised handler for react router URLS
#
# This routes all URLs to the same path, so that React can handle the path etc
##
class ReactRouterHandler(tornado.web.StaticFileHandler):
def initialize(self, path):
self.path = path
self.absolute_path = path
self.dirname = path.parent
self.filename = path.name
super().initialize(self.dirname)
def get(self, path=None, include_body=True):
return super().get(self.path, include_body)

68
mopidy_iris/system.py
View File

@ -1,17 +1,47 @@
from threading import Thread
import os, logging, subprocess, json
import logging, os, pathlib, subprocess, json
# import logger
logger = logging.getLogger(__name__)
class IrisSystemError(Exception):
pass
class IrisSystemPermissionError(IrisSystemError):
reason = "Permission denied"
def __init__(self, path):
message = "Password-less access to %s was refused. Check your /etc/sudoers file." % path.as_uri()
logger.error(message)
super().__init__(message)
class IrisSystemThread(Thread):
_USE_SUDO = True
def __init__(self, action, callback):
Thread.__init__(self)
self.action = action
self.callback = callback
self.path = os.path.dirname(__file__)
self.script_path = pathlib.Path(__file__).parent / "system.sh"
def get_command(self, action=None, *, non_interactive=False):
if self._USE_SUDO:
if non_interactive:
args = [b'sudo -n']
else:
args = [b'sudo']
else:
args = []
if action is None:
action = self.action
args = args + [bytes(self.script_path), action.encode()]
return args
##
# Run the defined action
##
@ -20,34 +50,32 @@ class IrisSystemThread(Thread):
try:
self.can_run()
except Exception as e:
except IrisSystemError as e:
logger.error(e)
error = {
'message': "Permission denied",
'description': str(e)
'message': e.reason,
'description': e.message
}
return {
'error': error
}
logger.debug("sudo "+ self.path +"/system.sh "+ self.action)
proc = subprocess.Popen(["sudo", self.path+"/system.sh", self.action],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
command = self.get_command()
logger.debug("Running '%s'", os.fsdecode(b' '.join(command)))
proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout,stderr = proc.communicate()
if stderr:
logger.error(stderr.decode())
self.callback(None, { 'error': stderr.decode() })
error_string = os.fsdecode(stderr)
logger.error(error_string)
self.callback(None, {'error': error_string})
else:
logger.info(stdout.decode())
self.callback({ 'output': stdout.decode() }, None)
response_string = os.fsdecode(stdout)
logger.info(response_string)
self.callback({'output': response_string}, None)
##
# Check if we have access to the system script (system.sh)
@ -55,14 +83,14 @@ class IrisSystemThread(Thread):
# @return boolean or exception
##
def can_run(self, *args, **kwargs):
# Attempt an empty call to our system file
process = subprocess.Popen("sudo -n "+self.path+"/system.sh check", stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
command_bytes = b' '.join(self.get_command('check', non_interactive=True))
process = subprocess.Popen(command_bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
result, error = process.communicate()
exitCode = process.wait()
# Some kind of failure, so we can't run any commands this way
if exitCode > 0:
raise Exception("Password-less access to "+self.path+"/system.sh was refused. Check your /etc/sudoers file.")
raise IrisSystemPermissionError(self.script_path)
else:
return True

98
tests/test_handlers.py Normal file
View File

@ -0,0 +1,98 @@
import pytest
from asyncio import Future
from unittest import mock
import tornado.testing
import tornado.web
from tornado.httpclient import HTTPResponse
from tornado.escape import json_decode
from mopidy_iris import handlers
from mopidy_iris import core
from mopidy_iris.mem import iris
def async_return_helper(result):
f = Future()
f.set_result(result)
return f
class HttpHandlerTest(tornado.testing.AsyncHTTPTestCase):
@pytest.fixture(autouse=True)
def inject_fixtures(self, caplog):
self._caplog = caplog
def get_app(self):
http_handler = handlers.HttpHandler
# http_handler.handle_result = mock.Mock()
# self.handler_mock = http_handler.handle_result
return tornado.web.Application(
[
(
r"/(.*)",
http_handler,
{
'core': None,
'config': {},
},
)
]
)
def test_get_method(self):
with mock.patch("time.time", return_value=100):
response = self.fetch("/test", method="GET")
assert 200 == response.code
result = json_decode(response.body)
assert "2.0" == result["jsonrpc"]
assert "test" == result["method"]
assert 100 == result["id"]
assert "Running test... please wait" == result["result"]["message"]
def test_get_method_headers(self):
response = self.fetch("/test", method="GET")
assert response.headers["Access-Control-Allow-Origin"] == "*"
assert "Origin" in response.headers["Access-Control-Allow-Headers"]
def test_get_unknown_method_is_error(self):
response = self.fetch("/baz", method="GET")
assert 400 == response.code
error = json_decode(response.body)["error"]
assert "Method baz does not exist" == error["message"]
@mock.patch.object(handlers, "iris")
def test_get_method_called(self, iris_mock):
iris_mock.foo = mock.Mock()
response = self.fetch("/foo", method="GET")
iris_mock.foo.assert_called_once()
assert 200 == response.code
@mock.patch.object(handlers, "iris")
def test_get_method_any_exception_handled(self, iris_mock):
iris_mock.foo = mock.Mock(side_effect=Exception("bar"))
response = self.fetch("/foo", method="GET")
iris_mock.foo.assert_called_once()
assert 200 == response.code
assert "bar" in self._caplog.text
@mock.patch.object(handlers.iris, "do_fetch")
def test_get_method_with_fetch(self, fetch_mock):
iris.config = {"spotify" : {"client_id": 123, "client_secret": 456}}
result = mock.Mock(spec=HTTPResponse, body='{"expires_in":88}')
fetch_mock.return_value = async_return_helper(result)
response = self.fetch("/refresh_spotify_token", method="GET")
assert 200 == response.code
assert len(response.body) > 0
result = json_decode(response.body)
assert "refresh_spotify_token" == result["method"]
assert 88 == result["result"]["spotify_token"]["expires_in"]

86
tests/test_system.py Normal file
View File

@ -0,0 +1,86 @@
import pathlib, pytest, subprocess
from unittest import mock
from mopidy_iris.system import IrisSystemThread, IrisSystemPermissionError
def test_system_sh_path():
iris_system = IrisSystemThread('foo', None)
assert iris_system.script_path.is_file()
assert iris_system.script_path.name == "system.sh"
def test_can_run():
iris_system = IrisSystemThread('foo', None)
iris_system._USE_SUDO = False
assert iris_system.can_run() is True
@pytest.fixture
def popen_mock():
patcher = mock.patch("subprocess.Popen", spec=True)
yield patcher.start()
patcher.stop()
@pytest.fixture
def process_mock(popen_mock):
mock_process = popen_mock.return_value
mock_process.communicate.return_value = ('', None)
mock_process.wait.return_value = 0
yield mock_process
def test_can_run_args(popen_mock, process_mock):
IrisSystemThread('foo', None).can_run()
popen_mock.assert_called_once_with(
mock.ANY,
shell=True,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE
)
def test_can_run_uses_sudo_non_interactive(popen_mock, process_mock):
IrisSystemThread('foo', None).can_run()
popen_mock.assert_called_once()
assert popen_mock.call_args[0][0].startswith(b"sudo -n ")
def test_can_run_calls_script_check(popen_mock, process_mock):
IrisSystemThread('foo', None).can_run()
assert popen_mock.call_args[0][0].endswith(b"system.sh check")
def test_can_run_sudo_refused_raises(popen_mock, process_mock, caplog):
process_mock.wait.return_value = 1
iris_system = IrisSystemThread('foo', None)
with pytest.raises(IrisSystemPermissionError) as excinfo:
iris_system.can_run()
error_message = (
"Password-less access to %s was refused. "
"Check your /etc/sudoers file." % iris_system.script_path.as_uri()
)
assert error_message in str(excinfo.value)
assert error_message in caplog.text
def test_run_args(popen_mock, process_mock):
iris_system = IrisSystemThread('foo', mock.Mock())
iris_system.can_run = mock.Mock(return_value = True)
iris_system.run()
popen_mock.assert_called_once_with(
mock.ANY,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE
)
def test_run_uses_sudo(popen_mock, process_mock):
iris_system = IrisSystemThread('foo', mock.Mock())
iris_system.can_run = mock.Mock(return_value = True)
iris_system.run()
popen_mock.assert_called_once()
assert popen_mock.call_args[0][0][0] == b"sudo"