diff --git a/mopidy_iris/system.py b/mopidy_iris/system.py index 37b40045..a7bcf10f 100755 --- a/mopidy_iris/system.py +++ b/mopidy_iris/system.py @@ -1,5 +1,5 @@ from threading import Thread -import logging, pathlib, subprocess, json +import logging, os, pathlib, subprocess, json # import logger logger = logging.getLogger(__name__) @@ -14,6 +14,7 @@ class IrisSystemPermissionError(IrisSystemError): 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) @@ -22,16 +23,34 @@ class IrisSystemMissingError(IrisSystemError): def __init__(self, path): message = "Unable to access %s." % 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.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 ## @@ -52,11 +71,9 @@ class IrisSystemThread(Thread): 'error': error } - logger.debug("sudo %s %s", self.script_path.as_uri(), self.action) - - proc = subprocess.Popen([b"sudo", bytes(self.script_path), self.action.encode()], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) + 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() @@ -69,8 +86,6 @@ class IrisSystemThread(Thread): logger.info(response_string) self.callback({'output': response_string}, None) - - ## # Check if we have access to the system script (system.sh) # @@ -81,7 +96,8 @@ class IrisSystemThread(Thread): raise IrisSystemMissingError(self.script_path) # Attempt an empty call to our system file - process = subprocess.Popen(b"sudo -n %s check" % bytes(self.script_path), 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() diff --git a/tests/test_system.py b/tests/test_system.py new file mode 100644 index 00000000..087a2280 --- /dev/null +++ b/tests/test_system.py @@ -0,0 +1,98 @@ +import pathlib, pytest, subprocess +from unittest import mock + +from mopidy_iris.system import IrisSystemThread, IrisSystemMissingError, 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_script_missing_raises(tmp_path, caplog): + iris_system = IrisSystemThread('foo', None) + iris_system.script_path = tmp_path + + with pytest.raises(IrisSystemMissingError) as excinfo: + iris_system.can_run() + + error_message = "Unable to access %s." % tmp_path.as_uri() + assert error_message in str(excinfo.value) + assert error_message in caplog.text + + +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"