mirror of
https://github.com/apache/cloudstack.git
synced 2025-10-26 08:42:29 +01:00
973b0e28fd
2023-06-27 08:42:08 [INFO] File:[/github/workspace/test/integration/plugins/storpool/TestEncryptedVolumes.py] 2023-06-27 08:42:08 [ERROR] Found errors in [flake8] linter! 2023-06-27 08:42:08 [ERROR] Error code: 1. Command output: ------ /github/workspace/test/integration/plugins/storpool/TestEncryptedVolumes.py:681:113: W292 no newline at end of file ------ 2023-06-27 08:42:08 [INFO] --------------------------- 2023-06-27 08:42:08 [INFO] File:[/github/workspace/test/integration/plugins/storpool/sp_util.py] 2023-06-27 08:42:08 [ERROR] Found errors in [flake8] linter! 2023-06-27 08:42:08 [ERROR] Error code: 1. Command output: ------ /github/workspace/test/integration/plugins/storpool/sp_util.py:798:31: W292 no newline at end of file ------
682 lines
24 KiB
Python
682 lines
24 KiB
Python
# Licensed to the Apache Software Foundation (ASF) under one
|
|
# or more contributor license agreements. See the NOTICE file
|
|
# distributed with this work for additional information
|
|
# regarding copyright ownership. The ASF licenses this file
|
|
# to you under the Apache License, Version 2.0 (the
|
|
# "License"); you may not use this file except in compliance
|
|
# with the License. You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing,
|
|
# software distributed under the License is distributed on an
|
|
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
# KIND, either express or implied. See the License for the
|
|
# specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
# Import Local Modules
|
|
from marvin.codes import FAILED, KVM, PASS, XEN_SERVER, RUNNING
|
|
from nose.plugins.attrib import attr
|
|
from marvin.cloudstackTestCase import cloudstackTestCase
|
|
from marvin.lib.utils import random_gen, cleanup_resources, validateList, is_snapshot_on_nfs, isAlmostEqual
|
|
from marvin.lib.base import (Account,
|
|
Cluster,
|
|
Configurations,
|
|
ServiceOffering,
|
|
Snapshot,
|
|
StoragePool,
|
|
Template,
|
|
VirtualMachine,
|
|
VmSnapshot,
|
|
Volume,
|
|
SecurityGroup,
|
|
Role,
|
|
DiskOffering,
|
|
)
|
|
from marvin.lib.common import (get_zone,
|
|
get_domain,
|
|
get_template,
|
|
list_disk_offering,
|
|
list_hosts,
|
|
list_snapshots,
|
|
list_storage_pools,
|
|
list_volumes,
|
|
list_virtual_machines,
|
|
list_configurations,
|
|
list_service_offering,
|
|
list_clusters,
|
|
list_zones)
|
|
from marvin.cloudstackAPI import (listOsTypes,
|
|
listTemplates,
|
|
listHosts,
|
|
createTemplate,
|
|
createVolume,
|
|
getVolumeSnapshotDetails,
|
|
resizeVolume,
|
|
listZones,
|
|
migrateVirtualMachine,
|
|
findHostsForMigration,
|
|
revertSnapshot,
|
|
deleteSnapshot)
|
|
from marvin.sshClient import SshClient
|
|
|
|
import time
|
|
import pprint
|
|
import random
|
|
import subprocess
|
|
from storpool import spapi
|
|
from storpool import sptypes
|
|
import unittest
|
|
|
|
import uuid
|
|
from sp_util import (TestData, StorPoolHelper)
|
|
|
|
class TestEncryptedVolumes(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(TestEncryptedVolumes, cls).setUpClass()
|
|
try:
|
|
cls.setUpCloudStack()
|
|
except Exception:
|
|
cls.cleanUpCloudStack()
|
|
raise
|
|
|
|
@classmethod
|
|
def setUpCloudStack(cls):
|
|
testClient = super(TestEncryptedVolumes, cls).getClsTestClient()
|
|
|
|
cls._cleanup = []
|
|
|
|
config = cls.getClsConfig()
|
|
StorPoolHelper.logger = cls
|
|
cls.logger = StorPoolHelper.logger
|
|
|
|
cls.apiclient = testClient.getApiClient()
|
|
|
|
zone = config.zones[0]
|
|
assert zone is not None
|
|
cls.zone = list_zones(cls.apiclient, name=zone.name)[0]
|
|
|
|
cls.hostConfig = cls.config.__dict__["zones"][0].__dict__["pods"][0].__dict__["clusters"][0].__dict__["hosts"][0].__dict__
|
|
|
|
cls.spapi = spapi.Api(host=zone.spEndpoint, port=zone.spEndpointPort, auth=zone.spAuthToken, multiCluster=True)
|
|
cls.helper = StorPoolHelper()
|
|
|
|
cls.unsupportedHypervisor = False
|
|
cls.hypervisor = testClient.getHypervisorInfo()
|
|
if cls.hypervisor.lower() in ("hyperv", "lxc"):
|
|
cls.unsupportedHypervisor = True
|
|
return
|
|
|
|
cls.services = testClient.getParsedTestDataConfig()
|
|
|
|
# Get Zone, Domain and templates
|
|
cls.domain = get_domain(cls.apiclient)
|
|
|
|
td = TestData()
|
|
cls.testdata = td.testdata
|
|
|
|
cls.sp_template_1 = "ssd"
|
|
storpool_primary_storage = {
|
|
"name": cls.sp_template_1,
|
|
"zoneid": cls.zone.id,
|
|
"url": "SP_API_HTTP=%s:%s;SP_AUTH_TOKEN=%s;SP_TEMPLATE=%s" % (zone.spEndpoint, zone.spEndpointPort, zone.spAuthToken, cls.sp_template_1),
|
|
"scope": "zone",
|
|
"capacitybytes": 564325555333,
|
|
"capacityiops": 155466,
|
|
"hypervisor": "kvm",
|
|
"provider": "StorPool",
|
|
"tags": cls.sp_template_1
|
|
}
|
|
|
|
cls.storpool_primary_storage = storpool_primary_storage
|
|
|
|
storage_pool = list_storage_pools(
|
|
cls.apiclient,
|
|
name=storpool_primary_storage["name"]
|
|
)
|
|
|
|
if storage_pool is None:
|
|
newTemplate = sptypes.VolumeTemplateCreateDesc(name=storpool_primary_storage["name"], placeAll="virtual",
|
|
placeTail="virtual", placeHead="virtual", replication=1)
|
|
template_on_local = cls.spapi.volumeTemplateCreate(newTemplate)
|
|
|
|
storage_pool = StoragePool.create(cls.apiclient, storpool_primary_storage)
|
|
else:
|
|
storage_pool = storage_pool[0]
|
|
cls.primary_storage = storage_pool
|
|
|
|
storpool_service_offerings_ssd = {
|
|
"name": "ssd-encrypted",
|
|
"displaytext": "SP_CO_2 (Min IOPS = 10,000; Max IOPS = 15,000)",
|
|
"cpunumber": 1,
|
|
"cpuspeed": 500,
|
|
"memory": 512,
|
|
"storagetype": "shared",
|
|
"customizediops": False,
|
|
"hypervisorsnapshotreserve": 200,
|
|
"encryptroot": True,
|
|
"tags": cls.sp_template_1
|
|
}
|
|
|
|
service_offerings_ssd = list_service_offering(
|
|
cls.apiclient,
|
|
name=storpool_service_offerings_ssd["name"]
|
|
)
|
|
|
|
if service_offerings_ssd is None:
|
|
service_offerings_ssd = ServiceOffering.create(cls.apiclient, storpool_service_offerings_ssd, encryptroot=True)
|
|
else:
|
|
service_offerings_ssd = service_offerings_ssd[0]
|
|
|
|
cls.service_offering = service_offerings_ssd
|
|
cls.debug(pprint.pformat(cls.service_offering))
|
|
|
|
cls.sp_template_2 = "ssd2"
|
|
|
|
storpool_primary_storage2 = {
|
|
"name": cls.sp_template_2,
|
|
"zoneid": cls.zone.id,
|
|
"url": "SP_API_HTTP=%s:%s;SP_AUTH_TOKEN=%s;SP_TEMPLATE=%s" % (zone.spEndpoint, zone.spEndpointPort, zone.spAuthToken, cls.sp_template_2),
|
|
"scope": "zone",
|
|
"capacitybytes": 564325555333,
|
|
"capacityiops": 1554,
|
|
"hypervisor": "kvm",
|
|
"provider": "StorPool",
|
|
"tags": cls.sp_template_2
|
|
}
|
|
|
|
cls.storpool_primary_storage2 = storpool_primary_storage2
|
|
storage_pool = list_storage_pools(
|
|
cls.apiclient,
|
|
name=storpool_primary_storage2["name"]
|
|
)
|
|
|
|
if storage_pool is None:
|
|
newTemplate = sptypes.VolumeTemplateCreateDesc(name=storpool_primary_storage2["name"], placeAll="virtual",
|
|
placeTail="virtual", placeHead="virtual", replication=1)
|
|
template_on_local = cls.spapi.volumeTemplateCreate(newTemplate)
|
|
storage_pool = StoragePool.create(cls.apiclient, storpool_primary_storage2)
|
|
|
|
else:
|
|
storage_pool = storage_pool[0]
|
|
cls.primary_storage2 = storage_pool
|
|
|
|
storpool_service_offerings_ssd2 = {
|
|
"name": "ssd2-encrypted",
|
|
"displaytext": "SP_CO_2",
|
|
"cpunumber": 1,
|
|
"cpuspeed": 500,
|
|
"memory": 512,
|
|
"storagetype": "shared",
|
|
"customizediops": False,
|
|
"encryptroot": True,
|
|
"tags": cls.sp_template_2
|
|
}
|
|
|
|
service_offerings_ssd2 = list_service_offering(
|
|
cls.apiclient,
|
|
name=storpool_service_offerings_ssd2["name"]
|
|
)
|
|
|
|
if service_offerings_ssd2 is None:
|
|
service_offerings_ssd2 = ServiceOffering.create(cls.apiclient, storpool_service_offerings_ssd2, encryptroot=True)
|
|
else:
|
|
service_offerings_ssd2 = service_offerings_ssd2[0]
|
|
|
|
cls.service_offering2 = service_offerings_ssd2
|
|
|
|
cls.disk_offerings_ssd2_encrypted = list_disk_offering(
|
|
cls.apiclient,
|
|
name=cls.testdata[TestData.diskOfferingEncrypted2]["name"]
|
|
)
|
|
if cls.disk_offerings_ssd2_encrypted is None:
|
|
cls.disk_offerings_ssd2_encrypted = DiskOffering.create(cls.apiclient, cls.testdata[TestData.diskOfferingEncrypted2], encrypt=True)
|
|
else:
|
|
cls.disk_offerings_ssd2_encrypted = cls.disk_offerings_ssd2_encrypted[0]
|
|
|
|
cls.disk_offering_ssd_encrypted = list_disk_offering(
|
|
cls.apiclient,
|
|
name=cls.testdata[TestData.diskOfferingEncrypted]["name"]
|
|
)
|
|
|
|
if cls.disk_offering_ssd_encrypted is None:
|
|
cls.disk_offering_ssd_encrypted = DiskOffering.create(cls.apiclient, cls.testdata[TestData.diskOfferingEncrypted], encrypt=True)
|
|
else:
|
|
cls.disk_offering_ssd_encrypted = cls.disk_offering_ssd_encrypted[0]
|
|
|
|
template = get_template(
|
|
cls.apiclient,
|
|
cls.zone.id,
|
|
account="system"
|
|
)
|
|
|
|
if template == FAILED:
|
|
assert False, "get_template() failed to return template\
|
|
with description %s" % cls.services["ostype"]
|
|
|
|
cls.services["domainid"] = cls.domain.id
|
|
cls.services["small"]["zoneid"] = cls.zone.id
|
|
cls.services["templates"]["ostypeid"] = template.ostypeid
|
|
cls.services["zoneid"] = cls.zone.id
|
|
|
|
role = Role.list(cls.apiclient, name='Root Admin')
|
|
|
|
cls.account = Account.create(
|
|
cls.apiclient,
|
|
cls.services["account"],
|
|
domainid=cls.domain.id,
|
|
roleid= 1
|
|
)
|
|
|
|
securitygroup = SecurityGroup.list(cls.apiclient, account=cls.account.name, domainid=cls.account.domainid)[0]
|
|
cls.helper.set_securityGroups(cls.apiclient, account=cls.account.name, domainid=cls.account.domainid,
|
|
id=securitygroup.id)
|
|
cls._cleanup.append(cls.account)
|
|
|
|
cls.volume_1 = Volume.create(
|
|
cls.apiclient,
|
|
{"diskname": "StorPoolEncryptedDiskLiveMigrate"},
|
|
zoneid=cls.zone.id,
|
|
diskofferingid=cls.disk_offering_ssd_encrypted.id,
|
|
account=cls.account.name,
|
|
domainid=cls.account.domainid,
|
|
)
|
|
|
|
cls.volume_2 = Volume.create(
|
|
cls.apiclient,
|
|
{"diskname": "StorPoolEncryptedDiskVMSnapshot"},
|
|
zoneid=cls.zone.id,
|
|
diskofferingid=cls.disk_offering_ssd_encrypted.id,
|
|
account=cls.account.name,
|
|
domainid=cls.account.domainid,
|
|
)
|
|
|
|
cls.virtual_machine = VirtualMachine.create(
|
|
cls.apiclient,
|
|
{"name": "StorPool-LiveMigrate-VM%s" % uuid.uuid4()},
|
|
accountid=cls.account.name,
|
|
domainid=cls.account.domainid,
|
|
zoneid=cls.zone.id,
|
|
templateid=template.id,
|
|
serviceofferingid=cls.service_offering.id,
|
|
hypervisor=cls.hypervisor,
|
|
rootdisksize=10
|
|
)
|
|
|
|
cls.virtual_machine2 = VirtualMachine.create(
|
|
cls.apiclient,
|
|
{"name": "StorPool-VMSnapshots%s" % uuid.uuid4()},
|
|
accountid=cls.account.name,
|
|
domainid=cls.account.domainid,
|
|
zoneid=cls.zone.id,
|
|
templateid=template.id,
|
|
serviceofferingid=cls.service_offering.id,
|
|
hypervisor=cls.hypervisor,
|
|
rootdisksize=10
|
|
)
|
|
|
|
cls.virtual_machine3 = VirtualMachine.create(
|
|
cls.apiclient,
|
|
{"name": "StorPool-VolumeSnapshots%s" % uuid.uuid4()},
|
|
accountid=cls.account.name,
|
|
domainid=cls.account.domainid,
|
|
zoneid=cls.zone.id,
|
|
templateid=template.id,
|
|
serviceofferingid=cls.service_offering.id,
|
|
hypervisor=cls.hypervisor,
|
|
rootdisksize=10
|
|
)
|
|
|
|
cls.template = template
|
|
cls.hostid = cls.virtual_machine.hostid
|
|
cls.random_data_0 = random_gen(size=100)
|
|
cls.test_dir = "/tmp"
|
|
cls.random_data = "random.data"
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.cleanUpCloudStack()
|
|
|
|
@classmethod
|
|
def cleanUpCloudStack(cls):
|
|
try:
|
|
cleanup_resources(cls.apiclient, cls._cleanup)
|
|
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
|
|
if self.unsupportedHypervisor:
|
|
self.skipTest("Skipping test because unsupported hypervisor\
|
|
%s" % self.hypervisor)
|
|
return
|
|
|
|
def tearDown(self):
|
|
return
|
|
|
|
|
|
# live migrate VM with encrypted volumes to another host
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
|
|
def test_01_live_migrate_vm(self):
|
|
'''
|
|
Live Migrate VM to another host with encrypted volumes
|
|
'''
|
|
self.virtual_machine.attach_volume(
|
|
self.apiclient,
|
|
self.volume_1
|
|
)
|
|
|
|
volumes = list_volumes(
|
|
self.apiclient,
|
|
virtualmachineid = self.virtual_machine.id,
|
|
)
|
|
|
|
vm_host = list_hosts(self.apiclient, id=self.virtual_machine.hostid)[0]
|
|
self.logger.debug(vm_host)
|
|
# sshc = SshClient(
|
|
# host=vm_host.name,
|
|
# port=22,
|
|
# user=None,
|
|
# passwd=None)
|
|
#
|
|
# for volume in volumes:
|
|
# cmd = 'blkid %s' % volume.path
|
|
# result = sshc.execute(cmd)
|
|
# if "LUKS" not in result:
|
|
# self.fail("The volume isn't encrypted %s" % volume)
|
|
|
|
|
|
dest_host_cmd = findHostsForMigration.findHostsForMigrationCmd()
|
|
dest_host_cmd.virtualmachineid = self.virtual_machine.id
|
|
host = self.apiclient.findHostsForMigration(dest_host_cmd)[0]
|
|
|
|
cmd = migrateVirtualMachine.migrateVirtualMachineCmd()
|
|
cmd.virtualmachineid = self.virtual_machine.id
|
|
cmd.hostid = host.id
|
|
self.apiclient.migrateVirtualMachine(cmd)
|
|
|
|
# VM snapshot
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
|
|
def test_02_vm_snapshot(self):
|
|
self.virtual_machine2.attach_volume(
|
|
self.apiclient,
|
|
self.volume_2
|
|
)
|
|
|
|
try:
|
|
ssh_client = self.virtual_machine2.get_ssh_client(reconnect=True)
|
|
|
|
cmds = [
|
|
"echo %s > %s/%s" %
|
|
(self.random_data_0, self.test_dir, self.random_data),
|
|
"sync",
|
|
"sleep 1",
|
|
"sync",
|
|
"sleep 1",
|
|
"cat %s/%s" %
|
|
(self.test_dir, self.random_data)
|
|
]
|
|
|
|
for c in cmds:
|
|
self.debug(c)
|
|
result = ssh_client.execute(c)
|
|
self.debug(result)
|
|
except Exception:
|
|
self.fail("SSH failed for Virtual machine: %s" %
|
|
self.virtual_machine2.ipaddress)
|
|
self.assertEqual(
|
|
self.random_data_0,
|
|
result[0],
|
|
"Check the random data has be write into temp file!"
|
|
)
|
|
|
|
time.sleep(30)
|
|
MemorySnapshot = False
|
|
vm_snapshot = VmSnapshot.create(
|
|
self.apiclient,
|
|
self.virtual_machine2.id,
|
|
MemorySnapshot,
|
|
"TestSnapshot",
|
|
"Display Text"
|
|
)
|
|
self.assertEqual(
|
|
vm_snapshot.state,
|
|
"Ready",
|
|
"Check the snapshot of vm is ready!"
|
|
)
|
|
|
|
# Revert VM snapshot
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
|
|
def test_03_revert_vm_snapshots(self):
|
|
"""Test to revert VM snapshots
|
|
"""
|
|
|
|
try:
|
|
ssh_client = self.virtual_machine2.get_ssh_client(reconnect=True)
|
|
|
|
cmds = [
|
|
"rm -rf %s/%s" % (self.test_dir, self.random_data),
|
|
"ls %s/%s" % (self.test_dir, self.random_data)
|
|
]
|
|
|
|
for c in cmds:
|
|
self.debug(c)
|
|
result = ssh_client.execute(c)
|
|
self.debug(result)
|
|
|
|
except Exception:
|
|
self.fail("SSH failed for Virtual machine: %s" %
|
|
self.virtual_machine2.ipaddress)
|
|
|
|
if str(result[0]).index("No such file or directory") == -1:
|
|
self.fail("Check the random data has be delete from temp file!")
|
|
|
|
time.sleep(30)
|
|
|
|
list_snapshot_response = VmSnapshot.list(
|
|
self.apiclient,
|
|
virtualmachineid=self.virtual_machine2.id,
|
|
listall=True)
|
|
|
|
self.assertEqual(
|
|
isinstance(list_snapshot_response, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
list_snapshot_response,
|
|
None,
|
|
"Check if snapshot exists in ListSnapshot"
|
|
)
|
|
|
|
self.assertEqual(
|
|
list_snapshot_response[0].state,
|
|
"Ready",
|
|
"Check the snapshot of vm is ready!"
|
|
)
|
|
|
|
self.virtual_machine2.stop(self.apiclient, forced=True)
|
|
|
|
VmSnapshot.revertToSnapshot(
|
|
self.apiclient,
|
|
list_snapshot_response[0].id
|
|
)
|
|
|
|
self.virtual_machine2.start(self.apiclient)
|
|
|
|
try:
|
|
ssh_client = self.virtual_machine2.get_ssh_client(reconnect=True)
|
|
|
|
cmds = [
|
|
"cat %s/%s" % (self.test_dir, self.random_data)
|
|
]
|
|
|
|
for c in cmds:
|
|
self.debug(c)
|
|
result = ssh_client.execute(c)
|
|
self.debug(result)
|
|
|
|
except Exception:
|
|
self.fail("SSH failed for Virtual machine: %s" %
|
|
self.virtual_machine2.ipaddress)
|
|
|
|
self.assertEqual(
|
|
self.random_data_0,
|
|
result[0],
|
|
"Check the random data is equal with the ramdom file!"
|
|
)
|
|
|
|
# Delete VM snapshot
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
|
|
def test_04_delete_vm_snapshots(self):
|
|
"""Test to delete vm snapshots
|
|
"""
|
|
|
|
list_snapshot_response = VmSnapshot.list(
|
|
self.apiclient,
|
|
virtualmachineid=self.virtual_machine2.id,
|
|
listall=True)
|
|
|
|
self.assertEqual(
|
|
isinstance(list_snapshot_response, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
list_snapshot_response,
|
|
None,
|
|
"Check if snapshot exists in ListSnapshot"
|
|
)
|
|
VmSnapshot.deleteVMSnapshot(
|
|
self.apiclient,
|
|
list_snapshot_response[0].id)
|
|
|
|
time.sleep(30)
|
|
|
|
list_snapshot_response = VmSnapshot.list(
|
|
self.apiclient,
|
|
#vmid=self.virtual_machine.id,
|
|
virtualmachineid=self.virtual_machine2.id,
|
|
listall=False)
|
|
self.debug('list_snapshot_response -------------------- %s' % list_snapshot_response)
|
|
|
|
self.assertIsNone(list_snapshot_response, "snapshot is already deleted")
|
|
|
|
# Take volume snapshot
|
|
@unittest.expectedFailure
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
|
|
def test_05_snapshot_volume_with_secondary(self):
|
|
'''
|
|
Test Create snapshot and backup to secondary
|
|
'''
|
|
backup_config = Configurations.update(self.apiclient,
|
|
name = "sp.bypass.secondary.storage",
|
|
value = "false")
|
|
volume = list_volumes(
|
|
self.apiclient,
|
|
virtualmachineid = self.virtual_machine3.id,
|
|
type = "ROOT",
|
|
listall = True,
|
|
)
|
|
|
|
snapshot = Snapshot.create(
|
|
self.apiclient,
|
|
volume_id = volume[0].id,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid,
|
|
)
|
|
|
|
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
|
|
def test_06_snapshot_volume_on_primary(self):
|
|
'''
|
|
Test Create snapshot and backup to secondary
|
|
'''
|
|
backup_config = Configurations.update(self.apiclient,
|
|
name = "sp.bypass.secondary.storage",
|
|
value = "true")
|
|
volume = list_volumes(
|
|
self.apiclient,
|
|
virtualmachineid = self.virtual_machine3.id,
|
|
type = "ROOT",
|
|
listall = True,
|
|
)
|
|
snapshot = Snapshot.create(
|
|
self.apiclient,
|
|
volume_id = volume[0].id,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid,
|
|
)
|
|
try:
|
|
cmd = getVolumeSnapshotDetails.getVolumeSnapshotDetailsCmd()
|
|
cmd.snapshotid = snapshot.id
|
|
snapshot_details = self.apiclient.getVolumeSnapshotDetails(cmd)
|
|
flag = False
|
|
for s in snapshot_details:
|
|
if s["snapshotDetailsName"] == snapshot.id:
|
|
name = s["snapshotDetailsValue"].split("/")[3]
|
|
sp_snapshot = self.spapi.snapshotList(snapshotName = "~" + name)
|
|
flag = True
|
|
if flag == False:
|
|
raise Exception("Could not find snapshot in snapshot_details")
|
|
except spapi.ApiError as err:
|
|
raise Exception(err)
|
|
self.assertIsNotNone(snapshot, "Could not create snapshot")
|
|
# Rever Volume snapshot
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
|
|
def test_07_revert_volume_on_primary(self):
|
|
volume = list_volumes(
|
|
self.apiclient,
|
|
virtualmachineid = self.virtual_machine3.id,
|
|
type = "ROOT",
|
|
listall = True,
|
|
)[0]
|
|
snapshot = list_snapshots(
|
|
self.apiclient,
|
|
volumeid = volume.id,
|
|
listall=True
|
|
)[0]
|
|
self.virtual_machine3.stop(self.apiclient, forced=True)
|
|
|
|
cmd = revertSnapshot.revertSnapshotCmd()
|
|
cmd.id = snapshot.id
|
|
revertcmd = self.apiclient.revertSnapshot(cmd)
|
|
|
|
# Delete volume snapshot
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
|
|
def test_08_delete_volume_on_primary(self):
|
|
volume = list_volumes(
|
|
self.apiclient,
|
|
virtualmachineid = self.virtual_machine3.id,
|
|
type = "ROOT",
|
|
listall = True,
|
|
)[0]
|
|
snapshot = list_snapshots(
|
|
self.apiclient,
|
|
volumeid = volume.id,
|
|
listall=True
|
|
)[0]
|
|
cmd = deleteSnapshot.deleteSnapshotCmd()
|
|
cmd.id = snapshot.id
|
|
self.apiclient.deleteSnapshot(cmd)
|
|
|
|
# Live migrate encrypted volume
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
|
|
def test_09_live_migrate_volume(self):
|
|
volume = list_volumes(
|
|
self.apiclient,
|
|
virtualmachineid = self.virtual_machine.id,
|
|
type = "ROOT",
|
|
listall = True,
|
|
)[0]
|
|
|
|
Volume.migrate(self.apiclient, volumeid=volume.id, storageid=self.primary_storage2.id, livemigrate=True)
|