Copilot commented on code in PR #11931:
URL: https://github.com/apache/cloudstack/pull/11931#discussion_r4218641817
##########
test/integration/smoke/test_backup_recovery_nas.py:
##########
@@ -72,14 +113,22 @@ def setUpClass(cls):
cls._cleanup = [cls.account]
- # Create NAS backup repository and offering. Use the same directory as
the storage pool
+ # Create NAS backup repository and offering.
cls.backup_repository = BackupRepository.add(cls.api_client,
zoneid=cls.zone.id, name="Nas",
-
address=cls.storage_pool.ipaddress + ":" + cls.storage_pool.path,
+
address=cls.nas_repository_address,
provider="nas",
type="nfs",)
cls._cleanup.append(cls.backup_repository)
+ # Match the external offering to the repository just created above by
externalid
+ # (== the repository's own id for the nas provider) rather than
blindly taking
+ # index 0 -- a stray repository left over from an earlier interrupted
run (e.g.
+ # one whose backups didn't get cleaned up, so its own teardown
couldn't remove
+ # it either) sorts alongside the new one, and index 0 has no guarantee
of being
+ # the one this run owns.
cls.provider_offerings = BackupOffering.listExternal(cls.api_client,
cls.zone.id)
- cls.backup_offering = BackupOffering.importExisting(cls.api_client,
cls.zone.id, cls.provider_offerings[0].externalid,
-
cls.provider_offerings[0].name, cls.provider_offerings[0].description)
+ matching_offerings = [o for o in cls.provider_offerings if
o.externalid == cls.backup_repository.id]
+ provider_offering = matching_offerings[0] if matching_offerings else
cls.provider_offerings[0]
+ cls.backup_offering = BackupOffering.importExisting(cls.api_client,
cls.zone.id, provider_offering.externalid,
+
provider_offering.name, provider_offering.description)
Review Comment:
If `BackupOffering.listExternal(...)` returns `None` or an empty list,
indexing `[0]` will raise. Add an explicit assertion/skip with a clear reason
when no external offerings are available (e.g., provider not properly
configured) before attempting to select a matching offering.
##########
test/integration/smoke/test_list_storage_pools.py:
##########
@@ -62,26 +62,31 @@ def test_01_list_storage_pools_clusterid_filter(self):
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
- self.assertGreater(
- len(storage_pools),
- 0,
- "Length of storage pools should greater than 0"
- )
+ # Filter may return zone-wide pools (with clusterid=None) alongside
cluster-scoped pools
+ cluster_scoped_found = False
for storage_pool in storage_pools:
- self.assertEqual(
- storage_pool.clusterid,
- self.cluster.id,
- "Cluster id should be equal to the cluster id passed in the
filter"
- )
+ if storage_pool.clusterid is not None:
+ # Only assert clusterid for cluster-scoped pools
+ cluster_scoped_found = True
+ self.assertEqual(
+ storage_pool.clusterid,
+ self.cluster.id,
+ "Cluster id should be equal to the cluster id passed in
the filter"
+ )
Review Comment:
`cluster_scoped_found` is set but never asserted, so the test can pass
vacuously even if the filter returns only zone-wide pools (or returns unrelated
pools with `clusterid=None`). Add an assertion that at least one cluster-scoped
pool was returned (or explicitly `skipTest` when the environment has no
cluster-scoped pools) so the `clusterid` filter behavior is actually validated.
##########
test/integration/smoke/test_list_storage_pools.py:
##########
@@ -264,26 +269,31 @@ def test_07_list_storage_pools_podid_filter(self):
isinstance(storage_pools, list),
"Storage pool response type should be a list"
)
- self.assertGreater(
- len(storage_pools),
- 0,
- "Length of storage pools should greater than 0"
- )
+ # Filter may return zone-wide pools (with podid=None) alongside
pod-scoped pools
+ pod_scoped_found = False
for storage_pool in storage_pools:
- self.assertEqual(
- storage_pool.podid,
- self.cluster.podid,
- "Pod id should be equal to the pod id passed in the filter"
- )
+ if storage_pool.podid is not None:
+ # Only assert podid for pod-scoped pools
+ pod_scoped_found = True
+ self.assertEqual(
+ storage_pool.podid,
+ self.cluster.podid,
+ "Pod id should be equal to the pod id passed in the filter"
+ )
Review Comment:
Same issue as the cluster filter test: `pod_scoped_found` is never asserted,
so the test may pass without validating filtering. Assert `pod_scoped_found`
(or skip when the environment lacks pod-scoped pools) to ensure the `podid`
filter is meaningfully tested.
##########
test/integration/smoke/test_backup_recovery_nas.py:
##########
@@ -594,3 +662,79 @@ def test_stopped_vm_falls_back_to_full(self):
finally:
self._set_full_every(original_full_every)
self.backup_offering.removeOffering(self.apiclient, self.vm.id)
+
+ # ------------------------------------------------------------------
+ # Restore-volume-and-attach regression (PR apache/cloudstack#14007)
+ # ------------------------------------------------------------------
+ # This test exercises the fixed path end to end via
restoreVolumeFromBackupAndAttachToVM,
+ # on whichever primary storage this environment is running with (NFS or
Ceph/RBD).
+
+ @attr(tags=["advanced", "backup"], required_hardware="true")
+ def test_restore_volume_and_attach_to_vm(self):
+ """
+ Test restoring the ROOT and DATADISK volumes of a backup and attaching
them
+ to a different Instance (restoreVolumeFromBackupAndAttachToVM).
+ """
+ target_vm = None
+ self.backup_offering.assignOffering(self.apiclient, self.vm.id)
+ try:
+ ssh_client_vm = self.vm.get_ssh_client(reconnect=True)
+ ssh_client_vm.execute("echo restore-attach-marker >
/root/restore_attach_marker.txt; sync")
+
+ Backup.create(self.apiclient, self.vm.id, "restore_attach_backup")
+
+ backups = Backup.list(self.apiclient, self.vm.id)
+ self.assertEqual(len(backups), 1, "There should exist only one
backup for the VM")
+ backup = backups[0]
Review Comment:
Two reliability issues: (1) `assertEqual(len(backups), 1)` will fail if the
environment already has backups for the VM (e.g., from previous incomplete
runs). Prefer selecting the backup by name/ID returned from `Backup.create` (or
filtering by `name == \"restore_attach_backup\"`). (2) `Backup.delete` is not
in the `finally` block, so a failure before line 733 leaves the backup behind
and can pollute subsequent runs. Move deletion into `finally` with a guard that
`backup` was created.
##########
test/integration/smoke/test_list_storage_pools.py:
##########
@@ -394,3 +404,64 @@ def test_11_list_storage_pools_no_filter(self):
0,
"Length of storage pools should greater than 0"
)
+
+ @attr(tags=["advanced", "advancedns", "smoke", "basic"],
required_hardware="false")
+ def test_12_list_storage_pools_zone_wide(self):
+ """ Test listing zone-wide vs cluster-scoped storage pools
+ Zone-wide pools (scope=ZONE) are not bound to a specific cluster.
+ Cluster-scoped pools are bound to clusters.
+ """
+ # List all storage pools and categorize them
+ storage_pools = StoragePool.list(self.apiclient)
+ self.assertTrue(
+ isinstance(storage_pools, list),
+ "Storage pool response type should be a list"
+ )
+
+ zone_wide_pools = [p for p in storage_pools if hasattr(p, 'scope') and
p.scope == "ZONE"]
+ cluster_scoped_pools = [p for p in storage_pools if hasattr(p,
'scope') and p.scope != "ZONE"]
+
Review Comment:
Pools without a `scope` attribute are silently ignored and won’t be
validated by either branch, which can make the test pass while missing (or
regressing) an expected API field. Consider asserting that all returned pools
expose `scope` (if it’s required), or include pools without `scope` in one
category and fail with a clear message so regressions don’t get masked.
##########
test/integration/smoke/test_snapshots.py:
##########
@@ -262,7 +262,7 @@ def test_02_list_snapshots_with_removed_data_store(self):
volume_pool_response = list_storage_pools(self.apiclient,
id=vol_res[0].storageid)
volume_pool = volume_pool_response[0]
- if volume_pool.type.lower() != 'networkfilesystem':
+ if volume_pool.type not in ("NetworkFilesystem", "RBD"):
Review Comment:
Both checks are now case-sensitive, but earlier code used
`volume_pool.type.lower()`, suggesting API responses may vary in casing. This
can cause incorrect skips/failures if the type comes back as e.g.
`networkfilesystem`/`rbd`. Normalize case (e.g., compare `type.lower()` against
lowercased allow-list) consistently for both `volume_pool.type` and
`storage.type`.
##########
test/integration/smoke/test_vm_snapshots.py:
##########
@@ -47,6 +47,11 @@ def setUpClass(cls):
cls.unsupportedHypervisor = True
return
+ list_volume_pool_response = list_storage_pools(cls.apiclient)
+ volume_pool = list_volume_pool_response[0]
+ if volume_pool.type == "RBD":
+ cls.skipTest(cls, reason="VM snapshot is unsupported for VMs on
RBD storage pool")
Review Comment:
Same issue as above: using the first storage pool to decide whether to skip
can incorrectly skip (or not skip) in environments with multiple pools. Use the
storage pool actually backing the VM under test (derive from the deployed VM’s
root volume `storageid`), or pick an `Up` pool and ensure it represents the
pool the test will consume. Add a defensive check for an empty result to avoid
`IndexError`.
##########
test/integration/smoke/test_backup_recovery_nas.py:
##########
@@ -594,3 +662,79 @@ def test_stopped_vm_falls_back_to_full(self):
finally:
self._set_full_every(original_full_every)
self.backup_offering.removeOffering(self.apiclient, self.vm.id)
+
+ # ------------------------------------------------------------------
+ # Restore-volume-and-attach regression (PR apache/cloudstack#14007)
+ # ------------------------------------------------------------------
+ # This test exercises the fixed path end to end via
restoreVolumeFromBackupAndAttachToVM,
+ # on whichever primary storage this environment is running with (NFS or
Ceph/RBD).
+
+ @attr(tags=["advanced", "backup"], required_hardware="true")
+ def test_restore_volume_and_attach_to_vm(self):
+ """
+ Test restoring the ROOT and DATADISK volumes of a backup and attaching
them
+ to a different Instance (restoreVolumeFromBackupAndAttachToVM).
+ """
+ target_vm = None
+ self.backup_offering.assignOffering(self.apiclient, self.vm.id)
+ try:
+ ssh_client_vm = self.vm.get_ssh_client(reconnect=True)
+ ssh_client_vm.execute("echo restore-attach-marker >
/root/restore_attach_marker.txt; sync")
+
+ Backup.create(self.apiclient, self.vm.id, "restore_attach_backup")
+
+ backups = Backup.list(self.apiclient, self.vm.id)
+ self.assertEqual(len(backups), 1, "There should exist only one
backup for the VM")
+ backup = backups[0]
+
+ volumes = Volume.list(self.apiclient, virtualmachineid=self.vm.id,
listall=True)
+ self.assertTrue(isinstance(volumes, list), "List volumes should
return a valid list")
+ root_disk_id = None
+ data_disk_id = None
+ for volume in volumes:
+ if volume.type == 'ROOT':
+ root_disk_id = volume.id
+ elif volume.type == 'DATADISK':
+ data_disk_id = volume.id
+ self.assertIsNotNone(root_disk_id, "The backed up VM should have a
ROOT volume")
+
+ # Target Instance that will receive the restored volumes. The nas
provider
+ # (unlike KBOSS) requires the target Instance to be stopped before
a volume
+ # can be restored and attached to it.
+ target_vm = VirtualMachine.create(
+ self.apiclient, self.services["small"],
accountid=self.account.name,
+ domainid=self.account.domainid,
serviceofferingid=self.offering.id,
+ mode=self.services["mode"]
+ )
+ target_vm.stop(self.apiclient, forced=True)
+
+ # Restore and attach the ROOT volume backup as an extra disk.
+ Backup.restoreVolumeFromBackupAndAttachToVM(
+ self.apiclient, backupid=backup.id, volumeid=root_disk_id,
virtualmachineid=target_vm.id
+ )
+ target_volumes = Volume.list(self.apiclient,
virtualmachineid=target_vm.id, listall=True)
+ self.assertTrue(isinstance(target_volumes, list), "List volumes
should return a valid list")
+ self.assertEqual(2, len(target_volumes),
+ "Target Instance should have its own ROOT volume plus the
restored volume")
+
+ if data_disk_id:
+ # Restore and attach the DATADISK volume backup as well.
+ Backup.restoreVolumeFromBackupAndAttachToVM(
+ self.apiclient, backupid=backup.id, volumeid=data_disk_id,
virtualmachineid=target_vm.id
+ )
+ target_volumes = Volume.list(self.apiclient,
virtualmachineid=target_vm.id, listall=True)
+ self.assertEqual(3, len(target_volumes),
+ "Target Instance should have 3 volumes after restoring
both the ROOT and DATADISK backups")
+
+ # Start the target Instance to verify the restored disk(s) are
actually
+ # usable and libvirt accepted the attach-device/attach-disk calls.
+ target_vm.start(self.apiclient)
+
+ Backup.delete(self.apiclient, backup.id)
+ finally:
+ if target_vm is not None:
+ try:
+ target_vm.delete(self.apiclient)
+ except Exception:
+ pass
+ self.backup_offering.removeOffering(self.apiclient, self.vm.id)
Review Comment:
Two reliability issues: (1) `assertEqual(len(backups), 1)` will fail if the
environment already has backups for the VM (e.g., from previous incomplete
runs). Prefer selecting the backup by name/ID returned from `Backup.create` (or
filtering by `name == \"restore_attach_backup\"`). (2) `Backup.delete` is not
in the `finally` block, so a failure before line 733 leaves the backup behind
and can pollute subsequent runs. Move deletion into `finally` with a guard that
`backup` was created.
##########
test/integration/smoke/test_snapshots.py:
##########
@@ -291,9 +291,8 @@ def test_02_list_snapshots_with_removed_data_store(self):
'Up',
"Check primary storage state"
)
- self.assertEqual(
- storage.type,
- 'NetworkFilesystem',
+ self.assertTrue(
+ storage.type in ("NetworkFilesystem", "RBD"),
"Check storage pool type"
Review Comment:
Both checks are now case-sensitive, but earlier code used
`volume_pool.type.lower()`, suggesting API responses may vary in casing. This
can cause incorrect skips/failures if the type comes back as e.g.
`networkfilesystem`/`rbd`. Normalize case (e.g., compare `type.lower()` against
lowercased allow-list) consistently for both `volume_pool.type` and
`storage.type`.
##########
test/integration/smoke/test_volumes.py:
##########
@@ -1111,6 +1117,11 @@ def setUpClass(cls):
cls.unsupportedHypervisor = True
return
+ list_volume_pool_response = list_storage_pools(cls.apiclient)
+ volume_pool = list_volume_pool_response[0]
+ if volume_pool.type == "RBD":
+ cls.skipTest(cls, reason="Volume encryption is unsupported for
volumes on RBD storage pool")
Review Comment:
This skips the entire class based on `list_storage_pools()[0]`, which can be
the wrong pool in mixed setups (or even a disabled/non-default pool), causing
false skips. Prefer determining the pool used by the test VM/volume (e.g.,
create the volume/VM first and check its `storageid`), or at minimum select an
`Up` pool (and only skip when the relevant pool type is RBD). Also guard
against an empty `list_storage_pools` response to avoid `IndexError`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
