From 669d5f5612840c9ed6449d91ee5aae97842cac72 Mon Sep 17 00:00:00 2001 From: Armando Migliaccio Date: Mon, 13 Dec 2010 18:22:56 +0000 Subject: second round for unit testing framework --- nova/tests/virt_unittest.py | 65 ++++++++++++++++++++++++++++++++------------- 1 file changed, 47 insertions(+), 18 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/virt_unittest.py b/nova/tests/virt_unittest.py index ba3fba83b..611022632 100644 --- a/nova/tests/virt_unittest.py +++ b/nova/tests/virt_unittest.py @@ -30,6 +30,9 @@ from nova.virt.xenapi import volume_utils FLAGS = flags.FLAGS flags.DECLARE('instances_path', 'nova.compute.manager') +# Those are XenAPI related +flags.DECLARE('target_host', 'nova.virt.xenapi_conn') +FLAGS.target_host = '127.0.0.1' class LibvirtConnTestCase(test.TrialTestCase): @@ -270,36 +273,62 @@ class XenAPIVolumeTestCase(test.TrialTestCase): self.helper = volume_utils.VolumeHelper self.helper.late_import() + 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(context.get_admin_context(), vol) + def test_create_iscsi_storage_raise_no_exception(self): - info = self.helper.parse_volume_info(None, None) - label = 'SR-' - description = '' - self.helper.create_iscsi_storage_blocking(self.session, - info, - label, - description) + vol = self._create_volume() + info = yield self.helper.parse_volume_info(vol['ec2_id'], '/dev/sdc') + label = None # For testing new SRs + description = 'Test-SR' + self.session.fail_next_call = False + sr_ref = self.helper.create_iscsi_storage_blocking(self.session, + info, + label, + description) + self.assertEqual(sr_ref, self.session.SR.FAKE_REF) + db.volume_destroy(context.get_admin_context(), vol['id']) def test_create_iscsi_storage_raise_unable_to_create_sr_exception(self): - info = self.helper.parse_volume_info(None, None) - label = None + vol = self._create_volume() + info = yield self.helper.parse_volume_info(vol['ec2_id'], '/dev/sdc') + label = None # For testing new SRs description = None - self.assertFailure(self.helper.create_iscsi_storage_blocking(self.session, - info, - label, - description), - StorageError) + self.session.fail_next_call = True + self.assertRaises(volume_utils.StorageError, + self.helper.create_iscsi_storage_blocking, + self.session, + info, + label, + description) def test_find_sr_from_vbd_raise_no_exception(self): - pass + sr_ref = yield self.helper.find_sr_from_vbd(self.session, + self.session.VBD.FAKE_REF) + self.assertEqual(sr_ref, self.session.SR.FAKE_REF) - def test_destroy_iscsi_storage_raise_no_exception(self): + def test_destroy_iscsi_storage(self): pass def test_introduce_vdi_raise_no_exception(self): - pass + sr_ref = self.session.SR.FAKE_REF + self.helper.introduce_vdi_blocking(self.session, sr_ref) def test_introduce_vdi_raise_unable_get_vdi_record_exception(self): - pass + sr_ref = self.session.SR.FAKE_REF + self.session.fail_next_call = True + self.assertRaises(volume_utils.StorageError, + self.helper.introduce_vdi_blocking, + self.session, sr_ref) def tearDown(self): super(XenAPIVolumeTestCase, self).tearDown() -- cgit From fe667352c3e25c744a989ca45f4f9ed472778ae3 Mon Sep 17 00:00:00 2001 From: Armando Migliaccio Date: Mon, 13 Dec 2010 18:43:24 +0000 Subject: removing imports that should have not been there --- nova/tests/virt_unittest.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/virt_unittest.py b/nova/tests/virt_unittest.py index 611022632..1095662c3 100644 --- a/nova/tests/virt_unittest.py +++ b/nova/tests/virt_unittest.py @@ -317,7 +317,8 @@ class XenAPIVolumeTestCase(test.TrialTestCase): self.assertEqual(sr_ref, self.session.SR.FAKE_REF) def test_destroy_iscsi_storage(self): - pass + sr_ref = self.session.SR.FAKE_REF + self.helper.destroy_iscsi_storage_blocking(self.session, sr_ref) def test_introduce_vdi_raise_no_exception(self): sr_ref = self.session.SR.FAKE_REF -- cgit From e30801445f8b543d78494ca63be60f85b94d3a53 Mon Sep 17 00:00:00 2001 From: Armando Migliaccio Date: Mon, 13 Dec 2010 20:31:33 +0000 Subject: moving xenapi unittests changes into another branch --- nova/tests/virt_unittest.py | 76 --------------------------------------------- 1 file changed, 76 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/virt_unittest.py b/nova/tests/virt_unittest.py index 1095662c3..d49383fb7 100644 --- a/nova/tests/virt_unittest.py +++ b/nova/tests/virt_unittest.py @@ -25,14 +25,9 @@ from nova import utils from nova.api.ec2 import cloud from nova.auth import manager from nova.virt import libvirt_conn -from nova.virt.xenapi import fake -from nova.virt.xenapi import volume_utils FLAGS = flags.FLAGS flags.DECLARE('instances_path', 'nova.compute.manager') -# Those are XenAPI related -flags.DECLARE('target_host', 'nova.virt.xenapi_conn') -FLAGS.target_host = '127.0.0.1' class LibvirtConnTestCase(test.TrialTestCase): @@ -262,74 +257,3 @@ class NWFilterTestCase(test.TrialTestCase): d.addCallback(lambda _: self.teardown_security_group()) return d - - -class XenAPIVolumeTestCase(test.TrialTestCase): - - def setUp(self): - super(XenAPIVolumeTestCase, self).setUp() - self.flags(xenapi_use_fake_session=True) - self.session = fake.FakeXenAPISession() - self.helper = volume_utils.VolumeHelper - self.helper.late_import() - - 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(context.get_admin_context(), vol) - - def test_create_iscsi_storage_raise_no_exception(self): - vol = self._create_volume() - info = yield self.helper.parse_volume_info(vol['ec2_id'], '/dev/sdc') - label = None # For testing new SRs - description = 'Test-SR' - self.session.fail_next_call = False - sr_ref = self.helper.create_iscsi_storage_blocking(self.session, - info, - label, - description) - self.assertEqual(sr_ref, self.session.SR.FAKE_REF) - db.volume_destroy(context.get_admin_context(), vol['id']) - - def test_create_iscsi_storage_raise_unable_to_create_sr_exception(self): - vol = self._create_volume() - info = yield self.helper.parse_volume_info(vol['ec2_id'], '/dev/sdc') - label = None # For testing new SRs - description = None - self.session.fail_next_call = True - self.assertRaises(volume_utils.StorageError, - self.helper.create_iscsi_storage_blocking, - self.session, - info, - label, - description) - - def test_find_sr_from_vbd_raise_no_exception(self): - sr_ref = yield self.helper.find_sr_from_vbd(self.session, - self.session.VBD.FAKE_REF) - self.assertEqual(sr_ref, self.session.SR.FAKE_REF) - - def test_destroy_iscsi_storage(self): - sr_ref = self.session.SR.FAKE_REF - self.helper.destroy_iscsi_storage_blocking(self.session, sr_ref) - - def test_introduce_vdi_raise_no_exception(self): - sr_ref = self.session.SR.FAKE_REF - self.helper.introduce_vdi_blocking(self.session, sr_ref) - - def test_introduce_vdi_raise_unable_get_vdi_record_exception(self): - sr_ref = self.session.SR.FAKE_REF - self.session.fail_next_call = True - self.assertRaises(volume_utils.StorageError, - self.helper.introduce_vdi_blocking, - self.session, sr_ref) - - def tearDown(self): - super(XenAPIVolumeTestCase, self).tearDown() -- cgit From dada56794679b213b2d80e4e1f907a212b73f54e Mon Sep 17 00:00:00 2001 From: Armando Migliaccio Date: Wed, 15 Dec 2010 17:50:05 +0000 Subject: * code cleanup * revised unittest approach * added stubout and a number of tests --- nova/tests/xenapi_unittest.py | 136 ++++++++++++++++++++++++++++++++++++++---- 1 file changed, 126 insertions(+), 10 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/xenapi_unittest.py b/nova/tests/xenapi_unittest.py index c9be17327..b9955a946 100644 --- a/nova/tests/xenapi_unittest.py +++ b/nova/tests/xenapi_unittest.py @@ -31,7 +31,7 @@ # under the License. -import mox +import stubout import uuid from twisted.internet import defer @@ -51,20 +51,34 @@ from nova.virt.xenapi import fake from nova.virt.xenapi import volume_utils from nova.virt.xenapi import vm_utils from nova.virt.xenapi import volumeops +from boto.ec2.volume import Volume FLAGS = flags.FLAGS +def stubout_session(stubs, cls): + def fake_import(self): + fake_module = 'nova.virt.xenapi.fake' + from_list = ['fake'] + return __import__(fake_module, globals(), locals(), from_list, -1) + + stubs.Set(xenapi_conn.XenAPISession, '_create_session', + lambda s, url: cls(url)) + stubs.Set(xenapi_conn.XenAPISession, 'get_imported_xenapi', + fake_import) + + class XenAPIVolumeTestCase(test.TrialTestCase): """ - This uses Ewan's fake session approach + Unit tests for VM operations """ def setUp(self): super(XenAPIVolumeTestCase, self).setUp() - FLAGS.xenapi_use_fake_session = True + self.stubs = stubout.StubOutForTesting() FLAGS.target_host = '127.0.0.1' FLAGS.xenapi_connection_url = 'test_url' FLAGS.xenapi_connection_password = 'test_pass' + fake.reset() def _create_volume(self, size='0'): """Create a volume object.""" @@ -78,11 +92,12 @@ class XenAPIVolumeTestCase(test.TrialTestCase): vol['attach_status'] = "detached" return db.volume_create(context.get_admin_context(), vol) - def test_create_iscsi_storage_raise_no_exception(self): - fake.reset() + def test_create_iscsi_storage(self): + """ This shows how to test helper classes' methods """ + stubout_session(self.stubs, FakeSessionForVolumeTests) session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') helper = volume_utils.VolumeHelper - helper.late_import(FLAGS) + helper.XenAPI = session.get_imported_xenapi() vol = self._create_volume() info = yield helper.parse_volume_info(vol['ec2_id'], '/dev/sdc') label = 'SR-%s' % vol['ec2_id'] @@ -93,8 +108,25 @@ class XenAPIVolumeTestCase(test.TrialTestCase): description) db.volume_destroy(context.get_admin_context(), vol['id']) + def test_parse_volume_info_raise_exception(self): + """ This shows how to test helper classes' methods """ + stubout_session(self.stubs, FakeSessionForVolumeTests) + session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') + helper = volume_utils.VolumeHelper + helper.XenAPI = session.get_imported_xenapi() + vol = self._create_volume() + # oops, wrong mount point! + info = helper.parse_volume_info(vol['ec2_id'], '/dev/sd') + + def check(exc): + self.assertIsInstance(exc.value, volume_utils.StorageError) + + info.addErrback(check) + db.volume_destroy(context.get_admin_context(), vol['id']) + def test_attach_volume(self): - fake.reset() + """ This shows how to test Ops classes' methods """ + stubout_session(self.stubs, FakeSessionForVolumeTests) conn = xenapi_conn.get_connection(False) volume = self._create_volume() instance = FakeInstance(1, 'fake', 'fake', 1, 2, 3, @@ -116,13 +148,34 @@ class XenAPIVolumeTestCase(test.TrialTestCase): result.addCallback(check) return result + def test_attach_volume_raise_exception(self): + """ This shows how to test when exceptions are raised """ + stubout_session(self.stubs, FakeSessionForVolumeFailedTests) + conn = xenapi_conn.get_connection(False) + volume = self._create_volume() + instance = FakeInstance(1, 'fake', 'fake', 1, 2, 3, + 'm1.large', 'aa:bb:cc:dd:ee:ff') + fake.create_vm(instance.name, 'Running') + result = conn.attach_volume(instance.name, volume['ec2_id'], + '/dev/sdc') + + def check(exc): + if exc: + pass + else: + self.fail('Oops, no exception has been raised!') + + result.addErrback(check) + return result + def tearDown(self): super(XenAPIVolumeTestCase, self).tearDown() + self.stubs.UnsetAll() class XenAPIVMTestCase(test.TrialTestCase): """ - This uses Ewan's fake session approach + Unit tests for VM operations """ def setUp(self): super(XenAPIVMTestCase, self).setUp() @@ -131,19 +184,20 @@ class XenAPIVMTestCase(test.TrialTestCase): admin=True) self.project = self.manager.create_project('fake', 'fake', 'fake') self.network = utils.import_object(FLAGS.network_manager) - FLAGS.xenapi_use_fake_session = True + self.stubs = stubout.StubOutForTesting() FLAGS.xenapi_connection_url = 'test_url' FLAGS.xenapi_connection_password = 'test_pass' fake.reset() fake.create_network('fake', FLAGS.flat_network_bridge) def test_list_instances_0(self): + stubout_session(self.stubs, FakeSessionForVMTests) conn = xenapi_conn.get_connection(False) instances = conn.list_instances() self.assertEquals(instances, []) - #test_list_instances_0.skip = "E" def test_spawn(self): + stubout_session(self.stubs, FakeSessionForVMTests) conn = xenapi_conn.get_connection(False) instance = FakeInstance(1, self.project.id, self.user.id, 1, 2, 3, 'm1.large', 'aa:bb:cc:dd:ee:ff') @@ -186,6 +240,7 @@ class XenAPIVMTestCase(test.TrialTestCase): super(XenAPIVMTestCase, self).tearDown() self.manager.delete_project(self.project) self.manager.delete_user(self.user) + self.stubs.UnsetAll() class FakeInstance(): @@ -199,3 +254,64 @@ class FakeInstance(): self.ramdisk_id = ramdisk_id self.instance_type = instance_type self.mac_address = mac_address + + +class FakeSessionForVMTests(fake.SessionBase): + def __init__(self, uri): + super(FakeSessionForVMTests, self).__init__(uri) + + def network_get_all_records_where(self, _1, _2): + return self.xenapi.network.get_all_records() + + def host_call_plugin(self, _1, _2, _3, _4, _5): + return '' + + def VM_start(self, _1, ref, _2, _3): + vm = fake.get_record('VM', ref) + if vm['power_state'] != 'Halted': + raise fake.Failure(['VM_BAD_POWER_STATE', ref, 'Halted', + vm['power_state']]) + vm['power_state'] = 'Running' + + +class FakeSessionForVolumeTests(fake.SessionBase): + def __init__(self, uri): + super(FakeSessionForVolumeTests, self).__init__(uri) + + def VBD_plug(self, _1, _2): + #FIXME(armando):make proper plug + pass + + def PBD_unplug(self, _1, _2): + #FIXME(armando):make proper unplug + pass + + def SR_forget(self, _1, _2): + #FIXME(armando):make proper forget + pass + + def VDI_introduce(self, _1, uuid, _2, _3, _4, _5, + _6, _7, _8, _9, _10, _11): + #FIXME(armando):make proper introduce + valid_vdi = False + refs = fake.get_all('VDI') + for ref in refs: + rec = fake.get_record('VDI', ref) + if rec['uuid'] == uuid: + valid_vdi = True + if not valid_vdi: + raise fake.Failure([['INVALID_VDI', 'session', self._session]]) + + +class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests): + def __init__(self, uri): + super(FakeSessionForVolumeFailedTests, self).__init__(uri) + + def VDI_introduce(self, _1, uuid, _2, _3, _4, _5, + _6, _7, _8, _9, _10, _11): + # test failure + raise fake.Failure([['INVALID_VDI', 'session', self._session]]) + + def VBD_plug(self, _1, _2): + # test failure + raise fake.Failure([['INVALID_VBD', 'session', self._session]]) -- cgit From a4db44b94e611798d57ad59f4d4dbb5fb00516db Mon Sep 17 00:00:00 2001 From: Armando Migliaccio Date: Thu, 16 Dec 2010 16:33:38 +0000 Subject: Removed FakeInstance and introduced stubout for DB. Code clean-up --- nova/tests/db/__init__.py | 33 +++++++++ nova/tests/db/fakes.py | 61 ++++++++++++++++ nova/tests/xenapi/__init__.py | 33 +++++++++ nova/tests/xenapi/stubs.py | 98 +++++++++++++++++++++++++ nova/tests/xenapi_unittest.py | 165 +++++++++++------------------------------- 5 files changed, 267 insertions(+), 123 deletions(-) create mode 100644 nova/tests/db/__init__.py create mode 100644 nova/tests/db/fakes.py create mode 100644 nova/tests/xenapi/__init__.py create mode 100644 nova/tests/xenapi/stubs.py (limited to 'nova/tests') diff --git a/nova/tests/db/__init__.py b/nova/tests/db/__init__.py new file mode 100644 index 000000000..a157d7592 --- /dev/null +++ b/nova/tests/db/__init__.py @@ -0,0 +1,33 @@ +# 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. + +# 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. + +"""Stubouts, mocks and fixtures for the test suite""" diff --git a/nova/tests/db/fakes.py b/nova/tests/db/fakes.py new file mode 100644 index 000000000..b3fb56c69 --- /dev/null +++ b/nova/tests/db/fakes.py @@ -0,0 +1,61 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 OpenStack, LLC +# All Rights Reserved. +# +# 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. + +"""Stubouts, mocks and fixtures for the test suite""" + +import time + +from nova import db +from nova import utils +from nova.compute import instance_types + + +def stub_out_db_instance_api(stubs): + """ Stubs out the db API for creating Instances """ + + class FakeInstance(object): + """ Stubs out the Instance model """ + def __init__(self, values): + self.values = values + + def __getattr__(self, name): + return self.values[name] + + def fake_create(values): + """ Stubs out the db.instance_create method """ + + type_data = instance_types.INSTANCE_TYPES[values['instance_type']] + + base_options = { + 'name': values['name'], + 'reservation_id': utils.generate_uid('r'), + 'image_id': values['image_id'], + 'kernel_id': values['kernel_id'], + 'ramdisk_id': values['ramdisk_id'], + 'state_description': 'scheduling', + 'user_id': values['user_id'], + 'project_id': values['project_id'], + 'launch_time': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), + 'instance_type': values['instance_type'], + 'memory_mb': type_data['memory_mb'], + 'mac_address': values['mac_address'], + 'vcpus': type_data['vcpus'], + 'local_gb': type_data['local_gb'], + } + return FakeInstance(base_options) + + stubs.Set(db, 'instance_create', fake_create) diff --git a/nova/tests/xenapi/__init__.py b/nova/tests/xenapi/__init__.py new file mode 100644 index 000000000..a157d7592 --- /dev/null +++ b/nova/tests/xenapi/__init__.py @@ -0,0 +1,33 @@ +# 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. + +# 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. + +"""Stubouts, mocks and fixtures for the test suite""" diff --git a/nova/tests/xenapi/stubs.py b/nova/tests/xenapi/stubs.py new file mode 100644 index 000000000..525189388 --- /dev/null +++ b/nova/tests/xenapi/stubs.py @@ -0,0 +1,98 @@ +# 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. + +"""Stubouts, mocks and fixtures for the test suite""" + +from nova.virt import xenapi_conn +from nova.virt.xenapi import fake + + +def stubout_session(stubs, cls): + """ Stubs out two methods from XenAPISession """ + def fake_import(self): + """ Stubs out get_imported_xenapi of XenAPISession """ + fake_module = 'nova.virt.xenapi.fake' + from_list = ['fake'] + return __import__(fake_module, globals(), locals(), from_list, -1) + + stubs.Set(xenapi_conn.XenAPISession, '_create_session', + lambda s, url: cls(url)) + stubs.Set(xenapi_conn.XenAPISession, 'get_imported_xenapi', + fake_import) + + +class FakeSessionForVMTests(fake.SessionBase): + """ Stubs out a XenAPISession for VM tests """ + def __init__(self, uri): + super(FakeSessionForVMTests, self).__init__(uri) + + def network_get_all_records_where(self, _1, _2): + return self.xenapi.network.get_all_records() + + def host_call_plugin(self, _1, _2, _3, _4, _5): + return '' + + def VM_start(self, _1, ref, _2, _3): + vm = fake.get_record('VM', ref) + if vm['power_state'] != 'Halted': + raise fake.Failure(['VM_BAD_POWER_STATE', ref, 'Halted', + vm['power_state']]) + vm['power_state'] = 'Running' + + +class FakeSessionForVolumeTests(fake.SessionBase): + """ Stubs out a XenAPISession for Volume tests """ + def __init__(self, uri): + super(FakeSessionForVolumeTests, self).__init__(uri) + + def VBD_plug(self, _1, _2): + #FIXME(armando):make proper plug + pass + + def PBD_unplug(self, _1, _2): + #FIXME(armando):make proper unplug + pass + + def SR_forget(self, _1, _2): + #FIXME(armando):make proper forget + pass + + def VDI_introduce(self, _1, uuid, _2, _3, _4, _5, + _6, _7, _8, _9, _10, _11): + #FIXME(armando):make proper introduce + valid_vdi = False + refs = fake.get_all('VDI') + for ref in refs: + rec = fake.get_record('VDI', ref) + if rec['uuid'] == uuid: + valid_vdi = True + if not valid_vdi: + raise fake.Failure([['INVALID_VDI', 'session', self._session]]) + + +class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests): + """ Stubs out a XenAPISession for Volume tests: it injects failures """ + def __init__(self, uri): + super(FakeSessionForVolumeFailedTests, self).__init__(uri) + + def VDI_introduce(self, _1, uuid, _2, _3, _4, _5, + _6, _7, _8, _9, _10, _11): + # This is for testing failure + raise fake.Failure([['INVALID_VDI', 'session', self._session]]) + + def VBD_plug(self, _1, _2): + # This is for testing failure + raise fake.Failure([['INVALID_VBD', 'session', self._session]]) diff --git a/nova/tests/xenapi_unittest.py b/nova/tests/xenapi_unittest.py index b9955a946..c2612a4c5 100644 --- a/nova/tests/xenapi_unittest.py +++ b/nova/tests/xenapi_unittest.py @@ -1,21 +1,5 @@ # 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. - -# 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 @@ -30,16 +14,14 @@ # License for the specific language governing permissions and limitations # under the License. +""" +Test suite for XenAPI +""" import stubout -import uuid - -from twisted.internet import defer -from twisted.internet import threads from nova import db from nova import context -from nova import exception from nova import flags from nova import test from nova import utils @@ -49,28 +31,15 @@ from nova.compute import power_state from nova.virt import xenapi_conn from nova.virt.xenapi import fake from nova.virt.xenapi import volume_utils -from nova.virt.xenapi import vm_utils -from nova.virt.xenapi import volumeops -from boto.ec2.volume import Volume +from nova.tests.db import fakes +from nova.tests.xenapi import stubs FLAGS = flags.FLAGS -def stubout_session(stubs, cls): - def fake_import(self): - fake_module = 'nova.virt.xenapi.fake' - from_list = ['fake'] - return __import__(fake_module, globals(), locals(), from_list, -1) - - stubs.Set(xenapi_conn.XenAPISession, '_create_session', - lambda s, url: cls(url)) - stubs.Set(xenapi_conn.XenAPISession, 'get_imported_xenapi', - fake_import) - - class XenAPIVolumeTestCase(test.TrialTestCase): """ - Unit tests for VM operations + Unit tests for Volume operations """ def setUp(self): super(XenAPIVolumeTestCase, self).setUp() @@ -78,7 +47,17 @@ class XenAPIVolumeTestCase(test.TrialTestCase): FLAGS.target_host = '127.0.0.1' FLAGS.xenapi_connection_url = 'test_url' FLAGS.xenapi_connection_password = 'test_pass' + fakes.stub_out_db_instance_api(self.stubs) fake.reset() + self.values = {'name': 1, + 'project_id': 'fake', + 'user_id': 'fake', + 'image_id': 1, + 'kernel_id': 2, + 'ramdisk_id': 3, + 'instance_type': 'm1.large', + 'mac_address': 'aa:bb:cc:dd:ee:ff', + } def _create_volume(self, size='0'): """Create a volume object.""" @@ -94,7 +73,7 @@ class XenAPIVolumeTestCase(test.TrialTestCase): def test_create_iscsi_storage(self): """ This shows how to test helper classes' methods """ - stubout_session(self.stubs, FakeSessionForVolumeTests) + stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests) session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') helper = volume_utils.VolumeHelper helper.XenAPI = session.get_imported_xenapi() @@ -106,11 +85,13 @@ class XenAPIVolumeTestCase(test.TrialTestCase): info, label, description) + srs = fake.get_all('SR') + self.assertEqual(sr_ref, srs[0]) db.volume_destroy(context.get_admin_context(), vol['id']) def test_parse_volume_info_raise_exception(self): """ This shows how to test helper classes' methods """ - stubout_session(self.stubs, FakeSessionForVolumeTests) + stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests) session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') helper = volume_utils.VolumeHelper helper.XenAPI = session.get_imported_xenapi() @@ -119,6 +100,7 @@ class XenAPIVolumeTestCase(test.TrialTestCase): info = helper.parse_volume_info(vol['ec2_id'], '/dev/sd') def check(exc): + """ handler """ self.assertIsInstance(exc.value, volume_utils.StorageError) info.addErrback(check) @@ -126,16 +108,16 @@ class XenAPIVolumeTestCase(test.TrialTestCase): def test_attach_volume(self): """ This shows how to test Ops classes' methods """ - stubout_session(self.stubs, FakeSessionForVolumeTests) + stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests) conn = xenapi_conn.get_connection(False) volume = self._create_volume() - instance = FakeInstance(1, 'fake', 'fake', 1, 2, 3, - 'm1.large', 'aa:bb:cc:dd:ee:ff') + instance = db.instance_create(self.values) fake.create_vm(instance.name, 'Running') result = conn.attach_volume(instance.name, volume['ec2_id'], '/dev/sdc') def check(_): + """ handler """ # check that the VM has a VBD attached to it # Get XenAPI reference for the VM vms = fake.get_all('VM') @@ -150,16 +132,17 @@ class XenAPIVolumeTestCase(test.TrialTestCase): def test_attach_volume_raise_exception(self): """ This shows how to test when exceptions are raised """ - stubout_session(self.stubs, FakeSessionForVolumeFailedTests) + stubs.stubout_session(self.stubs, + stubs.FakeSessionForVolumeFailedTests) conn = xenapi_conn.get_connection(False) volume = self._create_volume() - instance = FakeInstance(1, 'fake', 'fake', 1, 2, 3, - 'm1.large', 'aa:bb:cc:dd:ee:ff') + instance = db.instance_create(self.values) fake.create_vm(instance.name, 'Running') result = conn.attach_volume(instance.name, volume['ec2_id'], '/dev/sdc') def check(exc): + """ handler """ if exc: pass else: @@ -188,22 +171,32 @@ class XenAPIVMTestCase(test.TrialTestCase): FLAGS.xenapi_connection_url = 'test_url' FLAGS.xenapi_connection_password = 'test_pass' fake.reset() + fakes.stub_out_db_instance_api(self.stubs) fake.create_network('fake', FLAGS.flat_network_bridge) def test_list_instances_0(self): - stubout_session(self.stubs, FakeSessionForVMTests) + stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests) conn = xenapi_conn.get_connection(False) instances = conn.list_instances() self.assertEquals(instances, []) def test_spawn(self): - stubout_session(self.stubs, FakeSessionForVMTests) + stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests) + values = {'name': 1, + 'project_id': self.project.id, + 'user_id': self.user.id, + 'image_id': 1, + 'kernel_id': 2, + 'ramdisk_id': 3, + 'instance_type': 'm1.large', + 'mac_address': 'aa:bb:cc:dd:ee:ff', + } conn = xenapi_conn.get_connection(False) - instance = FakeInstance(1, self.project.id, self.user.id, 1, 2, 3, - 'm1.large', 'aa:bb:cc:dd:ee:ff') + instance = db.instance_create(values) result = conn.spawn(instance) def check(_): + """ handler """ instances = conn.list_instances() self.assertEquals(instances, [1]) @@ -241,77 +234,3 @@ class XenAPIVMTestCase(test.TrialTestCase): self.manager.delete_project(self.project) self.manager.delete_user(self.user) self.stubs.UnsetAll() - - -class FakeInstance(): - def __init__(self, name, project_id, user_id, image_id, kernel_id, - ramdisk_id, instance_type, mac_address): - self.name = name - self.project_id = project_id - self.user_id = user_id - self.image_id = image_id - self.kernel_id = kernel_id - self.ramdisk_id = ramdisk_id - self.instance_type = instance_type - self.mac_address = mac_address - - -class FakeSessionForVMTests(fake.SessionBase): - def __init__(self, uri): - super(FakeSessionForVMTests, self).__init__(uri) - - def network_get_all_records_where(self, _1, _2): - return self.xenapi.network.get_all_records() - - def host_call_plugin(self, _1, _2, _3, _4, _5): - return '' - - def VM_start(self, _1, ref, _2, _3): - vm = fake.get_record('VM', ref) - if vm['power_state'] != 'Halted': - raise fake.Failure(['VM_BAD_POWER_STATE', ref, 'Halted', - vm['power_state']]) - vm['power_state'] = 'Running' - - -class FakeSessionForVolumeTests(fake.SessionBase): - def __init__(self, uri): - super(FakeSessionForVolumeTests, self).__init__(uri) - - def VBD_plug(self, _1, _2): - #FIXME(armando):make proper plug - pass - - def PBD_unplug(self, _1, _2): - #FIXME(armando):make proper unplug - pass - - def SR_forget(self, _1, _2): - #FIXME(armando):make proper forget - pass - - def VDI_introduce(self, _1, uuid, _2, _3, _4, _5, - _6, _7, _8, _9, _10, _11): - #FIXME(armando):make proper introduce - valid_vdi = False - refs = fake.get_all('VDI') - for ref in refs: - rec = fake.get_record('VDI', ref) - if rec['uuid'] == uuid: - valid_vdi = True - if not valid_vdi: - raise fake.Failure([['INVALID_VDI', 'session', self._session]]) - - -class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests): - def __init__(self, uri): - super(FakeSessionForVolumeFailedTests, self).__init__(uri) - - def VDI_introduce(self, _1, uuid, _2, _3, _4, _5, - _6, _7, _8, _9, _10, _11): - # test failure - raise fake.Failure([['INVALID_VDI', 'session', self._session]]) - - def VBD_plug(self, _1, _2): - # test failure - raise fake.Failure([['INVALID_VBD', 'session', self._session]]) -- cgit From 8152acf7c3df83a04591fdafb21201965da7bfad Mon Sep 17 00:00:00 2001 From: Armando Migliaccio Date: Thu, 16 Dec 2010 17:47:48 +0000 Subject: fake session clean-up --- nova/tests/db/__init__.py | 16 ---------------- nova/tests/xenapi/__init__.py | 16 ---------------- 2 files changed, 32 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/db/__init__.py b/nova/tests/db/__init__.py index a157d7592..dcd81b743 100644 --- a/nova/tests/db/__init__.py +++ b/nova/tests/db/__init__.py @@ -1,21 +1,5 @@ # 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. - -# 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 diff --git a/nova/tests/xenapi/__init__.py b/nova/tests/xenapi/__init__.py index a157d7592..dcd81b743 100644 --- a/nova/tests/xenapi/__init__.py +++ b/nova/tests/xenapi/__init__.py @@ -1,21 +1,5 @@ # 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. - -# 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 -- cgit From db96fd559d28bcfdf8cc29d79b9afca6dea1cfb7 Mon Sep 17 00:00:00 2001 From: Armando Migliaccio Date: Thu, 16 Dec 2010 18:44:42 +0000 Subject: reviewed the FIXMEs, and spotted an uncaught exception in volume_utils...yay! --- nova/tests/virt_unittest.py | 2 -- nova/tests/xenapi/stubs.py | 24 +++++++++--------------- nova/tests/xenapi_unittest.py | 5 ++--- 3 files changed, 11 insertions(+), 20 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/virt_unittest.py b/nova/tests/virt_unittest.py index 52843b703..d49383fb7 100644 --- a/nova/tests/virt_unittest.py +++ b/nova/tests/virt_unittest.py @@ -25,8 +25,6 @@ from nova import utils from nova.api.ec2 import cloud from nova.auth import manager from nova.virt import libvirt_conn -from nova.virt.xenapi import fake -from nova.virt.xenapi import volume_utils FLAGS = flags.FLAGS flags.DECLARE('instances_path', 'nova.compute.manager') diff --git a/nova/tests/xenapi/stubs.py b/nova/tests/xenapi/stubs.py index 525189388..11dd535d4 100644 --- a/nova/tests/xenapi/stubs.py +++ b/nova/tests/xenapi/stubs.py @@ -58,21 +58,12 @@ class FakeSessionForVolumeTests(fake.SessionBase): def __init__(self, uri): super(FakeSessionForVolumeTests, self).__init__(uri) - def VBD_plug(self, _1, _2): - #FIXME(armando):make proper plug - pass - - def PBD_unplug(self, _1, _2): - #FIXME(armando):make proper unplug - pass - - def SR_forget(self, _1, _2): - #FIXME(armando):make proper forget - pass + def VBD_plug(self, _1, ref): + rec = fake.get_record('VBD', ref) + rec['currently-attached'] = True def VDI_introduce(self, _1, uuid, _2, _3, _4, _5, _6, _7, _8, _9, _10, _11): - #FIXME(armando):make proper introduce valid_vdi = False refs = fake.get_all('VDI') for ref in refs: @@ -93,6 +84,9 @@ class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests): # This is for testing failure raise fake.Failure([['INVALID_VDI', 'session', self._session]]) - def VBD_plug(self, _1, _2): - # This is for testing failure - raise fake.Failure([['INVALID_VBD', 'session', self._session]]) + def PBD_unplug(self, _1, ref): + rec = fake.get_record('PBD', ref) + rec['currently-attached'] = False + + def SR_forget(self, _1, ref): + pass diff --git a/nova/tests/xenapi_unittest.py b/nova/tests/xenapi_unittest.py index c2612a4c5..839d6aa44 100644 --- a/nova/tests/xenapi_unittest.py +++ b/nova/tests/xenapi_unittest.py @@ -141,14 +141,13 @@ class XenAPIVolumeTestCase(test.TrialTestCase): result = conn.attach_volume(instance.name, volume['ec2_id'], '/dev/sdc') - def check(exc): + def check_exception(exc): """ handler """ if exc: pass else: self.fail('Oops, no exception has been raised!') - - result.addErrback(check) + result.addErrback(check_exception) return result def tearDown(self): -- cgit From 8c343e1b4b92aa7b1062acebe8eaea402bc6ab4a Mon Sep 17 00:00:00 2001 From: Eric Day Date: Thu, 16 Dec 2010 17:05:54 -0800 Subject: First pass at converting run_tests.py to nosetests. The network and objctstore tests don't yet work. Also, we need to manually remove the sqlite file between runs. --- nova/tests/access_unittest.py | 127 -------------- nova/tests/api/__init__.py | 81 --------- nova/tests/api/test.py | 81 +++++++++ nova/tests/api_integration.py | 54 ------ nova/tests/api_unittest.py | 338 ------------------------------------- nova/tests/auth_unittest.py | 352 --------------------------------------- nova/tests/cloud_unittest.py | 332 ------------------------------------ nova/tests/compute_unittest.py | 155 ----------------- nova/tests/flags_unittest.py | 102 ------------ nova/tests/misc_unittest.py | 55 ------ nova/tests/quota_unittest.py | 153 ----------------- nova/tests/rpc_unittest.py | 103 ------------ nova/tests/scheduler_unittest.py | 246 --------------------------- nova/tests/service_unittest.py | 227 ------------------------- nova/tests/test_access.py | 127 ++++++++++++++ nova/tests/test_api.py | 338 +++++++++++++++++++++++++++++++++++++ nova/tests/test_auth.py | 352 +++++++++++++++++++++++++++++++++++++++ nova/tests/test_cloud.py | 332 ++++++++++++++++++++++++++++++++++++ nova/tests/test_compute.py | 155 +++++++++++++++++ nova/tests/test_flags.py | 102 ++++++++++++ nova/tests/test_misc.py | 55 ++++++ nova/tests/test_quota.py | 153 +++++++++++++++++ nova/tests/test_rpc.py | 103 ++++++++++++ nova/tests/test_scheduler.py | 246 +++++++++++++++++++++++++++ nova/tests/test_service.py | 227 +++++++++++++++++++++++++ nova/tests/test_twistd.py | 53 ++++++ nova/tests/test_virt.py | 258 ++++++++++++++++++++++++++++ nova/tests/test_volume.py | 175 +++++++++++++++++++ nova/tests/twistd_unittest.py | 53 ------ nova/tests/virt_unittest.py | 258 ---------------------------- nova/tests/volume_unittest.py | 175 ------------------- 31 files changed, 2757 insertions(+), 2811 deletions(-) delete mode 100644 nova/tests/access_unittest.py create mode 100644 nova/tests/api/test.py delete mode 100644 nova/tests/api_integration.py delete mode 100644 nova/tests/api_unittest.py delete mode 100644 nova/tests/auth_unittest.py delete mode 100644 nova/tests/cloud_unittest.py delete mode 100644 nova/tests/compute_unittest.py delete mode 100644 nova/tests/flags_unittest.py delete mode 100644 nova/tests/misc_unittest.py delete mode 100644 nova/tests/quota_unittest.py delete mode 100644 nova/tests/rpc_unittest.py delete mode 100644 nova/tests/scheduler_unittest.py delete mode 100644 nova/tests/service_unittest.py create mode 100644 nova/tests/test_access.py create mode 100644 nova/tests/test_api.py create mode 100644 nova/tests/test_auth.py create mode 100644 nova/tests/test_cloud.py create mode 100644 nova/tests/test_compute.py create mode 100644 nova/tests/test_flags.py create mode 100644 nova/tests/test_misc.py create mode 100644 nova/tests/test_quota.py create mode 100644 nova/tests/test_rpc.py create mode 100644 nova/tests/test_scheduler.py create mode 100644 nova/tests/test_service.py create mode 100644 nova/tests/test_twistd.py create mode 100644 nova/tests/test_virt.py create mode 100644 nova/tests/test_volume.py delete mode 100644 nova/tests/twistd_unittest.py delete mode 100644 nova/tests/virt_unittest.py delete mode 100644 nova/tests/volume_unittest.py (limited to 'nova/tests') diff --git a/nova/tests/access_unittest.py b/nova/tests/access_unittest.py deleted file mode 100644 index 58fdea3b5..000000000 --- a/nova/tests/access_unittest.py +++ /dev/null @@ -1,127 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. - -import unittest -import logging -import webob - -from nova import context -from nova import exception -from nova import flags -from nova import test -from nova.api import ec2 -from nova.auth import manager - - -FLAGS = flags.FLAGS - - -class Context(object): - pass - - -class AccessTestCase(test.TestCase): - def setUp(self): - super(AccessTestCase, self).setUp() - um = manager.AuthManager() - self.context = context.get_admin_context() - # Make test users - self.testadmin = um.create_user('testadmin') - self.testpmsys = um.create_user('testpmsys') - self.testnet = um.create_user('testnet') - self.testsys = um.create_user('testsys') - # Assign some rules - um.add_role('testadmin', 'cloudadmin') - um.add_role('testpmsys', 'sysadmin') - um.add_role('testnet', 'netadmin') - um.add_role('testsys', 'sysadmin') - - # Make a test project - self.project = um.create_project('testproj', - 'testpmsys', - 'a test project', - ['testpmsys', 'testnet', 'testsys']) - self.project.add_role(self.testnet, 'netadmin') - self.project.add_role(self.testsys, 'sysadmin') - #user is set in each test - - def noopWSGIApp(environ, start_response): - start_response('200 OK', []) - return [''] - - self.mw = ec2.Authorizer(noopWSGIApp) - self.mw.action_roles = {'str': { - '_allow_all': ['all'], - '_allow_none': [], - '_allow_project_manager': ['projectmanager'], - '_allow_sys_and_net': ['sysadmin', 'netadmin'], - '_allow_sysadmin': ['sysadmin']}} - - def tearDown(self): - um = manager.AuthManager() - # Delete the test project - um.delete_project('testproj') - # Delete the test user - um.delete_user('testadmin') - um.delete_user('testpmsys') - um.delete_user('testnet') - um.delete_user('testsys') - super(AccessTestCase, self).tearDown() - - def response_status(self, user, methodName): - ctxt = context.RequestContext(user, self.project) - environ = {'ec2.context': ctxt, - 'ec2.controller': 'some string', - 'ec2.action': methodName} - req = webob.Request.blank('/', environ) - resp = req.get_response(self.mw) - return resp.status_int - - def shouldAllow(self, user, methodName): - self.assertEqual(200, self.response_status(user, methodName)) - - def shouldDeny(self, user, methodName): - self.assertEqual(401, self.response_status(user, methodName)) - - def test_001_allow_all(self): - users = [self.testadmin, self.testpmsys, self.testnet, self.testsys] - for user in users: - self.shouldAllow(user, '_allow_all') - - def test_002_allow_none(self): - self.shouldAllow(self.testadmin, '_allow_none') - users = [self.testpmsys, self.testnet, self.testsys] - for user in users: - self.shouldDeny(user, '_allow_none') - - def test_003_allow_project_manager(self): - for user in [self.testadmin, self.testpmsys]: - self.shouldAllow(user, '_allow_project_manager') - for user in [self.testnet, self.testsys]: - self.shouldDeny(user, '_allow_project_manager') - - def test_004_allow_sys_and_net(self): - for user in [self.testadmin, self.testnet, self.testsys]: - self.shouldAllow(user, '_allow_sys_and_net') - # denied because it doesn't have the per project sysadmin - for user in [self.testpmsys]: - self.shouldDeny(user, '_allow_sys_and_net') - -if __name__ == "__main__": - # TODO: Implement use_fake as an option - unittest.main() diff --git a/nova/tests/api/__init__.py b/nova/tests/api/__init__.py index 9caa8c9d0..e69de29bb 100644 --- a/nova/tests/api/__init__.py +++ b/nova/tests/api/__init__.py @@ -1,81 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 OpenStack LLC. -# All Rights Reserved. -# -# 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 for the root WSGI middleware for all API controllers. -""" - -import unittest - -import stubout -import webob -import webob.dec - -import nova.exception -from nova import api -from nova.tests.api.fakes import APIStub - - -class Test(unittest.TestCase): - - def setUp(self): - self.stubs = stubout.StubOutForTesting() - - def tearDown(self): - self.stubs.UnsetAll() - - def _request(self, url, subdomain, **kwargs): - environ_keys = {'HTTP_HOST': '%s.example.com' % subdomain} - environ_keys.update(kwargs) - req = webob.Request.blank(url, environ_keys) - return req.get_response(api.API('ec2')) - - def test_openstack(self): - self.stubs.Set(api.openstack, 'API', APIStub) - result = self._request('/v1.0/cloud', 'api') - self.assertEqual(result.body, "/cloud") - - def test_ec2(self): - self.stubs.Set(api.ec2, 'API', APIStub) - result = self._request('/services/cloud', 'ec2') - self.assertEqual(result.body, "/cloud") - - def test_not_found(self): - self.stubs.Set(api.ec2, 'API', APIStub) - self.stubs.Set(api.openstack, 'API', APIStub) - result = self._request('/test/cloud', 'ec2') - self.assertNotEqual(result.body, "/cloud") - - def test_query_api_versions(self): - result = self._request('/', 'api') - self.assertTrue('CURRENT' in result.body) - - def test_metadata(self): - def go(url): - result = self._request(url, 'ec2', REMOTE_ADDR='128.192.151.2') - # Each should get to the ORM layer and fail to find the IP - self.assertRaises(nova.exception.NotFound, go, '/latest/') - self.assertRaises(nova.exception.NotFound, go, '/2009-04-04/') - self.assertRaises(nova.exception.NotFound, go, '/1.0/') - - def test_ec2_root(self): - result = self._request('/', 'ec2') - self.assertTrue('2007-12-15\n' in result.body) - - -if __name__ == '__main__': - unittest.main() diff --git a/nova/tests/api/test.py b/nova/tests/api/test.py new file mode 100644 index 000000000..9caa8c9d0 --- /dev/null +++ b/nova/tests/api/test.py @@ -0,0 +1,81 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 OpenStack LLC. +# All Rights Reserved. +# +# 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 for the root WSGI middleware for all API controllers. +""" + +import unittest + +import stubout +import webob +import webob.dec + +import nova.exception +from nova import api +from nova.tests.api.fakes import APIStub + + +class Test(unittest.TestCase): + + def setUp(self): + self.stubs = stubout.StubOutForTesting() + + def tearDown(self): + self.stubs.UnsetAll() + + def _request(self, url, subdomain, **kwargs): + environ_keys = {'HTTP_HOST': '%s.example.com' % subdomain} + environ_keys.update(kwargs) + req = webob.Request.blank(url, environ_keys) + return req.get_response(api.API('ec2')) + + def test_openstack(self): + self.stubs.Set(api.openstack, 'API', APIStub) + result = self._request('/v1.0/cloud', 'api') + self.assertEqual(result.body, "/cloud") + + def test_ec2(self): + self.stubs.Set(api.ec2, 'API', APIStub) + result = self._request('/services/cloud', 'ec2') + self.assertEqual(result.body, "/cloud") + + def test_not_found(self): + self.stubs.Set(api.ec2, 'API', APIStub) + self.stubs.Set(api.openstack, 'API', APIStub) + result = self._request('/test/cloud', 'ec2') + self.assertNotEqual(result.body, "/cloud") + + def test_query_api_versions(self): + result = self._request('/', 'api') + self.assertTrue('CURRENT' in result.body) + + def test_metadata(self): + def go(url): + result = self._request(url, 'ec2', REMOTE_ADDR='128.192.151.2') + # Each should get to the ORM layer and fail to find the IP + self.assertRaises(nova.exception.NotFound, go, '/latest/') + self.assertRaises(nova.exception.NotFound, go, '/2009-04-04/') + self.assertRaises(nova.exception.NotFound, go, '/1.0/') + + def test_ec2_root(self): + result = self._request('/', 'ec2') + self.assertTrue('2007-12-15\n' in result.body) + + +if __name__ == '__main__': + unittest.main() diff --git a/nova/tests/api_integration.py b/nova/tests/api_integration.py deleted file mode 100644 index 54403c655..000000000 --- a/nova/tests/api_integration.py +++ /dev/null @@ -1,54 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. - - -import boto -from boto.ec2.regioninfo import RegionInfo -import unittest - - -ACCESS_KEY = 'fake' -SECRET_KEY = 'fake' -CLC_IP = '127.0.0.1' -CLC_PORT = 8773 -REGION = 'test' - - -def get_connection(): - return boto.connect_ec2( - aws_access_key_id=ACCESS_KEY, - aws_secret_access_key=SECRET_KEY, - is_secure=False, - region=RegionInfo(None, REGION, CLC_IP), - port=CLC_PORT, - path='/services/Cloud', - debug=99) - - -class APIIntegrationTests(unittest.TestCase): - def test_001_get_all_images(self): - conn = get_connection() - res = conn.get_all_images() - - -if __name__ == '__main__': - unittest.main() - -#print conn.get_all_key_pairs() -#print conn.create_key_pair -#print conn.create_security_group('name', 'description') diff --git a/nova/tests/api_unittest.py b/nova/tests/api_unittest.py deleted file mode 100644 index 33d4cb294..000000000 --- a/nova/tests/api_unittest.py +++ /dev/null @@ -1,338 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. - -"""Unit tests for the API endpoint""" - -import boto -from boto.ec2 import regioninfo -import httplib -import random -import StringIO -import webob - -from nova import context -from nova import flags -from nova import test -from nova import api -from nova.api.ec2 import cloud -from nova.api.ec2 import apirequest -from nova.auth import manager - - -class FakeHttplibSocket(object): - """a fake socket implementation for httplib.HTTPResponse, trivial""" - def __init__(self, response_string): - self._buffer = StringIO.StringIO(response_string) - - def makefile(self, _mode, _other): - """Returns the socket's internal buffer""" - return self._buffer - - -class FakeHttplibConnection(object): - """A fake httplib.HTTPConnection for boto to use - - requests made via this connection actually get translated and routed into - our WSGI app, we then wait for the response and turn it back into - the httplib.HTTPResponse that boto expects. - """ - def __init__(self, app, host, is_secure=False): - self.app = app - self.host = host - - def request(self, method, path, data, headers): - req = webob.Request.blank(path) - req.method = method - req.body = data - req.headers = headers - req.headers['Accept'] = 'text/html' - req.host = self.host - # Call the WSGI app, get the HTTP response - resp = str(req.get_response(self.app)) - # For some reason, the response doesn't have "HTTP/1.0 " prepended; I - # guess that's a function the web server usually provides. - resp = "HTTP/1.0 %s" % resp - sock = FakeHttplibSocket(resp) - self.http_response = httplib.HTTPResponse(sock) - self.http_response.begin() - - def getresponse(self): - return self.http_response - - def close(self): - """Required for compatibility with boto/tornado""" - pass - - -class XmlConversionTestCase(test.TrialTestCase): - """Unit test api xml conversion""" - def test_number_conversion(self): - conv = apirequest._try_convert - self.assertEqual(conv('None'), None) - self.assertEqual(conv('True'), True) - self.assertEqual(conv('False'), False) - self.assertEqual(conv('0'), 0) - self.assertEqual(conv('42'), 42) - self.assertEqual(conv('3.14'), 3.14) - self.assertEqual(conv('-57.12'), -57.12) - self.assertEqual(conv('0x57'), 0x57) - self.assertEqual(conv('-0x57'), -0x57) - self.assertEqual(conv('-'), '-') - self.assertEqual(conv('-0'), 0) - - -class ApiEc2TestCase(test.TrialTestCase): - """Unit test for the cloud controller on an EC2 API""" - def setUp(self): - super(ApiEc2TestCase, self).setUp() - - self.manager = manager.AuthManager() - - self.host = '127.0.0.1' - - self.app = api.API('ec2') - - def expect_http(self, host=None, is_secure=False): - """Returns a new EC2 connection""" - self.ec2 = boto.connect_ec2( - aws_access_key_id='fake', - aws_secret_access_key='fake', - is_secure=False, - region=regioninfo.RegionInfo(None, 'test', self.host), - port=8773, - path='/services/Cloud') - - self.mox.StubOutWithMock(self.ec2, 'new_http_connection') - http = FakeHttplibConnection( - self.app, '%s:8773' % (self.host), False) - # pylint: disable-msg=E1103 - self.ec2.new_http_connection(host, is_secure).AndReturn(http) - return http - - def test_describe_instances(self): - """Test that, after creating a user and a project, the describe - instances call to the API works properly""" - self.expect_http() - self.mox.ReplayAll() - user = self.manager.create_user('fake', 'fake', 'fake') - project = self.manager.create_project('fake', 'fake', 'fake') - self.assertEqual(self.ec2.get_all_instances(), []) - self.manager.delete_project(project) - self.manager.delete_user(user) - - def test_get_all_key_pairs(self): - """Test that, after creating a user and project and generating - a key pair, that the API call to list key pairs works properly""" - self.expect_http() - self.mox.ReplayAll() - keyname = "".join(random.choice("sdiuisudfsdcnpaqwertasd") \ - for x in range(random.randint(4, 8))) - user = self.manager.create_user('fake', 'fake', 'fake') - project = self.manager.create_project('fake', 'fake', 'fake') - # NOTE(vish): create depends on pool, so call helper directly - cloud._gen_key(context.get_admin_context(), user.id, keyname) - - rv = self.ec2.get_all_key_pairs() - results = [k for k in rv if k.name == keyname] - self.assertEquals(len(results), 1) - self.manager.delete_project(project) - self.manager.delete_user(user) - - def test_get_all_security_groups(self): - """Test that we can retrieve security groups""" - self.expect_http() - self.mox.ReplayAll() - user = self.manager.create_user('fake', 'fake', 'fake', admin=True) - project = self.manager.create_project('fake', 'fake', 'fake') - - rv = self.ec2.get_all_security_groups() - - self.assertEquals(len(rv), 1) - self.assertEquals(rv[0].name, 'default') - - self.manager.delete_project(project) - self.manager.delete_user(user) - - def test_create_delete_security_group(self): - """Test that we can create a security group""" - self.expect_http() - self.mox.ReplayAll() - user = self.manager.create_user('fake', 'fake', 'fake', admin=True) - project = self.manager.create_project('fake', 'fake', 'fake') - - # At the moment, you need both of these to actually be netadmin - self.manager.add_role('fake', 'netadmin') - project.add_role('fake', 'netadmin') - - security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd") - for x in range(random.randint(4, 8))) - - self.ec2.create_security_group(security_group_name, 'test group') - - self.expect_http() - self.mox.ReplayAll() - - rv = self.ec2.get_all_security_groups() - self.assertEquals(len(rv), 2) - self.assertTrue(security_group_name in [group.name for group in rv]) - - self.expect_http() - self.mox.ReplayAll() - - self.ec2.delete_security_group(security_group_name) - - self.manager.delete_project(project) - self.manager.delete_user(user) - - def test_authorize_revoke_security_group_cidr(self): - """ - Test that we can add and remove CIDR based rules - to a security group - """ - self.expect_http() - self.mox.ReplayAll() - user = self.manager.create_user('fake', 'fake', 'fake') - project = self.manager.create_project('fake', 'fake', 'fake') - - # At the moment, you need both of these to actually be netadmin - self.manager.add_role('fake', 'netadmin') - project.add_role('fake', 'netadmin') - - security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd") - for x in range(random.randint(4, 8))) - - group = self.ec2.create_security_group(security_group_name, - 'test group') - - self.expect_http() - self.mox.ReplayAll() - group.connection = self.ec2 - - group.authorize('tcp', 80, 81, '0.0.0.0/0') - - self.expect_http() - self.mox.ReplayAll() - - rv = self.ec2.get_all_security_groups() - # I don't bother checkng that we actually find it here, - # because the create/delete unit test further up should - # be good enough for that. - for group in rv: - if group.name == security_group_name: - self.assertEquals(len(group.rules), 1) - self.assertEquals(int(group.rules[0].from_port), 80) - self.assertEquals(int(group.rules[0].to_port), 81) - self.assertEquals(len(group.rules[0].grants), 1) - self.assertEquals(str(group.rules[0].grants[0]), '0.0.0.0/0') - - self.expect_http() - self.mox.ReplayAll() - group.connection = self.ec2 - - group.revoke('tcp', 80, 81, '0.0.0.0/0') - - self.expect_http() - self.mox.ReplayAll() - - self.ec2.delete_security_group(security_group_name) - - self.expect_http() - self.mox.ReplayAll() - group.connection = self.ec2 - - rv = self.ec2.get_all_security_groups() - - self.assertEqual(len(rv), 1) - self.assertEqual(rv[0].name, 'default') - - self.manager.delete_project(project) - self.manager.delete_user(user) - - return - - def test_authorize_revoke_security_group_foreign_group(self): - """ - Test that we can grant and revoke another security group access - to a security group - """ - self.expect_http() - self.mox.ReplayAll() - user = self.manager.create_user('fake', 'fake', 'fake', admin=True) - project = self.manager.create_project('fake', 'fake', 'fake') - - # At the moment, you need both of these to actually be netadmin - self.manager.add_role('fake', 'netadmin') - project.add_role('fake', 'netadmin') - - rand_string = 'sdiuisudfsdcnpaqwertasd' - security_group_name = "".join(random.choice(rand_string) - for x in range(random.randint(4, 8))) - other_security_group_name = "".join(random.choice(rand_string) - for x in range(random.randint(4, 8))) - - group = self.ec2.create_security_group(security_group_name, - 'test group') - - self.expect_http() - self.mox.ReplayAll() - - other_group = self.ec2.create_security_group(other_security_group_name, - 'some other group') - - self.expect_http() - self.mox.ReplayAll() - group.connection = self.ec2 - - group.authorize(src_group=other_group) - - self.expect_http() - self.mox.ReplayAll() - - rv = self.ec2.get_all_security_groups() - - # I don't bother checkng that we actually find it here, - # because the create/delete unit test further up should - # be good enough for that. - for group in rv: - if group.name == security_group_name: - self.assertEquals(len(group.rules), 1) - self.assertEquals(len(group.rules[0].grants), 1) - self.assertEquals(str(group.rules[0].grants[0]), '%s-%s' % - (other_security_group_name, 'fake')) - - self.expect_http() - self.mox.ReplayAll() - - rv = self.ec2.get_all_security_groups() - - for group in rv: - if group.name == security_group_name: - self.expect_http() - self.mox.ReplayAll() - group.connection = self.ec2 - group.revoke(src_group=other_group) - - self.expect_http() - self.mox.ReplayAll() - - self.ec2.delete_security_group(security_group_name) - - self.manager.delete_project(project) - self.manager.delete_user(user) - - return diff --git a/nova/tests/auth_unittest.py b/nova/tests/auth_unittest.py deleted file mode 100644 index 4508d6721..000000000 --- a/nova/tests/auth_unittest.py +++ /dev/null @@ -1,352 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. - -import logging -from M2Crypto import X509 -import unittest - -from nova import crypto -from nova import flags -from nova import test -from nova.auth import manager -from nova.api.ec2 import cloud - -FLAGS = flags.FLAGS - - -class user_generator(object): - def __init__(self, manager, **user_state): - if 'name' not in user_state: - user_state['name'] = 'test1' - self.manager = manager - self.user = manager.create_user(**user_state) - - def __enter__(self): - return self.user - - def __exit__(self, value, type, trace): - self.manager.delete_user(self.user) - - -class project_generator(object): - def __init__(self, manager, **project_state): - if 'name' not in project_state: - project_state['name'] = 'testproj' - if 'manager_user' not in project_state: - project_state['manager_user'] = 'test1' - self.manager = manager - self.project = manager.create_project(**project_state) - - def __enter__(self): - return self.project - - def __exit__(self, value, type, trace): - self.manager.delete_project(self.project) - - -class user_and_project_generator(object): - def __init__(self, manager, user_state={}, project_state={}): - self.manager = manager - if 'name' not in user_state: - user_state['name'] = 'test1' - if 'name' not in project_state: - project_state['name'] = 'testproj' - if 'manager_user' not in project_state: - project_state['manager_user'] = 'test1' - self.user = manager.create_user(**user_state) - self.project = manager.create_project(**project_state) - - def __enter__(self): - return (self.user, self.project) - - def __exit__(self, value, type, trace): - self.manager.delete_user(self.user) - self.manager.delete_project(self.project) - - -class AuthManagerTestCase(object): - def setUp(self): - FLAGS.auth_driver = self.auth_driver - super(AuthManagerTestCase, self).setUp() - self.flags(connection_type='fake') - self.manager = manager.AuthManager(new=True) - - def test_create_and_find_user(self): - with user_generator(self.manager): - self.assert_(self.manager.get_user('test1')) - - def test_create_and_find_with_properties(self): - with user_generator(self.manager, name="herbert", secret="classified", - access="private-party"): - u = self.manager.get_user('herbert') - self.assertEqual('herbert', u.id) - self.assertEqual('herbert', u.name) - self.assertEqual('classified', u.secret) - self.assertEqual('private-party', u.access) - - def test_004_signature_is_valid(self): - #self.assertTrue(self.manager.authenticate(**boto.generate_url ...? )) - pass - #raise NotImplementedError - - def test_005_can_get_credentials(self): - return - credentials = self.manager.get_user('test1').get_credentials() - self.assertEqual(credentials, - 'export EC2_ACCESS_KEY="access"\n' + - 'export EC2_SECRET_KEY="secret"\n' + - 'export EC2_URL="http://127.0.0.1:8773/services/Cloud"\n' + - 'export S3_URL="http://127.0.0.1:3333/"\n' + - 'export EC2_USER_ID="test1"\n') - - def test_can_list_users(self): - with user_generator(self.manager): - with user_generator(self.manager, name="test2"): - users = self.manager.get_users() - self.assert_(filter(lambda u: u.id == 'test1', users)) - self.assert_(filter(lambda u: u.id == 'test2', users)) - self.assert_(not filter(lambda u: u.id == 'test3', users)) - - def test_can_add_and_remove_user_role(self): - with user_generator(self.manager): - self.assertFalse(self.manager.has_role('test1', 'itsec')) - self.manager.add_role('test1', 'itsec') - self.assertTrue(self.manager.has_role('test1', 'itsec')) - self.manager.remove_role('test1', 'itsec') - self.assertFalse(self.manager.has_role('test1', 'itsec')) - - def test_can_create_and_get_project(self): - with user_and_project_generator(self.manager) as (u, p): - self.assert_(self.manager.get_user('test1')) - self.assert_(self.manager.get_user('test1')) - self.assert_(self.manager.get_project('testproj')) - - def test_can_list_projects(self): - with user_and_project_generator(self.manager): - with project_generator(self.manager, name="testproj2"): - projects = self.manager.get_projects() - self.assert_(filter(lambda p: p.name == 'testproj', projects)) - self.assert_(filter(lambda p: p.name == 'testproj2', projects)) - self.assert_(not filter(lambda p: p.name == 'testproj3', - projects)) - - def test_can_create_and_get_project_with_attributes(self): - with user_generator(self.manager): - with project_generator(self.manager, description='A test project'): - project = self.manager.get_project('testproj') - self.assertEqual('A test project', project.description) - - def test_can_create_project_with_manager(self): - with user_and_project_generator(self.manager) as (user, project): - self.assertEqual('test1', project.project_manager_id) - self.assertTrue(self.manager.is_project_manager(user, project)) - - def test_create_project_assigns_manager_to_members(self): - with user_and_project_generator(self.manager) as (user, project): - self.assertTrue(self.manager.is_project_member(user, project)) - - def test_no_extra_project_members(self): - with user_generator(self.manager, name='test2') as baduser: - with user_and_project_generator(self.manager) as (user, project): - self.assertFalse(self.manager.is_project_member(baduser, - project)) - - def test_no_extra_project_managers(self): - with user_generator(self.manager, name='test2') as baduser: - with user_and_project_generator(self.manager) as (user, project): - self.assertFalse(self.manager.is_project_manager(baduser, - project)) - - def test_can_add_user_to_project(self): - with user_generator(self.manager, name='test2') as user: - with user_and_project_generator(self.manager) as (_user, project): - self.manager.add_to_project(user, project) - project = self.manager.get_project('testproj') - self.assertTrue(self.manager.is_project_member(user, project)) - - def test_can_remove_user_from_project(self): - with user_generator(self.manager, name='test2') as user: - with user_and_project_generator(self.manager) as (_user, project): - self.manager.add_to_project(user, project) - project = self.manager.get_project('testproj') - self.assertTrue(self.manager.is_project_member(user, project)) - self.manager.remove_from_project(user, project) - project = self.manager.get_project('testproj') - self.assertFalse(self.manager.is_project_member(user, project)) - - def test_can_add_remove_user_with_role(self): - with user_generator(self.manager, name='test2') as user: - with user_and_project_generator(self.manager) as (_user, project): - # NOTE(todd): after modifying users you must reload project - self.manager.add_to_project(user, project) - project = self.manager.get_project('testproj') - self.manager.add_role(user, 'developer', project) - self.assertTrue(self.manager.is_project_member(user, project)) - self.manager.remove_from_project(user, project) - project = self.manager.get_project('testproj') - self.assertFalse(self.manager.has_role(user, 'developer', - project)) - self.assertFalse(self.manager.is_project_member(user, project)) - - def test_can_generate_x509(self): - # NOTE(todd): this doesn't assert against the auth manager - # so it probably belongs in crypto_unittest - # but I'm leaving it where I found it. - with user_and_project_generator(self.manager) as (user, project): - # NOTE(todd): Should mention why we must setup controller first - # (somebody please clue me in) - cloud_controller = cloud.CloudController() - cloud_controller.setup() - _key, cert_str = self.manager._generate_x509_cert('test1', - 'testproj') - logging.debug(cert_str) - - # Need to verify that it's signed by the right intermediate CA - full_chain = crypto.fetch_ca(project_id='testproj', chain=True) - int_cert = crypto.fetch_ca(project_id='testproj', chain=False) - cloud_cert = crypto.fetch_ca() - logging.debug("CA chain:\n\n =====\n%s\n\n=====" % full_chain) - signed_cert = X509.load_cert_string(cert_str) - chain_cert = X509.load_cert_string(full_chain) - int_cert = X509.load_cert_string(int_cert) - cloud_cert = X509.load_cert_string(cloud_cert) - self.assertTrue(signed_cert.verify(chain_cert.get_pubkey())) - self.assertTrue(signed_cert.verify(int_cert.get_pubkey())) - if not FLAGS.use_intermediate_ca: - self.assertTrue(signed_cert.verify(cloud_cert.get_pubkey())) - else: - self.assertFalse(signed_cert.verify(cloud_cert.get_pubkey())) - - def test_adding_role_to_project_is_ignored_unless_added_to_user(self): - with user_and_project_generator(self.manager) as (user, project): - self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) - self.manager.add_role(user, 'sysadmin', project) - # NOTE(todd): it will still show up in get_user_roles(u, project) - self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) - self.manager.add_role(user, 'sysadmin') - self.assertTrue(self.manager.has_role(user, 'sysadmin', project)) - - def test_add_user_role_doesnt_infect_project_roles(self): - with user_and_project_generator(self.manager) as (user, project): - self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) - self.manager.add_role(user, 'sysadmin') - self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) - - def test_can_list_user_roles(self): - with user_and_project_generator(self.manager) as (user, project): - self.manager.add_role(user, 'sysadmin') - roles = self.manager.get_user_roles(user) - self.assertTrue('sysadmin' in roles) - self.assertFalse('netadmin' in roles) - - def test_can_list_project_roles(self): - with user_and_project_generator(self.manager) as (user, project): - self.manager.add_role(user, 'sysadmin') - self.manager.add_role(user, 'sysadmin', project) - self.manager.add_role(user, 'netadmin', project) - project_roles = self.manager.get_user_roles(user, project) - self.assertTrue('sysadmin' in project_roles) - self.assertTrue('netadmin' in project_roles) - # has role should be false user-level role is missing - self.assertFalse(self.manager.has_role(user, 'netadmin', project)) - - def test_can_remove_user_roles(self): - with user_and_project_generator(self.manager) as (user, project): - self.manager.add_role(user, 'sysadmin') - self.assertTrue(self.manager.has_role(user, 'sysadmin')) - self.manager.remove_role(user, 'sysadmin') - self.assertFalse(self.manager.has_role(user, 'sysadmin')) - - def test_removing_user_role_hides_it_from_project(self): - with user_and_project_generator(self.manager) as (user, project): - self.manager.add_role(user, 'sysadmin') - self.manager.add_role(user, 'sysadmin', project) - self.assertTrue(self.manager.has_role(user, 'sysadmin', project)) - self.manager.remove_role(user, 'sysadmin') - self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) - - def test_can_remove_project_role_but_keep_user_role(self): - with user_and_project_generator(self.manager) as (user, project): - self.manager.add_role(user, 'sysadmin') - self.manager.add_role(user, 'sysadmin', project) - self.assertTrue(self.manager.has_role(user, 'sysadmin')) - self.manager.remove_role(user, 'sysadmin', project) - self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) - self.assertTrue(self.manager.has_role(user, 'sysadmin')) - - def test_can_retrieve_project_by_user(self): - with user_and_project_generator(self.manager) as (user, project): - self.assertEqual(1, len(self.manager.get_projects('test1'))) - - def test_can_modify_project(self): - with user_and_project_generator(self.manager): - with user_generator(self.manager, name='test2'): - self.manager.modify_project('testproj', 'test2', 'new desc') - project = self.manager.get_project('testproj') - self.assertEqual('test2', project.project_manager_id) - self.assertEqual('new desc', project.description) - - def test_can_delete_project(self): - with user_generator(self.manager): - self.manager.create_project('testproj', 'test1') - self.assert_(self.manager.get_project('testproj')) - self.manager.delete_project('testproj') - projectlist = self.manager.get_projects() - self.assert_(not filter(lambda p: p.name == 'testproj', - projectlist)) - - def test_can_delete_user(self): - self.manager.create_user('test1') - self.assert_(self.manager.get_user('test1')) - self.manager.delete_user('test1') - userlist = self.manager.get_users() - self.assert_(not filter(lambda u: u.id == 'test1', userlist)) - - def test_can_modify_users(self): - with user_generator(self.manager): - self.manager.modify_user('test1', 'access', 'secret', True) - user = self.manager.get_user('test1') - self.assertEqual('access', user.access) - self.assertEqual('secret', user.secret) - self.assertTrue(user.is_admin()) - - -class AuthManagerLdapTestCase(AuthManagerTestCase, test.TestCase): - auth_driver = 'nova.auth.ldapdriver.FakeLdapDriver' - - def __init__(self, *args, **kwargs): - AuthManagerTestCase.__init__(self) - test.TestCase.__init__(self, *args, **kwargs) - import nova.auth.fakeldap as fakeldap - FLAGS.redis_db = 8 - if FLAGS.flush_db: - logging.info("Flushing redis datastore") - try: - r = fakeldap.Redis.instance() - r.flushdb() - except: - self.skip = True - - -class AuthManagerDbTestCase(AuthManagerTestCase, test.TestCase): - auth_driver = 'nova.auth.dbdriver.DbDriver' - - -if __name__ == "__main__": - # TODO: Implement use_fake as an option - unittest.main() diff --git a/nova/tests/cloud_unittest.py b/nova/tests/cloud_unittest.py deleted file mode 100644 index 53a762310..000000000 --- a/nova/tests/cloud_unittest.py +++ /dev/null @@ -1,332 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. - -from base64 import b64decode -import json -import logging -from M2Crypto import BIO -from M2Crypto import RSA -import os -import StringIO -import tempfile -import time - -from eventlet import greenthread -from xml.etree import ElementTree - -from nova import context -from nova import crypto -from nova import db -from nova import flags -from nova import rpc -from nova import test -from nova import utils -from nova.auth import manager -from nova.compute import power_state -from nova.api.ec2 import cloud -from nova.objectstore import image - - -FLAGS = flags.FLAGS - -# Temp dirs for working with image attributes through the cloud controller -# (stole this from objectstore_unittest.py) -OSS_TEMPDIR = tempfile.mkdtemp(prefix='test_oss-') -IMAGES_PATH = os.path.join(OSS_TEMPDIR, 'images') -os.makedirs(IMAGES_PATH) - - -class CloudTestCase(test.TestCase): - def setUp(self): - super(CloudTestCase, self).setUp() - self.flags(connection_type='fake', images_path=IMAGES_PATH) - - self.conn = rpc.Connection.instance() - logging.getLogger().setLevel(logging.DEBUG) - - # set up our cloud - self.cloud = cloud.CloudController() - - # set up a service - self.compute = utils.import_object(FLAGS.compute_manager) - self.compute_consumer = rpc.AdapterConsumer(connection=self.conn, - topic=FLAGS.compute_topic, - proxy=self.compute) - self.compute_consumer.attach_to_eventlet() - self.network = utils.import_object(FLAGS.network_manager) - self.network_consumer = rpc.AdapterConsumer(connection=self.conn, - topic=FLAGS.network_topic, - proxy=self.network) - self.network_consumer.attach_to_eventlet() - - self.manager = manager.AuthManager() - self.user = self.manager.create_user('admin', 'admin', 'admin', True) - self.project = self.manager.create_project('proj', 'admin', 'proj') - self.context = context.RequestContext(user=self.user, - project=self.project) - - def tearDown(self): - self.manager.delete_project(self.project) - self.manager.delete_user(self.user) - super(CloudTestCase, self).tearDown() - - def _create_key(self, name): - # NOTE(vish): create depends on pool, so just call helper directly - return cloud._gen_key(self.context, self.context.user.id, name) - - def test_describe_addresses(self): - """Makes sure describe addresses runs without raising an exception""" - address = "10.10.10.10" - db.floating_ip_create(self.context, - {'address': address, - 'host': FLAGS.host}) - self.cloud.allocate_address(self.context) - self.cloud.describe_addresses(self.context) - self.cloud.release_address(self.context, - public_ip=address) - greenthread.sleep(0.3) - db.floating_ip_destroy(self.context, address) - - def test_associate_disassociate_address(self): - """Verifies associate runs cleanly without raising an exception""" - address = "10.10.10.10" - db.floating_ip_create(self.context, - {'address': address, - 'host': FLAGS.host}) - self.cloud.allocate_address(self.context) - inst = db.instance_create(self.context, {}) - fixed = self.network.allocate_fixed_ip(self.context, inst['id']) - ec2_id = cloud.internal_id_to_ec2_id(inst['internal_id']) - self.cloud.associate_address(self.context, - instance_id=ec2_id, - public_ip=address) - self.cloud.disassociate_address(self.context, - public_ip=address) - self.cloud.release_address(self.context, - public_ip=address) - greenthread.sleep(0.3) - self.network.deallocate_fixed_ip(self.context, fixed) - db.instance_destroy(self.context, inst['id']) - db.floating_ip_destroy(self.context, address) - - def test_describe_volumes(self): - """Makes sure describe_volumes works and filters results.""" - vol1 = db.volume_create(self.context, {}) - vol2 = db.volume_create(self.context, {}) - result = self.cloud.describe_volumes(self.context) - self.assertEqual(len(result['volumeSet']), 2) - result = self.cloud.describe_volumes(self.context, - volume_id=[vol2['ec2_id']]) - self.assertEqual(len(result['volumeSet']), 1) - self.assertEqual(result['volumeSet'][0]['volumeId'], vol2['ec2_id']) - db.volume_destroy(self.context, vol1['id']) - db.volume_destroy(self.context, vol2['id']) - - def test_console_output(self): - image_id = FLAGS.default_image - instance_type = FLAGS.default_instance_type - max_count = 1 - kwargs = {'image_id': image_id, - 'instance_type': instance_type, - 'max_count': max_count} - rv = yield self.cloud.run_instances(self.context, **kwargs) - instance_id = rv['instancesSet'][0]['instanceId'] - output = yield self.cloud.get_console_output(context=self.context, - instance_id=[instance_id]) - self.assertEquals(b64decode(output['output']), 'FAKE CONSOLE OUTPUT') - # TODO(soren): We need this until we can stop polling in the rpc code - # for unit tests. - greenthread.sleep(0.3) - rv = yield self.cloud.terminate_instances(self.context, [instance_id]) - - def test_key_generation(self): - result = self._create_key('test') - private_key = result['private_key'] - key = RSA.load_key_string(private_key, callback=lambda: None) - bio = BIO.MemoryBuffer() - public_key = db.key_pair_get(self.context, - self.context.user.id, - 'test')['public_key'] - key.save_pub_key_bio(bio) - converted = crypto.ssl_pub_to_ssh_pub(bio.read()) - # assert key fields are equal - self.assertEqual(public_key.split(" ")[1].strip(), - converted.split(" ")[1].strip()) - - def test_describe_key_pairs(self): - self._create_key('test1') - self._create_key('test2') - result = self.cloud.describe_key_pairs(self.context) - keys = result["keypairsSet"] - self.assertTrue(filter(lambda k: k['keyName'] == 'test1', keys)) - self.assertTrue(filter(lambda k: k['keyName'] == 'test2', keys)) - - def test_delete_key_pair(self): - self._create_key('test') - self.cloud.delete_key_pair(self.context, 'test') - - def test_run_instances(self): - if FLAGS.connection_type == 'fake': - logging.debug("Can't test instances without a real virtual env.") - return - image_id = FLAGS.default_image - instance_type = FLAGS.default_instance_type - max_count = 1 - kwargs = {'image_id': image_id, - 'instance_type': instance_type, - 'max_count': max_count} - rv = yield self.cloud.run_instances(self.context, **kwargs) - # TODO: check for proper response - instance_id = rv['reservationSet'][0].keys()[0] - instance = rv['reservationSet'][0][instance_id][0] - logging.debug("Need to watch instance %s until it's running..." % - instance['instance_id']) - while True: - greenthread.sleep(1) - info = self.cloud._get_instance(instance['instance_id']) - logging.debug(info['state']) - if info['state'] == power_state.RUNNING: - break - self.assert_(rv) - - if connection_type != 'fake': - time.sleep(45) # Should use boto for polling here - for reservations in rv['reservationSet']: - # for res_id in reservations.keys(): - # logging.debug(reservations[res_id]) - # for instance in reservations[res_id]: - for instance in reservations[reservations.keys()[0]]: - instance_id = instance['instance_id'] - logging.debug("Terminating instance %s" % instance_id) - rv = yield self.compute.terminate_instance(instance_id) - - def test_instance_update_state(self): - def instance(num): - return { - 'reservation_id': 'r-1', - 'instance_id': 'i-%s' % num, - 'image_id': 'ami-%s' % num, - 'private_dns_name': '10.0.0.%s' % num, - 'dns_name': '10.0.0%s' % num, - 'ami_launch_index': str(num), - 'instance_type': 'fake', - 'availability_zone': 'fake', - 'key_name': None, - 'kernel_id': 'fake', - 'ramdisk_id': 'fake', - 'groups': ['default'], - 'product_codes': None, - 'state': 0x01, - 'user_data': ''} - rv = self.cloud._format_describe_instances(self.context) - self.assert_(len(rv['reservationSet']) == 0) - - # simulate launch of 5 instances - # self.cloud.instances['pending'] = {} - #for i in xrange(5): - # inst = instance(i) - # self.cloud.instances['pending'][inst['instance_id']] = inst - - #rv = self.cloud._format_instances(self.admin) - #self.assert_(len(rv['reservationSet']) == 1) - #self.assert_(len(rv['reservationSet'][0]['instances_set']) == 5) - # report 4 nodes each having 1 of the instances - #for i in xrange(4): - # self.cloud.update_state('instances', - # {('node-%s' % i): {('i-%s' % i): - # instance(i)}}) - - # one instance should be pending still - #self.assert_(len(self.cloud.instances['pending'].keys()) == 1) - - # check that the reservations collapse - #rv = self.cloud._format_instances(self.admin) - #self.assert_(len(rv['reservationSet']) == 1) - #self.assert_(len(rv['reservationSet'][0]['instances_set']) == 5) - - # check that we can get metadata for each instance - #for i in xrange(4): - # data = self.cloud.get_metadata(instance(i)['private_dns_name']) - # self.assert_(data['meta-data']['ami-id'] == 'ami-%s' % i) - - @staticmethod - def _fake_set_image_description(ctxt, image_id, description): - from nova.objectstore import handler - - class req: - pass - - request = req() - request.context = ctxt - request.args = {'image_id': [image_id], - 'description': [description]} - - resource = handler.ImagesResource() - resource.render_POST(request) - - def test_user_editable_image_endpoint(self): - pathdir = os.path.join(FLAGS.images_path, 'ami-testing') - os.mkdir(pathdir) - info = {'isPublic': False} - with open(os.path.join(pathdir, 'info.json'), 'w') as f: - json.dump(info, f) - img = image.Image('ami-testing') - # self.cloud.set_image_description(self.context, 'ami-testing', - # 'Foo Img') - # NOTE(vish): Above won't work unless we start objectstore or create - # a fake version of api/ec2/images.py conn that can - # call methods directly instead of going through boto. - # for now, just cheat and call the method directly - self._fake_set_image_description(self.context, 'ami-testing', - 'Foo Img') - self.assertEqual('Foo Img', img.metadata['description']) - self._fake_set_image_description(self.context, 'ami-testing', '') - self.assertEqual('', img.metadata['description']) - - def test_update_of_instance_display_fields(self): - inst = db.instance_create(self.context, {}) - ec2_id = cloud.internal_id_to_ec2_id(inst['internal_id']) - self.cloud.update_instance(self.context, ec2_id, - display_name='c00l 1m4g3') - inst = db.instance_get(self.context, inst['id']) - self.assertEqual('c00l 1m4g3', inst['display_name']) - db.instance_destroy(self.context, inst['id']) - - def test_update_of_instance_wont_update_private_fields(self): - inst = db.instance_create(self.context, {}) - self.cloud.update_instance(self.context, inst['id'], - mac_address='DE:AD:BE:EF') - inst = db.instance_get(self.context, inst['id']) - self.assertEqual(None, inst['mac_address']) - db.instance_destroy(self.context, inst['id']) - - def test_update_of_volume_display_fields(self): - vol = db.volume_create(self.context, {}) - self.cloud.update_volume(self.context, vol['id'], - display_name='c00l v0lum3') - vol = db.volume_get(self.context, vol['id']) - self.assertEqual('c00l v0lum3', vol['display_name']) - db.volume_destroy(self.context, vol['id']) - - def test_update_of_volume_wont_update_private_fields(self): - vol = db.volume_create(self.context, {}) - self.cloud.update_volume(self.context, vol['id'], - mountpoint='/not/here') - vol = db.volume_get(self.context, vol['id']) - self.assertEqual(None, vol['mountpoint']) - db.volume_destroy(self.context, vol['id']) diff --git a/nova/tests/compute_unittest.py b/nova/tests/compute_unittest.py deleted file mode 100644 index c6353d357..000000000 --- a/nova/tests/compute_unittest.py +++ /dev/null @@ -1,155 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. -""" -Tests For Compute -""" - -import datetime -import logging - -from nova import context -from nova import db -from nova import exception -from nova import flags -from nova import test -from nova import utils -from nova.auth import manager -from nova.compute import api as compute_api - - -FLAGS = flags.FLAGS - - -class ComputeTestCase(test.TestCase): - """Test case for compute""" - def setUp(self): - logging.getLogger().setLevel(logging.DEBUG) - super(ComputeTestCase, self).setUp() - self.flags(connection_type='fake', - network_manager='nova.network.manager.FlatManager') - self.compute = utils.import_object(FLAGS.compute_manager) - self.compute_api = compute_api.ComputeAPI() - self.manager = manager.AuthManager() - self.user = self.manager.create_user('fake', 'fake', 'fake') - self.project = self.manager.create_project('fake', 'fake', 'fake') - self.context = context.get_admin_context() - - def tearDown(self): - self.manager.delete_user(self.user) - self.manager.delete_project(self.project) - super(ComputeTestCase, self).tearDown() - - def _create_instance(self): - """Create a test instance""" - inst = {} - inst['image_id'] = 'ami-test' - inst['reservation_id'] = 'r-fakeres' - inst['launch_time'] = '10' - inst['user_id'] = self.user.id - inst['project_id'] = self.project.id - inst['instance_type'] = 'm1.tiny' - inst['mac_address'] = utils.generate_mac() - inst['ami_launch_index'] = 0 - return db.instance_create(self.context, inst)['id'] - - def test_create_instance_defaults_display_name(self): - """Verify that an instance cannot be created without a display_name.""" - cases = [dict(), dict(display_name=None)] - for instance in cases: - ref = self.compute_api.create_instances(self.context, - FLAGS.default_instance_type, None, **instance) - try: - self.assertNotEqual(ref[0].display_name, None) - finally: - db.instance_destroy(self.context, ref[0]['id']) - - def test_create_instance_associates_security_groups(self): - """Make sure create_instances associates security groups""" - values = {'name': 'default', - 'description': 'default', - 'user_id': self.user.id, - 'project_id': self.project.id} - group = db.security_group_create(self.context, values) - ref = self.compute_api.create_instances(self.context, - FLAGS.default_instance_type, None, security_group=['default']) - try: - self.assertEqual(len(ref[0]['security_groups']), 1) - finally: - db.security_group_destroy(self.context, group['id']) - db.instance_destroy(self.context, ref[0]['id']) - - def test_run_terminate(self): - """Make sure it is possible to run and terminate instance""" - instance_id = self._create_instance() - - self.compute.run_instance(self.context, instance_id) - - instances = db.instance_get_all(context.get_admin_context()) - logging.info("Running instances: %s", instances) - self.assertEqual(len(instances), 1) - - self.compute.terminate_instance(self.context, instance_id) - - instances = db.instance_get_all(context.get_admin_context()) - logging.info("After terminating instances: %s", instances) - self.assertEqual(len(instances), 0) - - def test_run_terminate_timestamps(self): - """Make sure timestamps are set for launched and destroyed""" - instance_id = self._create_instance() - instance_ref = db.instance_get(self.context, instance_id) - self.assertEqual(instance_ref['launched_at'], None) - self.assertEqual(instance_ref['deleted_at'], None) - launch = datetime.datetime.utcnow() - self.compute.run_instance(self.context, instance_id) - instance_ref = db.instance_get(self.context, instance_id) - self.assert_(instance_ref['launched_at'] > launch) - self.assertEqual(instance_ref['deleted_at'], None) - terminate = datetime.datetime.utcnow() - self.compute.terminate_instance(self.context, instance_id) - self.context = self.context.elevated(True) - instance_ref = db.instance_get(self.context, instance_id) - self.assert_(instance_ref['launched_at'] < terminate) - self.assert_(instance_ref['deleted_at'] > terminate) - - def test_reboot(self): - """Ensure instance can be rebooted""" - instance_id = self._create_instance() - self.compute.run_instance(self.context, instance_id) - self.compute.reboot_instance(self.context, instance_id) - self.compute.terminate_instance(self.context, instance_id) - - def test_console_output(self): - """Make sure we can get console output from instance""" - instance_id = self._create_instance() - self.compute.run_instance(self.context, instance_id) - - console = self.compute.get_console_output(self.context, - instance_id) - self.assert_(console) - self.compute.terminate_instance(self.context, instance_id) - - def test_run_instance_existing(self): - """Ensure failure when running an instance that already exists""" - instance_id = self._create_instance() - self.compute.run_instance(self.context, instance_id) - self.assertRaises(exception.Error, - self.compute.run_instance, - self.context, - instance_id) - self.compute.terminate_instance(self.context, instance_id) diff --git a/nova/tests/flags_unittest.py b/nova/tests/flags_unittest.py deleted file mode 100644 index 707300fcf..000000000 --- a/nova/tests/flags_unittest.py +++ /dev/null @@ -1,102 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. - -from nova import exception -from nova import flags -from nova import test - -FLAGS = flags.FLAGS -flags.DEFINE_string('flags_unittest', 'foo', 'for testing purposes only') - - -class FlagsTestCase(test.TestCase): - - def setUp(self): - super(FlagsTestCase, self).setUp() - self.FLAGS = flags.FlagValues() - self.global_FLAGS = flags.FLAGS - - def test_define(self): - self.assert_('string' not in self.FLAGS) - self.assert_('int' not in self.FLAGS) - self.assert_('false' not in self.FLAGS) - self.assert_('true' not in self.FLAGS) - - flags.DEFINE_string('string', 'default', 'desc', - flag_values=self.FLAGS) - flags.DEFINE_integer('int', 1, 'desc', flag_values=self.FLAGS) - flags.DEFINE_bool('false', False, 'desc', flag_values=self.FLAGS) - flags.DEFINE_bool('true', True, 'desc', flag_values=self.FLAGS) - - self.assert_(self.FLAGS['string']) - self.assert_(self.FLAGS['int']) - self.assert_(self.FLAGS['false']) - self.assert_(self.FLAGS['true']) - self.assertEqual(self.FLAGS.string, 'default') - self.assertEqual(self.FLAGS.int, 1) - self.assertEqual(self.FLAGS.false, False) - self.assertEqual(self.FLAGS.true, True) - - argv = ['flags_test', - '--string', 'foo', - '--int', '2', - '--false', - '--notrue'] - - self.FLAGS(argv) - self.assertEqual(self.FLAGS.string, 'foo') - self.assertEqual(self.FLAGS.int, 2) - self.assertEqual(self.FLAGS.false, True) - self.assertEqual(self.FLAGS.true, False) - - def test_declare(self): - self.assert_('answer' not in self.global_FLAGS) - flags.DECLARE('answer', 'nova.tests.declare_flags') - self.assert_('answer' in self.global_FLAGS) - self.assertEqual(self.global_FLAGS.answer, 42) - - # Make sure we don't overwrite anything - self.global_FLAGS.answer = 256 - self.assertEqual(self.global_FLAGS.answer, 256) - flags.DECLARE('answer', 'nova.tests.declare_flags') - self.assertEqual(self.global_FLAGS.answer, 256) - - def test_runtime_and_unknown_flags(self): - self.assert_('runtime_answer' not in self.global_FLAGS) - - argv = ['flags_test', '--runtime_answer=60', 'extra_arg'] - args = self.global_FLAGS(argv) - self.assertEqual(len(args), 2) - self.assertEqual(args[1], 'extra_arg') - - self.assert_('runtime_answer' not in self.global_FLAGS) - - import nova.tests.runtime_flags - - self.assert_('runtime_answer' in self.global_FLAGS) - self.assertEqual(self.global_FLAGS.runtime_answer, 60) - - def test_flag_leak_left(self): - self.assertEqual(FLAGS.flags_unittest, 'foo') - FLAGS.flags_unittest = 'bar' - self.assertEqual(FLAGS.flags_unittest, 'bar') - - def test_flag_leak_right(self): - self.assertEqual(FLAGS.flags_unittest, 'foo') - FLAGS.flags_unittest = 'bar' - self.assertEqual(FLAGS.flags_unittest, 'bar') diff --git a/nova/tests/misc_unittest.py b/nova/tests/misc_unittest.py deleted file mode 100644 index 3d947427a..000000000 --- a/nova/tests/misc_unittest.py +++ /dev/null @@ -1,55 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 OpenStack LLC -# -# 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. - -import os - -from nova import test -from nova.utils import parse_mailmap, str_dict_replace - - -class ProjectTestCase(test.TestCase): - def test_authors_up_to_date(self): - if os.path.exists('../.bzr'): - contributors = set() - - mailmap = parse_mailmap('../.mailmap') - - import bzrlib.workingtree - tree = bzrlib.workingtree.WorkingTree.open('..') - tree.lock_read() - try: - parents = tree.get_parent_ids() - g = tree.branch.repository.get_graph() - for p in parents[1:]: - rev_ids = [r for r, _ in g.iter_ancestry(parents) - if r != "null:"] - revs = tree.branch.repository.get_revisions(rev_ids) - for r in revs: - for author in r.get_apparent_authors(): - email = author.split(' ')[-1] - contributors.add(str_dict_replace(email, mailmap)) - - authors_file = open('../Authors', 'r').read() - - missing = set() - for contributor in contributors: - if not contributor in authors_file: - missing.add(contributor) - - self.assertTrue(len(missing) == 0, - '%r not listed in Authors' % missing) - finally: - tree.unlock() diff --git a/nova/tests/quota_unittest.py b/nova/tests/quota_unittest.py deleted file mode 100644 index 8cf2a5e54..000000000 --- a/nova/tests/quota_unittest.py +++ /dev/null @@ -1,153 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. - -import logging - -from nova import context -from nova import db -from nova import exception -from nova import flags -from nova import quota -from nova import test -from nova import utils -from nova.auth import manager -from nova.api.ec2 import cloud - - -FLAGS = flags.FLAGS - - -class QuotaTestCase(test.TestCase): - def setUp(self): - logging.getLogger().setLevel(logging.DEBUG) - super(QuotaTestCase, self).setUp() - self.flags(connection_type='fake', - quota_instances=2, - quota_cores=4, - quota_volumes=2, - quota_gigabytes=20, - quota_floating_ips=1) - - self.cloud = cloud.CloudController() - self.manager = manager.AuthManager() - self.user = self.manager.create_user('admin', 'admin', 'admin', True) - self.project = self.manager.create_project('admin', 'admin', 'admin') - self.network = utils.import_object(FLAGS.network_manager) - self.context = context.RequestContext(project=self.project, - user=self.user) - - def tearDown(self): - manager.AuthManager().delete_project(self.project) - manager.AuthManager().delete_user(self.user) - super(QuotaTestCase, self).tearDown() - - def _create_instance(self, cores=2): - """Create a test instance""" - inst = {} - inst['image_id'] = 'ami-test' - inst['reservation_id'] = 'r-fakeres' - inst['user_id'] = self.user.id - inst['project_id'] = self.project.id - inst['instance_type'] = 'm1.large' - inst['vcpus'] = cores - inst['mac_address'] = utils.generate_mac() - return db.instance_create(self.context, inst)['id'] - - def _create_volume(self, size=10): - """Create a test volume""" - vol = {} - vol['user_id'] = self.user.id - vol['project_id'] = self.project.id - vol['size'] = size - return db.volume_create(self.context, vol)['id'] - - def test_quota_overrides(self): - """Make sure overriding a projects quotas works""" - num_instances = quota.allowed_instances(self.context, 100, 'm1.small') - self.assertEqual(num_instances, 2) - db.quota_create(self.context, {'project_id': self.project.id, - 'instances': 10}) - num_instances = quota.allowed_instances(self.context, 100, 'm1.small') - self.assertEqual(num_instances, 4) - db.quota_update(self.context, self.project.id, {'cores': 100}) - num_instances = quota.allowed_instances(self.context, 100, 'm1.small') - self.assertEqual(num_instances, 10) - db.quota_destroy(self.context, self.project.id) - - def test_too_many_instances(self): - instance_ids = [] - for i in range(FLAGS.quota_instances): - instance_id = self._create_instance() - instance_ids.append(instance_id) - self.assertRaises(quota.QuotaError, self.cloud.run_instances, - self.context, - min_count=1, - max_count=1, - instance_type='m1.small', - image_id='fake') - for instance_id in instance_ids: - db.instance_destroy(self.context, instance_id) - - def test_too_many_cores(self): - instance_ids = [] - instance_id = self._create_instance(cores=4) - instance_ids.append(instance_id) - self.assertRaises(quota.QuotaError, self.cloud.run_instances, - self.context, - min_count=1, - max_count=1, - instance_type='m1.small', - image_id='fake') - for instance_id in instance_ids: - db.instance_destroy(self.context, instance_id) - - def test_too_many_volumes(self): - volume_ids = [] - for i in range(FLAGS.quota_volumes): - volume_id = self._create_volume() - volume_ids.append(volume_id) - self.assertRaises(quota.QuotaError, self.cloud.create_volume, - self.context, - size=10) - for volume_id in volume_ids: - db.volume_destroy(self.context, volume_id) - - def test_too_many_gigabytes(self): - volume_ids = [] - volume_id = self._create_volume(size=20) - volume_ids.append(volume_id) - self.assertRaises(quota.QuotaError, - self.cloud.create_volume, - self.context, - size=10) - for volume_id in volume_ids: - db.volume_destroy(self.context, volume_id) - - def test_too_many_addresses(self): - address = '192.168.0.100' - db.floating_ip_create(context.get_admin_context(), - {'address': address, 'host': FLAGS.host}) - float_addr = self.network.allocate_floating_ip(self.context, - self.project.id) - # NOTE(vish): This assert never fails. When cloud attempts to - # make an rpc.call, the test just finishes with OK. It - # appears to be something in the magic inline callbacks - # that is breaking. - self.assertRaises(quota.QuotaError, self.cloud.allocate_address, - self.context) - db.floating_ip_destroy(context.get_admin_context(), address) diff --git a/nova/tests/rpc_unittest.py b/nova/tests/rpc_unittest.py deleted file mode 100644 index a2495e65a..000000000 --- a/nova/tests/rpc_unittest.py +++ /dev/null @@ -1,103 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. -""" -Unit Tests for remote procedure calls using queue -""" -import logging - -from nova import context -from nova import flags -from nova import rpc -from nova import test - - -FLAGS = flags.FLAGS - - -class RpcTestCase(test.TestCase): - """Test cases for rpc""" - def setUp(self): - super(RpcTestCase, self).setUp() - self.conn = rpc.Connection.instance() - self.receiver = TestReceiver() - self.consumer = rpc.AdapterConsumer(connection=self.conn, - topic='test', - proxy=self.receiver) - self.consumer.attach_to_eventlet() - self.context = context.get_admin_context() - - def test_call_succeed(self): - """Get a value through rpc call""" - value = 42 - result = rpc.call(self.context, 'test', {"method": "echo", - "args": {"value": value}}) - self.assertEqual(value, result) - - def test_context_passed(self): - """Makes sure a context is passed through rpc call""" - value = 42 - result = rpc.call(self.context, - 'test', {"method": "context", - "args": {"value": value}}) - self.assertEqual(self.context.to_dict(), result) - - def test_call_exception(self): - """Test that exception gets passed back properly - - rpc.call returns a RemoteError object. The value of the - exception is converted to a string, so we convert it back - to an int in the test. - """ - value = 42 - self.assertRaises(rpc.RemoteError, - rpc.call, - self.context, - 'test', - {"method": "fail", - "args": {"value": value}}) - try: - rpc.call(self.context, - 'test', - {"method": "fail", - "args": {"value": value}}) - self.fail("should have thrown rpc.RemoteError") - except rpc.RemoteError as exc: - self.assertEqual(int(exc.value), value) - - -class TestReceiver(object): - """Simple Proxy class so the consumer has methods to call - - Uses static methods because we aren't actually storing any state""" - - @staticmethod - def echo(context, value): - """Simply returns whatever value is sent in""" - logging.debug("Received %s", value) - return value - - @staticmethod - def context(context, value): - """Returns dictionary version of context""" - logging.debug("Received %s", context) - return context.to_dict() - - @staticmethod - def fail(context, value): - """Raises an exception with the value sent in""" - raise Exception(value) diff --git a/nova/tests/scheduler_unittest.py b/nova/tests/scheduler_unittest.py deleted file mode 100644 index d1756b8fb..000000000 --- a/nova/tests/scheduler_unittest.py +++ /dev/null @@ -1,246 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. -""" -Tests For Scheduler -""" - -from nova import context -from nova import db -from nova import flags -from nova import service -from nova import test -from nova import rpc -from nova import utils -from nova.auth import manager as auth_manager -from nova.scheduler import manager -from nova.scheduler import driver - - -FLAGS = flags.FLAGS -flags.DECLARE('max_cores', 'nova.scheduler.simple') - - -class TestDriver(driver.Scheduler): - """Scheduler Driver for Tests""" - def schedule(context, topic, *args, **kwargs): - return 'fallback_host' - - def schedule_named_method(context, topic, num): - return 'named_host' - - -class SchedulerTestCase(test.TestCase): - """Test case for scheduler""" - def setUp(self): - super(SchedulerTestCase, self).setUp() - self.flags(scheduler_driver='nova.tests.scheduler_unittest.TestDriver') - - def test_fallback(self): - scheduler = manager.SchedulerManager() - self.mox.StubOutWithMock(rpc, 'cast', use_mock_anything=True) - ctxt = context.get_admin_context() - rpc.cast(ctxt, - 'topic.fallback_host', - {'method': 'noexist', - 'args': {'num': 7}}) - self.mox.ReplayAll() - scheduler.noexist(ctxt, 'topic', num=7) - - def test_named_method(self): - scheduler = manager.SchedulerManager() - self.mox.StubOutWithMock(rpc, 'cast', use_mock_anything=True) - ctxt = context.get_admin_context() - rpc.cast(ctxt, - 'topic.named_host', - {'method': 'named_method', - 'args': {'num': 7}}) - self.mox.ReplayAll() - scheduler.named_method(ctxt, 'topic', num=7) - - -class SimpleDriverTestCase(test.TestCase): - """Test case for simple driver""" - def setUp(self): - super(SimpleDriverTestCase, self).setUp() - self.flags(connection_type='fake', - max_cores=4, - max_gigabytes=4, - network_manager='nova.network.manager.FlatManager', - volume_driver='nova.volume.driver.FakeISCSIDriver', - scheduler_driver='nova.scheduler.simple.SimpleScheduler') - self.scheduler = manager.SchedulerManager() - self.manager = auth_manager.AuthManager() - self.user = self.manager.create_user('fake', 'fake', 'fake') - self.project = self.manager.create_project('fake', 'fake', 'fake') - self.context = context.get_admin_context() - - def tearDown(self): - self.manager.delete_user(self.user) - self.manager.delete_project(self.project) - - def _create_instance(self): - """Create a test instance""" - inst = {} - inst['image_id'] = 'ami-test' - inst['reservation_id'] = 'r-fakeres' - inst['user_id'] = self.user.id - inst['project_id'] = self.project.id - inst['instance_type'] = 'm1.tiny' - inst['mac_address'] = utils.generate_mac() - inst['ami_launch_index'] = 0 - inst['vcpus'] = 1 - return db.instance_create(self.context, inst)['id'] - - def _create_volume(self): - """Create a test volume""" - vol = {} - vol['image_id'] = 'ami-test' - vol['reservation_id'] = 'r-fakeres' - vol['size'] = 1 - return db.volume_create(self.context, vol)['id'] - - def test_hosts_are_up(self): - """Ensures driver can find the hosts that are up""" - # NOTE(vish): constructing service without create method - # because we are going to use it without queue - compute1 = service.Service('host1', - 'nova-compute', - 'compute', - FLAGS.compute_manager) - compute1.start() - compute2 = service.Service('host2', - 'nova-compute', - 'compute', - FLAGS.compute_manager) - compute2.start() - hosts = self.scheduler.driver.hosts_up(self.context, 'compute') - self.assertEqual(len(hosts), 2) - compute1.kill() - compute2.kill() - - def test_least_busy_host_gets_instance(self): - """Ensures the host with less cores gets the next one""" - compute1 = service.Service('host1', - 'nova-compute', - 'compute', - FLAGS.compute_manager) - compute1.start() - compute2 = service.Service('host2', - 'nova-compute', - 'compute', - FLAGS.compute_manager) - compute2.start() - instance_id1 = self._create_instance() - compute1.run_instance(self.context, instance_id1) - instance_id2 = self._create_instance() - host = self.scheduler.driver.schedule_run_instance(self.context, - instance_id2) - self.assertEqual(host, 'host2') - compute1.terminate_instance(self.context, instance_id1) - db.instance_destroy(self.context, instance_id2) - compute1.kill() - compute2.kill() - - def test_too_many_cores(self): - """Ensures we don't go over max cores""" - compute1 = service.Service('host1', - 'nova-compute', - 'compute', - FLAGS.compute_manager) - compute1.start() - compute2 = service.Service('host2', - 'nova-compute', - 'compute', - FLAGS.compute_manager) - compute2.start() - instance_ids1 = [] - instance_ids2 = [] - for index in xrange(FLAGS.max_cores): - instance_id = self._create_instance() - compute1.run_instance(self.context, instance_id) - instance_ids1.append(instance_id) - instance_id = self._create_instance() - compute2.run_instance(self.context, instance_id) - instance_ids2.append(instance_id) - instance_id = self._create_instance() - self.assertRaises(driver.NoValidHost, - self.scheduler.driver.schedule_run_instance, - self.context, - instance_id) - for instance_id in instance_ids1: - compute1.terminate_instance(self.context, instance_id) - for instance_id in instance_ids2: - compute2.terminate_instance(self.context, instance_id) - compute1.kill() - compute2.kill() - - def test_least_busy_host_gets_volume(self): - """Ensures the host with less gigabytes gets the next one""" - volume1 = service.Service('host1', - 'nova-volume', - 'volume', - FLAGS.volume_manager) - volume1.start() - volume2 = service.Service('host2', - 'nova-volume', - 'volume', - FLAGS.volume_manager) - volume2.start() - volume_id1 = self._create_volume() - volume1.create_volume(self.context, volume_id1) - volume_id2 = self._create_volume() - host = self.scheduler.driver.schedule_create_volume(self.context, - volume_id2) - self.assertEqual(host, 'host2') - volume1.delete_volume(self.context, volume_id1) - db.volume_destroy(self.context, volume_id2) - volume1.kill() - volume2.kill() - - def test_too_many_gigabytes(self): - """Ensures we don't go over max gigabytes""" - volume1 = service.Service('host1', - 'nova-volume', - 'volume', - FLAGS.volume_manager) - volume1.start() - volume2 = service.Service('host2', - 'nova-volume', - 'volume', - FLAGS.volume_manager) - volume2.start() - volume_ids1 = [] - volume_ids2 = [] - for index in xrange(FLAGS.max_gigabytes): - volume_id = self._create_volume() - volume1.create_volume(self.context, volume_id) - volume_ids1.append(volume_id) - volume_id = self._create_volume() - volume2.create_volume(self.context, volume_id) - volume_ids2.append(volume_id) - volume_id = self._create_volume() - self.assertRaises(driver.NoValidHost, - self.scheduler.driver.schedule_create_volume, - self.context, - volume_id) - for volume_id in volume_ids1: - volume1.delete_volume(self.context, volume_id) - for volume_id in volume_ids2: - volume2.delete_volume(self.context, volume_id) - volume1.kill() - volume2.kill() diff --git a/nova/tests/service_unittest.py b/nova/tests/service_unittest.py deleted file mode 100644 index 47c092f8e..000000000 --- a/nova/tests/service_unittest.py +++ /dev/null @@ -1,227 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. - -""" -Unit Tests for remote procedure calls using queue -""" - -import mox - -from nova import exception -from nova import flags -from nova import rpc -from nova import test -from nova import service -from nova import manager - -FLAGS = flags.FLAGS -flags.DEFINE_string("fake_manager", "nova.tests.service_unittest.FakeManager", - "Manager for testing") - - -class FakeManager(manager.Manager): - """Fake manager for tests""" - def test_method(self): - return 'manager' - - -class ExtendedService(service.Service): - def test_method(self): - return 'service' - - -class ServiceManagerTestCase(test.TestCase): - """Test cases for Services""" - - def test_attribute_error_for_no_manager(self): - serv = service.Service('test', - 'test', - 'test', - 'nova.tests.service_unittest.FakeManager') - self.assertRaises(AttributeError, getattr, serv, 'test_method') - - def test_message_gets_to_manager(self): - serv = service.Service('test', - 'test', - 'test', - 'nova.tests.service_unittest.FakeManager') - serv.start() - self.assertEqual(serv.test_method(), 'manager') - - def test_override_manager_method(self): - serv = ExtendedService('test', - 'test', - 'test', - 'nova.tests.service_unittest.FakeManager') - serv.start() - self.assertEqual(serv.test_method(), 'service') - - -class ServiceTestCase(test.TestCase): - """Test cases for Services""" - - def setUp(self): - super(ServiceTestCase, self).setUp() - self.mox.StubOutWithMock(service, 'db') - - def test_create(self): - host = 'foo' - binary = 'nova-fake' - topic = 'fake' - - # NOTE(vish): Create was moved out of mox replay to make sure that - # the looping calls are created in StartService. - app = service.Service.create(host=host, binary=binary) - - self.mox.StubOutWithMock(rpc, - 'AdapterConsumer', - use_mock_anything=True) - rpc.AdapterConsumer(connection=mox.IgnoreArg(), - topic=topic, - proxy=mox.IsA(service.Service)).AndReturn( - rpc.AdapterConsumer) - - rpc.AdapterConsumer(connection=mox.IgnoreArg(), - topic='%s.%s' % (topic, host), - proxy=mox.IsA(service.Service)).AndReturn( - rpc.AdapterConsumer) - - rpc.AdapterConsumer.attach_to_eventlet() - rpc.AdapterConsumer.attach_to_eventlet() - - service_create = {'host': host, - 'binary': binary, - 'topic': topic, - 'report_count': 0} - service_ref = {'host': host, - 'binary': binary, - 'report_count': 0, - 'id': 1} - - service.db.service_get_by_args(mox.IgnoreArg(), - host, - binary).AndRaise(exception.NotFound()) - service.db.service_create(mox.IgnoreArg(), - service_create).AndReturn(service_ref) - self.mox.ReplayAll() - - app.start() - app.stop() - self.assert_(app) - - # We're testing sort of weird behavior in how report_state decides - # whether it is disconnected, it looks for a variable on itself called - # 'model_disconnected' and report_state doesn't really do much so this - # these are mostly just for coverage - def test_report_state_no_service(self): - host = 'foo' - binary = 'bar' - topic = 'test' - service_create = {'host': host, - 'binary': binary, - 'topic': topic, - 'report_count': 0} - service_ref = {'host': host, - 'binary': binary, - 'topic': topic, - 'report_count': 0, - 'id': 1} - - service.db.service_get_by_args(mox.IgnoreArg(), - host, - binary).AndRaise(exception.NotFound()) - service.db.service_create(mox.IgnoreArg(), - service_create).AndReturn(service_ref) - service.db.service_get(mox.IgnoreArg(), - service_ref['id']).AndReturn(service_ref) - service.db.service_update(mox.IgnoreArg(), service_ref['id'], - mox.ContainsKeyValue('report_count', 1)) - - self.mox.ReplayAll() - serv = service.Service(host, - binary, - topic, - 'nova.tests.service_unittest.FakeManager') - serv.start() - serv.report_state() - - def test_report_state_newly_disconnected(self): - host = 'foo' - binary = 'bar' - topic = 'test' - service_create = {'host': host, - 'binary': binary, - 'topic': topic, - 'report_count': 0} - service_ref = {'host': host, - 'binary': binary, - 'topic': topic, - 'report_count': 0, - 'id': 1} - - service.db.service_get_by_args(mox.IgnoreArg(), - host, - binary).AndRaise(exception.NotFound()) - service.db.service_create(mox.IgnoreArg(), - service_create).AndReturn(service_ref) - service.db.service_get(mox.IgnoreArg(), - mox.IgnoreArg()).AndRaise(Exception()) - - self.mox.ReplayAll() - serv = service.Service(host, - binary, - topic, - 'nova.tests.service_unittest.FakeManager') - serv.start() - serv.report_state() - self.assert_(serv.model_disconnected) - - def test_report_state_newly_connected(self): - host = 'foo' - binary = 'bar' - topic = 'test' - service_create = {'host': host, - 'binary': binary, - 'topic': topic, - 'report_count': 0} - service_ref = {'host': host, - 'binary': binary, - 'topic': topic, - 'report_count': 0, - 'id': 1} - - service.db.service_get_by_args(mox.IgnoreArg(), - host, - binary).AndRaise(exception.NotFound()) - service.db.service_create(mox.IgnoreArg(), - service_create).AndReturn(service_ref) - service.db.service_get(mox.IgnoreArg(), - service_ref['id']).AndReturn(service_ref) - service.db.service_update(mox.IgnoreArg(), service_ref['id'], - mox.ContainsKeyValue('report_count', 1)) - - self.mox.ReplayAll() - serv = service.Service(host, - binary, - topic, - 'nova.tests.service_unittest.FakeManager') - serv.start() - serv.model_disconnected = True - serv.report_state() - - self.assert_(not serv.model_disconnected) diff --git a/nova/tests/test_access.py b/nova/tests/test_access.py new file mode 100644 index 000000000..58fdea3b5 --- /dev/null +++ b/nova/tests/test_access.py @@ -0,0 +1,127 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. + +import unittest +import logging +import webob + +from nova import context +from nova import exception +from nova import flags +from nova import test +from nova.api import ec2 +from nova.auth import manager + + +FLAGS = flags.FLAGS + + +class Context(object): + pass + + +class AccessTestCase(test.TestCase): + def setUp(self): + super(AccessTestCase, self).setUp() + um = manager.AuthManager() + self.context = context.get_admin_context() + # Make test users + self.testadmin = um.create_user('testadmin') + self.testpmsys = um.create_user('testpmsys') + self.testnet = um.create_user('testnet') + self.testsys = um.create_user('testsys') + # Assign some rules + um.add_role('testadmin', 'cloudadmin') + um.add_role('testpmsys', 'sysadmin') + um.add_role('testnet', 'netadmin') + um.add_role('testsys', 'sysadmin') + + # Make a test project + self.project = um.create_project('testproj', + 'testpmsys', + 'a test project', + ['testpmsys', 'testnet', 'testsys']) + self.project.add_role(self.testnet, 'netadmin') + self.project.add_role(self.testsys, 'sysadmin') + #user is set in each test + + def noopWSGIApp(environ, start_response): + start_response('200 OK', []) + return [''] + + self.mw = ec2.Authorizer(noopWSGIApp) + self.mw.action_roles = {'str': { + '_allow_all': ['all'], + '_allow_none': [], + '_allow_project_manager': ['projectmanager'], + '_allow_sys_and_net': ['sysadmin', 'netadmin'], + '_allow_sysadmin': ['sysadmin']}} + + def tearDown(self): + um = manager.AuthManager() + # Delete the test project + um.delete_project('testproj') + # Delete the test user + um.delete_user('testadmin') + um.delete_user('testpmsys') + um.delete_user('testnet') + um.delete_user('testsys') + super(AccessTestCase, self).tearDown() + + def response_status(self, user, methodName): + ctxt = context.RequestContext(user, self.project) + environ = {'ec2.context': ctxt, + 'ec2.controller': 'some string', + 'ec2.action': methodName} + req = webob.Request.blank('/', environ) + resp = req.get_response(self.mw) + return resp.status_int + + def shouldAllow(self, user, methodName): + self.assertEqual(200, self.response_status(user, methodName)) + + def shouldDeny(self, user, methodName): + self.assertEqual(401, self.response_status(user, methodName)) + + def test_001_allow_all(self): + users = [self.testadmin, self.testpmsys, self.testnet, self.testsys] + for user in users: + self.shouldAllow(user, '_allow_all') + + def test_002_allow_none(self): + self.shouldAllow(self.testadmin, '_allow_none') + users = [self.testpmsys, self.testnet, self.testsys] + for user in users: + self.shouldDeny(user, '_allow_none') + + def test_003_allow_project_manager(self): + for user in [self.testadmin, self.testpmsys]: + self.shouldAllow(user, '_allow_project_manager') + for user in [self.testnet, self.testsys]: + self.shouldDeny(user, '_allow_project_manager') + + def test_004_allow_sys_and_net(self): + for user in [self.testadmin, self.testnet, self.testsys]: + self.shouldAllow(user, '_allow_sys_and_net') + # denied because it doesn't have the per project sysadmin + for user in [self.testpmsys]: + self.shouldDeny(user, '_allow_sys_and_net') + +if __name__ == "__main__": + # TODO: Implement use_fake as an option + unittest.main() diff --git a/nova/tests/test_api.py b/nova/tests/test_api.py new file mode 100644 index 000000000..33d4cb294 --- /dev/null +++ b/nova/tests/test_api.py @@ -0,0 +1,338 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. + +"""Unit tests for the API endpoint""" + +import boto +from boto.ec2 import regioninfo +import httplib +import random +import StringIO +import webob + +from nova import context +from nova import flags +from nova import test +from nova import api +from nova.api.ec2 import cloud +from nova.api.ec2 import apirequest +from nova.auth import manager + + +class FakeHttplibSocket(object): + """a fake socket implementation for httplib.HTTPResponse, trivial""" + def __init__(self, response_string): + self._buffer = StringIO.StringIO(response_string) + + def makefile(self, _mode, _other): + """Returns the socket's internal buffer""" + return self._buffer + + +class FakeHttplibConnection(object): + """A fake httplib.HTTPConnection for boto to use + + requests made via this connection actually get translated and routed into + our WSGI app, we then wait for the response and turn it back into + the httplib.HTTPResponse that boto expects. + """ + def __init__(self, app, host, is_secure=False): + self.app = app + self.host = host + + def request(self, method, path, data, headers): + req = webob.Request.blank(path) + req.method = method + req.body = data + req.headers = headers + req.headers['Accept'] = 'text/html' + req.host = self.host + # Call the WSGI app, get the HTTP response + resp = str(req.get_response(self.app)) + # For some reason, the response doesn't have "HTTP/1.0 " prepended; I + # guess that's a function the web server usually provides. + resp = "HTTP/1.0 %s" % resp + sock = FakeHttplibSocket(resp) + self.http_response = httplib.HTTPResponse(sock) + self.http_response.begin() + + def getresponse(self): + return self.http_response + + def close(self): + """Required for compatibility with boto/tornado""" + pass + + +class XmlConversionTestCase(test.TrialTestCase): + """Unit test api xml conversion""" + def test_number_conversion(self): + conv = apirequest._try_convert + self.assertEqual(conv('None'), None) + self.assertEqual(conv('True'), True) + self.assertEqual(conv('False'), False) + self.assertEqual(conv('0'), 0) + self.assertEqual(conv('42'), 42) + self.assertEqual(conv('3.14'), 3.14) + self.assertEqual(conv('-57.12'), -57.12) + self.assertEqual(conv('0x57'), 0x57) + self.assertEqual(conv('-0x57'), -0x57) + self.assertEqual(conv('-'), '-') + self.assertEqual(conv('-0'), 0) + + +class ApiEc2TestCase(test.TrialTestCase): + """Unit test for the cloud controller on an EC2 API""" + def setUp(self): + super(ApiEc2TestCase, self).setUp() + + self.manager = manager.AuthManager() + + self.host = '127.0.0.1' + + self.app = api.API('ec2') + + def expect_http(self, host=None, is_secure=False): + """Returns a new EC2 connection""" + self.ec2 = boto.connect_ec2( + aws_access_key_id='fake', + aws_secret_access_key='fake', + is_secure=False, + region=regioninfo.RegionInfo(None, 'test', self.host), + port=8773, + path='/services/Cloud') + + self.mox.StubOutWithMock(self.ec2, 'new_http_connection') + http = FakeHttplibConnection( + self.app, '%s:8773' % (self.host), False) + # pylint: disable-msg=E1103 + self.ec2.new_http_connection(host, is_secure).AndReturn(http) + return http + + def test_describe_instances(self): + """Test that, after creating a user and a project, the describe + instances call to the API works properly""" + self.expect_http() + self.mox.ReplayAll() + user = self.manager.create_user('fake', 'fake', 'fake') + project = self.manager.create_project('fake', 'fake', 'fake') + self.assertEqual(self.ec2.get_all_instances(), []) + self.manager.delete_project(project) + self.manager.delete_user(user) + + def test_get_all_key_pairs(self): + """Test that, after creating a user and project and generating + a key pair, that the API call to list key pairs works properly""" + self.expect_http() + self.mox.ReplayAll() + keyname = "".join(random.choice("sdiuisudfsdcnpaqwertasd") \ + for x in range(random.randint(4, 8))) + user = self.manager.create_user('fake', 'fake', 'fake') + project = self.manager.create_project('fake', 'fake', 'fake') + # NOTE(vish): create depends on pool, so call helper directly + cloud._gen_key(context.get_admin_context(), user.id, keyname) + + rv = self.ec2.get_all_key_pairs() + results = [k for k in rv if k.name == keyname] + self.assertEquals(len(results), 1) + self.manager.delete_project(project) + self.manager.delete_user(user) + + def test_get_all_security_groups(self): + """Test that we can retrieve security groups""" + self.expect_http() + self.mox.ReplayAll() + user = self.manager.create_user('fake', 'fake', 'fake', admin=True) + project = self.manager.create_project('fake', 'fake', 'fake') + + rv = self.ec2.get_all_security_groups() + + self.assertEquals(len(rv), 1) + self.assertEquals(rv[0].name, 'default') + + self.manager.delete_project(project) + self.manager.delete_user(user) + + def test_create_delete_security_group(self): + """Test that we can create a security group""" + self.expect_http() + self.mox.ReplayAll() + user = self.manager.create_user('fake', 'fake', 'fake', admin=True) + project = self.manager.create_project('fake', 'fake', 'fake') + + # At the moment, you need both of these to actually be netadmin + self.manager.add_role('fake', 'netadmin') + project.add_role('fake', 'netadmin') + + security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd") + for x in range(random.randint(4, 8))) + + self.ec2.create_security_group(security_group_name, 'test group') + + self.expect_http() + self.mox.ReplayAll() + + rv = self.ec2.get_all_security_groups() + self.assertEquals(len(rv), 2) + self.assertTrue(security_group_name in [group.name for group in rv]) + + self.expect_http() + self.mox.ReplayAll() + + self.ec2.delete_security_group(security_group_name) + + self.manager.delete_project(project) + self.manager.delete_user(user) + + def test_authorize_revoke_security_group_cidr(self): + """ + Test that we can add and remove CIDR based rules + to a security group + """ + self.expect_http() + self.mox.ReplayAll() + user = self.manager.create_user('fake', 'fake', 'fake') + project = self.manager.create_project('fake', 'fake', 'fake') + + # At the moment, you need both of these to actually be netadmin + self.manager.add_role('fake', 'netadmin') + project.add_role('fake', 'netadmin') + + security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd") + for x in range(random.randint(4, 8))) + + group = self.ec2.create_security_group(security_group_name, + 'test group') + + self.expect_http() + self.mox.ReplayAll() + group.connection = self.ec2 + + group.authorize('tcp', 80, 81, '0.0.0.0/0') + + self.expect_http() + self.mox.ReplayAll() + + rv = self.ec2.get_all_security_groups() + # I don't bother checkng that we actually find it here, + # because the create/delete unit test further up should + # be good enough for that. + for group in rv: + if group.name == security_group_name: + self.assertEquals(len(group.rules), 1) + self.assertEquals(int(group.rules[0].from_port), 80) + self.assertEquals(int(group.rules[0].to_port), 81) + self.assertEquals(len(group.rules[0].grants), 1) + self.assertEquals(str(group.rules[0].grants[0]), '0.0.0.0/0') + + self.expect_http() + self.mox.ReplayAll() + group.connection = self.ec2 + + group.revoke('tcp', 80, 81, '0.0.0.0/0') + + self.expect_http() + self.mox.ReplayAll() + + self.ec2.delete_security_group(security_group_name) + + self.expect_http() + self.mox.ReplayAll() + group.connection = self.ec2 + + rv = self.ec2.get_all_security_groups() + + self.assertEqual(len(rv), 1) + self.assertEqual(rv[0].name, 'default') + + self.manager.delete_project(project) + self.manager.delete_user(user) + + return + + def test_authorize_revoke_security_group_foreign_group(self): + """ + Test that we can grant and revoke another security group access + to a security group + """ + self.expect_http() + self.mox.ReplayAll() + user = self.manager.create_user('fake', 'fake', 'fake', admin=True) + project = self.manager.create_project('fake', 'fake', 'fake') + + # At the moment, you need both of these to actually be netadmin + self.manager.add_role('fake', 'netadmin') + project.add_role('fake', 'netadmin') + + rand_string = 'sdiuisudfsdcnpaqwertasd' + security_group_name = "".join(random.choice(rand_string) + for x in range(random.randint(4, 8))) + other_security_group_name = "".join(random.choice(rand_string) + for x in range(random.randint(4, 8))) + + group = self.ec2.create_security_group(security_group_name, + 'test group') + + self.expect_http() + self.mox.ReplayAll() + + other_group = self.ec2.create_security_group(other_security_group_name, + 'some other group') + + self.expect_http() + self.mox.ReplayAll() + group.connection = self.ec2 + + group.authorize(src_group=other_group) + + self.expect_http() + self.mox.ReplayAll() + + rv = self.ec2.get_all_security_groups() + + # I don't bother checkng that we actually find it here, + # because the create/delete unit test further up should + # be good enough for that. + for group in rv: + if group.name == security_group_name: + self.assertEquals(len(group.rules), 1) + self.assertEquals(len(group.rules[0].grants), 1) + self.assertEquals(str(group.rules[0].grants[0]), '%s-%s' % + (other_security_group_name, 'fake')) + + self.expect_http() + self.mox.ReplayAll() + + rv = self.ec2.get_all_security_groups() + + for group in rv: + if group.name == security_group_name: + self.expect_http() + self.mox.ReplayAll() + group.connection = self.ec2 + group.revoke(src_group=other_group) + + self.expect_http() + self.mox.ReplayAll() + + self.ec2.delete_security_group(security_group_name) + + self.manager.delete_project(project) + self.manager.delete_user(user) + + return diff --git a/nova/tests/test_auth.py b/nova/tests/test_auth.py new file mode 100644 index 000000000..4508d6721 --- /dev/null +++ b/nova/tests/test_auth.py @@ -0,0 +1,352 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. + +import logging +from M2Crypto import X509 +import unittest + +from nova import crypto +from nova import flags +from nova import test +from nova.auth import manager +from nova.api.ec2 import cloud + +FLAGS = flags.FLAGS + + +class user_generator(object): + def __init__(self, manager, **user_state): + if 'name' not in user_state: + user_state['name'] = 'test1' + self.manager = manager + self.user = manager.create_user(**user_state) + + def __enter__(self): + return self.user + + def __exit__(self, value, type, trace): + self.manager.delete_user(self.user) + + +class project_generator(object): + def __init__(self, manager, **project_state): + if 'name' not in project_state: + project_state['name'] = 'testproj' + if 'manager_user' not in project_state: + project_state['manager_user'] = 'test1' + self.manager = manager + self.project = manager.create_project(**project_state) + + def __enter__(self): + return self.project + + def __exit__(self, value, type, trace): + self.manager.delete_project(self.project) + + +class user_and_project_generator(object): + def __init__(self, manager, user_state={}, project_state={}): + self.manager = manager + if 'name' not in user_state: + user_state['name'] = 'test1' + if 'name' not in project_state: + project_state['name'] = 'testproj' + if 'manager_user' not in project_state: + project_state['manager_user'] = 'test1' + self.user = manager.create_user(**user_state) + self.project = manager.create_project(**project_state) + + def __enter__(self): + return (self.user, self.project) + + def __exit__(self, value, type, trace): + self.manager.delete_user(self.user) + self.manager.delete_project(self.project) + + +class AuthManagerTestCase(object): + def setUp(self): + FLAGS.auth_driver = self.auth_driver + super(AuthManagerTestCase, self).setUp() + self.flags(connection_type='fake') + self.manager = manager.AuthManager(new=True) + + def test_create_and_find_user(self): + with user_generator(self.manager): + self.assert_(self.manager.get_user('test1')) + + def test_create_and_find_with_properties(self): + with user_generator(self.manager, name="herbert", secret="classified", + access="private-party"): + u = self.manager.get_user('herbert') + self.assertEqual('herbert', u.id) + self.assertEqual('herbert', u.name) + self.assertEqual('classified', u.secret) + self.assertEqual('private-party', u.access) + + def test_004_signature_is_valid(self): + #self.assertTrue(self.manager.authenticate(**boto.generate_url ...? )) + pass + #raise NotImplementedError + + def test_005_can_get_credentials(self): + return + credentials = self.manager.get_user('test1').get_credentials() + self.assertEqual(credentials, + 'export EC2_ACCESS_KEY="access"\n' + + 'export EC2_SECRET_KEY="secret"\n' + + 'export EC2_URL="http://127.0.0.1:8773/services/Cloud"\n' + + 'export S3_URL="http://127.0.0.1:3333/"\n' + + 'export EC2_USER_ID="test1"\n') + + def test_can_list_users(self): + with user_generator(self.manager): + with user_generator(self.manager, name="test2"): + users = self.manager.get_users() + self.assert_(filter(lambda u: u.id == 'test1', users)) + self.assert_(filter(lambda u: u.id == 'test2', users)) + self.assert_(not filter(lambda u: u.id == 'test3', users)) + + def test_can_add_and_remove_user_role(self): + with user_generator(self.manager): + self.assertFalse(self.manager.has_role('test1', 'itsec')) + self.manager.add_role('test1', 'itsec') + self.assertTrue(self.manager.has_role('test1', 'itsec')) + self.manager.remove_role('test1', 'itsec') + self.assertFalse(self.manager.has_role('test1', 'itsec')) + + def test_can_create_and_get_project(self): + with user_and_project_generator(self.manager) as (u, p): + self.assert_(self.manager.get_user('test1')) + self.assert_(self.manager.get_user('test1')) + self.assert_(self.manager.get_project('testproj')) + + def test_can_list_projects(self): + with user_and_project_generator(self.manager): + with project_generator(self.manager, name="testproj2"): + projects = self.manager.get_projects() + self.assert_(filter(lambda p: p.name == 'testproj', projects)) + self.assert_(filter(lambda p: p.name == 'testproj2', projects)) + self.assert_(not filter(lambda p: p.name == 'testproj3', + projects)) + + def test_can_create_and_get_project_with_attributes(self): + with user_generator(self.manager): + with project_generator(self.manager, description='A test project'): + project = self.manager.get_project('testproj') + self.assertEqual('A test project', project.description) + + def test_can_create_project_with_manager(self): + with user_and_project_generator(self.manager) as (user, project): + self.assertEqual('test1', project.project_manager_id) + self.assertTrue(self.manager.is_project_manager(user, project)) + + def test_create_project_assigns_manager_to_members(self): + with user_and_project_generator(self.manager) as (user, project): + self.assertTrue(self.manager.is_project_member(user, project)) + + def test_no_extra_project_members(self): + with user_generator(self.manager, name='test2') as baduser: + with user_and_project_generator(self.manager) as (user, project): + self.assertFalse(self.manager.is_project_member(baduser, + project)) + + def test_no_extra_project_managers(self): + with user_generator(self.manager, name='test2') as baduser: + with user_and_project_generator(self.manager) as (user, project): + self.assertFalse(self.manager.is_project_manager(baduser, + project)) + + def test_can_add_user_to_project(self): + with user_generator(self.manager, name='test2') as user: + with user_and_project_generator(self.manager) as (_user, project): + self.manager.add_to_project(user, project) + project = self.manager.get_project('testproj') + self.assertTrue(self.manager.is_project_member(user, project)) + + def test_can_remove_user_from_project(self): + with user_generator(self.manager, name='test2') as user: + with user_and_project_generator(self.manager) as (_user, project): + self.manager.add_to_project(user, project) + project = self.manager.get_project('testproj') + self.assertTrue(self.manager.is_project_member(user, project)) + self.manager.remove_from_project(user, project) + project = self.manager.get_project('testproj') + self.assertFalse(self.manager.is_project_member(user, project)) + + def test_can_add_remove_user_with_role(self): + with user_generator(self.manager, name='test2') as user: + with user_and_project_generator(self.manager) as (_user, project): + # NOTE(todd): after modifying users you must reload project + self.manager.add_to_project(user, project) + project = self.manager.get_project('testproj') + self.manager.add_role(user, 'developer', project) + self.assertTrue(self.manager.is_project_member(user, project)) + self.manager.remove_from_project(user, project) + project = self.manager.get_project('testproj') + self.assertFalse(self.manager.has_role(user, 'developer', + project)) + self.assertFalse(self.manager.is_project_member(user, project)) + + def test_can_generate_x509(self): + # NOTE(todd): this doesn't assert against the auth manager + # so it probably belongs in crypto_unittest + # but I'm leaving it where I found it. + with user_and_project_generator(self.manager) as (user, project): + # NOTE(todd): Should mention why we must setup controller first + # (somebody please clue me in) + cloud_controller = cloud.CloudController() + cloud_controller.setup() + _key, cert_str = self.manager._generate_x509_cert('test1', + 'testproj') + logging.debug(cert_str) + + # Need to verify that it's signed by the right intermediate CA + full_chain = crypto.fetch_ca(project_id='testproj', chain=True) + int_cert = crypto.fetch_ca(project_id='testproj', chain=False) + cloud_cert = crypto.fetch_ca() + logging.debug("CA chain:\n\n =====\n%s\n\n=====" % full_chain) + signed_cert = X509.load_cert_string(cert_str) + chain_cert = X509.load_cert_string(full_chain) + int_cert = X509.load_cert_string(int_cert) + cloud_cert = X509.load_cert_string(cloud_cert) + self.assertTrue(signed_cert.verify(chain_cert.get_pubkey())) + self.assertTrue(signed_cert.verify(int_cert.get_pubkey())) + if not FLAGS.use_intermediate_ca: + self.assertTrue(signed_cert.verify(cloud_cert.get_pubkey())) + else: + self.assertFalse(signed_cert.verify(cloud_cert.get_pubkey())) + + def test_adding_role_to_project_is_ignored_unless_added_to_user(self): + with user_and_project_generator(self.manager) as (user, project): + self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) + self.manager.add_role(user, 'sysadmin', project) + # NOTE(todd): it will still show up in get_user_roles(u, project) + self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) + self.manager.add_role(user, 'sysadmin') + self.assertTrue(self.manager.has_role(user, 'sysadmin', project)) + + def test_add_user_role_doesnt_infect_project_roles(self): + with user_and_project_generator(self.manager) as (user, project): + self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) + self.manager.add_role(user, 'sysadmin') + self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) + + def test_can_list_user_roles(self): + with user_and_project_generator(self.manager) as (user, project): + self.manager.add_role(user, 'sysadmin') + roles = self.manager.get_user_roles(user) + self.assertTrue('sysadmin' in roles) + self.assertFalse('netadmin' in roles) + + def test_can_list_project_roles(self): + with user_and_project_generator(self.manager) as (user, project): + self.manager.add_role(user, 'sysadmin') + self.manager.add_role(user, 'sysadmin', project) + self.manager.add_role(user, 'netadmin', project) + project_roles = self.manager.get_user_roles(user, project) + self.assertTrue('sysadmin' in project_roles) + self.assertTrue('netadmin' in project_roles) + # has role should be false user-level role is missing + self.assertFalse(self.manager.has_role(user, 'netadmin', project)) + + def test_can_remove_user_roles(self): + with user_and_project_generator(self.manager) as (user, project): + self.manager.add_role(user, 'sysadmin') + self.assertTrue(self.manager.has_role(user, 'sysadmin')) + self.manager.remove_role(user, 'sysadmin') + self.assertFalse(self.manager.has_role(user, 'sysadmin')) + + def test_removing_user_role_hides_it_from_project(self): + with user_and_project_generator(self.manager) as (user, project): + self.manager.add_role(user, 'sysadmin') + self.manager.add_role(user, 'sysadmin', project) + self.assertTrue(self.manager.has_role(user, 'sysadmin', project)) + self.manager.remove_role(user, 'sysadmin') + self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) + + def test_can_remove_project_role_but_keep_user_role(self): + with user_and_project_generator(self.manager) as (user, project): + self.manager.add_role(user, 'sysadmin') + self.manager.add_role(user, 'sysadmin', project) + self.assertTrue(self.manager.has_role(user, 'sysadmin')) + self.manager.remove_role(user, 'sysadmin', project) + self.assertFalse(self.manager.has_role(user, 'sysadmin', project)) + self.assertTrue(self.manager.has_role(user, 'sysadmin')) + + def test_can_retrieve_project_by_user(self): + with user_and_project_generator(self.manager) as (user, project): + self.assertEqual(1, len(self.manager.get_projects('test1'))) + + def test_can_modify_project(self): + with user_and_project_generator(self.manager): + with user_generator(self.manager, name='test2'): + self.manager.modify_project('testproj', 'test2', 'new desc') + project = self.manager.get_project('testproj') + self.assertEqual('test2', project.project_manager_id) + self.assertEqual('new desc', project.description) + + def test_can_delete_project(self): + with user_generator(self.manager): + self.manager.create_project('testproj', 'test1') + self.assert_(self.manager.get_project('testproj')) + self.manager.delete_project('testproj') + projectlist = self.manager.get_projects() + self.assert_(not filter(lambda p: p.name == 'testproj', + projectlist)) + + def test_can_delete_user(self): + self.manager.create_user('test1') + self.assert_(self.manager.get_user('test1')) + self.manager.delete_user('test1') + userlist = self.manager.get_users() + self.assert_(not filter(lambda u: u.id == 'test1', userlist)) + + def test_can_modify_users(self): + with user_generator(self.manager): + self.manager.modify_user('test1', 'access', 'secret', True) + user = self.manager.get_user('test1') + self.assertEqual('access', user.access) + self.assertEqual('secret', user.secret) + self.assertTrue(user.is_admin()) + + +class AuthManagerLdapTestCase(AuthManagerTestCase, test.TestCase): + auth_driver = 'nova.auth.ldapdriver.FakeLdapDriver' + + def __init__(self, *args, **kwargs): + AuthManagerTestCase.__init__(self) + test.TestCase.__init__(self, *args, **kwargs) + import nova.auth.fakeldap as fakeldap + FLAGS.redis_db = 8 + if FLAGS.flush_db: + logging.info("Flushing redis datastore") + try: + r = fakeldap.Redis.instance() + r.flushdb() + except: + self.skip = True + + +class AuthManagerDbTestCase(AuthManagerTestCase, test.TestCase): + auth_driver = 'nova.auth.dbdriver.DbDriver' + + +if __name__ == "__main__": + # TODO: Implement use_fake as an option + unittest.main() diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py new file mode 100644 index 000000000..53a762310 --- /dev/null +++ b/nova/tests/test_cloud.py @@ -0,0 +1,332 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. + +from base64 import b64decode +import json +import logging +from M2Crypto import BIO +from M2Crypto import RSA +import os +import StringIO +import tempfile +import time + +from eventlet import greenthread +from xml.etree import ElementTree + +from nova import context +from nova import crypto +from nova import db +from nova import flags +from nova import rpc +from nova import test +from nova import utils +from nova.auth import manager +from nova.compute import power_state +from nova.api.ec2 import cloud +from nova.objectstore import image + + +FLAGS = flags.FLAGS + +# Temp dirs for working with image attributes through the cloud controller +# (stole this from objectstore_unittest.py) +OSS_TEMPDIR = tempfile.mkdtemp(prefix='test_oss-') +IMAGES_PATH = os.path.join(OSS_TEMPDIR, 'images') +os.makedirs(IMAGES_PATH) + + +class CloudTestCase(test.TestCase): + def setUp(self): + super(CloudTestCase, self).setUp() + self.flags(connection_type='fake', images_path=IMAGES_PATH) + + self.conn = rpc.Connection.instance() + logging.getLogger().setLevel(logging.DEBUG) + + # set up our cloud + self.cloud = cloud.CloudController() + + # set up a service + self.compute = utils.import_object(FLAGS.compute_manager) + self.compute_consumer = rpc.AdapterConsumer(connection=self.conn, + topic=FLAGS.compute_topic, + proxy=self.compute) + self.compute_consumer.attach_to_eventlet() + self.network = utils.import_object(FLAGS.network_manager) + self.network_consumer = rpc.AdapterConsumer(connection=self.conn, + topic=FLAGS.network_topic, + proxy=self.network) + self.network_consumer.attach_to_eventlet() + + self.manager = manager.AuthManager() + self.user = self.manager.create_user('admin', 'admin', 'admin', True) + self.project = self.manager.create_project('proj', 'admin', 'proj') + self.context = context.RequestContext(user=self.user, + project=self.project) + + def tearDown(self): + self.manager.delete_project(self.project) + self.manager.delete_user(self.user) + super(CloudTestCase, self).tearDown() + + def _create_key(self, name): + # NOTE(vish): create depends on pool, so just call helper directly + return cloud._gen_key(self.context, self.context.user.id, name) + + def test_describe_addresses(self): + """Makes sure describe addresses runs without raising an exception""" + address = "10.10.10.10" + db.floating_ip_create(self.context, + {'address': address, + 'host': FLAGS.host}) + self.cloud.allocate_address(self.context) + self.cloud.describe_addresses(self.context) + self.cloud.release_address(self.context, + public_ip=address) + greenthread.sleep(0.3) + db.floating_ip_destroy(self.context, address) + + def test_associate_disassociate_address(self): + """Verifies associate runs cleanly without raising an exception""" + address = "10.10.10.10" + db.floating_ip_create(self.context, + {'address': address, + 'host': FLAGS.host}) + self.cloud.allocate_address(self.context) + inst = db.instance_create(self.context, {}) + fixed = self.network.allocate_fixed_ip(self.context, inst['id']) + ec2_id = cloud.internal_id_to_ec2_id(inst['internal_id']) + self.cloud.associate_address(self.context, + instance_id=ec2_id, + public_ip=address) + self.cloud.disassociate_address(self.context, + public_ip=address) + self.cloud.release_address(self.context, + public_ip=address) + greenthread.sleep(0.3) + self.network.deallocate_fixed_ip(self.context, fixed) + db.instance_destroy(self.context, inst['id']) + db.floating_ip_destroy(self.context, address) + + def test_describe_volumes(self): + """Makes sure describe_volumes works and filters results.""" + vol1 = db.volume_create(self.context, {}) + vol2 = db.volume_create(self.context, {}) + result = self.cloud.describe_volumes(self.context) + self.assertEqual(len(result['volumeSet']), 2) + result = self.cloud.describe_volumes(self.context, + volume_id=[vol2['ec2_id']]) + self.assertEqual(len(result['volumeSet']), 1) + self.assertEqual(result['volumeSet'][0]['volumeId'], vol2['ec2_id']) + db.volume_destroy(self.context, vol1['id']) + db.volume_destroy(self.context, vol2['id']) + + def test_console_output(self): + image_id = FLAGS.default_image + instance_type = FLAGS.default_instance_type + max_count = 1 + kwargs = {'image_id': image_id, + 'instance_type': instance_type, + 'max_count': max_count} + rv = yield self.cloud.run_instances(self.context, **kwargs) + instance_id = rv['instancesSet'][0]['instanceId'] + output = yield self.cloud.get_console_output(context=self.context, + instance_id=[instance_id]) + self.assertEquals(b64decode(output['output']), 'FAKE CONSOLE OUTPUT') + # TODO(soren): We need this until we can stop polling in the rpc code + # for unit tests. + greenthread.sleep(0.3) + rv = yield self.cloud.terminate_instances(self.context, [instance_id]) + + def test_key_generation(self): + result = self._create_key('test') + private_key = result['private_key'] + key = RSA.load_key_string(private_key, callback=lambda: None) + bio = BIO.MemoryBuffer() + public_key = db.key_pair_get(self.context, + self.context.user.id, + 'test')['public_key'] + key.save_pub_key_bio(bio) + converted = crypto.ssl_pub_to_ssh_pub(bio.read()) + # assert key fields are equal + self.assertEqual(public_key.split(" ")[1].strip(), + converted.split(" ")[1].strip()) + + def test_describe_key_pairs(self): + self._create_key('test1') + self._create_key('test2') + result = self.cloud.describe_key_pairs(self.context) + keys = result["keypairsSet"] + self.assertTrue(filter(lambda k: k['keyName'] == 'test1', keys)) + self.assertTrue(filter(lambda k: k['keyName'] == 'test2', keys)) + + def test_delete_key_pair(self): + self._create_key('test') + self.cloud.delete_key_pair(self.context, 'test') + + def test_run_instances(self): + if FLAGS.connection_type == 'fake': + logging.debug("Can't test instances without a real virtual env.") + return + image_id = FLAGS.default_image + instance_type = FLAGS.default_instance_type + max_count = 1 + kwargs = {'image_id': image_id, + 'instance_type': instance_type, + 'max_count': max_count} + rv = yield self.cloud.run_instances(self.context, **kwargs) + # TODO: check for proper response + instance_id = rv['reservationSet'][0].keys()[0] + instance = rv['reservationSet'][0][instance_id][0] + logging.debug("Need to watch instance %s until it's running..." % + instance['instance_id']) + while True: + greenthread.sleep(1) + info = self.cloud._get_instance(instance['instance_id']) + logging.debug(info['state']) + if info['state'] == power_state.RUNNING: + break + self.assert_(rv) + + if connection_type != 'fake': + time.sleep(45) # Should use boto for polling here + for reservations in rv['reservationSet']: + # for res_id in reservations.keys(): + # logging.debug(reservations[res_id]) + # for instance in reservations[res_id]: + for instance in reservations[reservations.keys()[0]]: + instance_id = instance['instance_id'] + logging.debug("Terminating instance %s" % instance_id) + rv = yield self.compute.terminate_instance(instance_id) + + def test_instance_update_state(self): + def instance(num): + return { + 'reservation_id': 'r-1', + 'instance_id': 'i-%s' % num, + 'image_id': 'ami-%s' % num, + 'private_dns_name': '10.0.0.%s' % num, + 'dns_name': '10.0.0%s' % num, + 'ami_launch_index': str(num), + 'instance_type': 'fake', + 'availability_zone': 'fake', + 'key_name': None, + 'kernel_id': 'fake', + 'ramdisk_id': 'fake', + 'groups': ['default'], + 'product_codes': None, + 'state': 0x01, + 'user_data': ''} + rv = self.cloud._format_describe_instances(self.context) + self.assert_(len(rv['reservationSet']) == 0) + + # simulate launch of 5 instances + # self.cloud.instances['pending'] = {} + #for i in xrange(5): + # inst = instance(i) + # self.cloud.instances['pending'][inst['instance_id']] = inst + + #rv = self.cloud._format_instances(self.admin) + #self.assert_(len(rv['reservationSet']) == 1) + #self.assert_(len(rv['reservationSet'][0]['instances_set']) == 5) + # report 4 nodes each having 1 of the instances + #for i in xrange(4): + # self.cloud.update_state('instances', + # {('node-%s' % i): {('i-%s' % i): + # instance(i)}}) + + # one instance should be pending still + #self.assert_(len(self.cloud.instances['pending'].keys()) == 1) + + # check that the reservations collapse + #rv = self.cloud._format_instances(self.admin) + #self.assert_(len(rv['reservationSet']) == 1) + #self.assert_(len(rv['reservationSet'][0]['instances_set']) == 5) + + # check that we can get metadata for each instance + #for i in xrange(4): + # data = self.cloud.get_metadata(instance(i)['private_dns_name']) + # self.assert_(data['meta-data']['ami-id'] == 'ami-%s' % i) + + @staticmethod + def _fake_set_image_description(ctxt, image_id, description): + from nova.objectstore import handler + + class req: + pass + + request = req() + request.context = ctxt + request.args = {'image_id': [image_id], + 'description': [description]} + + resource = handler.ImagesResource() + resource.render_POST(request) + + def test_user_editable_image_endpoint(self): + pathdir = os.path.join(FLAGS.images_path, 'ami-testing') + os.mkdir(pathdir) + info = {'isPublic': False} + with open(os.path.join(pathdir, 'info.json'), 'w') as f: + json.dump(info, f) + img = image.Image('ami-testing') + # self.cloud.set_image_description(self.context, 'ami-testing', + # 'Foo Img') + # NOTE(vish): Above won't work unless we start objectstore or create + # a fake version of api/ec2/images.py conn that can + # call methods directly instead of going through boto. + # for now, just cheat and call the method directly + self._fake_set_image_description(self.context, 'ami-testing', + 'Foo Img') + self.assertEqual('Foo Img', img.metadata['description']) + self._fake_set_image_description(self.context, 'ami-testing', '') + self.assertEqual('', img.metadata['description']) + + def test_update_of_instance_display_fields(self): + inst = db.instance_create(self.context, {}) + ec2_id = cloud.internal_id_to_ec2_id(inst['internal_id']) + self.cloud.update_instance(self.context, ec2_id, + display_name='c00l 1m4g3') + inst = db.instance_get(self.context, inst['id']) + self.assertEqual('c00l 1m4g3', inst['display_name']) + db.instance_destroy(self.context, inst['id']) + + def test_update_of_instance_wont_update_private_fields(self): + inst = db.instance_create(self.context, {}) + self.cloud.update_instance(self.context, inst['id'], + mac_address='DE:AD:BE:EF') + inst = db.instance_get(self.context, inst['id']) + self.assertEqual(None, inst['mac_address']) + db.instance_destroy(self.context, inst['id']) + + def test_update_of_volume_display_fields(self): + vol = db.volume_create(self.context, {}) + self.cloud.update_volume(self.context, vol['id'], + display_name='c00l v0lum3') + vol = db.volume_get(self.context, vol['id']) + self.assertEqual('c00l v0lum3', vol['display_name']) + db.volume_destroy(self.context, vol['id']) + + def test_update_of_volume_wont_update_private_fields(self): + vol = db.volume_create(self.context, {}) + self.cloud.update_volume(self.context, vol['id'], + mountpoint='/not/here') + vol = db.volume_get(self.context, vol['id']) + self.assertEqual(None, vol['mountpoint']) + db.volume_destroy(self.context, vol['id']) diff --git a/nova/tests/test_compute.py b/nova/tests/test_compute.py new file mode 100644 index 000000000..c6353d357 --- /dev/null +++ b/nova/tests/test_compute.py @@ -0,0 +1,155 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. +""" +Tests For Compute +""" + +import datetime +import logging + +from nova import context +from nova import db +from nova import exception +from nova import flags +from nova import test +from nova import utils +from nova.auth import manager +from nova.compute import api as compute_api + + +FLAGS = flags.FLAGS + + +class ComputeTestCase(test.TestCase): + """Test case for compute""" + def setUp(self): + logging.getLogger().setLevel(logging.DEBUG) + super(ComputeTestCase, self).setUp() + self.flags(connection_type='fake', + network_manager='nova.network.manager.FlatManager') + self.compute = utils.import_object(FLAGS.compute_manager) + self.compute_api = compute_api.ComputeAPI() + self.manager = manager.AuthManager() + self.user = self.manager.create_user('fake', 'fake', 'fake') + self.project = self.manager.create_project('fake', 'fake', 'fake') + self.context = context.get_admin_context() + + def tearDown(self): + self.manager.delete_user(self.user) + self.manager.delete_project(self.project) + super(ComputeTestCase, self).tearDown() + + def _create_instance(self): + """Create a test instance""" + inst = {} + inst['image_id'] = 'ami-test' + inst['reservation_id'] = 'r-fakeres' + inst['launch_time'] = '10' + inst['user_id'] = self.user.id + inst['project_id'] = self.project.id + inst['instance_type'] = 'm1.tiny' + inst['mac_address'] = utils.generate_mac() + inst['ami_launch_index'] = 0 + return db.instance_create(self.context, inst)['id'] + + def test_create_instance_defaults_display_name(self): + """Verify that an instance cannot be created without a display_name.""" + cases = [dict(), dict(display_name=None)] + for instance in cases: + ref = self.compute_api.create_instances(self.context, + FLAGS.default_instance_type, None, **instance) + try: + self.assertNotEqual(ref[0].display_name, None) + finally: + db.instance_destroy(self.context, ref[0]['id']) + + def test_create_instance_associates_security_groups(self): + """Make sure create_instances associates security groups""" + values = {'name': 'default', + 'description': 'default', + 'user_id': self.user.id, + 'project_id': self.project.id} + group = db.security_group_create(self.context, values) + ref = self.compute_api.create_instances(self.context, + FLAGS.default_instance_type, None, security_group=['default']) + try: + self.assertEqual(len(ref[0]['security_groups']), 1) + finally: + db.security_group_destroy(self.context, group['id']) + db.instance_destroy(self.context, ref[0]['id']) + + def test_run_terminate(self): + """Make sure it is possible to run and terminate instance""" + instance_id = self._create_instance() + + self.compute.run_instance(self.context, instance_id) + + instances = db.instance_get_all(context.get_admin_context()) + logging.info("Running instances: %s", instances) + self.assertEqual(len(instances), 1) + + self.compute.terminate_instance(self.context, instance_id) + + instances = db.instance_get_all(context.get_admin_context()) + logging.info("After terminating instances: %s", instances) + self.assertEqual(len(instances), 0) + + def test_run_terminate_timestamps(self): + """Make sure timestamps are set for launched and destroyed""" + instance_id = self._create_instance() + instance_ref = db.instance_get(self.context, instance_id) + self.assertEqual(instance_ref['launched_at'], None) + self.assertEqual(instance_ref['deleted_at'], None) + launch = datetime.datetime.utcnow() + self.compute.run_instance(self.context, instance_id) + instance_ref = db.instance_get(self.context, instance_id) + self.assert_(instance_ref['launched_at'] > launch) + self.assertEqual(instance_ref['deleted_at'], None) + terminate = datetime.datetime.utcnow() + self.compute.terminate_instance(self.context, instance_id) + self.context = self.context.elevated(True) + instance_ref = db.instance_get(self.context, instance_id) + self.assert_(instance_ref['launched_at'] < terminate) + self.assert_(instance_ref['deleted_at'] > terminate) + + def test_reboot(self): + """Ensure instance can be rebooted""" + instance_id = self._create_instance() + self.compute.run_instance(self.context, instance_id) + self.compute.reboot_instance(self.context, instance_id) + self.compute.terminate_instance(self.context, instance_id) + + def test_console_output(self): + """Make sure we can get console output from instance""" + instance_id = self._create_instance() + self.compute.run_instance(self.context, instance_id) + + console = self.compute.get_console_output(self.context, + instance_id) + self.assert_(console) + self.compute.terminate_instance(self.context, instance_id) + + def test_run_instance_existing(self): + """Ensure failure when running an instance that already exists""" + instance_id = self._create_instance() + self.compute.run_instance(self.context, instance_id) + self.assertRaises(exception.Error, + self.compute.run_instance, + self.context, + instance_id) + self.compute.terminate_instance(self.context, instance_id) diff --git a/nova/tests/test_flags.py b/nova/tests/test_flags.py new file mode 100644 index 000000000..707300fcf --- /dev/null +++ b/nova/tests/test_flags.py @@ -0,0 +1,102 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. + +from nova import exception +from nova import flags +from nova import test + +FLAGS = flags.FLAGS +flags.DEFINE_string('flags_unittest', 'foo', 'for testing purposes only') + + +class FlagsTestCase(test.TestCase): + + def setUp(self): + super(FlagsTestCase, self).setUp() + self.FLAGS = flags.FlagValues() + self.global_FLAGS = flags.FLAGS + + def test_define(self): + self.assert_('string' not in self.FLAGS) + self.assert_('int' not in self.FLAGS) + self.assert_('false' not in self.FLAGS) + self.assert_('true' not in self.FLAGS) + + flags.DEFINE_string('string', 'default', 'desc', + flag_values=self.FLAGS) + flags.DEFINE_integer('int', 1, 'desc', flag_values=self.FLAGS) + flags.DEFINE_bool('false', False, 'desc', flag_values=self.FLAGS) + flags.DEFINE_bool('true', True, 'desc', flag_values=self.FLAGS) + + self.assert_(self.FLAGS['string']) + self.assert_(self.FLAGS['int']) + self.assert_(self.FLAGS['false']) + self.assert_(self.FLAGS['true']) + self.assertEqual(self.FLAGS.string, 'default') + self.assertEqual(self.FLAGS.int, 1) + self.assertEqual(self.FLAGS.false, False) + self.assertEqual(self.FLAGS.true, True) + + argv = ['flags_test', + '--string', 'foo', + '--int', '2', + '--false', + '--notrue'] + + self.FLAGS(argv) + self.assertEqual(self.FLAGS.string, 'foo') + self.assertEqual(self.FLAGS.int, 2) + self.assertEqual(self.FLAGS.false, True) + self.assertEqual(self.FLAGS.true, False) + + def test_declare(self): + self.assert_('answer' not in self.global_FLAGS) + flags.DECLARE('answer', 'nova.tests.declare_flags') + self.assert_('answer' in self.global_FLAGS) + self.assertEqual(self.global_FLAGS.answer, 42) + + # Make sure we don't overwrite anything + self.global_FLAGS.answer = 256 + self.assertEqual(self.global_FLAGS.answer, 256) + flags.DECLARE('answer', 'nova.tests.declare_flags') + self.assertEqual(self.global_FLAGS.answer, 256) + + def test_runtime_and_unknown_flags(self): + self.assert_('runtime_answer' not in self.global_FLAGS) + + argv = ['flags_test', '--runtime_answer=60', 'extra_arg'] + args = self.global_FLAGS(argv) + self.assertEqual(len(args), 2) + self.assertEqual(args[1], 'extra_arg') + + self.assert_('runtime_answer' not in self.global_FLAGS) + + import nova.tests.runtime_flags + + self.assert_('runtime_answer' in self.global_FLAGS) + self.assertEqual(self.global_FLAGS.runtime_answer, 60) + + def test_flag_leak_left(self): + self.assertEqual(FLAGS.flags_unittest, 'foo') + FLAGS.flags_unittest = 'bar' + self.assertEqual(FLAGS.flags_unittest, 'bar') + + def test_flag_leak_right(self): + self.assertEqual(FLAGS.flags_unittest, 'foo') + FLAGS.flags_unittest = 'bar' + self.assertEqual(FLAGS.flags_unittest, 'bar') diff --git a/nova/tests/test_misc.py b/nova/tests/test_misc.py new file mode 100644 index 000000000..33c1777d5 --- /dev/null +++ b/nova/tests/test_misc.py @@ -0,0 +1,55 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 OpenStack LLC +# +# 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. + +import os + +from nova import test +from nova.utils import parse_mailmap, str_dict_replace + + +class ProjectTestCase(test.TestCase): + def test_authors_up_to_date(self): + if os.path.exists('.bzr'): + contributors = set() + + mailmap = parse_mailmap('.mailmap') + + import bzrlib.workingtree + tree = bzrlib.workingtree.WorkingTree.open('.') + tree.lock_read() + try: + parents = tree.get_parent_ids() + g = tree.branch.repository.get_graph() + for p in parents[1:]: + rev_ids = [r for r, _ in g.iter_ancestry(parents) + if r != "null:"] + revs = tree.branch.repository.get_revisions(rev_ids) + for r in revs: + for author in r.get_apparent_authors(): + email = author.split(' ')[-1] + contributors.add(str_dict_replace(email, mailmap)) + + authors_file = open('Authors', 'r').read() + + missing = set() + for contributor in contributors: + if not contributor in authors_file: + missing.add(contributor) + + self.assertTrue(len(missing) == 0, + '%r not listed in Authors' % missing) + finally: + tree.unlock() diff --git a/nova/tests/test_quota.py b/nova/tests/test_quota.py new file mode 100644 index 000000000..8cf2a5e54 --- /dev/null +++ b/nova/tests/test_quota.py @@ -0,0 +1,153 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. + +import logging + +from nova import context +from nova import db +from nova import exception +from nova import flags +from nova import quota +from nova import test +from nova import utils +from nova.auth import manager +from nova.api.ec2 import cloud + + +FLAGS = flags.FLAGS + + +class QuotaTestCase(test.TestCase): + def setUp(self): + logging.getLogger().setLevel(logging.DEBUG) + super(QuotaTestCase, self).setUp() + self.flags(connection_type='fake', + quota_instances=2, + quota_cores=4, + quota_volumes=2, + quota_gigabytes=20, + quota_floating_ips=1) + + self.cloud = cloud.CloudController() + self.manager = manager.AuthManager() + self.user = self.manager.create_user('admin', 'admin', 'admin', True) + self.project = self.manager.create_project('admin', 'admin', 'admin') + self.network = utils.import_object(FLAGS.network_manager) + self.context = context.RequestContext(project=self.project, + user=self.user) + + def tearDown(self): + manager.AuthManager().delete_project(self.project) + manager.AuthManager().delete_user(self.user) + super(QuotaTestCase, self).tearDown() + + def _create_instance(self, cores=2): + """Create a test instance""" + inst = {} + inst['image_id'] = 'ami-test' + inst['reservation_id'] = 'r-fakeres' + inst['user_id'] = self.user.id + inst['project_id'] = self.project.id + inst['instance_type'] = 'm1.large' + inst['vcpus'] = cores + inst['mac_address'] = utils.generate_mac() + return db.instance_create(self.context, inst)['id'] + + def _create_volume(self, size=10): + """Create a test volume""" + vol = {} + vol['user_id'] = self.user.id + vol['project_id'] = self.project.id + vol['size'] = size + return db.volume_create(self.context, vol)['id'] + + def test_quota_overrides(self): + """Make sure overriding a projects quotas works""" + num_instances = quota.allowed_instances(self.context, 100, 'm1.small') + self.assertEqual(num_instances, 2) + db.quota_create(self.context, {'project_id': self.project.id, + 'instances': 10}) + num_instances = quota.allowed_instances(self.context, 100, 'm1.small') + self.assertEqual(num_instances, 4) + db.quota_update(self.context, self.project.id, {'cores': 100}) + num_instances = quota.allowed_instances(self.context, 100, 'm1.small') + self.assertEqual(num_instances, 10) + db.quota_destroy(self.context, self.project.id) + + def test_too_many_instances(self): + instance_ids = [] + for i in range(FLAGS.quota_instances): + instance_id = self._create_instance() + instance_ids.append(instance_id) + self.assertRaises(quota.QuotaError, self.cloud.run_instances, + self.context, + min_count=1, + max_count=1, + instance_type='m1.small', + image_id='fake') + for instance_id in instance_ids: + db.instance_destroy(self.context, instance_id) + + def test_too_many_cores(self): + instance_ids = [] + instance_id = self._create_instance(cores=4) + instance_ids.append(instance_id) + self.assertRaises(quota.QuotaError, self.cloud.run_instances, + self.context, + min_count=1, + max_count=1, + instance_type='m1.small', + image_id='fake') + for instance_id in instance_ids: + db.instance_destroy(self.context, instance_id) + + def test_too_many_volumes(self): + volume_ids = [] + for i in range(FLAGS.quota_volumes): + volume_id = self._create_volume() + volume_ids.append(volume_id) + self.assertRaises(quota.QuotaError, self.cloud.create_volume, + self.context, + size=10) + for volume_id in volume_ids: + db.volume_destroy(self.context, volume_id) + + def test_too_many_gigabytes(self): + volume_ids = [] + volume_id = self._create_volume(size=20) + volume_ids.append(volume_id) + self.assertRaises(quota.QuotaError, + self.cloud.create_volume, + self.context, + size=10) + for volume_id in volume_ids: + db.volume_destroy(self.context, volume_id) + + def test_too_many_addresses(self): + address = '192.168.0.100' + db.floating_ip_create(context.get_admin_context(), + {'address': address, 'host': FLAGS.host}) + float_addr = self.network.allocate_floating_ip(self.context, + self.project.id) + # NOTE(vish): This assert never fails. When cloud attempts to + # make an rpc.call, the test just finishes with OK. It + # appears to be something in the magic inline callbacks + # that is breaking. + self.assertRaises(quota.QuotaError, self.cloud.allocate_address, + self.context) + db.floating_ip_destroy(context.get_admin_context(), address) diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py new file mode 100644 index 000000000..a2495e65a --- /dev/null +++ b/nova/tests/test_rpc.py @@ -0,0 +1,103 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. +""" +Unit Tests for remote procedure calls using queue +""" +import logging + +from nova import context +from nova import flags +from nova import rpc +from nova import test + + +FLAGS = flags.FLAGS + + +class RpcTestCase(test.TestCase): + """Test cases for rpc""" + def setUp(self): + super(RpcTestCase, self).setUp() + self.conn = rpc.Connection.instance() + self.receiver = TestReceiver() + self.consumer = rpc.AdapterConsumer(connection=self.conn, + topic='test', + proxy=self.receiver) + self.consumer.attach_to_eventlet() + self.context = context.get_admin_context() + + def test_call_succeed(self): + """Get a value through rpc call""" + value = 42 + result = rpc.call(self.context, 'test', {"method": "echo", + "args": {"value": value}}) + self.assertEqual(value, result) + + def test_context_passed(self): + """Makes sure a context is passed through rpc call""" + value = 42 + result = rpc.call(self.context, + 'test', {"method": "context", + "args": {"value": value}}) + self.assertEqual(self.context.to_dict(), result) + + def test_call_exception(self): + """Test that exception gets passed back properly + + rpc.call returns a RemoteError object. The value of the + exception is converted to a string, so we convert it back + to an int in the test. + """ + value = 42 + self.assertRaises(rpc.RemoteError, + rpc.call, + self.context, + 'test', + {"method": "fail", + "args": {"value": value}}) + try: + rpc.call(self.context, + 'test', + {"method": "fail", + "args": {"value": value}}) + self.fail("should have thrown rpc.RemoteError") + except rpc.RemoteError as exc: + self.assertEqual(int(exc.value), value) + + +class TestReceiver(object): + """Simple Proxy class so the consumer has methods to call + + Uses static methods because we aren't actually storing any state""" + + @staticmethod + def echo(context, value): + """Simply returns whatever value is sent in""" + logging.debug("Received %s", value) + return value + + @staticmethod + def context(context, value): + """Returns dictionary version of context""" + logging.debug("Received %s", context) + return context.to_dict() + + @staticmethod + def fail(context, value): + """Raises an exception with the value sent in""" + raise Exception(value) diff --git a/nova/tests/test_scheduler.py b/nova/tests/test_scheduler.py new file mode 100644 index 000000000..c56f69698 --- /dev/null +++ b/nova/tests/test_scheduler.py @@ -0,0 +1,246 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. +""" +Tests For Scheduler +""" + +from nova import context +from nova import db +from nova import flags +from nova import service +from nova import test +from nova import rpc +from nova import utils +from nova.auth import manager as auth_manager +from nova.scheduler import manager +from nova.scheduler import driver + + +FLAGS = flags.FLAGS +flags.DECLARE('max_cores', 'nova.scheduler.simple') + + +class TestDriver(driver.Scheduler): + """Scheduler Driver for Tests""" + def schedule(context, topic, *args, **kwargs): + return 'fallback_host' + + def schedule_named_method(context, topic, num): + return 'named_host' + + +class SchedulerTestCase(test.TestCase): + """Test case for scheduler""" + def setUp(self): + super(SchedulerTestCase, self).setUp() + self.flags(scheduler_driver='nova.tests.test_scheduler.TestDriver') + + def test_fallback(self): + scheduler = manager.SchedulerManager() + self.mox.StubOutWithMock(rpc, 'cast', use_mock_anything=True) + ctxt = context.get_admin_context() + rpc.cast(ctxt, + 'topic.fallback_host', + {'method': 'noexist', + 'args': {'num': 7}}) + self.mox.ReplayAll() + scheduler.noexist(ctxt, 'topic', num=7) + + def test_named_method(self): + scheduler = manager.SchedulerManager() + self.mox.StubOutWithMock(rpc, 'cast', use_mock_anything=True) + ctxt = context.get_admin_context() + rpc.cast(ctxt, + 'topic.named_host', + {'method': 'named_method', + 'args': {'num': 7}}) + self.mox.ReplayAll() + scheduler.named_method(ctxt, 'topic', num=7) + + +class SimpleDriverTestCase(test.TestCase): + """Test case for simple driver""" + def setUp(self): + super(SimpleDriverTestCase, self).setUp() + self.flags(connection_type='fake', + max_cores=4, + max_gigabytes=4, + network_manager='nova.network.manager.FlatManager', + volume_driver='nova.volume.driver.FakeISCSIDriver', + scheduler_driver='nova.scheduler.simple.SimpleScheduler') + self.scheduler = manager.SchedulerManager() + self.manager = auth_manager.AuthManager() + self.user = self.manager.create_user('fake', 'fake', 'fake') + self.project = self.manager.create_project('fake', 'fake', 'fake') + self.context = context.get_admin_context() + + def tearDown(self): + self.manager.delete_user(self.user) + self.manager.delete_project(self.project) + + def _create_instance(self): + """Create a test instance""" + inst = {} + inst['image_id'] = 'ami-test' + inst['reservation_id'] = 'r-fakeres' + inst['user_id'] = self.user.id + inst['project_id'] = self.project.id + inst['instance_type'] = 'm1.tiny' + inst['mac_address'] = utils.generate_mac() + inst['ami_launch_index'] = 0 + inst['vcpus'] = 1 + return db.instance_create(self.context, inst)['id'] + + def _create_volume(self): + """Create a test volume""" + vol = {} + vol['image_id'] = 'ami-test' + vol['reservation_id'] = 'r-fakeres' + vol['size'] = 1 + return db.volume_create(self.context, vol)['id'] + + def test_hosts_are_up(self): + """Ensures driver can find the hosts that are up""" + # NOTE(vish): constructing service without create method + # because we are going to use it without queue + compute1 = service.Service('host1', + 'nova-compute', + 'compute', + FLAGS.compute_manager) + compute1.start() + compute2 = service.Service('host2', + 'nova-compute', + 'compute', + FLAGS.compute_manager) + compute2.start() + hosts = self.scheduler.driver.hosts_up(self.context, 'compute') + self.assertEqual(len(hosts), 2) + compute1.kill() + compute2.kill() + + def test_least_busy_host_gets_instance(self): + """Ensures the host with less cores gets the next one""" + compute1 = service.Service('host1', + 'nova-compute', + 'compute', + FLAGS.compute_manager) + compute1.start() + compute2 = service.Service('host2', + 'nova-compute', + 'compute', + FLAGS.compute_manager) + compute2.start() + instance_id1 = self._create_instance() + compute1.run_instance(self.context, instance_id1) + instance_id2 = self._create_instance() + host = self.scheduler.driver.schedule_run_instance(self.context, + instance_id2) + self.assertEqual(host, 'host2') + compute1.terminate_instance(self.context, instance_id1) + db.instance_destroy(self.context, instance_id2) + compute1.kill() + compute2.kill() + + def test_too_many_cores(self): + """Ensures we don't go over max cores""" + compute1 = service.Service('host1', + 'nova-compute', + 'compute', + FLAGS.compute_manager) + compute1.start() + compute2 = service.Service('host2', + 'nova-compute', + 'compute', + FLAGS.compute_manager) + compute2.start() + instance_ids1 = [] + instance_ids2 = [] + for index in xrange(FLAGS.max_cores): + instance_id = self._create_instance() + compute1.run_instance(self.context, instance_id) + instance_ids1.append(instance_id) + instance_id = self._create_instance() + compute2.run_instance(self.context, instance_id) + instance_ids2.append(instance_id) + instance_id = self._create_instance() + self.assertRaises(driver.NoValidHost, + self.scheduler.driver.schedule_run_instance, + self.context, + instance_id) + for instance_id in instance_ids1: + compute1.terminate_instance(self.context, instance_id) + for instance_id in instance_ids2: + compute2.terminate_instance(self.context, instance_id) + compute1.kill() + compute2.kill() + + def test_least_busy_host_gets_volume(self): + """Ensures the host with less gigabytes gets the next one""" + volume1 = service.Service('host1', + 'nova-volume', + 'volume', + FLAGS.volume_manager) + volume1.start() + volume2 = service.Service('host2', + 'nova-volume', + 'volume', + FLAGS.volume_manager) + volume2.start() + volume_id1 = self._create_volume() + volume1.create_volume(self.context, volume_id1) + volume_id2 = self._create_volume() + host = self.scheduler.driver.schedule_create_volume(self.context, + volume_id2) + self.assertEqual(host, 'host2') + volume1.delete_volume(self.context, volume_id1) + db.volume_destroy(self.context, volume_id2) + volume1.kill() + volume2.kill() + + def test_too_many_gigabytes(self): + """Ensures we don't go over max gigabytes""" + volume1 = service.Service('host1', + 'nova-volume', + 'volume', + FLAGS.volume_manager) + volume1.start() + volume2 = service.Service('host2', + 'nova-volume', + 'volume', + FLAGS.volume_manager) + volume2.start() + volume_ids1 = [] + volume_ids2 = [] + for index in xrange(FLAGS.max_gigabytes): + volume_id = self._create_volume() + volume1.create_volume(self.context, volume_id) + volume_ids1.append(volume_id) + volume_id = self._create_volume() + volume2.create_volume(self.context, volume_id) + volume_ids2.append(volume_id) + volume_id = self._create_volume() + self.assertRaises(driver.NoValidHost, + self.scheduler.driver.schedule_create_volume, + self.context, + volume_id) + for volume_id in volume_ids1: + volume1.delete_volume(self.context, volume_id) + for volume_id in volume_ids2: + volume2.delete_volume(self.context, volume_id) + volume1.kill() + volume2.kill() diff --git a/nova/tests/test_service.py b/nova/tests/test_service.py new file mode 100644 index 000000000..b30838ad7 --- /dev/null +++ b/nova/tests/test_service.py @@ -0,0 +1,227 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. + +""" +Unit Tests for remote procedure calls using queue +""" + +import mox + +from nova import exception +from nova import flags +from nova import rpc +from nova import test +from nova import service +from nova import manager + +FLAGS = flags.FLAGS +flags.DEFINE_string("fake_manager", "nova.tests.test_service.FakeManager", + "Manager for testing") + + +class FakeManager(manager.Manager): + """Fake manager for tests""" + def test_method(self): + return 'manager' + + +class ExtendedService(service.Service): + def test_method(self): + return 'service' + + +class ServiceManagerTestCase(test.TestCase): + """Test cases for Services""" + + def test_attribute_error_for_no_manager(self): + serv = service.Service('test', + 'test', + 'test', + 'nova.tests.test_service.FakeManager') + self.assertRaises(AttributeError, getattr, serv, 'test_method') + + def test_message_gets_to_manager(self): + serv = service.Service('test', + 'test', + 'test', + 'nova.tests.test_service.FakeManager') + serv.start() + self.assertEqual(serv.test_method(), 'manager') + + def test_override_manager_method(self): + serv = ExtendedService('test', + 'test', + 'test', + 'nova.tests.test_service.FakeManager') + serv.start() + self.assertEqual(serv.test_method(), 'service') + + +class ServiceTestCase(test.TestCase): + """Test cases for Services""" + + def setUp(self): + super(ServiceTestCase, self).setUp() + self.mox.StubOutWithMock(service, 'db') + + def test_create(self): + host = 'foo' + binary = 'nova-fake' + topic = 'fake' + + # NOTE(vish): Create was moved out of mox replay to make sure that + # the looping calls are created in StartService. + app = service.Service.create(host=host, binary=binary) + + self.mox.StubOutWithMock(rpc, + 'AdapterConsumer', + use_mock_anything=True) + rpc.AdapterConsumer(connection=mox.IgnoreArg(), + topic=topic, + proxy=mox.IsA(service.Service)).AndReturn( + rpc.AdapterConsumer) + + rpc.AdapterConsumer(connection=mox.IgnoreArg(), + topic='%s.%s' % (topic, host), + proxy=mox.IsA(service.Service)).AndReturn( + rpc.AdapterConsumer) + + rpc.AdapterConsumer.attach_to_eventlet() + rpc.AdapterConsumer.attach_to_eventlet() + + service_create = {'host': host, + 'binary': binary, + 'topic': topic, + 'report_count': 0} + service_ref = {'host': host, + 'binary': binary, + 'report_count': 0, + 'id': 1} + + service.db.service_get_by_args(mox.IgnoreArg(), + host, + binary).AndRaise(exception.NotFound()) + service.db.service_create(mox.IgnoreArg(), + service_create).AndReturn(service_ref) + self.mox.ReplayAll() + + app.start() + app.stop() + self.assert_(app) + + # We're testing sort of weird behavior in how report_state decides + # whether it is disconnected, it looks for a variable on itself called + # 'model_disconnected' and report_state doesn't really do much so this + # these are mostly just for coverage + def test_report_state_no_service(self): + host = 'foo' + binary = 'bar' + topic = 'test' + service_create = {'host': host, + 'binary': binary, + 'topic': topic, + 'report_count': 0} + service_ref = {'host': host, + 'binary': binary, + 'topic': topic, + 'report_count': 0, + 'id': 1} + + service.db.service_get_by_args(mox.IgnoreArg(), + host, + binary).AndRaise(exception.NotFound()) + service.db.service_create(mox.IgnoreArg(), + service_create).AndReturn(service_ref) + service.db.service_get(mox.IgnoreArg(), + service_ref['id']).AndReturn(service_ref) + service.db.service_update(mox.IgnoreArg(), service_ref['id'], + mox.ContainsKeyValue('report_count', 1)) + + self.mox.ReplayAll() + serv = service.Service(host, + binary, + topic, + 'nova.tests.test_service.FakeManager') + serv.start() + serv.report_state() + + def test_report_state_newly_disconnected(self): + host = 'foo' + binary = 'bar' + topic = 'test' + service_create = {'host': host, + 'binary': binary, + 'topic': topic, + 'report_count': 0} + service_ref = {'host': host, + 'binary': binary, + 'topic': topic, + 'report_count': 0, + 'id': 1} + + service.db.service_get_by_args(mox.IgnoreArg(), + host, + binary).AndRaise(exception.NotFound()) + service.db.service_create(mox.IgnoreArg(), + service_create).AndReturn(service_ref) + service.db.service_get(mox.IgnoreArg(), + mox.IgnoreArg()).AndRaise(Exception()) + + self.mox.ReplayAll() + serv = service.Service(host, + binary, + topic, + 'nova.tests.test_service.FakeManager') + serv.start() + serv.report_state() + self.assert_(serv.model_disconnected) + + def test_report_state_newly_connected(self): + host = 'foo' + binary = 'bar' + topic = 'test' + service_create = {'host': host, + 'binary': binary, + 'topic': topic, + 'report_count': 0} + service_ref = {'host': host, + 'binary': binary, + 'topic': topic, + 'report_count': 0, + 'id': 1} + + service.db.service_get_by_args(mox.IgnoreArg(), + host, + binary).AndRaise(exception.NotFound()) + service.db.service_create(mox.IgnoreArg(), + service_create).AndReturn(service_ref) + service.db.service_get(mox.IgnoreArg(), + service_ref['id']).AndReturn(service_ref) + service.db.service_update(mox.IgnoreArg(), service_ref['id'], + mox.ContainsKeyValue('report_count', 1)) + + self.mox.ReplayAll() + serv = service.Service(host, + binary, + topic, + 'nova.tests.test_service.FakeManager') + serv.start() + serv.model_disconnected = True + serv.report_state() + + self.assert_(not serv.model_disconnected) diff --git a/nova/tests/test_twistd.py b/nova/tests/test_twistd.py new file mode 100644 index 000000000..75007b9c8 --- /dev/null +++ b/nova/tests/test_twistd.py @@ -0,0 +1,53 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. + +import StringIO +import sys + +from nova import twistd +from nova import exception +from nova import flags +from nova import test + + +FLAGS = flags.FLAGS + + +class TwistdTestCase(test.TrialTestCase): + def setUp(self): + super(TwistdTestCase, self).setUp() + self.Options = twistd.WrapTwistedOptions(twistd.TwistdServerOptions) + sys.stdout = StringIO.StringIO() + + def tearDown(self): + super(TwistdTestCase, self).tearDown() + sys.stdout = sys.__stdout__ + + def test_basic(self): + options = self.Options() + argv = options.parseOptions() + + def test_logfile(self): + options = self.Options() + argv = options.parseOptions(['--logfile=foo']) + self.assertEqual(FLAGS.logfile, 'foo') + + def test_help(self): + options = self.Options() + self.assertRaises(SystemExit, options.parseOptions, ['--help']) + self.assert_('pidfile' in sys.stdout.getvalue()) diff --git a/nova/tests/test_virt.py b/nova/tests/test_virt.py new file mode 100644 index 000000000..85e569858 --- /dev/null +++ b/nova/tests/test_virt.py @@ -0,0 +1,258 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 +# +# Copyright 2010 OpenStack LLC +# +# 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. + +from xml.etree.ElementTree import fromstring as xml_to_tree +from xml.dom.minidom import parseString as xml_to_dom + +from nova import context +from nova import db +from nova import flags +from nova import test +from nova import utils +from nova.api.ec2 import cloud +from nova.auth import manager +from nova.virt import libvirt_conn + +FLAGS = flags.FLAGS +flags.DECLARE('instances_path', 'nova.compute.manager') + + +class LibvirtConnTestCase(test.TestCase): + def setUp(self): + super(LibvirtConnTestCase, self).setUp() + self.manager = manager.AuthManager() + self.user = self.manager.create_user('fake', 'fake', 'fake', + admin=True) + self.project = self.manager.create_project('fake', 'fake', 'fake') + self.network = utils.import_object(FLAGS.network_manager) + FLAGS.instances_path = '' + + def test_get_uri_and_template(self): + ip = '10.11.12.13' + + instance = {'internal_id': 1, + 'memory_kb': '1024000', + 'basepath': '/some/path', + 'bridge_name': 'br100', + 'mac_address': '02:12:34:46:56:67', + 'vcpus': 2, + 'project_id': 'fake', + 'bridge': 'br101', + 'instance_type': 'm1.small'} + + user_context = context.RequestContext(project=self.project, + user=self.user) + instance_ref = db.instance_create(user_context, instance) + network_ref = self.network.get_network(user_context) + self.network.set_network_host(context.get_admin_context(), + network_ref['id']) + + fixed_ip = {'address': ip, + 'network_id': network_ref['id']} + + ctxt = context.get_admin_context() + fixed_ip_ref = db.fixed_ip_create(ctxt, fixed_ip) + db.fixed_ip_update(ctxt, ip, {'allocated': True, + 'instance_id': instance_ref['id']}) + + type_uri_map = {'qemu': ('qemu:///system', + [(lambda t: t.find('.').get('type'), 'qemu'), + (lambda t: t.find('./os/type').text, 'hvm'), + (lambda t: t.find('./devices/emulator'), None)]), + 'kvm': ('qemu:///system', + [(lambda t: t.find('.').get('type'), 'kvm'), + (lambda t: t.find('./os/type').text, 'hvm'), + (lambda t: t.find('./devices/emulator'), None)]), + 'uml': ('uml:///system', + [(lambda t: t.find('.').get('type'), 'uml'), + (lambda t: t.find('./os/type').text, 'uml')])} + + common_checks = [ + (lambda t: t.find('.').tag, 'domain'), + (lambda t: t.find('./devices/interface/filterref/parameter').\ + get('name'), 'IP'), + (lambda t: t.find('./devices/interface/filterref/parameter').\ + get('value'), '10.11.12.13')] + + for (libvirt_type, (expected_uri, checks)) in type_uri_map.iteritems(): + FLAGS.libvirt_type = libvirt_type + conn = libvirt_conn.LibvirtConnection(True) + + uri, _template, _rescue = conn.get_uri_and_templates() + self.assertEquals(uri, expected_uri) + + xml = conn.to_xml(instance_ref) + tree = xml_to_tree(xml) + for i, (check, expected_result) in enumerate(checks): + self.assertEqual(check(tree), + expected_result, + '%s failed check %d' % (xml, i)) + + for i, (check, expected_result) in enumerate(common_checks): + self.assertEqual(check(tree), + expected_result, + '%s failed common check %d' % (xml, i)) + + # Deliberately not just assigning this string to FLAGS.libvirt_uri and + # checking against that later on. This way we make sure the + # implementation doesn't fiddle around with the FLAGS. + testuri = 'something completely different' + FLAGS.libvirt_uri = testuri + for (libvirt_type, (expected_uri, checks)) in type_uri_map.iteritems(): + FLAGS.libvirt_type = libvirt_type + conn = libvirt_conn.LibvirtConnection(True) + uri, _template, _rescue = conn.get_uri_and_templates() + self.assertEquals(uri, testuri) + + def tearDown(self): + super(LibvirtConnTestCase, self).tearDown() + self.manager.delete_project(self.project) + self.manager.delete_user(self.user) + + +class NWFilterTestCase(test.TestCase): + + def setUp(self): + super(NWFilterTestCase, self).setUp() + + class Mock(object): + pass + + self.manager = manager.AuthManager() + self.user = self.manager.create_user('fake', 'fake', 'fake', + admin=True) + self.project = self.manager.create_project('fake', 'fake', 'fake') + self.context = context.RequestContext(self.user, self.project) + + self.fake_libvirt_connection = Mock() + + self.fw = libvirt_conn.NWFilterFirewall(self.fake_libvirt_connection) + + def tearDown(self): + self.manager.delete_project(self.project) + self.manager.delete_user(self.user) + + def test_cidr_rule_nwfilter_xml(self): + cloud_controller = cloud.CloudController() + cloud_controller.create_security_group(self.context, + 'testgroup', + 'test group description') + cloud_controller.authorize_security_group_ingress(self.context, + 'testgroup', + from_port='80', + to_port='81', + ip_protocol='tcp', + cidr_ip='0.0.0.0/0') + + security_group = db.security_group_get_by_name(self.context, + 'fake', + 'testgroup') + + xml = self.fw.security_group_to_nwfilter_xml(security_group.id) + + dom = xml_to_dom(xml) + self.assertEqual(dom.firstChild.tagName, 'filter') + + rules = dom.getElementsByTagName('rule') + self.assertEqual(len(rules), 1) + + # It's supposed to allow inbound traffic. + self.assertEqual(rules[0].getAttribute('action'), 'accept') + self.assertEqual(rules[0].getAttribute('direction'), 'in') + + # Must be lower priority than the base filter (which blocks everything) + self.assertTrue(int(rules[0].getAttribute('priority')) < 1000) + + ip_conditions = rules[0].getElementsByTagName('tcp') + self.assertEqual(len(ip_conditions), 1) + self.assertEqual(ip_conditions[0].getAttribute('srcipaddr'), '0.0.0.0') + self.assertEqual(ip_conditions[0].getAttribute('srcipmask'), '0.0.0.0') + self.assertEqual(ip_conditions[0].getAttribute('dstportstart'), '80') + self.assertEqual(ip_conditions[0].getAttribute('dstportend'), '81') + self.teardown_security_group() + + def teardown_security_group(self): + cloud_controller = cloud.CloudController() + cloud_controller.delete_security_group(self.context, 'testgroup') + + def setup_and_return_security_group(self): + cloud_controller = cloud.CloudController() + cloud_controller.create_security_group(self.context, + 'testgroup', + 'test group description') + cloud_controller.authorize_security_group_ingress(self.context, + 'testgroup', + from_port='80', + to_port='81', + ip_protocol='tcp', + cidr_ip='0.0.0.0/0') + + return db.security_group_get_by_name(self.context, 'fake', 'testgroup') + + def test_creates_base_rule_first(self): + # These come pre-defined by libvirt + self.defined_filters = ['no-mac-spoofing', + 'no-ip-spoofing', + 'no-arp-spoofing', + 'allow-dhcp-server'] + + self.recursive_depends = {} + for f in self.defined_filters: + self.recursive_depends[f] = [] + + def _filterDefineXMLMock(xml): + dom = xml_to_dom(xml) + name = dom.firstChild.getAttribute('name') + self.recursive_depends[name] = [] + for f in dom.getElementsByTagName('filterref'): + ref = f.getAttribute('filter') + self.assertTrue(ref in self.defined_filters, + ('%s referenced filter that does ' + + 'not yet exist: %s') % (name, ref)) + dependencies = [ref] + self.recursive_depends[ref] + self.recursive_depends[name] += dependencies + + self.defined_filters.append(name) + return True + + self.fake_libvirt_connection.nwfilterDefineXML = _filterDefineXMLMock + + instance_ref = db.instance_create(self.context, + {'user_id': 'fake', + 'project_id': 'fake'}) + inst_id = instance_ref['id'] + + def _ensure_all_called(): + instance_filter = 'nova-instance-%s' % instance_ref['name'] + secgroup_filter = 'nova-secgroup-%s' % self.security_group['id'] + for required in [secgroup_filter, 'allow-dhcp-server', + 'no-arp-spoofing', 'no-ip-spoofing', + 'no-mac-spoofing']: + self.assertTrue(required in + self.recursive_depends[instance_filter], + "Instance's filter does not include %s" % + required) + + self.security_group = self.setup_and_return_security_group() + + db.instance_add_security_group(self.context, inst_id, + self.security_group.id) + instance = db.instance_get(self.context, inst_id) + + d = self.fw.setup_nwfilters_for_instance(instance) + _ensure_all_called() + self.teardown_security_group() + return d diff --git a/nova/tests/test_volume.py b/nova/tests/test_volume.py new file mode 100644 index 000000000..b13455fb0 --- /dev/null +++ b/nova/tests/test_volume.py @@ -0,0 +1,175 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. +""" +Tests for Volume Code. + +""" +import logging + +from nova import context +from nova import exception +from nova import db +from nova import flags +from nova import test +from nova import utils + +FLAGS = flags.FLAGS + + +class VolumeTestCase(test.TestCase): + """Test Case for volumes.""" + + def setUp(self): + logging.getLogger().setLevel(logging.DEBUG) + super(VolumeTestCase, self).setUp() + self.compute = utils.import_object(FLAGS.compute_manager) + self.flags(connection_type='fake') + self.volume = utils.import_object(FLAGS.volume_manager) + self.context = context.get_admin_context() + + @staticmethod + def _create_volume(size='0'): + """Create a volume object.""" + vol = {} + vol['size'] = size + vol['user_id'] = 'fake' + vol['project_id'] = 'fake' + vol['availability_zone'] = FLAGS.storage_availability_zone + vol['status'] = "creating" + vol['attach_status'] = "detached" + return db.volume_create(context.get_admin_context(), vol)['id'] + + def test_create_delete_volume(self): + """Test volume can be created and deleted.""" + volume_id = self._create_volume() + self.volume.create_volume(self.context, volume_id) + self.assertEqual(volume_id, db.volume_get(context.get_admin_context(), + volume_id).id) + + self.volume.delete_volume(self.context, volume_id) + self.assertRaises(exception.NotFound, + db.volume_get, + self.context, + volume_id) + + def test_too_big_volume(self): + """Ensure failure if a too large of a volume is requested.""" + # FIXME(vish): validation needs to move into the data layer in + # volume_create + return True + try: + volume_id = self._create_volume('1001') + self.volume.create_volume(self.context, volume_id) + self.fail("Should have thrown TypeError") + except TypeError: + pass + + def test_too_many_volumes(self): + """Ensure that NoMoreTargets is raised when we run out of volumes.""" + vols = [] + total_slots = FLAGS.iscsi_num_targets + for _index in xrange(total_slots): + volume_id = self._create_volume() + self.volume.create_volume(self.context, volume_id) + vols.append(volume_id) + volume_id = self._create_volume() + self.assertRaises(db.NoMoreTargets, + self.volume.create_volume, + self.context, + volume_id) + db.volume_destroy(context.get_admin_context(), volume_id) + for volume_id in vols: + self.volume.delete_volume(self.context, volume_id) + + def test_run_attach_detach_volume(self): + """Make sure volume can be attached and detached from instance.""" + inst = {} + inst['image_id'] = 'ami-test' + inst['reservation_id'] = 'r-fakeres' + inst['launch_time'] = '10' + inst['user_id'] = 'fake' + inst['project_id'] = 'fake' + inst['instance_type'] = 'm1.tiny' + inst['mac_address'] = utils.generate_mac() + inst['ami_launch_index'] = 0 + instance_id = db.instance_create(self.context, inst)['id'] + mountpoint = "/dev/sdf" + volume_id = self._create_volume() + self.volume.create_volume(self.context, volume_id) + if FLAGS.fake_tests: + db.volume_attached(self.context, volume_id, instance_id, + mountpoint) + else: + self.compute.attach_volume(self.context, + instance_id, + volume_id, + mountpoint) + vol = db.volume_get(context.get_admin_context(), volume_id) + self.assertEqual(vol['status'], "in-use") + self.assertEqual(vol['attach_status'], "attached") + self.assertEqual(vol['mountpoint'], mountpoint) + instance_ref = db.volume_get_instance(self.context, volume_id) + self.assertEqual(instance_ref['id'], instance_id) + + self.assertRaises(exception.Error, + self.volume.delete_volume, + self.context, + volume_id) + if FLAGS.fake_tests: + db.volume_detached(self.context, volume_id) + else: + self.compute.detach_volume(self.context, + instance_id, + volume_id) + vol = db.volume_get(self.context, volume_id) + self.assertEqual(vol['status'], "available") + + self.volume.delete_volume(self.context, volume_id) + self.assertRaises(exception.Error, + db.volume_get, + self.context, + volume_id) + db.instance_destroy(self.context, instance_id) + + def test_concurrent_volumes_get_different_targets(self): + """Ensure multiple concurrent volumes get different targets.""" + volume_ids = [] + targets = [] + + def _check(volume_id): + """Make sure targets aren't duplicated.""" + volume_ids.append(volume_id) + admin_context = context.get_admin_context() + iscsi_target = db.volume_get_iscsi_target_num(admin_context, + volume_id) + self.assert_(iscsi_target not in targets) + targets.append(iscsi_target) + logging.debug("Target %s allocated", iscsi_target) + total_slots = FLAGS.iscsi_num_targets + for _index in xrange(total_slots): + volume_id = self._create_volume() + d = self.volume.create_volume(self.context, volume_id) + _check(d) + for volume_id in volume_ids: + self.volume.delete_volume(self.context, volume_id) + + def test_multi_node(self): + # TODO(termie): Figure out how to test with two nodes, + # each of them having a different FLAG for storage_node + # This will allow us to test cross-node interactions + pass diff --git a/nova/tests/twistd_unittest.py b/nova/tests/twistd_unittest.py deleted file mode 100644 index 75007b9c8..000000000 --- a/nova/tests/twistd_unittest.py +++ /dev/null @@ -1,53 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. - -import StringIO -import sys - -from nova import twistd -from nova import exception -from nova import flags -from nova import test - - -FLAGS = flags.FLAGS - - -class TwistdTestCase(test.TrialTestCase): - def setUp(self): - super(TwistdTestCase, self).setUp() - self.Options = twistd.WrapTwistedOptions(twistd.TwistdServerOptions) - sys.stdout = StringIO.StringIO() - - def tearDown(self): - super(TwistdTestCase, self).tearDown() - sys.stdout = sys.__stdout__ - - def test_basic(self): - options = self.Options() - argv = options.parseOptions() - - def test_logfile(self): - options = self.Options() - argv = options.parseOptions(['--logfile=foo']) - self.assertEqual(FLAGS.logfile, 'foo') - - def test_help(self): - options = self.Options() - self.assertRaises(SystemExit, options.parseOptions, ['--help']) - self.assert_('pidfile' in sys.stdout.getvalue()) diff --git a/nova/tests/virt_unittest.py b/nova/tests/virt_unittest.py deleted file mode 100644 index 85e569858..000000000 --- a/nova/tests/virt_unittest.py +++ /dev/null @@ -1,258 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 -# -# Copyright 2010 OpenStack LLC -# -# 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. - -from xml.etree.ElementTree import fromstring as xml_to_tree -from xml.dom.minidom import parseString as xml_to_dom - -from nova import context -from nova import db -from nova import flags -from nova import test -from nova import utils -from nova.api.ec2 import cloud -from nova.auth import manager -from nova.virt import libvirt_conn - -FLAGS = flags.FLAGS -flags.DECLARE('instances_path', 'nova.compute.manager') - - -class LibvirtConnTestCase(test.TestCase): - def setUp(self): - super(LibvirtConnTestCase, self).setUp() - self.manager = manager.AuthManager() - self.user = self.manager.create_user('fake', 'fake', 'fake', - admin=True) - self.project = self.manager.create_project('fake', 'fake', 'fake') - self.network = utils.import_object(FLAGS.network_manager) - FLAGS.instances_path = '' - - def test_get_uri_and_template(self): - ip = '10.11.12.13' - - instance = {'internal_id': 1, - 'memory_kb': '1024000', - 'basepath': '/some/path', - 'bridge_name': 'br100', - 'mac_address': '02:12:34:46:56:67', - 'vcpus': 2, - 'project_id': 'fake', - 'bridge': 'br101', - 'instance_type': 'm1.small'} - - user_context = context.RequestContext(project=self.project, - user=self.user) - instance_ref = db.instance_create(user_context, instance) - network_ref = self.network.get_network(user_context) - self.network.set_network_host(context.get_admin_context(), - network_ref['id']) - - fixed_ip = {'address': ip, - 'network_id': network_ref['id']} - - ctxt = context.get_admin_context() - fixed_ip_ref = db.fixed_ip_create(ctxt, fixed_ip) - db.fixed_ip_update(ctxt, ip, {'allocated': True, - 'instance_id': instance_ref['id']}) - - type_uri_map = {'qemu': ('qemu:///system', - [(lambda t: t.find('.').get('type'), 'qemu'), - (lambda t: t.find('./os/type').text, 'hvm'), - (lambda t: t.find('./devices/emulator'), None)]), - 'kvm': ('qemu:///system', - [(lambda t: t.find('.').get('type'), 'kvm'), - (lambda t: t.find('./os/type').text, 'hvm'), - (lambda t: t.find('./devices/emulator'), None)]), - 'uml': ('uml:///system', - [(lambda t: t.find('.').get('type'), 'uml'), - (lambda t: t.find('./os/type').text, 'uml')])} - - common_checks = [ - (lambda t: t.find('.').tag, 'domain'), - (lambda t: t.find('./devices/interface/filterref/parameter').\ - get('name'), 'IP'), - (lambda t: t.find('./devices/interface/filterref/parameter').\ - get('value'), '10.11.12.13')] - - for (libvirt_type, (expected_uri, checks)) in type_uri_map.iteritems(): - FLAGS.libvirt_type = libvirt_type - conn = libvirt_conn.LibvirtConnection(True) - - uri, _template, _rescue = conn.get_uri_and_templates() - self.assertEquals(uri, expected_uri) - - xml = conn.to_xml(instance_ref) - tree = xml_to_tree(xml) - for i, (check, expected_result) in enumerate(checks): - self.assertEqual(check(tree), - expected_result, - '%s failed check %d' % (xml, i)) - - for i, (check, expected_result) in enumerate(common_checks): - self.assertEqual(check(tree), - expected_result, - '%s failed common check %d' % (xml, i)) - - # Deliberately not just assigning this string to FLAGS.libvirt_uri and - # checking against that later on. This way we make sure the - # implementation doesn't fiddle around with the FLAGS. - testuri = 'something completely different' - FLAGS.libvirt_uri = testuri - for (libvirt_type, (expected_uri, checks)) in type_uri_map.iteritems(): - FLAGS.libvirt_type = libvirt_type - conn = libvirt_conn.LibvirtConnection(True) - uri, _template, _rescue = conn.get_uri_and_templates() - self.assertEquals(uri, testuri) - - def tearDown(self): - super(LibvirtConnTestCase, self).tearDown() - self.manager.delete_project(self.project) - self.manager.delete_user(self.user) - - -class NWFilterTestCase(test.TestCase): - - def setUp(self): - super(NWFilterTestCase, self).setUp() - - class Mock(object): - pass - - self.manager = manager.AuthManager() - self.user = self.manager.create_user('fake', 'fake', 'fake', - admin=True) - self.project = self.manager.create_project('fake', 'fake', 'fake') - self.context = context.RequestContext(self.user, self.project) - - self.fake_libvirt_connection = Mock() - - self.fw = libvirt_conn.NWFilterFirewall(self.fake_libvirt_connection) - - def tearDown(self): - self.manager.delete_project(self.project) - self.manager.delete_user(self.user) - - def test_cidr_rule_nwfilter_xml(self): - cloud_controller = cloud.CloudController() - cloud_controller.create_security_group(self.context, - 'testgroup', - 'test group description') - cloud_controller.authorize_security_group_ingress(self.context, - 'testgroup', - from_port='80', - to_port='81', - ip_protocol='tcp', - cidr_ip='0.0.0.0/0') - - security_group = db.security_group_get_by_name(self.context, - 'fake', - 'testgroup') - - xml = self.fw.security_group_to_nwfilter_xml(security_group.id) - - dom = xml_to_dom(xml) - self.assertEqual(dom.firstChild.tagName, 'filter') - - rules = dom.getElementsByTagName('rule') - self.assertEqual(len(rules), 1) - - # It's supposed to allow inbound traffic. - self.assertEqual(rules[0].getAttribute('action'), 'accept') - self.assertEqual(rules[0].getAttribute('direction'), 'in') - - # Must be lower priority than the base filter (which blocks everything) - self.assertTrue(int(rules[0].getAttribute('priority')) < 1000) - - ip_conditions = rules[0].getElementsByTagName('tcp') - self.assertEqual(len(ip_conditions), 1) - self.assertEqual(ip_conditions[0].getAttribute('srcipaddr'), '0.0.0.0') - self.assertEqual(ip_conditions[0].getAttribute('srcipmask'), '0.0.0.0') - self.assertEqual(ip_conditions[0].getAttribute('dstportstart'), '80') - self.assertEqual(ip_conditions[0].getAttribute('dstportend'), '81') - self.teardown_security_group() - - def teardown_security_group(self): - cloud_controller = cloud.CloudController() - cloud_controller.delete_security_group(self.context, 'testgroup') - - def setup_and_return_security_group(self): - cloud_controller = cloud.CloudController() - cloud_controller.create_security_group(self.context, - 'testgroup', - 'test group description') - cloud_controller.authorize_security_group_ingress(self.context, - 'testgroup', - from_port='80', - to_port='81', - ip_protocol='tcp', - cidr_ip='0.0.0.0/0') - - return db.security_group_get_by_name(self.context, 'fake', 'testgroup') - - def test_creates_base_rule_first(self): - # These come pre-defined by libvirt - self.defined_filters = ['no-mac-spoofing', - 'no-ip-spoofing', - 'no-arp-spoofing', - 'allow-dhcp-server'] - - self.recursive_depends = {} - for f in self.defined_filters: - self.recursive_depends[f] = [] - - def _filterDefineXMLMock(xml): - dom = xml_to_dom(xml) - name = dom.firstChild.getAttribute('name') - self.recursive_depends[name] = [] - for f in dom.getElementsByTagName('filterref'): - ref = f.getAttribute('filter') - self.assertTrue(ref in self.defined_filters, - ('%s referenced filter that does ' + - 'not yet exist: %s') % (name, ref)) - dependencies = [ref] + self.recursive_depends[ref] - self.recursive_depends[name] += dependencies - - self.defined_filters.append(name) - return True - - self.fake_libvirt_connection.nwfilterDefineXML = _filterDefineXMLMock - - instance_ref = db.instance_create(self.context, - {'user_id': 'fake', - 'project_id': 'fake'}) - inst_id = instance_ref['id'] - - def _ensure_all_called(): - instance_filter = 'nova-instance-%s' % instance_ref['name'] - secgroup_filter = 'nova-secgroup-%s' % self.security_group['id'] - for required in [secgroup_filter, 'allow-dhcp-server', - 'no-arp-spoofing', 'no-ip-spoofing', - 'no-mac-spoofing']: - self.assertTrue(required in - self.recursive_depends[instance_filter], - "Instance's filter does not include %s" % - required) - - self.security_group = self.setup_and_return_security_group() - - db.instance_add_security_group(self.context, inst_id, - self.security_group.id) - instance = db.instance_get(self.context, inst_id) - - d = self.fw.setup_nwfilters_for_instance(instance) - _ensure_all_called() - self.teardown_security_group() - return d diff --git a/nova/tests/volume_unittest.py b/nova/tests/volume_unittest.py deleted file mode 100644 index b13455fb0..000000000 --- a/nova/tests/volume_unittest.py +++ /dev/null @@ -1,175 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. -""" -Tests for Volume Code. - -""" -import logging - -from nova import context -from nova import exception -from nova import db -from nova import flags -from nova import test -from nova import utils - -FLAGS = flags.FLAGS - - -class VolumeTestCase(test.TestCase): - """Test Case for volumes.""" - - def setUp(self): - logging.getLogger().setLevel(logging.DEBUG) - super(VolumeTestCase, self).setUp() - self.compute = utils.import_object(FLAGS.compute_manager) - self.flags(connection_type='fake') - self.volume = utils.import_object(FLAGS.volume_manager) - self.context = context.get_admin_context() - - @staticmethod - def _create_volume(size='0'): - """Create a volume object.""" - vol = {} - vol['size'] = size - vol['user_id'] = 'fake' - vol['project_id'] = 'fake' - vol['availability_zone'] = FLAGS.storage_availability_zone - vol['status'] = "creating" - vol['attach_status'] = "detached" - return db.volume_create(context.get_admin_context(), vol)['id'] - - def test_create_delete_volume(self): - """Test volume can be created and deleted.""" - volume_id = self._create_volume() - self.volume.create_volume(self.context, volume_id) - self.assertEqual(volume_id, db.volume_get(context.get_admin_context(), - volume_id).id) - - self.volume.delete_volume(self.context, volume_id) - self.assertRaises(exception.NotFound, - db.volume_get, - self.context, - volume_id) - - def test_too_big_volume(self): - """Ensure failure if a too large of a volume is requested.""" - # FIXME(vish): validation needs to move into the data layer in - # volume_create - return True - try: - volume_id = self._create_volume('1001') - self.volume.create_volume(self.context, volume_id) - self.fail("Should have thrown TypeError") - except TypeError: - pass - - def test_too_many_volumes(self): - """Ensure that NoMoreTargets is raised when we run out of volumes.""" - vols = [] - total_slots = FLAGS.iscsi_num_targets - for _index in xrange(total_slots): - volume_id = self._create_volume() - self.volume.create_volume(self.context, volume_id) - vols.append(volume_id) - volume_id = self._create_volume() - self.assertRaises(db.NoMoreTargets, - self.volume.create_volume, - self.context, - volume_id) - db.volume_destroy(context.get_admin_context(), volume_id) - for volume_id in vols: - self.volume.delete_volume(self.context, volume_id) - - def test_run_attach_detach_volume(self): - """Make sure volume can be attached and detached from instance.""" - inst = {} - inst['image_id'] = 'ami-test' - inst['reservation_id'] = 'r-fakeres' - inst['launch_time'] = '10' - inst['user_id'] = 'fake' - inst['project_id'] = 'fake' - inst['instance_type'] = 'm1.tiny' - inst['mac_address'] = utils.generate_mac() - inst['ami_launch_index'] = 0 - instance_id = db.instance_create(self.context, inst)['id'] - mountpoint = "/dev/sdf" - volume_id = self._create_volume() - self.volume.create_volume(self.context, volume_id) - if FLAGS.fake_tests: - db.volume_attached(self.context, volume_id, instance_id, - mountpoint) - else: - self.compute.attach_volume(self.context, - instance_id, - volume_id, - mountpoint) - vol = db.volume_get(context.get_admin_context(), volume_id) - self.assertEqual(vol['status'], "in-use") - self.assertEqual(vol['attach_status'], "attached") - self.assertEqual(vol['mountpoint'], mountpoint) - instance_ref = db.volume_get_instance(self.context, volume_id) - self.assertEqual(instance_ref['id'], instance_id) - - self.assertRaises(exception.Error, - self.volume.delete_volume, - self.context, - volume_id) - if FLAGS.fake_tests: - db.volume_detached(self.context, volume_id) - else: - self.compute.detach_volume(self.context, - instance_id, - volume_id) - vol = db.volume_get(self.context, volume_id) - self.assertEqual(vol['status'], "available") - - self.volume.delete_volume(self.context, volume_id) - self.assertRaises(exception.Error, - db.volume_get, - self.context, - volume_id) - db.instance_destroy(self.context, instance_id) - - def test_concurrent_volumes_get_different_targets(self): - """Ensure multiple concurrent volumes get different targets.""" - volume_ids = [] - targets = [] - - def _check(volume_id): - """Make sure targets aren't duplicated.""" - volume_ids.append(volume_id) - admin_context = context.get_admin_context() - iscsi_target = db.volume_get_iscsi_target_num(admin_context, - volume_id) - self.assert_(iscsi_target not in targets) - targets.append(iscsi_target) - logging.debug("Target %s allocated", iscsi_target) - total_slots = FLAGS.iscsi_num_targets - for _index in xrange(total_slots): - volume_id = self._create_volume() - d = self.volume.create_volume(self.context, volume_id) - _check(d) - for volume_id in volume_ids: - self.volume.delete_volume(self.context, volume_id) - - def test_multi_node(self): - # TODO(termie): Figure out how to test with two nodes, - # each of them having a different FLAG for storage_node - # This will allow us to test cross-node interactions - pass -- cgit From b1d4579404f9e49fcdea23c21733fdf65edc1da3 Mon Sep 17 00:00:00 2001 From: Eric Day Date: Thu, 16 Dec 2010 17:29:26 -0800 Subject: Fixed network test (thanks Vish!) and fixed run_tests.sh. --- nova/tests/network_unittest.py | 347 ----------------------------------------- nova/tests/test_network.py | 347 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 347 insertions(+), 347 deletions(-) delete mode 100644 nova/tests/network_unittest.py create mode 100644 nova/tests/test_network.py (limited to 'nova/tests') diff --git a/nova/tests/network_unittest.py b/nova/tests/network_unittest.py deleted file mode 100644 index bcac20585..000000000 --- a/nova/tests/network_unittest.py +++ /dev/null @@ -1,347 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# 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. -""" -Unit Tests for network code -""" -import IPy -import os -import logging - -from nova import context -from nova import db -from nova import exception -from nova import flags -from nova import test -from nova import utils -from nova.auth import manager - -FLAGS = flags.FLAGS - - -class NetworkTestCase(test.TestCase): - """Test cases for network code""" - def setUp(self): - super(NetworkTestCase, self).setUp() - # NOTE(vish): if you change these flags, make sure to change the - # flags in the corresponding section in nova-dhcpbridge - self.flags(connection_type='fake', - fake_network=True, - network_size=16, - num_networks=5) - logging.getLogger().setLevel(logging.DEBUG) - self.manager = manager.AuthManager() - self.user = self.manager.create_user('netuser', 'netuser', 'netuser') - self.projects = [] - self.network = utils.import_object(FLAGS.network_manager) - self.context = context.RequestContext(project=None, user=self.user) - for i in range(5): - name = 'project%s' % i - project = self.manager.create_project(name, 'netuser', name) - self.projects.append(project) - # create the necessary network data for the project - user_context = context.RequestContext(project=self.projects[i], - user=self.user) - network_ref = self.network.get_network(user_context) - self.network.set_network_host(context.get_admin_context(), - network_ref['id']) - instance_ref = self._create_instance(0) - self.instance_id = instance_ref['id'] - instance_ref = self._create_instance(1) - self.instance2_id = instance_ref['id'] - - def tearDown(self): - super(NetworkTestCase, self).tearDown() - # TODO(termie): this should really be instantiating clean datastores - # in between runs, one failure kills all the tests - db.instance_destroy(context.get_admin_context(), self.instance_id) - db.instance_destroy(context.get_admin_context(), self.instance2_id) - for project in self.projects: - self.manager.delete_project(project) - self.manager.delete_user(self.user) - - def _create_instance(self, project_num, mac=None): - if not mac: - mac = utils.generate_mac() - project = self.projects[project_num] - self.context._project = project - self.context.project_id = project.id - return db.instance_create(self.context, - {'project_id': project.id, - 'mac_address': mac}) - - def _create_address(self, project_num, instance_id=None): - """Create an address in given project num""" - if instance_id is None: - instance_id = self.instance_id - self.context._project = self.projects[project_num] - self.context.project_id = self.projects[project_num].id - return self.network.allocate_fixed_ip(self.context, instance_id) - - def _deallocate_address(self, project_num, address): - self.context._project = self.projects[project_num] - self.context.project_id = self.projects[project_num].id - self.network.deallocate_fixed_ip(self.context, address) - - def test_public_network_association(self): - """Makes sure that we can allocaate a public ip""" - # TODO(vish): better way of adding floating ips - self.context._project = self.projects[0] - self.context.project_id = self.projects[0].id - pubnet = IPy.IP(flags.FLAGS.floating_range) - address = str(pubnet[0]) - try: - db.floating_ip_get_by_address(context.get_admin_context(), address) - except exception.NotFound: - db.floating_ip_create(context.get_admin_context(), - {'address': address, - 'host': FLAGS.host}) - float_addr = self.network.allocate_floating_ip(self.context, - self.projects[0].id) - fix_addr = self._create_address(0) - lease_ip(fix_addr) - self.assertEqual(float_addr, str(pubnet[0])) - self.network.associate_floating_ip(self.context, float_addr, fix_addr) - address = db.instance_get_floating_address(context.get_admin_context(), - self.instance_id) - self.assertEqual(address, float_addr) - self.network.disassociate_floating_ip(self.context, float_addr) - address = db.instance_get_floating_address(context.get_admin_context(), - self.instance_id) - self.assertEqual(address, None) - self.network.deallocate_floating_ip(self.context, float_addr) - self.network.deallocate_fixed_ip(self.context, fix_addr) - release_ip(fix_addr) - db.floating_ip_destroy(context.get_admin_context(), float_addr) - - def test_allocate_deallocate_fixed_ip(self): - """Makes sure that we can allocate and deallocate a fixed ip""" - address = self._create_address(0) - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) - lease_ip(address) - self._deallocate_address(0, address) - - # Doesn't go away until it's dhcp released - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) - - release_ip(address) - self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) - - def test_side_effects(self): - """Ensures allocating and releasing has no side effects""" - address = self._create_address(0) - address2 = self._create_address(1, self.instance2_id) - - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) - self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) - self.assertFalse(is_allocated_in_project(address, self.projects[1].id)) - - # Addresses are allocated before they're issued - lease_ip(address) - lease_ip(address2) - - self._deallocate_address(0, address) - release_ip(address) - self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) - - # First address release shouldn't affect the second - self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) - - self._deallocate_address(1, address2) - release_ip(address2) - self.assertFalse(is_allocated_in_project(address2, - self.projects[1].id)) - - def test_subnet_edge(self): - """Makes sure that private ips don't overlap""" - first = self._create_address(0) - lease_ip(first) - instance_ids = [] - for i in range(1, 5): - instance_ref = self._create_instance(i, mac=utils.generate_mac()) - instance_ids.append(instance_ref['id']) - address = self._create_address(i, instance_ref['id']) - instance_ref = self._create_instance(i, mac=utils.generate_mac()) - instance_ids.append(instance_ref['id']) - address2 = self._create_address(i, instance_ref['id']) - instance_ref = self._create_instance(i, mac=utils.generate_mac()) - instance_ids.append(instance_ref['id']) - address3 = self._create_address(i, instance_ref['id']) - lease_ip(address) - lease_ip(address2) - lease_ip(address3) - self.context._project = self.projects[i] - self.context.project_id = self.projects[i].id - self.assertFalse(is_allocated_in_project(address, - self.projects[0].id)) - self.assertFalse(is_allocated_in_project(address2, - self.projects[0].id)) - self.assertFalse(is_allocated_in_project(address3, - self.projects[0].id)) - self.network.deallocate_fixed_ip(self.context, address) - self.network.deallocate_fixed_ip(self.context, address2) - self.network.deallocate_fixed_ip(self.context, address3) - release_ip(address) - release_ip(address2) - release_ip(address3) - for instance_id in instance_ids: - db.instance_destroy(context.get_admin_context(), instance_id) - self.context._project = self.projects[0] - self.context.project_id = self.projects[0].id - self.network.deallocate_fixed_ip(self.context, first) - self._deallocate_address(0, first) - release_ip(first) - - def test_vpn_ip_and_port_looks_valid(self): - """Ensure the vpn ip and port are reasonable""" - self.assert_(self.projects[0].vpn_ip) - self.assert_(self.projects[0].vpn_port >= FLAGS.vpn_start) - self.assert_(self.projects[0].vpn_port <= FLAGS.vpn_start + - FLAGS.num_networks) - - def test_too_many_networks(self): - """Ensure error is raised if we run out of networks""" - projects = [] - networks_left = (FLAGS.num_networks - - db.network_count(context.get_admin_context())) - for i in range(networks_left): - project = self.manager.create_project('many%s' % i, self.user) - projects.append(project) - db.project_get_network(context.get_admin_context(), project.id) - project = self.manager.create_project('last', self.user) - projects.append(project) - self.assertRaises(db.NoMoreNetworks, - db.project_get_network, - context.get_admin_context(), - project.id) - for project in projects: - self.manager.delete_project(project) - - def test_ips_are_reused(self): - """Makes sure that ip addresses that are deallocated get reused""" - address = self._create_address(0) - lease_ip(address) - self.network.deallocate_fixed_ip(self.context, address) - release_ip(address) - - address2 = self._create_address(0) - self.assertEqual(address, address2) - lease_ip(address) - self.network.deallocate_fixed_ip(self.context, address2) - release_ip(address) - - def test_available_ips(self): - """Make sure the number of available ips for the network is correct - - The number of available IP addresses depends on the test - environment's setup. - - Network size is set in test fixture's setUp method. - - There are ips reserved at the bottom and top of the range. - services (network, gateway, CloudPipe, broadcast) - """ - network = db.project_get_network(context.get_admin_context(), - self.projects[0].id) - net_size = flags.FLAGS.network_size - admin_context = context.get_admin_context() - total_ips = (db.network_count_available_ips(admin_context, - network['id']) + - db.network_count_reserved_ips(admin_context, - network['id']) + - db.network_count_allocated_ips(admin_context, - network['id'])) - self.assertEqual(total_ips, net_size) - - def test_too_many_addresses(self): - """Test for a NoMoreAddresses exception when all fixed ips are used. - """ - admin_context = context.get_admin_context() - network = db.project_get_network(admin_context, self.projects[0].id) - num_available_ips = db.network_count_available_ips(admin_context, - network['id']) - addresses = [] - instance_ids = [] - for i in range(num_available_ips): - instance_ref = self._create_instance(0) - instance_ids.append(instance_ref['id']) - address = self._create_address(0, instance_ref['id']) - addresses.append(address) - lease_ip(address) - - ip_count = db.network_count_available_ips(context.get_admin_context(), - network['id']) - self.assertEqual(ip_count, 0) - self.assertRaises(db.NoMoreAddresses, - self.network.allocate_fixed_ip, - self.context, - 'foo') - - for i in range(num_available_ips): - self.network.deallocate_fixed_ip(self.context, addresses[i]) - release_ip(addresses[i]) - db.instance_destroy(context.get_admin_context(), instance_ids[i]) - ip_count = db.network_count_available_ips(context.get_admin_context(), - network['id']) - self.assertEqual(ip_count, num_available_ips) - - -def is_allocated_in_project(address, project_id): - """Returns true if address is in specified project""" - project_net = db.project_get_network(context.get_admin_context(), - project_id) - network = db.fixed_ip_get_network(context.get_admin_context(), address) - instance = db.fixed_ip_get_instance(context.get_admin_context(), address) - # instance exists until release - return instance is not None and network['id'] == project_net['id'] - - -def binpath(script): - """Returns the absolute path to a script in bin""" - return os.path.abspath(os.path.join(__file__, "../../../bin", script)) - - -def lease_ip(private_ip): - """Run add command on dhcpbridge""" - network_ref = db.fixed_ip_get_network(context.get_admin_context(), - private_ip) - instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), - private_ip) - cmd = "%s add %s %s fake" % (binpath('nova-dhcpbridge'), - instance_ref['mac_address'], - private_ip) - env = {'DNSMASQ_INTERFACE': network_ref['bridge'], - 'TESTING': '1', - 'FLAGFILE': FLAGS.dhcpbridge_flagfile} - (out, err) = utils.execute(cmd, addl_env=env) - logging.debug("ISSUE_IP: %s, %s ", out, err) - - -def release_ip(private_ip): - """Run del command on dhcpbridge""" - network_ref = db.fixed_ip_get_network(context.get_admin_context(), - private_ip) - instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), - private_ip) - cmd = "%s del %s %s fake" % (binpath('nova-dhcpbridge'), - instance_ref['mac_address'], - private_ip) - env = {'DNSMASQ_INTERFACE': network_ref['bridge'], - 'TESTING': '1', - 'FLAGFILE': FLAGS.dhcpbridge_flagfile} - (out, err) = utils.execute(cmd, addl_env=env) - logging.debug("RELEASE_IP: %s, %s ", out, err) diff --git a/nova/tests/test_network.py b/nova/tests/test_network.py new file mode 100644 index 000000000..bcac20585 --- /dev/null +++ b/nova/tests/test_network.py @@ -0,0 +1,347 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# 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. +""" +Unit Tests for network code +""" +import IPy +import os +import logging + +from nova import context +from nova import db +from nova import exception +from nova import flags +from nova import test +from nova import utils +from nova.auth import manager + +FLAGS = flags.FLAGS + + +class NetworkTestCase(test.TestCase): + """Test cases for network code""" + def setUp(self): + super(NetworkTestCase, self).setUp() + # NOTE(vish): if you change these flags, make sure to change the + # flags in the corresponding section in nova-dhcpbridge + self.flags(connection_type='fake', + fake_network=True, + network_size=16, + num_networks=5) + logging.getLogger().setLevel(logging.DEBUG) + self.manager = manager.AuthManager() + self.user = self.manager.create_user('netuser', 'netuser', 'netuser') + self.projects = [] + self.network = utils.import_object(FLAGS.network_manager) + self.context = context.RequestContext(project=None, user=self.user) + for i in range(5): + name = 'project%s' % i + project = self.manager.create_project(name, 'netuser', name) + self.projects.append(project) + # create the necessary network data for the project + user_context = context.RequestContext(project=self.projects[i], + user=self.user) + network_ref = self.network.get_network(user_context) + self.network.set_network_host(context.get_admin_context(), + network_ref['id']) + instance_ref = self._create_instance(0) + self.instance_id = instance_ref['id'] + instance_ref = self._create_instance(1) + self.instance2_id = instance_ref['id'] + + def tearDown(self): + super(NetworkTestCase, self).tearDown() + # TODO(termie): this should really be instantiating clean datastores + # in between runs, one failure kills all the tests + db.instance_destroy(context.get_admin_context(), self.instance_id) + db.instance_destroy(context.get_admin_context(), self.instance2_id) + for project in self.projects: + self.manager.delete_project(project) + self.manager.delete_user(self.user) + + def _create_instance(self, project_num, mac=None): + if not mac: + mac = utils.generate_mac() + project = self.projects[project_num] + self.context._project = project + self.context.project_id = project.id + return db.instance_create(self.context, + {'project_id': project.id, + 'mac_address': mac}) + + def _create_address(self, project_num, instance_id=None): + """Create an address in given project num""" + if instance_id is None: + instance_id = self.instance_id + self.context._project = self.projects[project_num] + self.context.project_id = self.projects[project_num].id + return self.network.allocate_fixed_ip(self.context, instance_id) + + def _deallocate_address(self, project_num, address): + self.context._project = self.projects[project_num] + self.context.project_id = self.projects[project_num].id + self.network.deallocate_fixed_ip(self.context, address) + + def test_public_network_association(self): + """Makes sure that we can allocaate a public ip""" + # TODO(vish): better way of adding floating ips + self.context._project = self.projects[0] + self.context.project_id = self.projects[0].id + pubnet = IPy.IP(flags.FLAGS.floating_range) + address = str(pubnet[0]) + try: + db.floating_ip_get_by_address(context.get_admin_context(), address) + except exception.NotFound: + db.floating_ip_create(context.get_admin_context(), + {'address': address, + 'host': FLAGS.host}) + float_addr = self.network.allocate_floating_ip(self.context, + self.projects[0].id) + fix_addr = self._create_address(0) + lease_ip(fix_addr) + self.assertEqual(float_addr, str(pubnet[0])) + self.network.associate_floating_ip(self.context, float_addr, fix_addr) + address = db.instance_get_floating_address(context.get_admin_context(), + self.instance_id) + self.assertEqual(address, float_addr) + self.network.disassociate_floating_ip(self.context, float_addr) + address = db.instance_get_floating_address(context.get_admin_context(), + self.instance_id) + self.assertEqual(address, None) + self.network.deallocate_floating_ip(self.context, float_addr) + self.network.deallocate_fixed_ip(self.context, fix_addr) + release_ip(fix_addr) + db.floating_ip_destroy(context.get_admin_context(), float_addr) + + def test_allocate_deallocate_fixed_ip(self): + """Makes sure that we can allocate and deallocate a fixed ip""" + address = self._create_address(0) + self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + lease_ip(address) + self._deallocate_address(0, address) + + # Doesn't go away until it's dhcp released + self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + + release_ip(address) + self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) + + def test_side_effects(self): + """Ensures allocating and releasing has no side effects""" + address = self._create_address(0) + address2 = self._create_address(1, self.instance2_id) + + self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) + self.assertFalse(is_allocated_in_project(address, self.projects[1].id)) + + # Addresses are allocated before they're issued + lease_ip(address) + lease_ip(address2) + + self._deallocate_address(0, address) + release_ip(address) + self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) + + # First address release shouldn't affect the second + self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) + + self._deallocate_address(1, address2) + release_ip(address2) + self.assertFalse(is_allocated_in_project(address2, + self.projects[1].id)) + + def test_subnet_edge(self): + """Makes sure that private ips don't overlap""" + first = self._create_address(0) + lease_ip(first) + instance_ids = [] + for i in range(1, 5): + instance_ref = self._create_instance(i, mac=utils.generate_mac()) + instance_ids.append(instance_ref['id']) + address = self._create_address(i, instance_ref['id']) + instance_ref = self._create_instance(i, mac=utils.generate_mac()) + instance_ids.append(instance_ref['id']) + address2 = self._create_address(i, instance_ref['id']) + instance_ref = self._create_instance(i, mac=utils.generate_mac()) + instance_ids.append(instance_ref['id']) + address3 = self._create_address(i, instance_ref['id']) + lease_ip(address) + lease_ip(address2) + lease_ip(address3) + self.context._project = self.projects[i] + self.context.project_id = self.projects[i].id + self.assertFalse(is_allocated_in_project(address, + self.projects[0].id)) + self.assertFalse(is_allocated_in_project(address2, + self.projects[0].id)) + self.assertFalse(is_allocated_in_project(address3, + self.projects[0].id)) + self.network.deallocate_fixed_ip(self.context, address) + self.network.deallocate_fixed_ip(self.context, address2) + self.network.deallocate_fixed_ip(self.context, address3) + release_ip(address) + release_ip(address2) + release_ip(address3) + for instance_id in instance_ids: + db.instance_destroy(context.get_admin_context(), instance_id) + self.context._project = self.projects[0] + self.context.project_id = self.projects[0].id + self.network.deallocate_fixed_ip(self.context, first) + self._deallocate_address(0, first) + release_ip(first) + + def test_vpn_ip_and_port_looks_valid(self): + """Ensure the vpn ip and port are reasonable""" + self.assert_(self.projects[0].vpn_ip) + self.assert_(self.projects[0].vpn_port >= FLAGS.vpn_start) + self.assert_(self.projects[0].vpn_port <= FLAGS.vpn_start + + FLAGS.num_networks) + + def test_too_many_networks(self): + """Ensure error is raised if we run out of networks""" + projects = [] + networks_left = (FLAGS.num_networks - + db.network_count(context.get_admin_context())) + for i in range(networks_left): + project = self.manager.create_project('many%s' % i, self.user) + projects.append(project) + db.project_get_network(context.get_admin_context(), project.id) + project = self.manager.create_project('last', self.user) + projects.append(project) + self.assertRaises(db.NoMoreNetworks, + db.project_get_network, + context.get_admin_context(), + project.id) + for project in projects: + self.manager.delete_project(project) + + def test_ips_are_reused(self): + """Makes sure that ip addresses that are deallocated get reused""" + address = self._create_address(0) + lease_ip(address) + self.network.deallocate_fixed_ip(self.context, address) + release_ip(address) + + address2 = self._create_address(0) + self.assertEqual(address, address2) + lease_ip(address) + self.network.deallocate_fixed_ip(self.context, address2) + release_ip(address) + + def test_available_ips(self): + """Make sure the number of available ips for the network is correct + + The number of available IP addresses depends on the test + environment's setup. + + Network size is set in test fixture's setUp method. + + There are ips reserved at the bottom and top of the range. + services (network, gateway, CloudPipe, broadcast) + """ + network = db.project_get_network(context.get_admin_context(), + self.projects[0].id) + net_size = flags.FLAGS.network_size + admin_context = context.get_admin_context() + total_ips = (db.network_count_available_ips(admin_context, + network['id']) + + db.network_count_reserved_ips(admin_context, + network['id']) + + db.network_count_allocated_ips(admin_context, + network['id'])) + self.assertEqual(total_ips, net_size) + + def test_too_many_addresses(self): + """Test for a NoMoreAddresses exception when all fixed ips are used. + """ + admin_context = context.get_admin_context() + network = db.project_get_network(admin_context, self.projects[0].id) + num_available_ips = db.network_count_available_ips(admin_context, + network['id']) + addresses = [] + instance_ids = [] + for i in range(num_available_ips): + instance_ref = self._create_instance(0) + instance_ids.append(instance_ref['id']) + address = self._create_address(0, instance_ref['id']) + addresses.append(address) + lease_ip(address) + + ip_count = db.network_count_available_ips(context.get_admin_context(), + network['id']) + self.assertEqual(ip_count, 0) + self.assertRaises(db.NoMoreAddresses, + self.network.allocate_fixed_ip, + self.context, + 'foo') + + for i in range(num_available_ips): + self.network.deallocate_fixed_ip(self.context, addresses[i]) + release_ip(addresses[i]) + db.instance_destroy(context.get_admin_context(), instance_ids[i]) + ip_count = db.network_count_available_ips(context.get_admin_context(), + network['id']) + self.assertEqual(ip_count, num_available_ips) + + +def is_allocated_in_project(address, project_id): + """Returns true if address is in specified project""" + project_net = db.project_get_network(context.get_admin_context(), + project_id) + network = db.fixed_ip_get_network(context.get_admin_context(), address) + instance = db.fixed_ip_get_instance(context.get_admin_context(), address) + # instance exists until release + return instance is not None and network['id'] == project_net['id'] + + +def binpath(script): + """Returns the absolute path to a script in bin""" + return os.path.abspath(os.path.join(__file__, "../../../bin", script)) + + +def lease_ip(private_ip): + """Run add command on dhcpbridge""" + network_ref = db.fixed_ip_get_network(context.get_admin_context(), + private_ip) + instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), + private_ip) + cmd = "%s add %s %s fake" % (binpath('nova-dhcpbridge'), + instance_ref['mac_address'], + private_ip) + env = {'DNSMASQ_INTERFACE': network_ref['bridge'], + 'TESTING': '1', + 'FLAGFILE': FLAGS.dhcpbridge_flagfile} + (out, err) = utils.execute(cmd, addl_env=env) + logging.debug("ISSUE_IP: %s, %s ", out, err) + + +def release_ip(private_ip): + """Run del command on dhcpbridge""" + network_ref = db.fixed_ip_get_network(context.get_admin_context(), + private_ip) + instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), + private_ip) + cmd = "%s del %s %s fake" % (binpath('nova-dhcpbridge'), + instance_ref['mac_address'], + private_ip) + env = {'DNSMASQ_INTERFACE': network_ref['bridge'], + 'TESTING': '1', + 'FLAGFILE': FLAGS.dhcpbridge_flagfile} + (out, err) = utils.execute(cmd, addl_env=env) + logging.debug("RELEASE_IP: %s, %s ", out, err) -- cgit From 6383f7f9f63e348a12adeff66a266ef796d98ded Mon Sep 17 00:00:00 2001 From: Cerberus Date: Fri, 17 Dec 2010 11:54:59 -0600 Subject: Some typo fixes --- nova/tests/api/openstack/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/__init__.py b/nova/tests/api/openstack/__init__.py index 2e357febe..4e4dfe4fc 100644 --- a/nova/tests/api/openstack/__init__.py +++ b/nova/tests/api/openstack/__init__.py @@ -17,7 +17,7 @@ import unittest -from nova.api.openstack import limited +from nova.api.openstack.common import limited from nova.api.openstack import RateLimitingMiddleware from nova.tests.api.fakes import APIStub from webob import Request -- cgit From 75d6de8a67db02f886636edfedcf3f3fc8cff9cc Mon Sep 17 00:00:00 2001 From: Cerberus Date: Fri, 17 Dec 2010 16:03:21 -0600 Subject: Some tweaks --- nova/tests/api/openstack/fakes.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/fakes.py b/nova/tests/api/openstack/fakes.py index 21b8aac1c..6c30761d6 100644 --- a/nova/tests/api/openstack/fakes.py +++ b/nova/tests/api/openstack/fakes.py @@ -173,7 +173,7 @@ class FakeToken(object): class FakeRequestContext(object): - def __init__(self, user, project): + def __init__(self, user, project, *args, **kwargs): self.user_id = 1 self.project_id = 1 -- cgit From 5b8362d0f56bdbeba7ee8292222863a501bad6af Mon Sep 17 00:00:00 2001 From: Cerberus Date: Fri, 17 Dec 2010 16:56:42 -0600 Subject: A few more tweaks to get the OS API tests passing --- nova/tests/api/openstack/fakes.py | 5 +++-- nova/tests/api/openstack/test_servers.py | 1 + 2 files changed, 4 insertions(+), 2 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/fakes.py b/nova/tests/api/openstack/fakes.py index 6c30761d6..96689d2cd 100644 --- a/nova/tests/api/openstack/fakes.py +++ b/nova/tests/api/openstack/fakes.py @@ -29,8 +29,9 @@ from nova import exception as exc from nova import flags from nova import utils import nova.api.openstack.auth -from nova.image import service from nova.image import glance +from nova.image import local +from nova.image import service from nova.tests import fake_flags from nova.wsgi import Router @@ -75,7 +76,7 @@ def stub_out_image_service(stubs): def fake_image_show(meh, context, id): return dict(kernelId=1, ramdiskId=1) - stubs.Set(nova.image.local.LocalImageService, 'show', fake_image_show) + stubs.Set(local.LocalImageService, 'show', fake_image_show) def stub_out_auth(stubs): diff --git a/nova/tests/api/openstack/test_servers.py b/nova/tests/api/openstack/test_servers.py index 8444b6fce..dcd2fe766 100644 --- a/nova/tests/api/openstack/test_servers.py +++ b/nova/tests/api/openstack/test_servers.py @@ -27,6 +27,7 @@ import nova.api.openstack from nova.api.openstack import servers import nova.db.api from nova.db.sqlalchemy.models import Instance +import nova.image import nova.rpc from nova.tests.api.openstack import fakes -- cgit From ca81b0c12a3853942e9ce85154c38dad381ead0e Mon Sep 17 00:00:00 2001 From: Armando Migliaccio Date: Sat, 18 Dec 2010 00:50:49 +0000 Subject: fixed unittests and further clean-up post-eventlet merge --- nova/tests/xenapi_unittest.py | 48 +++++++++++++++---------------------------- 1 file changed, 16 insertions(+), 32 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/xenapi_unittest.py b/nova/tests/xenapi_unittest.py index 839d6aa44..74401ce5d 100644 --- a/nova/tests/xenapi_unittest.py +++ b/nova/tests/xenapi_unittest.py @@ -78,13 +78,10 @@ class XenAPIVolumeTestCase(test.TrialTestCase): helper = volume_utils.VolumeHelper helper.XenAPI = session.get_imported_xenapi() vol = self._create_volume() - info = yield helper.parse_volume_info(vol['ec2_id'], '/dev/sdc') + info = helper.parse_volume_info(vol['ec2_id'], '/dev/sdc') label = 'SR-%s' % vol['ec2_id'] description = 'Test-SR' - sr_ref = helper.create_iscsi_storage_blocking(session, - info, - label, - description) + sr_ref = helper.create_iscsi_storage(session, info, label, description) srs = fake.get_all('SR') self.assertEqual(sr_ref, srs[0]) db.volume_destroy(context.get_admin_context(), vol['id']) @@ -97,13 +94,10 @@ class XenAPIVolumeTestCase(test.TrialTestCase): helper.XenAPI = session.get_imported_xenapi() vol = self._create_volume() # oops, wrong mount point! - info = helper.parse_volume_info(vol['ec2_id'], '/dev/sd') - - def check(exc): - """ handler """ - self.assertIsInstance(exc.value, volume_utils.StorageError) - - info.addErrback(check) + self.assertRaises(volume_utils.StorageError, + helper.parse_volume_info, + vol['ec2_id'], + '/dev/sd') db.volume_destroy(context.get_admin_context(), vol['id']) def test_attach_volume(self): @@ -116,8 +110,7 @@ class XenAPIVolumeTestCase(test.TrialTestCase): result = conn.attach_volume(instance.name, volume['ec2_id'], '/dev/sdc') - def check(_): - """ handler """ + def check(): # check that the VM has a VBD attached to it # Get XenAPI reference for the VM vms = fake.get_all('VM') @@ -127,8 +120,7 @@ class XenAPIVolumeTestCase(test.TrialTestCase): vm_ref = vbd['VM'] self.assertEqual(vm_ref, vms[0]) - result.addCallback(check) - return result + check() def test_attach_volume_raise_exception(self): """ This shows how to test when exceptions are raised """ @@ -138,17 +130,11 @@ class XenAPIVolumeTestCase(test.TrialTestCase): volume = self._create_volume() instance = db.instance_create(self.values) fake.create_vm(instance.name, 'Running') - result = conn.attach_volume(instance.name, volume['ec2_id'], - '/dev/sdc') - - def check_exception(exc): - """ handler """ - if exc: - pass - else: - self.fail('Oops, no exception has been raised!') - result.addErrback(check_exception) - return result + self.assertRaises(Exception, + conn.attach_volume, + instance.name, + volume['ec2_id'], + '/dev/sdc') def tearDown(self): super(XenAPIVolumeTestCase, self).tearDown() @@ -192,10 +178,9 @@ class XenAPIVMTestCase(test.TrialTestCase): } conn = xenapi_conn.get_connection(False) instance = db.instance_create(values) - result = conn.spawn(instance) + conn.spawn(instance) - def check(_): - """ handler """ + def check(): instances = conn.list_instances() self.assertEquals(instances, [1]) @@ -225,8 +210,7 @@ class XenAPIVMTestCase(test.TrialTestCase): # Check that the VM is running according to XenAPI. self.assertEquals(vm['power_state'], 'Running') - result.addCallback(check) - return result + check() def tearDown(self): super(XenAPIVMTestCase, self).tearDown() -- cgit From c7dada9b2f5811afbc7a181636d6be3ab8ab98cc Mon Sep 17 00:00:00 2001 From: Salvatore Orlando Date: Mon, 20 Dec 2010 15:41:43 +0000 Subject: Fixed pep8 errors --- nova/tests/virt_unittest.py | 43 ++++++++++++++++++++----------------------- 1 file changed, 20 insertions(+), 23 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/virt_unittest.py b/nova/tests/virt_unittest.py index 9bbba4ba9..e5d5f00c8 100644 --- a/nova/tests/virt_unittest.py +++ b/nova/tests/virt_unittest.py @@ -129,44 +129,41 @@ class LibvirtConnTestCase(test.TestCase): check_list.append(check) else: if expect_kernel: - check = (lambda t: t.find('./os/kernel').text.split('/' - )[1], 'kernel') + check = (lambda t: t.find('./os/kernel').text.split('/') + [1], 'kernel') else: check = (lambda t: t.find('./os/kernel'), None) check_list.append(check) if expect_ramdisk: - check = (lambda t: t.find('./os/initrd').text.split('/' - )[1], 'ramdisk') + check = (lambda t: t.find('./os/initrd').text.split('/') + [1], 'ramdisk') else: check = (lambda t: t.find('./os/initrd'), None) check_list.append(check) common_checks = [ (lambda t: t.find('.').tag, 'domain'), - (lambda t: t.find('./devices/interface/filterref/parameter' - ).get('name'), 'IP'), - (lambda t: t.find('./devices/interface/filterref/parameter' - ).get('value'), '10.11.12.13'), - (lambda t: t.findall('./devices/interface/filterref/parameter' - )[1].get('name'), 'DHCPSERVER'), - (lambda t: t.findall('./devices/interface/filterref/parameter' - )[1].get('value'), '10.0.0.1'), - (lambda t: t.find('./devices/serial/source').get('path' - ).split('/')[1], 'console.log'), + (lambda t: t.find('./devices/interface/filterref/parameter') + .get('name'), 'IP'), + (lambda t: t.find('./devices/interface/filterref/parameter') + .get('value'), '10.11.12.13'), + (lambda t: t.findall('./devices/interface/filterref/parameter') + [1].get('name'), 'DHCPSERVER'), + (lambda t: t.findall('./devices/interface/filterref/parameter') + [1].get('value'), '10.0.0.1'), + (lambda t: t.find('./devices/serial/source').get('path') + .split('/')[1], 'console.log'), (lambda t: t.find('./memory').text, '2097152')] if rescue: - common_checks += [(lambda t: t.findall('./devices/disk/source' - )[0].get('file').split('/')[1], - 'rescue-disk'), - (lambda t: t.findall('./devices/disk/source' - )[1].get('file').split('/')[1], - 'disk')] + common_checks += [(lambda t: t.findall('./devices/disk/source') + [0].get('file').split('/')[1], 'rescue-disk'), + (lambda t: t.findall('./devices/disk/source') + [1].get('file').split('/')[1], 'disk')] else: - common_checks += [(lambda t: t.findall('./devices/disk/source' - )[0].get('file').split('/')[1], - 'disk')] + common_checks += [(lambda t: t.findall('./devices/disk/source') + [0].get('file').split('/')[1], 'disk')] for (libvirt_type, (expected_uri, checks)) in type_uri_map.iteritems(): FLAGS.libvirt_type = libvirt_type -- cgit From aded4faba96e4de88f0294604927ef824cb249be Mon Sep 17 00:00:00 2001 From: Trey Morris Date: Mon, 20 Dec 2010 22:55:11 +0000 Subject: added suspend and resume --- nova/tests/api/openstack/test_servers.py | 32 ++++++++++++++++++++++++++++++-- nova/tests/compute_unittest.py | 8 ++++++++ 2 files changed, 38 insertions(+), 2 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/test_servers.py b/nova/tests/api/openstack/test_servers.py index 3820f5f27..5d23db588 100644 --- a/nova/tests/api/openstack/test_servers.py +++ b/nova/tests/api/openstack/test_servers.py @@ -88,9 +88,13 @@ class ServersTest(unittest.TestCase): self.stubs.Set(nova.db.api, 'instance_get_floating_address', instance_address) self.stubs.Set(nova.compute.api.ComputeAPI, 'pause', - fake_compute_api) + fake_compute_api) self.stubs.Set(nova.compute.api.ComputeAPI, 'unpause', - fake_compute_api) + fake_compute_api) + self.stubs.Set(nova.compute.api.ComputeAPI, 'suspend', + fake_compute_api) + self.stubs.Set(nova.compute.api.ComputeAPI, 'resume', + fake_compute_api) self.allow_admin = FLAGS.allow_admin_api def tearDown(self): @@ -246,6 +250,30 @@ class ServersTest(unittest.TestCase): res = req.get_response(nova.api.API('os')) self.assertEqual(res.status_int, 202) + def test_server_suspend(self): + FLAGS.allow_admin_api = True + body = dict(server=dict( + name='server_test', imageId=2, flavorId=2, metadata={}, + personality={})) + req = webob.Request.blank('/v1.0/servers/1/suspend') + req.method = 'POST' + req.content_type = 'application/json' + req.body = json.dumps(body) + res = req.get_response(nova.api.API('os')) + self.assertEqual(res.status_int, 202) + + def test_server_resume(self): + FLAGS.allow_admin_api = True + body = dict(server=dict( + name='server_test', imageId=2, flavorId=2, metadata={}, + personality={})) + req = webob.Request.blank('/v1.0/servers/1/resume') + req.method = 'POST' + req.content_type = 'application/json' + req.body = json.dumps(body) + res = req.get_response(nova.api.API('os')) + self.assertEqual(res.status_int, 202) + def test_server_reboot(self): body = dict(server=dict( name='server_test', imageId=2, flavorId=2, metadata={}, diff --git a/nova/tests/compute_unittest.py b/nova/tests/compute_unittest.py index 187ca31de..111a43cdc 100644 --- a/nova/tests/compute_unittest.py +++ b/nova/tests/compute_unittest.py @@ -135,6 +135,14 @@ class ComputeTestCase(test.TestCase): self.compute.unpause_instance(self.context, instance_id) self.compute.terminate_instance(self.context, instance_id) + def test_suspend(self): + """ensure instance can be suspended""" + instance_id = self._create_instance() + self.compute.run_instance(self.context, instance_id) + self.compute.suspend_instance(self.context, instance_id) + self.compute.resume_instance(self.context, instance_id) + self.compute.terminate_instance(self.context, instance_id) + def test_reboot(self): """Ensure instance can be rebooted""" instance_id = self._create_instance() -- cgit From e419c27a00a85b7daba42f580e332d31713ae271 Mon Sep 17 00:00:00 2001 From: Cerberus Date: Wed, 22 Dec 2010 13:33:26 -0600 Subject: Moved some things for testing --- nova/tests/api/openstack/__init__.py | 2 ++ 1 file changed, 2 insertions(+) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/__init__.py b/nova/tests/api/openstack/__init__.py index fffc57e67..efe73b8e2 100644 --- a/nova/tests/api/openstack/__init__.py +++ b/nova/tests/api/openstack/__init__.py @@ -27,6 +27,8 @@ from webob import Request FLAGS = flags.FLAGS +RateLimitingMiddleware = utils.import_class(FLAGS.os_api_ratelimiting) + class RateLimitingMiddlewareTest(unittest.TestCase): def test_get_action_name(self): -- cgit From 168cde072542f9f4df7e7eb26f6b632306c0b7d2 Mon Sep 17 00:00:00 2001 From: mdietz Date: Wed, 22 Dec 2010 19:52:13 +0000 Subject: Finished moving the middleware layers and fixed the API tests again --- nova/tests/api/openstack/__init__.py | 13 ++++++------- nova/tests/api/openstack/fakes.py | 15 +++++++++------ nova/tests/api/openstack/test_auth.py | 4 ++-- 3 files changed, 17 insertions(+), 15 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/__init__.py b/nova/tests/api/openstack/__init__.py index efe73b8e2..9e183bd0d 100644 --- a/nova/tests/api/openstack/__init__.py +++ b/nova/tests/api/openstack/__init__.py @@ -19,15 +19,14 @@ import unittest from nova import context from nova import flags +from nova.api.openstack.ratelimiting import RateLimitingMiddleware from nova.api.openstack.common import limited -from nova.api.openstack import RateLimitingMiddleware from nova.tests.api.fakes import APIStub from nova import utils from webob import Request FLAGS = flags.FLAGS -RateLimitingMiddleware = utils.import_class(FLAGS.os_api_ratelimiting) class RateLimitingMiddlewareTest(unittest.TestCase): @@ -37,7 +36,7 @@ class RateLimitingMiddlewareTest(unittest.TestCase): def verify(method, url, action_name): req = Request.blank(url) req.method = method - action = middleware._limiting_driver.get_action_name(req) + action = middleware.get_action_name(req) self.assertEqual(action, action_name) verify('PUT', '/servers/4', 'PUT') @@ -70,7 +69,7 @@ class RateLimitingMiddlewareTest(unittest.TestCase): middleware = RateLimitingMiddleware(APIStub()) self.exhaust(middleware, 'POST', '/servers/4', 'usr1', 10) self.exhaust(middleware, 'POST', '/images/4', 'usr2', 10) - self.assertTrue(set(middleware._limiting_driver.limiter._levels) == \ + self.assertTrue(set(middleware.limiter._levels) == \ set(['usr1:POST', 'usr1:POST servers', 'usr2:POST'])) def test_POST_servers_action_correctly_ratelimited(self): @@ -79,15 +78,15 @@ class RateLimitingMiddlewareTest(unittest.TestCase): for i in range(5): self.exhaust(middleware, 'POST', '/servers/4', 'usr1', 10) # Reset the 'POST' action counter. - del middleware._limiting_driver.limiter._levels['usr1:POST'] + del middleware.limiter._levels['usr1:POST'] # All 50 daily "POST servers" actions should be all used up self.exhaust(middleware, 'POST', '/servers/4', 'usr1', 0) def test_proxy_ctor_works(self): middleware = RateLimitingMiddleware(APIStub()) - self.assertEqual(middleware._limiting_driver.limiter.__class__.__name__, "Limiter") + self.assertEqual(middleware.limiter.__class__.__name__, "Limiter") middleware = RateLimitingMiddleware(APIStub(), service_host='foobar') - self.assertEqual(middleware._limiting_driver.limiter.__class__.__name__, "WSGIAppProxy") + self.assertEqual(middleware.limiter.__class__.__name__, "WSGIAppProxy") class LimiterTest(unittest.TestCase): diff --git a/nova/tests/api/openstack/fakes.py b/nova/tests/api/openstack/fakes.py index 96689d2cd..f773b26a7 100644 --- a/nova/tests/api/openstack/fakes.py +++ b/nova/tests/api/openstack/fakes.py @@ -29,6 +29,8 @@ from nova import exception as exc from nova import flags from nova import utils import nova.api.openstack.auth +from nova.api.openstack import auth +from nova.api.openstack import ratelimiting from nova.image import glance from nova.image import local from nova.image import service @@ -52,10 +54,11 @@ class FakeRouter(Router): return res -def fake_auth_init(self): +def fake_auth_init(self, application): self.db = FakeAuthDatabase() self.context = Context() self.auth = FakeAuthManager() + self.application = application @webob.dec.wsgify @@ -83,21 +86,21 @@ def stub_out_auth(stubs): def fake_auth_init(self, app): self.application = app - stubs.Set(nova.api.openstack.AuthMiddleware, + stubs.Set(nova.api.openstack.auth.AuthMiddleware, '__init__', fake_auth_init) - stubs.Set(nova.api.openstack.AuthMiddleware, + stubs.Set(nova.api.openstack.auth.AuthMiddleware, '__call__', fake_wsgi) def stub_out_rate_limiting(stubs): def fake_rate_init(self, app): - super(nova.api.openstack.RateLimitingMiddleware, self).__init__(app) + super(nova.api.openstack.ratelimiting.RateLimitingMiddleware, self).__init__(app) self.application = app - stubs.Set(nova.api.openstack.RateLimitingMiddleware, + stubs.Set(nova.api.openstack.ratelimiting.RateLimitingMiddleware, '__init__', fake_rate_init) - stubs.Set(nova.api.openstack.RateLimitingMiddleware, + stubs.Set(nova.api.openstack.ratelimiting.RateLimitingMiddleware, '__call__', fake_wsgi) diff --git a/nova/tests/api/openstack/test_auth.py b/nova/tests/api/openstack/test_auth.py index 7b427c2db..489a1dfbf 100644 --- a/nova/tests/api/openstack/test_auth.py +++ b/nova/tests/api/openstack/test_auth.py @@ -34,7 +34,7 @@ class Test(unittest.TestCase): def setUp(self): self.stubs = stubout.StubOutForTesting() - self.stubs.Set(nova.api.openstack.auth.BasicApiAuthManager, + self.stubs.Set(nova.api.openstack.auth.AuthMiddleware, '__init__', fakes.fake_auth_init) self.stubs.Set(context, 'RequestContext', fakes.FakeRequestContext) fakes.FakeAuthManager.auth_data = {} @@ -131,7 +131,7 @@ class Test(unittest.TestCase): class TestLimiter(unittest.TestCase): def setUp(self): self.stubs = stubout.StubOutForTesting() - self.stubs.Set(nova.api.openstack.auth.BasicApiAuthManager, + self.stubs.Set(nova.api.openstack.auth.AuthMiddleware, '__init__', fakes.fake_auth_init) self.stubs.Set(context, 'RequestContext', fakes.FakeRequestContext) fakes.FakeAuthManager.auth_data = {} -- cgit From ba6a99f926180d47870dcb18e4387d18cddad9b0 Mon Sep 17 00:00:00 2001 From: Cerberus Date: Thu, 23 Dec 2010 11:58:13 -0600 Subject: Superfluous images include and added basic routes for shared ip groups --- nova/tests/api/openstack/test_servers.py | 1 - 1 file changed, 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/test_servers.py b/nova/tests/api/openstack/test_servers.py index 97cf3ed65..3820f5f27 100644 --- a/nova/tests/api/openstack/test_servers.py +++ b/nova/tests/api/openstack/test_servers.py @@ -27,7 +27,6 @@ import nova.api.openstack from nova.api.openstack import servers import nova.db.api from nova.db.sqlalchemy.models import Instance -import nova.image import nova.rpc from nova.tests.api.openstack import fakes -- cgit From e275fbd8e16e6dc55c54072aa162815d522f9242 Mon Sep 17 00:00:00 2001 From: Cerberus Date: Thu, 23 Dec 2010 13:30:24 -0600 Subject: One more time --- nova/tests/api/openstack/fakes.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/fakes.py b/nova/tests/api/openstack/fakes.py index f773b26a7..79663e43a 100644 --- a/nova/tests/api/openstack/fakes.py +++ b/nova/tests/api/openstack/fakes.py @@ -94,7 +94,7 @@ def stub_out_auth(stubs): def stub_out_rate_limiting(stubs): def fake_rate_init(self, app): - super(nova.api.openstack.ratelimiting.RateLimitingMiddleware, self).__init__(app) + super(ratelimiting.RateLimitingMiddleware, self).__init__(app) self.application = app stubs.Set(nova.api.openstack.ratelimiting.RateLimitingMiddleware, -- cgit From a0ca9d4a9550370cc262574fbee097e5b70e408d Mon Sep 17 00:00:00 2001 From: Trey Morris Date: Thu, 23 Dec 2010 20:35:16 +0000 Subject: added _() for gettext and a couple of pep8s --- nova/tests/compute_unittest.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/compute_unittest.py b/nova/tests/compute_unittest.py index 111a43cdc..14954c3a2 100644 --- a/nova/tests/compute_unittest.py +++ b/nova/tests/compute_unittest.py @@ -100,13 +100,13 @@ class ComputeTestCase(test.TestCase): self.compute.run_instance(self.context, instance_id) instances = db.instance_get_all(context.get_admin_context()) - logging.info("Running instances: %s", instances) + logging.info(_("Running instances: %s"), instances) self.assertEqual(len(instances), 1) self.compute.terminate_instance(self.context, instance_id) instances = db.instance_get_all(context.get_admin_context()) - logging.info("After terminating instances: %s", instances) + logging.info(_("After terminating instances: %s"), instances) self.assertEqual(len(instances), 0) def test_run_terminate_timestamps(self): -- cgit From 257da8a0e5fd949f62232bf2eef9d91f36fc41ce Mon Sep 17 00:00:00 2001 From: Trey Morris Date: Thu, 23 Dec 2010 23:09:58 +0000 Subject: fixed the os api image test for glance --- nova/tests/api/openstack/test_images.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index f610cbf9c..e849b5ea6 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -223,6 +223,21 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase): res = req.get_response(nova.api.API('os')) res_dict = json.loads(res.body) + def _is_equivalent_subset(x, y): + if set(x) <= set(y): + for k, v in x.iteritems(): + if x[k] != y[k]: + return False + return True + return False + for image in res_dict['images']: - self.assertEquals(1, self.IMAGE_FIXTURES.count(image), - "image %s not in fixtures!" % str(image)) + for image_fixture in IMAGE_FIXTURES: + if _is_equivalent_subset(image, image_fixture): + break + else: + self.assertFalse("image %s not in fixtures!" % str(image)) + +# for image in res_dict['images']: +# self.assertEquals(1, self.IMAGE_FIXTURES.count(image), +# "image %s not in fixtures!" % str(image)) -- cgit From 391ab4dd63297afcc9449059bcadfe6ac5008b5f Mon Sep 17 00:00:00 2001 From: Trey Morris Date: Thu, 23 Dec 2010 23:13:03 +0000 Subject: typo --- nova/tests/api/openstack/test_images.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index e849b5ea6..1986a2b5f 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -232,7 +232,7 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase): return False for image in res_dict['images']: - for image_fixture in IMAGE_FIXTURES: + for image_fixture in self.IMAGE_FIXTURES: if _is_equivalent_subset(image, image_fixture): break else: -- cgit From d8d66d4c2c25d25892289e08ca52720f9d123d88 Mon Sep 17 00:00:00 2001 From: Trey Morris Date: Thu, 23 Dec 2010 23:19:24 +0000 Subject: trying again --- nova/tests/api/openstack/test_images.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index 1986a2b5f..16a910a3f 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -236,7 +236,7 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase): if _is_equivalent_subset(image, image_fixture): break else: - self.assertFalse("image %s not in fixtures!" % str(image)) + self.assertEquals(1,2,"image %s not in fixtures!" % str(image)) # for image in res_dict['images']: # self.assertEquals(1, self.IMAGE_FIXTURES.count(image), -- cgit From 26a8afd85233e142f97fdcc802c41b9a765efb32 Mon Sep 17 00:00:00 2001 From: Trey Morris Date: Thu, 23 Dec 2010 23:27:39 +0000 Subject: applied power state conversion to test --- nova/tests/api/openstack/test_images.py | 2 ++ 1 file changed, 2 insertions(+) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index 16a910a3f..ec0119bfe 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -227,6 +227,8 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase): if set(x) <= set(y): for k, v in x.iteritems(): if x[k] != y[k]: + if x[k] == 'active' and y[k] == 'available': + next return False return True return False -- cgit From f793e186910c1aec10759f5d05e305cf6889a02f Mon Sep 17 00:00:00 2001 From: Trey Morris Date: Thu, 23 Dec 2010 23:52:33 +0000 Subject: another typo --- nova/tests/api/openstack/test_images.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index ec0119bfe..d9845eb71 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -228,7 +228,7 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase): for k, v in x.iteritems(): if x[k] != y[k]: if x[k] == 'active' and y[k] == 'available': - next + continue return False return True return False -- cgit From f361891d100bbc9a5fb0f2ffd5424b6c69100133 Mon Sep 17 00:00:00 2001 From: Armando Migliaccio Date: Fri, 24 Dec 2010 19:17:02 +0000 Subject: remove requirement of sudo on tests --- nova/tests/test_xenapi.py | 1 + nova/tests/xenapi/stubs.py | 13 +++++++++++-- 2 files changed, 12 insertions(+), 2 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/test_xenapi.py b/nova/tests/test_xenapi.py index b5d3ea395..ed2e4ffde 100644 --- a/nova/tests/test_xenapi.py +++ b/nova/tests/test_xenapi.py @@ -48,6 +48,7 @@ class XenAPIVolumeTestCase(test.TestCase): FLAGS.xenapi_connection_url = 'test_url' FLAGS.xenapi_connection_password = 'test_pass' fakes.stub_out_db_instance_api(self.stubs) + stubs.stub_out_get_target(self.stubs) fake.reset() self.values = {'name': 1, 'id': 1, 'project_id': 'fake', diff --git a/nova/tests/xenapi/stubs.py b/nova/tests/xenapi/stubs.py index 1dacad6a3..a7e592fee 100644 --- a/nova/tests/xenapi/stubs.py +++ b/nova/tests/xenapi/stubs.py @@ -18,12 +18,13 @@ from nova.virt import xenapi_conn from nova.virt.xenapi import fake +from nova.virt.xenapi import volume_utils def stubout_session(stubs, cls): - """ Stubs out two methods from XenAPISession """ + """Stubs out two methods from XenAPISession""" def fake_import(self): - """ Stubs out get_imported_xenapi of XenAPISession """ + """Stubs out get_imported_xenapi of XenAPISession""" fake_module = 'nova.virt.xenapi.fake' from_list = ['fake'] return __import__(fake_module, globals(), locals(), from_list, -1) @@ -34,6 +35,14 @@ def stubout_session(stubs, cls): fake_import) +def stub_out_get_target(stubs): + """Stubs out _get_target in volume_utils""" + def fake_get_target(volume_id): + return (None, None) + + stubs.Set(volume_utils, '_get_target', fake_get_target) + + class FakeSessionForVMTests(fake.SessionBase): """ Stubs out a XenAPISession for VM tests """ def __init__(self, uri): -- cgit From 4b271b9e25ac2573cbb82f4b89434d608a91a8c7 Mon Sep 17 00:00:00 2001 From: Trey Morris Date: Mon, 27 Dec 2010 16:41:41 +0000 Subject: couple of pep8s --- nova/tests/api/openstack/test_images.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index d9845eb71..ae11f7531 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -238,7 +238,8 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase): if _is_equivalent_subset(image, image_fixture): break else: - self.assertEquals(1,2,"image %s not in fixtures!" % str(image)) + self.assertEquals(1, 2, "image %s not in fixtures!" % \ + str(image)) # for image in res_dict['images']: # self.assertEquals(1, self.IMAGE_FIXTURES.count(image), -- cgit From 404015903646a00901ad1310c2a7731f960fae75 Mon Sep 17 00:00:00 2001 From: Trey Morris Date: Mon, 27 Dec 2010 17:18:43 +0000 Subject: removed \ --- nova/tests/api/openstack/test_images.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index ae11f7531..f2c8bc935 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -238,7 +238,7 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase): if _is_equivalent_subset(image, image_fixture): break else: - self.assertEquals(1, 2, "image %s not in fixtures!" % \ + self.assertEquals(1, 2, "image %s not in fixtures!" % str(image)) # for image in res_dict['images']: -- cgit From d2ec717f7f819503f977c7a6f35e96867cc6c512 Mon Sep 17 00:00:00 2001 From: Trey Morris Date: Mon, 27 Dec 2010 20:06:45 +0000 Subject: renaming things to be a bit more descriptive --- nova/tests/api/openstack/test_images.py | 4 ---- 1 file changed, 4 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index f2c8bc935..1b4031217 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -240,7 +240,3 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase): else: self.assertEquals(1, 2, "image %s not in fixtures!" % str(image)) - -# for image in res_dict['images']: -# self.assertEquals(1, self.IMAGE_FIXTURES.count(image), -# "image %s not in fixtures!" % str(image)) -- cgit From eab0ce934e6296910b26c087e4268a65dc233a55 Mon Sep 17 00:00:00 2001 From: Ewan Mellor Date: Tue, 28 Dec 2010 01:37:04 +0000 Subject: Bug #694880: nova-compute now depends upon Cheetah even when not using libvirt Only import Cheetah when needed, as we do already with libvirt and libxml2. This ensures that users of other virt backends don't need Cheetah to run nova-compute. --- nova/tests/test_virt.py | 1 + 1 file changed, 1 insertion(+) (limited to 'nova/tests') diff --git a/nova/tests/test_virt.py b/nova/tests/test_virt.py index 1c155abe4..4aa489d08 100644 --- a/nova/tests/test_virt.py +++ b/nova/tests/test_virt.py @@ -33,6 +33,7 @@ flags.DECLARE('instances_path', 'nova.compute.manager') class LibvirtConnTestCase(test.TestCase): def setUp(self): super(LibvirtConnTestCase, self).setUp() + libvirt_conn._late_load_cheetah() self.flags(fake_call=True) self.manager = manager.AuthManager() self.user = self.manager.create_user('fake', 'fake', 'fake', -- cgit