diff options
Diffstat (limited to 'nova/tests/unit/test_nova_manage.py')
| -rw-r--r-- | nova/tests/unit/test_nova_manage.py | 467 |
1 files changed, 467 insertions, 0 deletions
diff --git a/nova/tests/unit/test_nova_manage.py b/nova/tests/unit/test_nova_manage.py new file mode 100644 index 0000000000..9ffaf66e81 --- /dev/null +++ b/nova/tests/unit/test_nova_manage.py @@ -0,0 +1,467 @@ +# Copyright 2011 OpenStack Foundation +# Copyright 2011 Ilya Alekseyev +# +# 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 + +import fixtures +import mock + +from nova.cmd import manage +from nova import context +from nova import db +from nova import exception +from nova.i18n import _ +from nova import test +from nova.tests.unit.db import fakes as db_fakes +from nova.tests.unit.objects import test_network + + +class FixedIpCommandsTestCase(test.TestCase): + def setUp(self): + super(FixedIpCommandsTestCase, self).setUp() + db_fakes.stub_out_db_network_api(self.stubs) + self.commands = manage.FixedIpCommands() + + def test_reserve(self): + self.commands.reserve('192.168.0.100') + address = db.fixed_ip_get_by_address(context.get_admin_context(), + '192.168.0.100') + self.assertEqual(address['reserved'], True) + + def test_reserve_nonexistent_address(self): + self.assertEqual(2, self.commands.reserve('55.55.55.55')) + + def test_unreserve(self): + self.commands.unreserve('192.168.0.100') + address = db.fixed_ip_get_by_address(context.get_admin_context(), + '192.168.0.100') + self.assertEqual(address['reserved'], False) + + def test_unreserve_nonexistent_address(self): + self.assertEqual(2, self.commands.unreserve('55.55.55.55')) + + def test_list(self): + self.useFixture(fixtures.MonkeyPatch('sys.stdout', + StringIO.StringIO())) + self.commands.list() + self.assertNotEqual(1, sys.stdout.getvalue().find('192.168.0.100')) + + def test_list_just_one_host(self): + def fake_fixed_ip_get_by_host(*args, **kwargs): + return [db_fakes.fixed_ip_fields] + + self.useFixture(fixtures.MonkeyPatch( + 'nova.db.fixed_ip_get_by_host', + fake_fixed_ip_get_by_host)) + self.useFixture(fixtures.MonkeyPatch('sys.stdout', + StringIO.StringIO())) + self.commands.list('banana') + self.assertNotEqual(1, sys.stdout.getvalue().find('192.168.0.100')) + + +class FloatingIpCommandsTestCase(test.TestCase): + def setUp(self): + super(FloatingIpCommandsTestCase, self).setUp() + db_fakes.stub_out_db_network_api(self.stubs) + self.commands = manage.FloatingIpCommands() + + def test_address_to_hosts(self): + def assert_loop(result, expected): + for ip in result: + self.assertIn(str(ip), expected) + + address_to_hosts = self.commands.address_to_hosts + # /32 and /31 + self.assertRaises(exception.InvalidInput, address_to_hosts, + '192.168.100.1/32') + self.assertRaises(exception.InvalidInput, address_to_hosts, + '192.168.100.1/31') + # /30 + expected = ["192.168.100.%s" % i for i in range(1, 3)] + result = address_to_hosts('192.168.100.0/30') + self.assertEqual(2, len(list(result))) + assert_loop(result, expected) + # /29 + expected = ["192.168.100.%s" % i for i in range(1, 7)] + result = address_to_hosts('192.168.100.0/29') + self.assertEqual(6, len(list(result))) + assert_loop(result, expected) + # /28 + expected = ["192.168.100.%s" % i for i in range(1, 15)] + result = address_to_hosts('192.168.100.0/28') + self.assertEqual(14, len(list(result))) + assert_loop(result, expected) + # /16 + result = address_to_hosts('192.168.100.0/16') + self.assertEqual(65534, len(list(result))) + # NOTE(dripton): I don't test /13 because it makes the test take 3s. + # /12 gives over a million IPs, which is ridiculous. + self.assertRaises(exception.InvalidInput, address_to_hosts, + '192.168.100.1/12') + + +class NetworkCommandsTestCase(test.TestCase): + def setUp(self): + super(NetworkCommandsTestCase, self).setUp() + self.commands = manage.NetworkCommands() + self.net = {'id': 0, + 'label': 'fake', + 'injected': False, + 'cidr': '192.168.0.0/24', + 'cidr_v6': 'dead:beef::/64', + 'multi_host': False, + 'gateway_v6': 'dead:beef::1', + 'netmask_v6': '64', + 'netmask': '255.255.255.0', + 'bridge': 'fa0', + 'bridge_interface': 'fake_fa0', + 'gateway': '192.168.0.1', + 'broadcast': '192.168.0.255', + 'dns1': '8.8.8.8', + 'dns2': '8.8.4.4', + 'vlan': 200, + 'vlan_start': 201, + 'vpn_public_address': '10.0.0.2', + 'vpn_public_port': '2222', + 'vpn_private_address': '192.168.0.2', + 'dhcp_start': '192.168.0.3', + 'project_id': 'fake_project', + 'host': 'fake_host', + 'uuid': 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'} + + def fake_network_get_by_cidr(context, cidr): + self.assertTrue(context.to_dict()['is_admin']) + self.assertEqual(cidr, self.fake_net['cidr']) + return db_fakes.FakeModel(dict(test_network.fake_network, + **self.fake_net)) + + def fake_network_get_by_uuid(context, uuid): + self.assertTrue(context.to_dict()['is_admin']) + self.assertEqual(uuid, self.fake_net['uuid']) + return db_fakes.FakeModel(dict(test_network.fake_network, + **self.fake_net)) + + def fake_network_update(context, network_id, values): + self.assertTrue(context.to_dict()['is_admin']) + self.assertEqual(network_id, self.fake_net['id']) + self.assertEqual(values, self.fake_update_value) + self.fake_network_get_by_cidr = fake_network_get_by_cidr + self.fake_network_get_by_uuid = fake_network_get_by_uuid + self.fake_network_update = fake_network_update + + def test_create(self): + + def fake_create_networks(obj, context, **kwargs): + self.assertTrue(context.to_dict()['is_admin']) + self.assertEqual(kwargs['label'], 'Test') + self.assertEqual(kwargs['cidr'], '10.2.0.0/24') + self.assertEqual(kwargs['multi_host'], False) + self.assertEqual(kwargs['num_networks'], 1) + self.assertEqual(kwargs['network_size'], 256) + self.assertEqual(kwargs['vlan'], 200) + self.assertEqual(kwargs['vlan_start'], 201) + self.assertEqual(kwargs['vpn_start'], 2000) + self.assertEqual(kwargs['cidr_v6'], 'fd00:2::/120') + self.assertEqual(kwargs['gateway'], '10.2.0.1') + self.assertEqual(kwargs['gateway_v6'], 'fd00:2::22') + self.assertEqual(kwargs['bridge'], 'br200') + self.assertEqual(kwargs['bridge_interface'], 'eth0') + self.assertEqual(kwargs['dns1'], '8.8.8.8') + self.assertEqual(kwargs['dns2'], '8.8.4.4') + self.flags(network_manager='nova.network.manager.VlanManager') + from nova.network import manager as net_manager + self.stubs.Set(net_manager.VlanManager, 'create_networks', + fake_create_networks) + self.commands.create( + label='Test', + cidr='10.2.0.0/24', + num_networks=1, + network_size=256, + multi_host='F', + vlan=200, + vlan_start=201, + vpn_start=2000, + cidr_v6='fd00:2::/120', + gateway='10.2.0.1', + gateway_v6='fd00:2::22', + bridge='br200', + bridge_interface='eth0', + dns1='8.8.8.8', + dns2='8.8.4.4', + uuid='aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa') + + def test_list(self): + + def fake_network_get_all(context): + return [db_fakes.FakeModel(self.net)] + self.stubs.Set(db, 'network_get_all', fake_network_get_all) + output = StringIO.StringIO() + sys.stdout = output + self.commands.list() + sys.stdout = sys.__stdout__ + result = output.getvalue() + _fmt = "\t".join(["%(id)-5s", "%(cidr)-18s", "%(cidr_v6)-15s", + "%(dhcp_start)-15s", "%(dns1)-15s", "%(dns2)-15s", + "%(vlan)-15s", "%(project_id)-15s", "%(uuid)-15s"]) + head = _fmt % {'id': _('id'), + 'cidr': _('IPv4'), + 'cidr_v6': _('IPv6'), + 'dhcp_start': _('start address'), + 'dns1': _('DNS1'), + 'dns2': _('DNS2'), + 'vlan': _('VlanID'), + 'project_id': _('project'), + 'uuid': _("uuid")} + body = _fmt % {'id': self.net['id'], + 'cidr': self.net['cidr'], + 'cidr_v6': self.net['cidr_v6'], + 'dhcp_start': self.net['dhcp_start'], + 'dns1': self.net['dns1'], + 'dns2': self.net['dns2'], + 'vlan': self.net['vlan'], + 'project_id': self.net['project_id'], + 'uuid': self.net['uuid']} + answer = '%s\n%s\n' % (head, body) + self.assertEqual(result, answer) + + def test_delete(self): + self.fake_net = self.net + self.fake_net['project_id'] = None + self.fake_net['host'] = None + self.stubs.Set(db, 'network_get_by_uuid', + self.fake_network_get_by_uuid) + + def fake_network_delete_safe(context, network_id): + self.assertTrue(context.to_dict()['is_admin']) + self.assertEqual(network_id, self.fake_net['id']) + self.stubs.Set(db, 'network_delete_safe', fake_network_delete_safe) + self.commands.delete(uuid=self.fake_net['uuid']) + + def test_delete_by_cidr(self): + self.fake_net = self.net + self.fake_net['project_id'] = None + self.fake_net['host'] = None + self.stubs.Set(db, 'network_get_by_cidr', + self.fake_network_get_by_cidr) + + def fake_network_delete_safe(context, network_id): + self.assertTrue(context.to_dict()['is_admin']) + self.assertEqual(network_id, self.fake_net['id']) + self.stubs.Set(db, 'network_delete_safe', fake_network_delete_safe) + self.commands.delete(fixed_range=self.fake_net['cidr']) + + def _test_modify_base(self, update_value, project, host, dis_project=None, + dis_host=None): + self.fake_net = self.net + self.fake_update_value = update_value + self.stubs.Set(db, 'network_get_by_cidr', + self.fake_network_get_by_cidr) + self.stubs.Set(db, 'network_update', self.fake_network_update) + self.commands.modify(self.fake_net['cidr'], project=project, host=host, + dis_project=dis_project, dis_host=dis_host) + + def test_modify_associate(self): + self._test_modify_base(update_value={'project_id': 'test_project', + 'host': 'test_host'}, + project='test_project', host='test_host') + + def test_modify_unchanged(self): + self._test_modify_base(update_value={}, project=None, host=None) + + def test_modify_disassociate(self): + self._test_modify_base(update_value={'project_id': None, 'host': None}, + project=None, host=None, dis_project=True, + dis_host=True) + + +class NeutronV2NetworkCommandsTestCase(test.TestCase): + def setUp(self): + super(NeutronV2NetworkCommandsTestCase, self).setUp() + self.flags(network_api_class='nova.network.neutronv2.api.API') + self.commands = manage.NetworkCommands() + + def test_create(self): + self.assertEqual(2, self.commands.create()) + + def test_list(self): + self.assertEqual(2, self.commands.list()) + + def test_delete(self): + self.assertEqual(2, self.commands.delete()) + + def test_modify(self): + self.assertEqual(2, self.commands.modify('192.168.0.1')) + + +class ProjectCommandsTestCase(test.TestCase): + def setUp(self): + super(ProjectCommandsTestCase, self).setUp() + self.commands = manage.ProjectCommands() + + def test_quota(self): + output = StringIO.StringIO() + sys.stdout = output + self.commands.quota(project_id='admin', + key='instances', + value='unlimited', + ) + + sys.stdout = sys.__stdout__ + result = output.getvalue() + print_format = "%-36s %-10s" % ('instances', 'unlimited') + self.assertEqual((print_format in result), True) + + def test_quota_update_invalid_key(self): + self.assertEqual(2, self.commands.quota('admin', 'volumes1', '10')) + + +class DBCommandsTestCase(test.TestCase): + def setUp(self): + super(DBCommandsTestCase, self).setUp() + self.commands = manage.DbCommands() + + def test_archive_deleted_rows_negative(self): + self.assertEqual(1, self.commands.archive_deleted_rows(-1)) + + +class ServiceCommandsTestCase(test.TestCase): + def setUp(self): + super(ServiceCommandsTestCase, self).setUp() + self.commands = manage.ServiceCommands() + + def test_service_enable_invalid_params(self): + self.assertEqual(2, self.commands.enable('nohost', 'noservice')) + + def test_service_disable_invalid_params(self): + self.assertEqual(2, self.commands.disable('nohost', 'noservice')) + + +class CellCommandsTestCase(test.TestCase): + def setUp(self): + super(CellCommandsTestCase, self).setUp() + self.commands = manage.CellCommands() + + def test_create_transport_hosts_multiple(self): + """Test the _create_transport_hosts method + when broker_hosts is set. + """ + brokers = "127.0.0.1:5672,127.0.0.2:5671" + thosts = self.commands._create_transport_hosts( + 'guest', 'devstack', + broker_hosts=brokers) + self.assertEqual(2, len(thosts)) + self.assertEqual('127.0.0.1', thosts[0].hostname) + self.assertEqual(5672, thosts[0].port) + self.assertEqual('127.0.0.2', thosts[1].hostname) + self.assertEqual(5671, thosts[1].port) + + def test_create_transport_hosts_single(self): + """Test the _create_transport_hosts method when hostname is passed.""" + thosts = self.commands._create_transport_hosts('guest', 'devstack', + hostname='127.0.0.1', + port=80) + self.assertEqual(1, len(thosts)) + self.assertEqual('127.0.0.1', thosts[0].hostname) + self.assertEqual(80, thosts[0].port) + + def test_create_transport_hosts_single_broker(self): + """Test the _create_transport_hosts method for single broker_hosts.""" + thosts = self.commands._create_transport_hosts( + 'guest', 'devstack', + broker_hosts='127.0.0.1:5672') + self.assertEqual(1, len(thosts)) + self.assertEqual('127.0.0.1', thosts[0].hostname) + self.assertEqual(5672, thosts[0].port) + + def test_create_transport_hosts_both(self): + """Test the _create_transport_hosts method when both broker_hosts + and hostname/port are passed. + """ + thosts = self.commands._create_transport_hosts( + 'guest', 'devstack', + broker_hosts='127.0.0.1:5672', + hostname='127.0.0.2', port=80) + self.assertEqual(1, len(thosts)) + self.assertEqual('127.0.0.1', thosts[0].hostname) + self.assertEqual(5672, thosts[0].port) + + def test_create_transport_hosts_wrong_val(self): + """Test the _create_transport_hosts method when broker_hosts + is wrongly sepcified + """ + self.assertRaises(ValueError, + self.commands._create_transport_hosts, + 'guest', 'devstack', + broker_hosts='127.0.0.1:5672,127.0.0.1') + + def test_create_transport_hosts_wrong_port_val(self): + """Test the _create_transport_hosts method when port in + broker_hosts is wrongly sepcified + """ + self.assertRaises(ValueError, + self.commands._create_transport_hosts, + 'guest', 'devstack', + broker_hosts='127.0.0.1:') + + def test_create_transport_hosts_wrong_port_arg(self): + """Test the _create_transport_hosts method when port + argument is wrongly sepcified + """ + self.assertRaises(ValueError, + self.commands._create_transport_hosts, + 'guest', 'devstack', + hostname='127.0.0.1', port='ab') + + @mock.patch.object(context, 'get_admin_context') + @mock.patch.object(db, 'cell_create') + def test_create_broker_hosts(self, mock_db_cell_create, mock_ctxt): + """Test the create function when broker_hosts is + passed + """ + cell_tp_url = "fake://guest:devstack@127.0.0.1:5432" + cell_tp_url += ",guest:devstack@127.0.0.2:9999/" + ctxt = mock.sentinel + mock_ctxt.return_value = mock.sentinel + self.commands.create("test", + broker_hosts='127.0.0.1:5432,127.0.0.2:9999', + woffset=0, wscale=0, + username="guest", password="devstack") + exp_values = {'name': "test", + 'is_parent': False, + 'transport_url': cell_tp_url, + 'weight_offset': 0.0, + 'weight_scale': 0.0} + mock_db_cell_create.assert_called_once_with(ctxt, exp_values) + + @mock.patch.object(context, 'get_admin_context') + @mock.patch.object(db, 'cell_create') + def test_create_hostname(self, mock_db_cell_create, mock_ctxt): + """Test the create function when hostname and port is + passed + """ + cell_tp_url = "fake://guest:devstack@127.0.0.1:9999/" + ctxt = mock.sentinel + mock_ctxt.return_value = mock.sentinel + self.commands.create("test", + hostname='127.0.0.1', port="9999", + woffset=0, wscale=0, + username="guest", password="devstack") + exp_values = {'name': "test", + 'is_parent': False, + 'transport_url': cell_tp_url, + 'weight_offset': 0.0, + 'weight_scale': 0.0} + mock_db_cell_create.assert_called_once_with(ctxt, exp_values) |