@@ -802,11 +802,11 @@ private void mockSourceVolume(String volUuid, String name) {
         Mockito.when(volumeDao.findByUuid(volUuid)).thenReturn(srcVolume);
     }
 
-    private void mockThinDiskOffering(Long diskOfferingId) {
+    private void mockThinDiskOffering(String diskOfferingUuid, Long diskOfferingId) {
         DiskOfferingVO diskOffering = mock(DiskOfferingVO.class);
         Mockito.when(diskOffering.getId()).thenReturn(diskOfferingId);
         Mockito.when(diskOffering.getProvisioningType()).thenReturn(Storage.ProvisioningType.THIN);
-        Mockito.when(diskOfferingDao.findByUuid(Mockito.anyString())).thenReturn(diskOffering);
+        Mockito.when(diskOfferingDao.findByUuid(diskOfferingUuid)).thenReturn(diskOffering);
     }
 
     private void mockStorPoolPoolByUuid(String dsUuid, Long poolId) {
@@ -877,7 +877,7 @@ public void restoreBackedUpVolumeUsesQcow2FormatAndRealPathForStorPool()
         String realStorPoolPath = "/dev/storpool-byid/t.t.t";
 
         mockSourceVolume(volUuid, "data-sp");
-        mockThinDiskOffering(6L);
+        mockThinDiskOffering("disk-offering-uuid", 6L);
         mockStorPoolPoolByUuid(dsUuid, poolId);
         mockHostByIp(hostIp, 8L);
         mockNasRepository(backupOfferingId);
 
@@ -481,6 +481,10 @@ public KVMPhysicalDisk createPhysicalDisk(String name, KVMStoragePool pool, Phys
             throw new CloudRuntimeException("Encrypted StorPool volumes are not supported");
         }
 
+        if (!PhysicalDiskFormat.RAW.equals(format)) {
+            throw new CloudRuntimeException(String.format("StorPool volumes can only be created as RAW, %s was requested", format));
+        }
+
         if (size <= 0) {
             throw new CloudRuntimeException("Size must be greater than 0 to create a StorPool volume");
         }
@@ -507,9 +511,19 @@ public KVMPhysicalDisk createPhysicalDisk(String name, KVMStoragePool pool, Phys
 
         String volumePath = StorPoolUtil.devPath(globalId);
         try {
-            attachOrDetachVolume("attach", "volume", volumePath);
+            if (!attachOrDetachVolume("attach", "volume", volumePath)) {
+                // The attach succeeded but the device never showed up; StorPool refuses to delete an attached volume.
+                attachOrDetachVolume("detach", "volume", volumePath);
+                throw new CloudRuntimeException(String.format("StorPool volume %s for %s was attached but its device did not appear", volumePath, name));
+            }
         } catch (CloudRuntimeException e) {
-            volumeDelete("~" + globalId);
+            try {
+                volumeDelete("~" + globalId);
+            } catch (CloudRuntimeException deleteEx) {
+                SP_LOG("Failed to delete StorPool volume ~%s for %s after a failed attach: %s", globalId, name, deleteEx.getMessage());
+                LOGGER.warn(String.format("Failed to delete StorPool volume ~%s for %s after a failed attach, it must be deleted manually: %s",
+                        globalId, name, deleteEx.getMessage()));
+            }
             throw e;
         }
 
 
@@ -566,8 +566,6 @@ print(gid)
   echo "$clone_path"
 }
 
-# Deletes a volume previously returned by sp_create_backup_source_disk. Best-effort:
-# logs on failure rather than aborting the backup, matching the wrapper's old cleanup semantics.
 sp_delete_backup_source_disk() {
   local clone_name="$1"
   log -ne "StorPool: releasing backup source volume $clone_name"
@@ -582,7 +580,7 @@ sp_cleanup_backup_source_disks() {
   for clone_name in "${SP_CLEANUP_VOLUMES[@]:-}"; do
     [[ -z "$clone_name" ]] && continue
     log -ne "StorPool: cleaning up backup source volume $clone_name left over on script exit"
-    sp_delete_backup_source_disk "$clone_name"
+    sp_delete_backup_source_disk "$clone_name" || true
   done
   SP_CLEANUP_VOLUMES=()
 }
@@ -926,6 +924,11 @@ backup_stopped_vm() {
 
   local -a disk_arr=()
   IFS=',' read -ra disk_arr <<< "$DISK_PATHS"
+  if [[ ${#disk_arr[@]} -eq 0 ]]; then
+    echo "No disk paths given for stopped VM $VM"
+    cleanup
+    exit 1
+  fi
 
   name="root"
   for disk in "${disk_arr[@]}"; do
@@ -951,7 +954,12 @@ backup_stopped_vm() {
         cleanup
         exit 1
       fi
-      sp_clone_name=$(sp_volume_name_from_path "$read_disk")
+      if ! sp_clone_name=$(sp_volume_name_from_path "$read_disk"); then
+        log -ne "Failed to resolve the StorPool volume name of backup source volume $read_disk, it must be deleted manually"
+        echo "Failed to resolve the StorPool volume name of backup source volume $read_disk"
+        cleanup
+        exit 1
+      fi
       # Register the clone here, in the parent shell, right after the command substitution
       # above returns — sp_create_backup_source_disk cannot do this itself (see its comment).
       SP_CLEANUP_VOLUMES+=("$sp_clone_name")
 
@@ -28,10 +28,8 @@
 
 from marvin.cloudstackTestCase import cloudstackTestCase
 from marvin.codes import FAILED, KVM
-from marvin.lib.base import (Account,
-                             Backup,
+from marvin.lib.base import (Backup,
                              BackupOffering,
-                             BackupRepository,
                              DiskOffering,
                              SecurityGroup,
                              ServiceOffering,
@@ -89,7 +87,8 @@ def _setup_common(cls):
         StorPoolHelper.logger = cls
 
         zone_cfg = config.zones[0]
-        assert zone_cfg is not None
+        if zone_cfg is None:
+            raise Exception("No zone configured in the Marvin config")
         cls.zone_cfg = zone_cfg
 
         testClient = super(TestNasBackupStorPool, cls).getClsTestClient()
@@ -106,8 +105,10 @@ def _setup_common(cls):
 
         cls.services = testClient.getParsedTestDataConfig()
         cls.domain = get_domain(cls.apiclient)
-        cls.zone = list_zones(cls.apiclient, name=zone_cfg.name)[0]
-        assert cls.zone is not None
+        zones = list_zones(cls.apiclient, name=zone_cfg.name)
+        if not zones:
+            raise Exception("Zone %s not found" % zone_cfg.name)
+        cls.zone = zones[0]
         cls.services["mode"] = cls.zone.networktype
 
         cls.hypervisor = get_hypervisor_type(cls.apiclient)
@@ -127,7 +128,7 @@ def _setup_common(cls):
 
         template = get_template(cls.apiclient, cls.zone.id, account="system")
         if template == FAILED:
-            assert False, "get_template() failed to return a system template"
+            raise Exception("get_template() failed to return a system template")
         cls.template = template
 
         offerings = BackupOffering.listByZone(cls.apiclient, cls.zone.id) or []
@@ -202,7 +203,7 @@ def _setup_common(cls):
     @classmethod
     def _get_or_create_disk_offering(cls, tag, displaytext):
         disk_offerings = list_disk_offering(cls.apiclient, name=tag)
-        if disk_offerings is None:
+        if not disk_offerings:
             disk_offering = DiskOffering.create(
                 cls.apiclient,
                 {
@@ -237,7 +238,8 @@ def _create_test_vm(cls, name, diskofferingid):
     @classmethod
     def _verify_setup_volumes(cls, vm):
         volumes = Volume.list(cls.apiclient, virtualmachineid=vm.id, listall=True)
-        assert isinstance(volumes, list) and len(volumes) >= 1
+        if not isinstance(volumes, list) or not volumes:
+            raise Exception("No volumes found for VM %s" % vm.id)
         for vol in volumes:
             if cls.nfs_storage_pool and vol.storageid == cls.nfs_storage_pool.id:
                 continue
@@ -678,7 +680,7 @@ def _setup_nfs_environment(cls):
     @classmethod
     def _get_or_create_nfs_service_offering(cls):
         service_offerings = list_service_offering(cls.apiclient, name=cls.nfs_tag)
-        if service_offerings is None:
+        if not service_offerings:
             service_offering = ServiceOffering.create(
                 cls.apiclient,
                 {
 
@@ -21,7 +21,6 @@
 from marvin.lib.base import (Account,
                              Cluster,
                              Configurations,
-                             ImageStore,
                              ServiceOffering,
                              Snapshot,
                              StoragePool,
@@ -66,6 +65,7 @@
 import subprocess
 import json
 from storpool import spapi
+from storpool.spapi import ApiError
 from storpool import sptypes
 
 class TestData():
@@ -696,7 +696,7 @@ def get_storpool_volume_by_path(cls, spapi, path):
         name = "~" + parts[3]
         try:
             volumes = spapi.volumeList(volumeName=name)
-        except Exception as err:
+        except ApiError as err:
             raise Exception("StorPool volume not found for path %s: %s" % (path, err))
         if not volumes:
             raise Exception("StorPool volume not found for path %s" % path)
@@ -706,20 +706,24 @@ def get_storpool_volume_by_path(cls, spapi, path):
     def check_storpool_volume_size(cls, volume, spapi):
         spvolume = cls.get_storpool_volume_by_path(spapi, volume.path)
         if spvolume.size != volume.size:
-            raise Exception("Storpool volume size is not the same as CloudStack db size")
+            raise Exception("StorPool volume size is not the same as CloudStack db size")
 
     @classmethod
     def check_storpool_volume_iops(cls, spapi, volume,):
-        spvolume = cls.get_storpool_volume_by_path(spapi, volume.path)
-        logging.debug(spvolume.iops)
-        logging.debug(volume.maxiops)
-        if spvolume.iops != volume.maxiops:
-            raise Exception("Storpool volume size is not the same as CloudStack db size")
+        name = volume.path.split("/")[3]
+        try:
+            spvolume = spapi.volumeList(volumeName = "~" + name)
+            logging.debug(spvolume[0].iops)
+            logging.debug(volume.maxiops)
+            if spvolume[0].iops != volume.maxiops:
+                raise Exception("Storpool volume size is not the same as CloudStack db size")
+        except spapi.ApiError as err:
+           raise Exception(err)
 
     @classmethod
     def verify_storpool_volume(cls, spapi, volume, check_size=True):
         """Assert the CloudStack volume exists on StorPool; optionally compare size."""
-        if not volume.path or not volume.path.startswith("/dev/storpool"):
+        if not volume.path or not volume.path.startswith("/dev/storpool-byid/"):
             raise Exception("Volume %s does not have a StorPool device path: %s" %
                             (volume.id, volume.path))
         spvolume = cls.get_storpool_volume_by_path(spapi, volume.path)