apache/cloudstack
1
0
Fork 0
mirror of https://github.com/apache/cloudstack.git synced 2025-10-26 08:42:29 +01:00
Code Issues Packages Projects Releases Wiki Activity

Added a test to verify that listSnapshots response returns zone id

Browse Source 
Made changes based on Comments
This closes #632
This commit is contained in:
manasaVeloori 2015-07-29 13:59:15 +05:30 committed by sanjeev
parent bccd344796
commit f45548bf0b
1 changed files with 77 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
test/integration/smoke/test_snapshots.py
View File

@ -31,6 +31,7 @@ from marvin.lib.common import (get_domain,
list_snapshots) list_snapshots)
from marvin.lib.decoratorGenerators import skipTestIf from marvin.lib.decoratorGenerators import skipTestIf
class TestSnapshotRootDisk(cloudstackTestCase): class TestSnapshotRootDisk(cloudstackTestCase):
@classmethod @classmethod
@ -52,12 +53,13 @@ class TestSnapshotRootDisk(cloudstackTestCase):
cls._cleanup = [] cls._cleanup = []
if not cls.hypervisorNotSupported: if not cls.hypervisorNotSupported:
template = get_template( template = get_template(
cls.apiclient, cls.apiclient,
cls.zone.id, cls.zone.id,
cls.services["ostype"] cls.services["ostype"]
) )
if template == FAILED: if template == FAILED:
assert False, "get_template() failed to return template with description %s" % cls.services["ostype"] assert False, "get_template() failed to return template with description %s" % cls.services[
"ostype"]
cls.services["domainid"] = cls.domain.id cls.services["domainid"] = cls.domain.id
cls.services["server_without_disk"]["zoneid"] = cls.zone.id cls.services["server_without_disk"]["zoneid"] = cls.zone.id
@ -66,32 +68,33 @@ class TestSnapshotRootDisk(cloudstackTestCase):
# Create VMs, NAT Rules etc # Create VMs, NAT Rules etc
cls.account = Account.create( cls.account = Account.create(
cls.apiclient, cls.apiclient,
cls.services["account"], cls.services["account"],
domainid=cls.domain.id domainid=cls.domain.id
) )
cls._cleanup.append(cls.account) cls._cleanup.append(cls.account)
cls.service_offering = ServiceOffering.create( cls.service_offering = ServiceOffering.create(
cls.apiclient, cls.apiclient,
cls.services["service_offerings"] cls.services["service_offerings"]
) )
cls._cleanup.append(cls.service_offering) cls._cleanup.append(cls.service_offering)
cls.virtual_machine = cls.virtual_machine_with_disk = \ cls.virtual_machine = cls.virtual_machine_with_disk = \
VirtualMachine.create( VirtualMachine.create(
cls.apiclient, cls.apiclient,
cls.services["server_without_disk"], cls.services["server_without_disk"],
templateid=template.id, templateid=template.id,
accountid=cls.account.name, accountid=cls.account.name,
domainid=cls.account.domainid, domainid=cls.account.domainid,
serviceofferingid=cls.service_offering.id, zoneid=cls.zone.id,
mode=cls.services["mode"] serviceofferingid=cls.service_offering.id,
) mode=cls.services["mode"]
)
return return
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
try: try:
#Cleanup resources used # Cleanup resources used
cleanup_resources(cls.apiclient, cls._cleanup) cleanup_resources(cls.apiclient, cls._cleanup)
except Exception as e: except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e) raise Exception("Warning: Exception during cleanup : %s" % e)
@ -105,14 +108,14 @@ class TestSnapshotRootDisk(cloudstackTestCase):
def tearDown(self): def tearDown(self):
try: try:
#Clean up, terminate the created instance, volumes and snapshots # Clean up, terminate the created instance, volumes and snapshots
cleanup_resources(self.apiclient, self.cleanup) cleanup_resources(self.apiclient, self.cleanup)
except Exception as e: except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e) raise Exception("Warning: Exception during cleanup : %s" % e)
return return
@skipTestIf("hypervisorNotSupported") @skipTestIf("hypervisorNotSupported")
@attr(tags = ["advanced", "advancedns", "smoke"], required_hardware="true") @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
def test_01_snapshot_root_disk(self): def test_01_snapshot_root_disk(self):
"""Test Snapshot Root Disk """Test Snapshot Root Disk
""" """
@ -123,65 +126,76 @@ class TestSnapshotRootDisk(cloudstackTestCase):
# the reqd volume under # the reqd volume under
# /secondary/snapshots//$account_id/$volumeid/$snapshot_uuid # /secondary/snapshots//$account_id/$volumeid/$snapshot_uuid
# 3. verify backup_snap_id was non null in the `snapshots` table # 3. verify backup_snap_id was non null in the `snapshots` table
# 4. Verify that zoneid is returned in listSnapshots API response
volumes = list_volumes( volumes = list_volumes(
self.apiclient, self.apiclient,
virtualmachineid=self.virtual_machine_with_disk.id, virtualmachineid=self.virtual_machine_with_disk.id,
type='ROOT', type='ROOT',
listall=True listall=True
) )
snapshot = Snapshot.create( snapshot = Snapshot.create(
self.apiclient, self.apiclient,
volumes[0].id, volumes[0].id,
account=self.account.name, account=self.account.name,
domainid=self.account.domainid domainid=self.account.domainid
) )
self.debug("Snapshot created: ID - %s" % snapshot.id) self.debug("Snapshot created: ID - %s" % snapshot.id)
snapshots = list_snapshots( snapshots = list_snapshots(
self.apiclient, self.apiclient,
id=snapshot.id id=snapshot.id
) )
self.assertEqual( self.assertEqual(
isinstance(snapshots, list), isinstance(snapshots, list),
True, True,
"Check list response returns a valid list" "Check list response returns a valid list"
) )
self.assertNotEqual( self.assertNotEqual(
snapshots, snapshots,
None, None,
"Check if result exists in list item call" "Check if result exists in list item call"
) )
self.assertEqual( self.assertEqual(
snapshots[0].id, snapshots[0].id,
snapshot.id, snapshot.id,
"Check resource id in list resources call" "Check resource id in list resources call"
) )
self.assertIsNotNone(snapshots[0].zoneid,
"Zone id is not none in listSnapshots")
self.assertEqual(
snapshots[0].zoneid,
self.zone.id,
"Check zone id in the list snapshots"
)
self.debug( self.debug(
"select backup_snap_id, account_id, volume_id from snapshots where uuid = '%s';" \ "select backup_snap_id, account_id, volume_id from snapshots where uuid = '%s';"
% str(snapshot.id) % str(snapshot.id)
) )
qresultset = self.dbclient.execute( qresultset = self.dbclient.execute(
"select backup_snap_id, account_id, volume_id from snapshots where uuid = '%s';" \ "select backup_snap_id, account_id, volume_id from snapshots where uuid = '%s';"
% str(snapshot.id) % str(snapshot.id)
) )
self.assertNotEqual( self.assertNotEqual(
len(qresultset), len(qresultset),
0, 0,
"Check DB Query result set" "Check DB Query result set"
) )
qresult = qresultset[0] qresult = qresultset[0]
snapshot_uuid = qresult[0] # backup_snap_id = snapshot UUID snapshot_uuid = qresult[0] # backup_snap_id = snapshot UUID
self.assertNotEqual( self.assertNotEqual(
str(snapshot_uuid), str(snapshot_uuid),
'NULL', 'NULL',
"Check if backup_snap_id is not null" "Check if backup_snap_id is not null"
) )
self.assertTrue(is_snapshot_on_nfs(self.apiclient, self.dbclient, self.config, self.zone.id, snapshot.id)) self.assertTrue(is_snapshot_on_nfs(
self.apiclient, self.dbclient, self.config, self.zone.id, snapshot.id))
return return