Added IrisSystemThread.get_command() helper and system tests.

_USE_SUDO is to make testing easier.

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()
 

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"