tests: QubesVM.run_for_stdio

qubes/tests/vm/qubesvm.py

@@ -1995,8 +1995,63 @@ class TC_90_QubesVM(QubesVMTestsMixin, qubes.tests.QubesTestCase):
                 'user:QUBESRPC test.service test-inst-vm')
             self.assertFalse(start_mock.called)
 
+    @unittest.mock.patch('qubes.vm.qubesvm.QubesVM.run')
+    def test_710_run_for_stdio(self, mock_run):
+        vm = self.get_vm(cls=qubes.vm.standalonevm.StandaloneVM,
+                         name='vm')
+
+        func_mock = unittest.mock.Mock()
+        mock_run.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(
+                vm.run_for_stdio('cat'))
+            func_mock.assert_called_once_with(
+                'cat',
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                stdin=subprocess.PIPE)
+            communicate_mock.assert_called_once_with(input=b'')
+            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(
+                vm.run_for_stdio('cat', input=b'abc'))
+            func_mock.assert_called_once_with(
+                'cat',
+                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(
+                    vm.run_for_stdio('cat'))
+            func_mock.assert_called_once_with(
+                'cat',
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                stdin=subprocess.PIPE)
+            communicate_mock.assert_called_once_with(input=b'')
+            self.assertEqual(exc.exception.returncode, 1)
+            self.assertEqual(exc.exception.output, b'stdout')
+            self.assertEqual(exc.exception.stderr, b'stderr')
+
     @unittest.mock.patch('qubes.vm.qubesvm.QubesVM.run_service')
-    def test_710_run_service_for_stdio(self, mock_run_service):
+    def test_711_run_service_for_stdio(self, mock_run_service):
         vm = self.get_vm(cls=qubes.vm.standalonevm.StandaloneVM,
                          name='vm')