diff --git a/bin/nova-manage b/bin/nova-manage index 5414e4e5e..6bda2ada5 100755 --- a/bin/nova-manage +++ b/bin/nova-manage @@ -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) diff --git a/nova/test.py b/nova/test.py index 8de4ddfca..1839fd1e4 100644 --- a/nova/test.py +++ b/nova/test.py @@ -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) diff --git a/nova/tests/test_db_api.py b/nova/tests/test_db_api.py index d28acfd72..aaddb08a6 100644 --- a/nova/tests/test_db_api.py +++ b/nova/tests/test_db_api.py @@ -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") diff --git a/nova/tests/test_xensm.py b/nova/tests/test_xensm.py new file mode 100644 index 000000000..639caf009 --- /dev/null +++ b/nova/tests/test_xensm.py @@ -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']))