Improve image mount failure handling

coriolis/osmorphing/windows.py

@@ -274,9 +274,19 @@ def _add_dism_driver(self, driver_path):
 
     def _mount_disk_image(self, path):
         LOG.info("Mounting disk image: %s" % path)
-        return self._conn.exec_ps_command(
+        drive_letter, stderr = self._conn.exec_ps_command(
             "(Mount-DiskImage '%s' -PassThru | Get-Volume).DriveLetter" %
-            path)
+            path,
+            include_stderr=True)
+        if not drive_letter:
+            # Couldn't mount the image, let's fetch the file size. It may
+            # help us spot invalid images.
+            file_size = self._conn.exec_ps_command(f"(ls '{path}').Length")
+            raise exception.CoriolisException(
+                f"Could not mount image: '{path}', "
+                f"file size: {file_size}, "
+                f"mount stderr: {stderr}")
+        return drive_letter
 
     def _dismount_disk_image(self, path):
         LOG.info("Unmounting disk image: %s" % path)

coriolis/tests/osmorphing/test_windows.py

@@ -203,14 +203,24 @@ def test__add_dism_driver_not_found(self, mock_get_worker_os_drive_path):
         mock_get_worker_os_drive_path.assert_called()
 
     def test__mount_disk_image(self):
+        self.conn.exec_ps_command.return_value = ("fake-letter", "fake stderr")
         result = self.morphing_tools._mount_disk_image(
             mock.sentinel.path)
 
         self.conn.exec_ps_command.assert_called_once_with(
             "(Mount-DiskImage '%s' -PassThru | Get-Volume).DriveLetter" %
-            mock.sentinel.path)
+            mock.sentinel.path,
+            include_stderr=True)
 
-        self.assertEqual(result, self.conn.exec_ps_command.return_value)
+        self.assertEqual("fake-letter", result)
+
+    def test__mount_disk_image_missing_letter(self):
+        self.conn.exec_ps_command.side_effect = [
+            ("", "fake stderr"), "fake-size"]
+        self.assertRaises(
+            exception.CoriolisException,
+            self.morphing_tools._mount_disk_image,
+            mock.sentinel.path)
 
     def test__dismount_disk_image(self):
         self.morphing_tools._dismount_disk_image(mock.sentinel.path)

coriolis/tests/test_wsman.py

@@ -153,7 +153,10 @@ def test_exec_command_exception(self):
     def test_exec_ps_command(self):
         self.conn.exec_command = mock.Mock()
         self.conn.exec_command.return_value = "std_out\n\n"
-        result = self.conn.exec_ps_command(self.cmd)
+        result = self.conn.exec_ps_command(
+            self.cmd,
+            include_stderr=False,
+        )
         self.conn.exec_command.assert_called_once_with(
             "powershell.exe",
             [
@@ -163,9 +166,30 @@ def test_exec_ps_command(self):
                 '-ExecutionPolicy', 'RemoteSigned',
             ],
             timeout=None,
-            sanitizable=False)
+            sanitizable=False,
+            include_stderr=False)
         self.assertEqual(result, "std_out")
 
+    def test_exec_ps_command_with_stderr(self):
+        self.conn.exec_command = mock.Mock()
+        self.conn.exec_command.return_value = "std_out\n\n", "stderr"
+        result = self.conn.exec_ps_command(
+            self.cmd,
+            include_stderr=True,
+        )
+        self.conn.exec_command.assert_called_once_with(
+            "powershell.exe",
+            [
+                "-EncodedCommand",
+                'dABlAHMAdABfAGMAbQBkAA==',
+                '-NonInteractive',
+                '-ExecutionPolicy', 'RemoteSigned',
+            ],
+            timeout=None,
+            sanitizable=False,
+            include_stderr=True)
+        self.assertEqual(result, ("std_out", "stderr"))
+
     def test_test_path(self):
         self.conn.exec_ps_command = mock.Mock()
         self.conn.exec_ps_command.return_value = "True"

coriolis/wsman.py

@@ -139,7 +139,14 @@ def _exec_command(self, cmd, args=[], timeout=None, sanitizable=True):
             if shell_id:
                 self._protocol.close_shell(shell_id)
 
-    def exec_command(self, cmd, args=[], timeout=None, sanitizable=True):
+    def exec_command(
+        self,
+        cmd,
+        args=[],
+        timeout=None,
+        sanitizable=True,
+        include_stderr=False,
+    ):
         # Our sanitization helpers do not work for base64 encoded commands,
         # in which case we'll avoid logging it so that we won't leak
         # sensitive information.
@@ -158,12 +165,20 @@ def exec_command(self, cmd, args=[], timeout=None, sanitizable=True):
                 "stdout: %s\nstd_err: %s" %
                 (sanitized_cmd, exit_code, std_out, std_err))
 
+        if include_stderr:
+            return std_out, std_err
         return std_out
 
-    def exec_ps_command(self, cmd, ignore_stdout=False, timeout=None):
+    def exec_ps_command(
+        self,
+        cmd,
+        ignore_stdout=False,
+        timeout=None,
+        include_stderr=False,
+    ):
         LOG.debug("Executing PS command: %s", strutils.mask_password(cmd))
         base64_cmd = base64.b64encode(cmd.encode('utf-16le')).decode()
-        return self.exec_command(
+        ret = self.exec_command(
             "powershell.exe",
             [
                 "-EncodedCommand",
@@ -173,7 +188,14 @@ def exec_ps_command(self, cmd, ignore_stdout=False, timeout=None):
                 "RemoteSigned",
             ],
             timeout=timeout,
-            sanitizable=False)[:-2]
+            sanitizable=False,
+            include_stderr=include_stderr)
+        if include_stderr:
+            stdout, stderr = ret
+            return stdout[:-2], stderr
+        else:
+            stdout = ret
+            return stdout[:-2]
 
     def test_path(self, remote_path):
         ret_val = self.exec_ps_command("Test-Path -Path \"%s\"" % remote_path)