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

CLOUDSTACK-4452 - Fixed method is_snapshot_on_nfs and moved it to utils

Browse Source 
Signed-off-by: Prasanna Santhanam <tsp@apache.org>
(cherry picked from commit 35ee38385ce0c54b3a16c3270db66d3eb3a56d04)
This commit is contained in:
Gaurav Aradhye 2013-08-22 02:07:09 -04:00 committed by Prasanna Santhanam
parent f724c91296
commit 684573939f
4 changed files with 97 additions and 321 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
test/integration/component/test_snapshot_gc.py
View File

@ -22,6 +22,7 @@ from marvin.integration.lib.utils import *
from marvin.integration.lib.base import *
from marvin.integration.lib.common import *
from marvin.remoteSSHClient import remoteSSHClient
from marvin.integration.lib.utils import is_snapshot_on_nfs
import os
@ -210,82 +211,6 @@ class TestAccountSnapshotClean(cloudstackTestCase):
raise Exception("Warning: Exception during cleanup : %s" % e)
return
def is_snapshot_on_nfs(self, snapshot_id):
"""
Checks whether a snapshot with id (not UUID) `snapshot_id` is present on the nfs storage
@param snapshot_id: id of the snapshot (not uuid)
@return: True if snapshot is found, False otherwise
"""
secondaryStores = ImageStore.list(self.apiclient, zoneid=self.zone.id)
self.assertTrue(isinstance(secondaryStores, list), "Not a valid response for listImageStores")
self.assertNotEqual(len(secondaryStores), 0, "No image stores found in zone %s" % self.zone.id)
secondaryStore = secondaryStores[0]
if str(secondaryStore.providername).lower() != "nfs":
self.skipTest("TODO: %s test works only against nfs secondary storage" % self._testMethodName)
qresultset = self.dbclient.execute(
"select install_path from snapshot_store_ref where snapshot_id='%s' and store_role='Image';" % snapshot_id
)
self.assertEqual(
isinstance(qresultset, list),
True,
"Invalid db query response for snapshot %s" % snapshot_id
)
self.assertNotEqual(
len(qresultset),
0,
"No such snapshot %s found in the cloudstack db" % snapshot_id
)
snapshotPath = qresultset[0][0]
nfsurl = secondaryStore.url
# parse_url = ['nfs:', '', '192.168.100.21', 'export', 'test']
from urllib2 import urlparse
parse_url = urlparse.urlsplit(nfsurl, scheme='nfs')
host, path = parse_url.netloc, parse_url.path
# Sleep to ensure that snapshot is reflected in sec storage
time.sleep(self.services["sleep"])
snapshots = []
try:
# Login to Secondary storage VM to check snapshot present on sec disk
ssh_client = remoteSSHClient(
self.config.mgtSvr[0].mgtSvrIp,
22,
self.config.mgtSvr[0].user,
self.config.mgtSvr[0].passwd,
)
cmds = [
"mkdir -p %s" % self.services["paths"]["mount_dir"],
"mount -t %s %s%s %s" % (
'nfs',
host,
path,
self.services["paths"]["mount_dir"]
),
"ls %s" % (
os.path.join(self.services["paths"]["mount_dir"], snapshotPath)
),
]
for c in cmds:
self.debug("command: %s" % c)
result = ssh_client.execute(c)
self.debug("Result: %s" % result)
snapshots.extend(result)
# Unmount the Sec Storage
cmds = [
"cd",
"umount %s" % (self.services["paths"]["mount_dir"]),
]
for c in cmds:
ssh_client.execute(c)
except Exception as e:
self.fail("SSH failed for management server: %s - %s" %
(self.config.mgtSvr[0].mgtSvrIp, e))
return snapshots.count(snapshot_id) == 1
@attr(speed = "slow")
@attr(tags = ["advanced", "advancedns", "basic", "sg"])
def test_02_accountSnapshotClean(self):
@ -382,7 +307,9 @@ class TestAccountSnapshotClean(cloudstackTestCase):
qresult = qresultset[0]
snapshot_id = qresult[0]
self.assertTrue(self.is_snapshot_on_nfs(snapshot_id), "Snapshot was not found no NFS")
self.assertTrue(is_snapshot_on_nfs(self.apiclient, self.dbclient, self.config.mgtSvr,
self.services["paths"], self.zone.id, snapshot_id),
"Snapshot was not found on NFS")
self.debug("Deleting account: %s" % self.account.name)
# Delete account
@ -400,5 +327,7 @@ class TestAccountSnapshotClean(cloudstackTestCase):
"List accounts should return empty list after account deletion"
)
self.assertFalse(self.is_snapshot_on_nfs(snapshot_id), "Snapshot was still found no NFS after account gc")
self.assertFalse(is_snapshot_on_nfs(self.apiclient, self.dbclient, self.config.mgtSvr,
self.services["paths"], self.zone.id, snapshot_id),
"Snapshot was still found no NFS after account gc")
return

80
test/integration/component/test_snapshot_limits.py
View File

@ -22,6 +22,7 @@ from marvin.integration.lib.utils import *
from marvin.integration.lib.base import *
from marvin.integration.lib.common import *
from marvin.remoteSSHClient import remoteSSHClient
from marvin.integration.lib.utils import is_snapshot_on_nfs
import os
@ -198,82 +199,6 @@ class TestSnapshotLimit(cloudstackTestCase):
raise Exception("Warning: Exception during cleanup : %s" % e)
return
def is_snapshot_on_nfs(self, snapshot_id):
"""
Checks whether a snapshot with id (not UUID) `snapshot_id` is present on the nfs storage
@param snapshot_id: id of the snapshot (not uuid)
@return: True if snapshot is found, False otherwise
"""
secondaryStores = ImageStore.list(self.apiclient, zoneid=self.zone.id)
self.assertTrue(isinstance(secondaryStores, list), "Not a valid response for listImageStores")
self.assertNotEqual(len(secondaryStores), 0, "No image stores found in zone %s" % self.zone.id)
secondaryStore = secondaryStores[0]
if str(secondaryStore.providername).lower() != "nfs":
self.skipTest("TODO: %s test works only against nfs secondary storage" % self._testMethodName)
qresultset = self.dbclient.execute(
"select install_path from snapshot_store_ref where snapshot_id='%s' and store_role='Image';" % snapshot_id
)
self.assertEqual(
isinstance(qresultset, list),
True,
"Invalid db query response for snapshot %s" % snapshot_id
)
self.assertNotEqual(
len(qresultset),
0,
"No such snapshot %s found in the cloudstack db" % snapshot_id
)
snapshotPath = qresultset[0][0]
nfsurl = secondaryStore.url
# parse_url = ['nfs:', '', '192.168.100.21', 'export', 'test']
from urllib2 import urlparse
parse_url = urlparse.urlsplit(nfsurl, scheme='nfs')
host, path = parse_url.netloc, parse_url.path
# Sleep to ensure that snapshot is reflected in sec storage
time.sleep(self.services["sleep"])
snapshots = []
try:
# Login to Secondary storage VM to check snapshot present on sec disk
ssh_client = remoteSSHClient(
self.config.mgtSvr[0].mgtSvrIp,
22,
self.config.mgtSvr[0].user,
self.config.mgtSvr[0].passwd,
)
cmds = [
"mkdir -p %s" % self.services["paths"]["mount_dir"],
"mount -t %s %s%s %s" % (
'nfs',
host,
path,
self.services["paths"]["mount_dir"]
),
"ls %s" % (
os.path.join(self.services["paths"]["mount_dir"], snapshotPath)
),
]
for c in cmds:
self.debug("command: %s" % c)
result = ssh_client.execute(c)
self.debug("Result: %s" % result)
snapshots.extend(result)
# Unmount the Sec Storage
cmds = [
"cd",
"umount %s" % (self.services["paths"]["mount_dir"]),
]
for c in cmds:
ssh_client.execute(c)
except Exception as e:
self.fail("SSH failed for management server: %s - %s" %
(self.config.mgtSvr[0].mgtSvrIp, e))
return snapshots.count(snapshot_id) == 1
@attr(speed = "slow")
@attr(tags = ["advanced", "advancedns"])
def test_04_snapshot_limit(self):
@ -383,5 +308,6 @@ class TestSnapshotLimit(cloudstackTestCase):
qresult = qresultset[0]
snapshot_id = qresult[0]
self.is_snapshot_on_nfs(snapshot_id)
self.assertTrue(is_snapshot_on_nfs(self.apiclient, self.dbclient, self.config.mgtSvr,
self.services["paths"], self.zone.id, snapshot_id))
return

175
test/integration/component/test_snapshots.py
View File

@ -24,6 +24,7 @@ from marvin.integration.lib.utils import *
from marvin.integration.lib.base import *
from marvin.integration.lib.common import *
from marvin.remoteSSHClient import remoteSSHClient
from marvin.integration.lib.utils import is_snapshot_on_nfs
import os
@ -174,16 +175,7 @@ class TestSnapshots(cloudstackTestCase):
serviceofferingid=cls.service_offering.id,
mode=cls.services["mode"]
)
cls.virtual_machine_without_disk = \
VirtualMachine.create(
cls.api_client,
cls.services["server_without_disk"],
templateid=cls.template.id,
accountid=cls.account.name,
domainid=cls.account.domainid,
serviceofferingid=cls.service_offering.id,
mode=cls.services["mode"]
)
cls._cleanup = [
cls.service_offering,
cls.disk_offering,
@ -191,82 +183,6 @@ class TestSnapshots(cloudstackTestCase):
]
return
def is_snapshot_on_nfs(self, snapshot_id):
"""
Checks whether a snapshot with id (not UUID) `snapshot_id` is present on the nfs storage
@param snapshot_id: id of the snapshot (not uuid)
@return: True if snapshot is found, False otherwise
"""
secondaryStores = ImageStore.list(self.apiclient, zoneid=self.zone.id)
self.assertTrue(isinstance(secondaryStores, list), "Not a valid response for listImageStores")
self.assertNotEqual(len(secondaryStores), 0, "No image stores found in zone %s" % self.zone.id)
secondaryStore = secondaryStores[0]
if str(secondaryStore.providername).lower() != "nfs":
self.skipTest("TODO: %s test works only against nfs secondary storage" % self._testMethodName)
qresultset = self.dbclient.execute(
"select install_path from snapshot_store_ref where snapshot_id='%s' and store_role='Image';" % snapshot_id
)
self.assertEqual(
isinstance(qresultset, list),
True,
"Invalid db query response for snapshot %s" % snapshot_id
)
self.assertNotEqual(
len(qresultset),
0,
"No such snapshot %s found in the cloudstack db" % snapshot_id
)
snapshotPath = qresultset[0][0]
nfsurl = secondaryStore.url
# parse_url = ['nfs:', '', '192.168.100.21', 'export', 'test']
from urllib2 import urlparse
parse_url = urlparse.urlsplit(nfsurl, scheme='nfs')
host, path = parse_url.netloc, parse_url.path
# Sleep to ensure that snapshot is reflected in sec storage
time.sleep(self.services["sleep"])
snapshots = []
try:
# Login to Secondary storage VM to check snapshot present on sec disk
ssh_client = remoteSSHClient(
self.config.mgtSvr[0].mgtSvrIp,
22,
self.config.mgtSvr[0].user,
self.config.mgtSvr[0].passwd,
)
cmds = [
"mkdir -p %s" % self.services["paths"]["mount_dir"],
"mount -t %s %s%s %s" % (
'nfs',
host,
path,
self.services["paths"]["mount_dir"]
),
"ls %s" % (
os.path.join(self.services["paths"]["mount_dir"], snapshotPath)
),
]
for c in cmds:
self.debug("command: %s" % c)
result = ssh_client.execute(c)
self.debug("Result: %s" % result)
snapshots.extend(result)
# Unmount the Sec Storage
cmds = [
"cd",
"umount %s" % (self.services["paths"]["mount_dir"]),
]
for c in cmds:
ssh_client.execute(c)
except Exception as e:
self.fail("SSH failed for management server: %s - %s" %
(self.config.mgtSvr[0].mgtSvrIp, e))
return snapshots.count(snapshot_id) == 1
@classmethod
def tearDownClass(cls):
try:
@ -352,7 +268,8 @@ class TestSnapshots(cloudstackTestCase):
'NULL',
"Check if backup_snap_id is not null"
)
self.assertTrue(self.is_snapshot_on_nfs(snapshot_uuid))
self.assertTrue(is_snapshot_on_nfs(self.apiclient, self.dbclient, self.config.mgtSvr,
self.services["paths"], self.zone.id, snapshot_uuid))
return
@attr(speed = "slow")
@ -620,7 +537,8 @@ class TestSnapshots(cloudstackTestCase):
qresult = qresultset[0]
snapshotid = qresult[0]
self.assertFalse(self.is_snapshot_on_nfs(snapshotid))
self.assertFalse(is_snapshot_on_nfs(self.apiclient, self.dbclient, self.config.mgtSvr,
self.services["paths"], self.zone.id, snapshotid))
return
@attr(speed = "slow")
@ -771,7 +689,7 @@ class TestSnapshots(cloudstackTestCase):
cmds = [
"mkdir -p %s" % self.services["paths"]["mount_dir"],
"mount %s1 %s" % (
self.services["rootdisk"],
self.services["volume"]["diskdevice"],
self.services["paths"]["mount_dir"]
),
"mkdir -p %s/%s/{%s,%s} " % (
@ -885,7 +803,7 @@ class TestSnapshots(cloudstackTestCase):
cmds = [
"mkdir -p %s" % self.services["paths"]["mount_dir"],
"mount %s1 %s" % (
self.services["rootdisk"],
self.services["volume"]["diskdevice"],
self.services["paths"]["mount_dir"]
)
]
@ -998,80 +916,6 @@ class TestCreateVMSnapshotTemplate(cloudstackTestCase):
raise Exception("Warning: Exception during cleanup : %s" % e)
return
def is_snapshot_on_nfs(self, snapshot_id):
"""
Checks whether a snapshot with id (not UUID) `snapshot_id` is present on the nfs storage
@param snapshot_id: id of the snapshot (not uuid)
@return: True if snapshot is found, False otherwise
"""
secondaryStores = ImageStore.list(self.apiclient, zoneid=self.zone.id)
self.assertTrue(isinstance(secondaryStores, list), "Not a valid response for listImageStores")
self.assertNotEqual(len(secondaryStores), 0, "No image stores found in zone %s" % self.zone.id)
secondaryStore = secondaryStores[0]
if str(secondaryStore.providername).lower() != "nfs":
self.skipTest("TODO: %s test works only against nfs secondary storage" % self._testMethodName)
qresultset = self.dbclient.execute(
"select install_path from snapshot_store_ref where snapshot_id='%s' and store_role='Image';" % snapshot_id
)
self.assertEqual(
isinstance(qresultset, list),
True,
"Invalid db query response for snapshot %s" % snapshot_id
)
self.assertNotEqual(
len(qresultset),
0,
"No such snapshot %s found in the cloudstack db" % snapshot_id
)
snapshotPath = qresultset[0][0]
nfsurl = secondaryStore.url
# parse_url = ['nfs:', '', '192.168.100.21', 'export', 'test']
from urllib2 import urlparse
parse_url = urlparse.urlsplit(nfsurl, scheme='nfs')
host, path = parse_url.netloc, parse_url.path
snapshots = []
try:
# Login to Secondary storage VM to check snapshot present on sec disk
ssh_client = remoteSSHClient(
self.config.mgtSvr[0].mgtSvrIp,
22,
self.config.mgtSvr[0].user,
self.config.mgtSvr[0].passwd,
)
cmds = [
"mkdir -p %s" % self.services["paths"]["mount_dir"],
"mount -t %s %s%s %s" % (
'nfs',
host,
path,
self.services["paths"]["mount_dir"]
),
"ls %s" % (
os.path.join(self.services["paths"]["mount_dir"], snapshotPath)
),
]
for c in cmds:
self.debug("command: %s" % c)
result = ssh_client.execute(c)
self.debug("Result: %s" % result)
snapshots.extend(result)
# Unmount the Sec Storage
cmds = [
"cd",
"umount %s" % (self.services["paths"]["mount_dir"]),
]
for c in cmds:
ssh_client.execute(c)
except Exception as e:
self.fail("SSH failed for management server: %s - %s" %
(self.services["mgmt_server"]["ipaddress"], e))
return snapshots.count(snapshot_id) == 1
@attr(speed = "slow")
@attr(tags = ["advanced", "advancedns"])
def test_01_createVM_snapshotTemplate(self):
@ -1217,7 +1061,8 @@ class TestCreateVMSnapshotTemplate(cloudstackTestCase):
'Running',
"Check list VM response for Running state"
)
self.assertTrue(self.is_snapshot_on_nfs(snapshot_uuid))
self.assertTrue(is_snapshot_on_nfs(self.apiclient, self.dbclient, self.config.mgtSvr,
self.services["paths"], self.zone.id, snapshot_uuid))
return

76
tools/marvin/marvin/integration/lib/utils.py
View File

@ -219,3 +219,79 @@ def xsplit(txt, seps):
for sep in seps[1:]: # we skip seps[0] because that's the default separator
txt = txt.replace(sep, default_sep)
return [i.strip() for i in txt.split(default_sep)]
def is_snapshot_on_nfs(api_client, db_client, config_mgtSvr,
dir_paths, zone_id, snapshot_id):
"""
Checks whether a snapshot with id (not UUID) `snapshot_id` is present on the nfs storage
@param snapshot_id: id of the snapshot (not uuid)
@return: True if snapshot is found, False otherwise
"""
from base import ImageStore
secondaryStores = ImageStore.list(api_client, zoneid=zone_id)
assert isinstance(secondaryStores, list), "Not a valid response for listImageStores"
assert len(secondaryStores) != 0, "No image stores found in zone %s" % zone_id
secondaryStore = secondaryStores[0]
if str(secondaryStore.providername).lower() != "nfs":
raise Exception("Test works only against nfs secondary storage")
qresultset = db_client.execute(
"select install_path from snapshot_store_ref where snapshot_id='%s' and store_role='Image';" % snapshot_id
)
assert isinstance(qresultset, list), "Invalid db query response for snapshot %s" % snapshot_id
assert len(qresultset) != 0, "No such snapshot %s found in the cloudstack db" % snapshot_id
snapshotPath = qresultset[0][0]
nfsurl = secondaryStore.url
# parse_url = ['nfs:', '', '192.168.100.21', 'export', 'test']
from urllib2 import urlparse
parse_url = urlparse.urlsplit(nfsurl, scheme='nfs')
host, path = parse_url.netloc, parse_url.path
snapshots = []
try:
# Login to Secondary storage VM to check snapshot present on sec disk
ssh_client = remoteSSHClient(
config_mgtSvr[0].mgtSvrIp,
22,
config_mgtSvr[0].user,
config_mgtSvr[0].passwd,
)
import os
cmds = [
"mkdir -p %s" % dir_paths["mount_dir"],
"mount -t %s %s%s %s" % (
'nfs',
host,
path,
dir_paths["mount_dir"]
),
"test -f %s && echo 'snapshot exists'" % (
os.path.join(dir_paths["mount_dir"], snapshotPath)
),
]
for c in cmds:
result = ssh_client.execute(c)
# Unmount the Sec Storage
cmds = [
"cd",
"umount %s" % (dir_paths["mount_dir"]),
]
for c in cmds:
ssh_client.execute(c)
except Exception as e:
raise Exception("SSH failed for management server: %s - %s" %
(config_mgtSvr[0].mgtSvrIp, e))
return 'snapshot exists' in result