diff --git a/nova/tests/test_melange_ipam_lib.py b/nova/tests/test_melange_ipam_lib.py new file mode 100644 index 00000000..031c1f10 --- /dev/null +++ b/nova/tests/test_melange_ipam_lib.py @@ -0,0 +1,237 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2012 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. + +from nova import db +from nova import exception +from nova import flags +from nova import test +from nova.network.quantum import melange_connection +from nova.network.quantum import melange_ipam_lib + +FLAGS = flags.FLAGS + + +class MelangeIpamLibTestCase(test.TestCase): + def setUp(self): + super(MelangeIpamLibTestCase, self).setUp() + self.m_conn = self.mox.CreateMock(melange_connection.MelangeConnection) + self.ipam = melange_ipam_lib.QuantumMelangeIPAMLib() + self.ipam.m_conn = self.m_conn + + def _block_list(self, id='id', cidr='cidr', network_id='network_id'): + return {'ip_blocks': [{'id': id, + 'cidr': cidr, + 'network_id': network_id}]} + + def test_allocate_fixed_ips_extracts_address(self): + self.m_conn.allocate_ip('quantum_net_id', 'network_tenant_id', + 'vif_ref_uuid', 'project_id', + 'vif_ref_address').AndReturn( + [{'address': 'ip_address'}]) + + self.mox.ReplayAll() + + ips = self.ipam.allocate_fixed_ips('context', + 'project_id', + 'quantum_net_id', + 'network_tenant_id', + {'uuid': 'vif_ref_uuid', + 'address': 'vif_ref_address'}) + self.assertEqual(ips[0], 'ip_address') + + def test_get_network_id_by_cidr_finds_block(self): + self.m_conn.get_blocks('tenant_id').AndReturn(self._block_list()) + + self.mox.ReplayAll() + + net_id = self.ipam.get_network_id_by_cidr('context', 'cidr', + 'tenant_id') + self.assertEqual(net_id, 'network_id') + + def test_get_network_id_by_cidr_raises_on_not_found(self): + self.m_conn.get_blocks('tenant_id').AndReturn({'ip_blocks': []}) + self.mox.ReplayAll() + self.assertRaises(exception.NotFound, + self.ipam.get_network_id_by_cidr, + 'context', 'cidr', 'tenant_id') + + def test_delete_subnets_by_net_id_deletes_block(self): + context = self.mox.CreateMockAnything() + context.elevated().AndReturn('elevated') + + self.m_conn.get_blocks('project_id').AndReturn( + self._block_list(id='block_id')) + self.m_conn.delete_block('block_id', 'project_id') + + self.mox.StubOutWithMock(db, 'network_get_by_uuid') + db.network_get_by_uuid('elevated', 'network_id').AndReturn( + {'id': 'network_id'}) + + self.mox.StubOutWithMock(db, 'network_delete_safe') + db.network_delete_safe(context, 'network_id') + + self.mox.ReplayAll() + self.ipam.delete_subnets_by_net_id(context, 'network_id', 'project_id') + + def test_get_networks_by_tenant_gets_all_networks(self): + block_list = self._block_list(network_id='net_1') + block_list['ip_blocks'] += self._block_list( + network_id='net_2')['ip_blocks'] + + self.m_conn.get_blocks('tenant_id').AndReturn(block_list) + + self.mox.StubOutWithMock(db, 'network_get_by_uuid') + db.network_get_by_uuid('admin_context', 'net_1').AndReturn('network1') + db.network_get_by_uuid('admin_context', 'net_2').AndReturn('network2') + + self.mox.ReplayAll() + values = self.ipam.get_networks_by_tenant('admin_context', 'tenant_id') + self.assertEquals(values, ['network1', 'network2']) + + def test_get_global_networks(self): + FLAGS.quantum_default_tenant_id = 'quantum_default_tenant_id' + self.mox.StubOutWithMock(self.ipam, 'get_networks_by_tenant') + self.ipam.get_networks_by_tenant('admin_context', + 'quantum_default_tenant_id') + + self.mox.ReplayAll() + + self.ipam.get_global_networks('admin_context') + + def test_get_project_networks(self): + context = self.mox.CreateMockAnything() + context.elevated().AndReturn('elevated') + + networks = [{'project_id': 1}, {'project_id': None}] + + self.mox.StubOutWithMock(db, 'network_get_all') + db.network_get_all('elevated').AndReturn(networks) + + self.mox.ReplayAll() + values = self.ipam.get_project_networks(context) + self.assertEquals(values, [networks[0]]) + + def test_get_project_and_global_net_ids__by_priority(self): + context = self.mox.CreateMockAnything() + context.elevated().AndReturn('elevated') + + FLAGS.quantum_default_tenant_id = 'default_tenant_id' + + net1 = {'uuid': 'net1_uuid', 'priority': 'net1_priority'} + net2 = {'uuid': 'net2_uuid', 'priority': 'net2_priority'} + + self.mox.StubOutWithMock(self.ipam, 'get_networks_by_tenant') + self.ipam.get_networks_by_tenant('elevated', + 'project_id').AndReturn([net1]) + self.ipam.get_networks_by_tenant('elevated', + 'default_tenant_id').AndReturn([net2]) + self.mox.ReplayAll() + self.ipam.get_project_and_global_net_ids(context, 'project_id') + + def test_get_tenant_id_by_net_id(self): + FLAGS.quantum_default_tenant_id = 'qdti' + + self.m_conn.get_allocated_ips('net_id', 'vif_id', + None).AndReturn('tenant_id') + self.mox.ReplayAll() + self.ipam.get_tenant_id_by_net_id('context', 'net_id', 'vif_id', + 'project_id') + + def test_get_subnets_by_net_id(self): + ips = [{'ip_block': {'network_id': 'network_id', + 'id': 'id', + 'cidr': 'cidr', + 'gateway': 'gateway', + 'broadcast': 'broadcast', + 'netmask': 'netmask', + 'dns1': 'dns1', + 'dns2': 'dns2'}, + 'version': 4}] + + self.m_conn.get_allocated_ips('net_id', 'vif_id', + 'tenant_id').AndReturn(ips) + + self.mox.ReplayAll() + value = self.ipam.get_subnets_by_net_id('context', 'tenant_id', + 'net_id', 'vif_id') + self.assertEquals(value[0]['cidr'], 'cidr') + + def test_get_routes_by_ip_block(self): + self.m_conn.get_routes('block_id', 'project_id') + self.mox.ReplayAll() + self.ipam.get_routes_by_ip_block('context', 'block_id', 'project_id') + + def test_get_v4_ips_by_interface(self): + self.mox.StubOutWithMock(self.ipam, '_get_ips_by_interface') + self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id', + 'project_id', 4) + self.mox.ReplayAll() + self.ipam.get_v4_ips_by_interface('context', 'net_id', 'vif_id', + 'project_id') + + def test_get_v6_ips_by_interface(self): + self.mox.StubOutWithMock(self.ipam, '_get_ips_by_interface') + self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id', + 'project_id', 6) + self.mox.ReplayAll() + self.ipam.get_v6_ips_by_interface('context', 'net_id', 'vif_id', + 'project_id') + + def test_get_ips_by_interface(self): + ips = [{'address': '10.10.10.10'}, {'address': '2001::CAFE'}] + self.m_conn.get_allocated_ips('net_id', 'vif_id', + 'tenant_id').AndReturn(ips) + self.m_conn.get_allocated_ips('net_id', 'vif_id', + 'tenant_id').AndReturn(ips) + self.mox.ReplayAll() + values = self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id', + 'tenant_id', 4) + self.assertEquals(values, ["10.10.10.10"]) + values = self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id', + 'tenant_id', 6) + self.assertEquals(values, ["2001::CAFE"]) + + def test_verify_subnet_exists(self): + blocks = {'ip_blocks': [{'network_id': 'quantum_net_id'}]} + self.m_conn.get_blocks('tenant_id').AndReturn(blocks) + self.mox.ReplayAll() + value = self.ipam.verify_subnet_exists('context', 'tenant_id', + 'quantum_net_id') + self.assertEquals(value, True) + + def test_deallocate_ips_by_vif(self): + self.m_conn.deallocate_ips('net_id', 'uuid', 'tenant_id') + self.mox.ReplayAll() + self.ipam.deallocate_ips_by_vif('context', 'tenant_id', 'net_id', + {'uuid': 'uuid'}) + + def test_get_allocated_ips(self): + ips = [{'address': 'ip_address', 'interface_id': 'interface_id'}] + self.m_conn.get_allocated_ips_for_network('subnet_id', + 'project_id').AndReturn(ips) + self.mox.ReplayAll() + self.ipam.get_allocated_ips('context', 'subnet_id', 'project_id') + + def test_create_vif(self): + self.m_conn.create_vif('vif_id', 'instance_id', 'project_id') + self.mox.ReplayAll() + self.ipam.create_vif('vif_id', 'instance_id', 'project_id') + + def test_get_floating_ips_by_fixed_address(self): + value = self.ipam.get_floating_ips_by_fixed_address('context', + 'fixed_address') + self.assertEquals(value, [])