Merge changes Ifebbdbf7,I00f431ed
* changes: Xen Storage Manager: tests for xensm volume driver SM volume driver: DB changes and tests
This commit is contained in:
Jenkins
@@ -1367,7 +1367,7 @@ class StorageManagerCommands(object):
|
||||
sys.exit(2)
|
||||
|
||||
try:
|
||||
flavors = db.sm_flavor_get(ctxt, flavor_label)
|
||||
flavors = db.sm_flavor_get_by_label(ctxt, flavor_label)
|
||||
except exception.NotFound as ex:
|
||||
print "error: %s" % ex
|
||||
sys.exit(2)
|
||||
|
||||
@@ -316,3 +316,12 @@ class TestCase(unittest.TestCase):
|
||||
raise AssertionError(exc_msg)
|
||||
except Exception:
|
||||
pass # Any other errors are fine
|
||||
|
||||
def assertIsInstance(self, a, b, *args, **kwargs):
|
||||
"""Python < v2.7 compatibility. Assert 'a' is Instance of 'b'"""
|
||||
try:
|
||||
f = super(TestCase, self).assertIsInstance
|
||||
except AttributeError:
|
||||
self.assertTrue(isinstance(a, b), *args, **kwargs)
|
||||
else:
|
||||
f(a, b, *args, **kwargs)
|
||||
|
||||
@@ -878,3 +878,189 @@ class InstanceDestroyConstraints(test.TestCase):
|
||||
ctx, instance['uuid'], constraint)
|
||||
instance = db.instance_get_by_uuid(ctx, instance['uuid'])
|
||||
self.assertFalse(instance['deleted'])
|
||||
|
||||
|
||||
def _get_sm_backend_params():
|
||||
config_params = ("name_label=testsmbackend "
|
||||
"server=localhost "
|
||||
"serverpath=/tmp/nfspath")
|
||||
params = dict(flavor_id=1,
|
||||
sr_uuid=None,
|
||||
sr_type='nfs',
|
||||
config_params=config_params)
|
||||
return params
|
||||
|
||||
|
||||
def _get_sm_flavor_params():
|
||||
params = dict(label="gold",
|
||||
description="automatic backups")
|
||||
return params
|
||||
|
||||
|
||||
class SMVolumeDBApiTestCase(test.TestCase):
|
||||
def setUp(self):
|
||||
super(SMVolumeDBApiTestCase, self).setUp()
|
||||
self.user_id = 'fake'
|
||||
self.project_id = 'fake'
|
||||
self.context = context.RequestContext(self.user_id, self.project_id)
|
||||
|
||||
def test_sm_backend_conf_create(self):
|
||||
params = _get_sm_backend_params()
|
||||
ctxt = context.get_admin_context()
|
||||
beconf = db.sm_backend_conf_create(ctxt,
|
||||
params)
|
||||
self.assertIsInstance(beconf['id'], int)
|
||||
|
||||
def test_sm_backend_conf_create_raise_duplicate(self):
|
||||
params = _get_sm_backend_params()
|
||||
ctxt = context.get_admin_context()
|
||||
beconf = db.sm_backend_conf_create(ctxt,
|
||||
params)
|
||||
self.assertIsInstance(beconf['id'], int)
|
||||
self.assertRaises(exception.Duplicate,
|
||||
db.sm_backend_conf_create,
|
||||
ctxt,
|
||||
params)
|
||||
|
||||
def test_sm_backend_conf_update(self):
|
||||
ctxt = context.get_admin_context()
|
||||
params = _get_sm_backend_params()
|
||||
beconf = db.sm_backend_conf_create(ctxt,
|
||||
params)
|
||||
beconf = db.sm_backend_conf_update(ctxt,
|
||||
beconf['id'],
|
||||
dict(sr_uuid="FA15E-1D"))
|
||||
self.assertEqual(beconf['sr_uuid'], "FA15E-1D")
|
||||
|
||||
def test_sm_backend_conf_update_raise_notfound(self):
|
||||
ctxt = context.get_admin_context()
|
||||
self.assertRaises(exception.NotFound,
|
||||
db.sm_backend_conf_update,
|
||||
ctxt,
|
||||
7,
|
||||
dict(sr_uuid="FA15E-1D"))
|
||||
|
||||
def test_sm_backend_conf_get(self):
|
||||
ctxt = context.get_admin_context()
|
||||
params = _get_sm_backend_params()
|
||||
beconf = db.sm_backend_conf_create(ctxt,
|
||||
params)
|
||||
val = db.sm_backend_conf_get(ctxt, beconf['id'])
|
||||
self.assertDictMatch(dict(val), dict(beconf))
|
||||
|
||||
def test_sm_backend_conf_get_raise_notfound(self):
|
||||
ctxt = context.get_admin_context()
|
||||
self.assertRaises(exception.NotFound,
|
||||
db.sm_backend_conf_get,
|
||||
ctxt,
|
||||
7)
|
||||
|
||||
def test_sm_backend_conf_get_by_sr(self):
|
||||
ctxt = context.get_admin_context()
|
||||
params = _get_sm_backend_params()
|
||||
beconf = db.sm_backend_conf_create(ctxt,
|
||||
params)
|
||||
val = db.sm_backend_conf_get_by_sr(ctxt, beconf['sr_uuid'])
|
||||
self.assertDictMatch(dict(val), dict(beconf))
|
||||
|
||||
def test_sm_backend_conf_get_by_sr_raise_notfound(self):
|
||||
ctxt = context.get_admin_context()
|
||||
self.assertRaises(exception.NotFound,
|
||||
db.sm_backend_conf_get_by_sr,
|
||||
ctxt,
|
||||
"FA15E-1D")
|
||||
|
||||
def test_sm_backend_conf_delete(self):
|
||||
ctxt = context.get_admin_context()
|
||||
params = _get_sm_backend_params()
|
||||
beconf = db.sm_backend_conf_create(ctxt,
|
||||
params)
|
||||
db.sm_backend_conf_delete(ctxt, beconf['id'])
|
||||
self.assertRaises(exception.NotFound,
|
||||
db.sm_backend_conf_get,
|
||||
ctxt,
|
||||
beconf['id'])
|
||||
|
||||
def test_sm_backend_conf_delete_nonexisting(self):
|
||||
ctxt = context.get_admin_context()
|
||||
self.assertNotRaises(None, db.sm_backend_conf_delete,
|
||||
ctxt, "FA15E-1D")
|
||||
|
||||
def test_sm_flavor_create(self):
|
||||
ctxt = context.get_admin_context()
|
||||
params = _get_sm_flavor_params()
|
||||
flav = db.sm_flavor_create(ctxt,
|
||||
params)
|
||||
self.assertIsInstance(flav['id'], int)
|
||||
|
||||
def sm_flavor_create_raise_duplicate(self):
|
||||
ctxt = context.get_admin_context()
|
||||
params = _get_sm_flavor_params()
|
||||
flav = db.sm_flavor_create(ctxt,
|
||||
params)
|
||||
self.assertRaises(exception.Duplicate,
|
||||
db.sm_flavor_create,
|
||||
params)
|
||||
|
||||
def test_sm_flavor_update(self):
|
||||
ctxt = context.get_admin_context()
|
||||
params = _get_sm_flavor_params()
|
||||
flav = db.sm_flavor_create(ctxt,
|
||||
params)
|
||||
newparms = dict(description="basic volumes")
|
||||
flav = db.sm_flavor_update(ctxt, flav['id'], newparms)
|
||||
self.assertEqual(flav['description'], "basic volumes")
|
||||
|
||||
def test_sm_flavor_update_raise_notfound(self):
|
||||
ctxt = context.get_admin_context()
|
||||
self.assertRaises(exception.NotFound,
|
||||
db.sm_flavor_update,
|
||||
ctxt,
|
||||
7,
|
||||
dict(description="fakedesc"))
|
||||
|
||||
def test_sm_flavor_delete(self):
|
||||
ctxt = context.get_admin_context()
|
||||
params = _get_sm_flavor_params()
|
||||
flav = db.sm_flavor_create(ctxt,
|
||||
params)
|
||||
db.sm_flavor_delete(ctxt, flav['id'])
|
||||
self.assertRaises(exception.NotFound,
|
||||
db.sm_flavor_get,
|
||||
ctxt,
|
||||
"gold")
|
||||
|
||||
def test_sm_flavor_delete_nonexisting(self):
|
||||
ctxt = context.get_admin_context()
|
||||
self.assertNotRaises(None, db.sm_flavor_delete,
|
||||
ctxt, 7)
|
||||
|
||||
def test_sm_flavor_get(self):
|
||||
ctxt = context.get_admin_context()
|
||||
params = _get_sm_flavor_params()
|
||||
flav = db.sm_flavor_create(ctxt,
|
||||
params)
|
||||
val = db.sm_flavor_get(ctxt, flav['id'])
|
||||
self.assertDictMatch(dict(val), dict(flav))
|
||||
|
||||
def test_sm_flavor_get_raise_notfound(self):
|
||||
ctxt = context.get_admin_context()
|
||||
self.assertRaises(exception.NotFound,
|
||||
db.sm_flavor_get,
|
||||
ctxt,
|
||||
7)
|
||||
|
||||
def test_sm_flavor_get_by_label(self):
|
||||
ctxt = context.get_admin_context()
|
||||
params = _get_sm_flavor_params()
|
||||
flav = db.sm_flavor_create(ctxt,
|
||||
params)
|
||||
val = db.sm_flavor_get_by_label(ctxt, flav['label'])
|
||||
self.assertDictMatch(dict(val), dict(flav))
|
||||
|
||||
def test_sm_flavor_get_by_label_raise_notfound(self):
|
||||
ctxt = context.get_admin_context()
|
||||
self.assertRaises(exception.NotFound,
|
||||
db.sm_flavor_get,
|
||||
ctxt,
|
||||
"fake")
|
||||
|
||||
147
nova/tests/test_xensm.py
Normal file
147
nova/tests/test_xensm.py
Normal file
@@ -0,0 +1,147 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright (c) 2010 Citrix Systems, Inc.
|
||||
#
|
||||
# Licensed 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.
|
||||
|
||||
"""Test suite for Xen Storage Manager Volume Driver."""
|
||||
|
||||
import os
|
||||
|
||||
from nova import context
|
||||
from nova import db
|
||||
from nova import exception
|
||||
from nova import flags
|
||||
from nova import log as logging
|
||||
from nova import test
|
||||
from nova.tests.xenapi import stubs
|
||||
from nova.virt.xenapi import connection as xenapi_conn
|
||||
from nova.virt.xenapi import fake as xenapi_fake
|
||||
from nova.virt.xenapi import volume_utils
|
||||
from nova.volume import xensm
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
FLAGS = flags.FLAGS
|
||||
|
||||
|
||||
class XenSMTestCase(test.TestCase):
|
||||
"""Unit tests for Xen Storage Manager Volume operations."""
|
||||
|
||||
def _get_sm_backend_params(self):
|
||||
config_params = ("name_label=testsmbackend "
|
||||
"server=localhost "
|
||||
"serverpath=/tmp/nfspath")
|
||||
params = dict(flavor_id=1,
|
||||
sr_uuid=None,
|
||||
sr_type='nfs',
|
||||
config_params=config_params)
|
||||
return params
|
||||
|
||||
def setUp(self):
|
||||
super(XenSMTestCase, self).setUp()
|
||||
self.user_id = 'fake'
|
||||
self.project_id = 'fake'
|
||||
self.context = context.RequestContext(self.user_id, self.project_id)
|
||||
self.flags(connection_type='xenapi',
|
||||
xenapi_connection_url='http://test_url',
|
||||
xenapi_connection_username='test_user',
|
||||
xenapi_connection_password='test_pass')
|
||||
stubs.stubout_session(self.stubs, xenapi_fake.SessionBase)
|
||||
xenapi_fake.reset()
|
||||
self.driver = xensm.XenSMDriver()
|
||||
self.driver.db = db
|
||||
|
||||
def _setup_step(self, ctxt):
|
||||
# Create a fake backend conf
|
||||
params = self._get_sm_backend_params()
|
||||
beconf = db.sm_backend_conf_create(ctxt,
|
||||
params)
|
||||
# Call setup, the one time operation that will create a backend SR
|
||||
self.driver.do_setup(ctxt)
|
||||
return beconf
|
||||
|
||||
def test_do_setup(self):
|
||||
ctxt = context.get_admin_context()
|
||||
beconf = self._setup_step(ctxt)
|
||||
beconf = db.sm_backend_conf_get(ctxt, beconf['id'])
|
||||
self.assertIsInstance(beconf['sr_uuid'], basestring)
|
||||
|
||||
def _create_volume(self, size='0'):
|
||||
"""Create a volume object."""
|
||||
vol = {}
|
||||
vol['size'] = size
|
||||
vol['user_id'] = 'fake'
|
||||
vol['project_id'] = 'fake'
|
||||
vol['host'] = 'localhost'
|
||||
vol['availability_zone'] = FLAGS.storage_availability_zone
|
||||
vol['status'] = "creating"
|
||||
vol['attach_status'] = "detached"
|
||||
return db.volume_create(self.context, vol)
|
||||
|
||||
def test_create_volume(self):
|
||||
ctxt = context.get_admin_context()
|
||||
beconf = self._setup_step(ctxt)
|
||||
volume = self._create_volume()
|
||||
self.assertNotRaises(None, self.driver.create_volume, volume)
|
||||
self.assertNotRaises(None,
|
||||
db.sm_volume_get,
|
||||
ctxt,
|
||||
volume['id'])
|
||||
|
||||
def test_local_path(self):
|
||||
ctxt = context.get_admin_context()
|
||||
volume = self._create_volume()
|
||||
val = self.driver.local_path(volume)
|
||||
self.assertIsInstance(val, basestring)
|
||||
|
||||
def test_delete_volume(self):
|
||||
ctxt = context.get_admin_context()
|
||||
beconf = self._setup_step(ctxt)
|
||||
volume = self._create_volume()
|
||||
self.driver.create_volume(volume)
|
||||
self.assertNotRaises(None, self.driver.delete_volume, volume)
|
||||
self.assertRaises(exception.NotFound,
|
||||
db.sm_volume_get,
|
||||
ctxt,
|
||||
volume['id'])
|
||||
|
||||
def test_delete_volume_raises_notfound(self):
|
||||
ctxt = context.get_admin_context()
|
||||
beconf = self._setup_step(ctxt)
|
||||
self.assertRaises(exception.NotFound,
|
||||
self.driver.delete_volume,
|
||||
{'id': "FA15E-1D"})
|
||||
|
||||
def _get_expected_conn(self, beconf, vol):
|
||||
expected = {}
|
||||
expected['volume_id'] = unicode(vol['id'])
|
||||
expected['flavor_id'] = beconf['flavor_id']
|
||||
expected['sr_uuid'] = unicode(beconf['sr_uuid'])
|
||||
expected['sr_type'] = unicode(beconf['sr_type'])
|
||||
return expected
|
||||
|
||||
def test_initialize_connection(self):
|
||||
ctxt = context.get_admin_context()
|
||||
beconf = self._setup_step(ctxt)
|
||||
beconf = db.sm_backend_conf_get(ctxt, beconf['id'])
|
||||
volume = self._create_volume()
|
||||
self.driver.create_volume(volume)
|
||||
expected = self._get_expected_conn(beconf, volume)
|
||||
conn = self.driver.initialize_connection(volume, 'fakeConnector')
|
||||
res = {}
|
||||
for key in ['volume_id', 'flavor_id', 'sr_uuid', 'sr_type']:
|
||||
res[key] = conn['data'][key]
|
||||
self.assertDictMatch(expected, res)
|
||||
self.assertEqual(set(conn['data']['introduce_sr_keys']),
|
||||
set([u'sr_type', u'server', u'serverpath']))
|
||||
Reference in New Issue
Block a user