tests: AdminVM.run_service*()

qubes/tests/vm/adminvm.py

@@ -17,10 +17,14 @@
 # You should have received a copy of the GNU Lesser General Public
 # License along with this library; if not, see <https://www.gnu.org/licenses/>.
 #
-
+import subprocess
 import unittest
 import unittest.mock
 
+import functools
+
+import asyncio
+
 import qubes
 import qubes.exc
 import qubes.vm
@@ -44,6 +48,24 @@ class TC_00_AdminVM(qubes.tests.QubesTestCase):
                 raise
             self.skipTest('setup failed')
 
+    def tearDown(self) -> None:
+        self.app.domains.clear()
+
+    def add_vm(self, name, cls=qubes.vm.qubesvm.QubesVM, **kwargs):
+        vm = cls(self.app, None,
+                 qid=kwargs.pop('qid', 1), name=qubes.tests.VMPREFIX + name,
+                 **kwargs)
+        self.app.domains[vm.qid] = vm
+        self.app.domains[vm.uuid] = vm
+        self.app.domains[vm.name] = vm
+        self.app.domains[vm] = vm
+        self.addCleanup(vm.close)
+        return vm
+
+    @asyncio.coroutine
+    def coroutine_mock(self, mock, *args, **kwargs):
+        return mock(*args, **kwargs)
+
     def cleanup_adminvm(self):
         self.vm.close()
         del self.vm
@@ -82,3 +104,87 @@ class TC_00_AdminVM(qubes.tests.QubesTestCase):
     def test_311_suspend(self):
         with self.assertRaises(qubes.exc.QubesException):
             self.vm.suspend()
+
+    @unittest.mock.patch('asyncio.create_subprocess_exec')
+    def test_700_run_service(self, mock_subprocess):
+        func_mock = unittest.mock.Mock()
+        mock_subprocess.side_effect = functools.partial(
+            self.coroutine_mock, func_mock)
+
+        self.add_vm('vm')
+
+        with self.subTest('running'):
+            self.loop.run_until_complete(self.vm.run_service('test.service'))
+            func_mock.assert_called_once_with(
+                '/usr/lib/qubes/qubes-rpc-multiplexer',
+                'test.service', 'dom0', 'name', 'dom0')
+
+        func_mock.reset_mock()
+        with self.subTest('other_user'):
+            self.loop.run_until_complete(
+                self.vm.run_service('test.service', user='other'))
+            func_mock.assert_called_once_with(
+                'runuser', '-u', 'other', '--',
+                '/usr/lib/qubes/qubes-rpc-multiplexer',
+                'test.service', 'dom0', 'name', 'dom0')
+
+            func_mock.reset_mock()
+        with self.subTest('other_source'):
+            self.loop.run_until_complete(
+                self.vm.run_service('test.service', source='test-inst-vm'))
+            func_mock.assert_called_once_with(
+                '/usr/lib/qubes/qubes-rpc-multiplexer',
+                'test.service', 'test-inst-vm', 'name', 'dom0')
+
+    @unittest.mock.patch('qubes.vm.adminvm.AdminVM.run_service')
+    def test_710_run_service_for_stdio(self, mock_run_service):
+
+        func_mock = unittest.mock.Mock()
+        mock_run_service.side_effect = functools.partial(
+            self.coroutine_mock, func_mock)
+        communicate_mock = unittest.mock.Mock()
+        func_mock.return_value.communicate.side_effect = functools.partial(
+            self.coroutine_mock, communicate_mock)
+        communicate_mock.return_value = (b'stdout', b'stderr')
+        func_mock.return_value.returncode = 0
+
+        with self.subTest('default'):
+            value = self.loop.run_until_complete(
+                self.vm.run_service_for_stdio('test.service'))
+            func_mock.assert_called_once_with(
+                'test.service',
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                stdin=subprocess.PIPE)
+            communicate_mock.assert_called_once_with(input=None)
+            self.assertEqual(value, (b'stdout', b'stderr'))
+
+        func_mock.reset_mock()
+        communicate_mock.reset_mock()
+        with self.subTest('with_input'):
+            value = self.loop.run_until_complete(
+                self.vm.run_service_for_stdio('test.service', input=b'abc'))
+            func_mock.assert_called_once_with(
+                'test.service',
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                stdin=subprocess.PIPE)
+            communicate_mock.assert_called_once_with(input=b'abc')
+            self.assertEqual(value, (b'stdout', b'stderr'))
+
+        func_mock.reset_mock()
+        communicate_mock.reset_mock()
+        with self.subTest('error'):
+            func_mock.return_value.returncode = 1
+            with self.assertRaises(subprocess.CalledProcessError) as exc:
+                self.loop.run_until_complete(
+                    self.vm.run_service_for_stdio('test.service'))
+            func_mock.assert_called_once_with(
+                'test.service',
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                stdin=subprocess.PIPE)
+            communicate_mock.assert_called_once_with(input=None)
+            self.assertEqual(exc.exception.returncode, 1)
+            self.assertEqual(exc.exception.output, b'stdout')
+            self.assertEqual(exc.exception.stderr, b'stderr')