# vim: tabstop=4 shiftwidth=4 softtabstop=4 # Copyright 2011 Rackspace # 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 mox from nova import context from nova import db from nova import exception from nova import log as logging from nova import test from nova.network import manager as network_manager from nova.tests import fake_network LOG = logging.getLogger('nova.tests.network') HOST = "testhost" class FakeModel(dict): """Represent a model from the db""" def __init__(self, *args, **kwargs): self.update(kwargs) def __getattr__(self, name): return self[name] networks = [{'id': 0, 'uuid': "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", 'label': 'test0', 'injected': False, 'multi_host': False, 'cidr': '192.168.0.0/24', 'cidr_v6': '2001:db8::/64', 'gateway_v6': '2001:db8::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': '192.168.0.1', 'dns2': '192.168.0.2', 'vlan': None, 'host': HOST, 'project_id': 'fake_project', 'vpn_public_address': '192.168.0.2'}, {'id': 1, 'uuid': "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", 'label': 'test1', 'injected': False, 'multi_host': False, 'cidr': '192.168.1.0/24', 'cidr_v6': '2001:db9::/64', 'gateway_v6': '2001:db9::1', 'netmask_v6': '64', 'netmask': '255.255.255.0', 'bridge': 'fa1', 'bridge_interface': 'fake_fa1', 'gateway': '192.168.1.1', 'broadcast': '192.168.1.255', 'dns1': '192.168.0.1', 'dns2': '192.168.0.2', 'vlan': None, 'host': HOST, 'project_id': 'fake_project', 'vpn_public_address': '192.168.1.2'}] fixed_ips = [{'id': 0, 'network_id': 0, 'address': '192.168.0.100', 'instance_id': 0, 'allocated': False, 'virtual_interface_id': 0, 'floating_ips': []}, {'id': 0, 'network_id': 1, 'address': '192.168.1.100', 'instance_id': 0, 'allocated': False, 'virtual_interface_id': 0, 'floating_ips': []}] flavor = {'id': 0, 'rxtx_cap': 3} floating_ip_fields = {'id': 0, 'address': '192.168.10.100', 'fixed_ip_id': 0, 'project_id': None, 'auto_assigned': False} vifs = [{'id': 0, 'address': 'DE:AD:BE:EF:00:00', 'uuid': '00000000-0000-0000-0000-0000000000000000', 'network_id': 0, 'network': FakeModel(**networks[0]), 'instance_id': 0}, {'id': 1, 'address': 'DE:AD:BE:EF:00:01', 'uuid': '00000000-0000-0000-0000-0000000000000001', 'network_id': 1, 'network': FakeModel(**networks[1]), 'instance_id': 0}, {'id': 2, 'address': 'DE:AD:BE:EF:00:02', 'uuid': '00000000-0000-0000-0000-0000000000000002', 'network_id': 2, 'network': None, 'instance_id': 0}] class FlatNetworkTestCase(test.TestCase): def setUp(self): super(FlatNetworkTestCase, self).setUp() self.network = network_manager.FlatManager(host=HOST) self.network.db = db self.context = context.RequestContext('testuser', 'testproject', is_admin=False) def test_get_instance_nw_info(self): fake_get_instance_nw_info = fake_network.fake_get_instance_nw_info nw_info = fake_get_instance_nw_info(self.stubs, 0, 2) self.assertFalse(nw_info) for i, (nw, info) in enumerate(nw_info): check = {'bridge': 'fake_br%d' % i, 'cidr': '192.168.%s.0/24' % i, 'cidr_v6': '2001:db8:0:%x::/64' % i, 'id': i, 'multi_host': False, 'injected': False, 'bridge_interface': 'fake_eth%d' % i, 'vlan': None} self.assertDictMatch(nw, check) check = {'broadcast': '192.168.%d.255' % i, 'dhcp_server': '192.168.%d.1' % i, 'dns': ['192.168.%d.3' % n, '192.168.%d.4' % n], 'gateway': '192.168.%d.1' % i, 'gateway6': '2001:db8:0:%x::1' % i, 'ip6s': 'DONTCARE', 'ips': 'DONTCARE', 'label': 'test%d' % i, 'mac': 'DE:AD:BE:EF:00:%02x' % i, 'vif_uuid': '00000000-0000-0000-0000-00000000000000%02d' % i, 'rxtx_cap': 3, 'should_create_vlan': False, 'should_create_bridge': False} self.assertDictMatch(info, check) check = [{'enabled': 'DONTCARE', 'ip': '2001:db8::dcad:beff:feef:%s' % i, 'netmask': '64'}] self.assertDictListMatch(info['ip6s'], check) num_fixed_ips = len(info['ips']) check = [{'enabled': 'DONTCARE', 'ip': '192.168.%d.1%02d' % (i, ip_num), 'netmask': '255.255.255.0'} for ip_num in xrange(num_fixed_ips)] self.assertDictListMatch(info['ips'], check) def test_validate_networks(self): self.mox.StubOutWithMock(db, 'network_get_all_by_uuids') self.mox.StubOutWithMock(db, "fixed_ip_get_by_address") requested_networks = [("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", "192.168.1.100")] db.network_get_all_by_uuids(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(networks) fixed_ips[1]['network'] = FakeModel(**networks[1]) fixed_ips[1]['instance'] = None db.fixed_ip_get_by_address(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(fixed_ips[1]) self.mox.ReplayAll() self.network.validate_networks(self.context, requested_networks) def test_validate_networks_none_requested_networks(self): self.network.validate_networks(self.context, None) def test_validate_networks_empty_requested_networks(self): requested_networks = [] self.mox.ReplayAll() self.network.validate_networks(self.context, requested_networks) def test_validate_networks_invalid_fixed_ip(self): self.mox.StubOutWithMock(db, 'network_get_all_by_uuids') requested_networks = [(1, "192.168.0.100.1")] db.network_get_all_by_uuids(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(networks) self.mox.ReplayAll() self.assertRaises(exception.FixedIpInvalid, self.network.validate_networks, None, requested_networks) def test_validate_networks_empty_fixed_ip(self): self.mox.StubOutWithMock(db, 'network_get_all_by_uuids') requested_networks = [(1, "")] db.network_get_all_by_uuids(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(networks) self.mox.ReplayAll() self.assertRaises(exception.FixedIpInvalid, self.network.validate_networks, None, requested_networks) def test_validate_networks_none_fixed_ip(self): self.mox.StubOutWithMock(db, 'network_get_all_by_uuids') requested_networks = [(1, None)] db.network_get_all_by_uuids(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(networks) self.mox.ReplayAll() self.network.validate_networks(None, requested_networks) def test_add_fixed_ip_instance_without_vpn_requested_networks(self): self.mox.StubOutWithMock(db, 'network_get') self.mox.StubOutWithMock(db, 'network_update') self.mox.StubOutWithMock(db, 'fixed_ip_associate_pool') self.mox.StubOutWithMock(db, 'instance_get') self.mox.StubOutWithMock(db, 'virtual_interface_get_by_instance_and_network') self.mox.StubOutWithMock(db, 'fixed_ip_update') db.fixed_ip_update(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()) db.virtual_interface_get_by_instance_and_network(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'id': 0}) db.instance_get(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'security_groups': [{'id': 0}]}) db.fixed_ip_associate_pool(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn('192.168.0.101') db.network_get(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(networks[0]) db.network_update(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()) self.mox.ReplayAll() self.network.add_fixed_ip_to_instance(self.context, 1, HOST, networks[0]['id']) class VlanNetworkTestCase(test.TestCase): def setUp(self): super(VlanNetworkTestCase, self).setUp() self.network = network_manager.VlanManager(host=HOST) self.network.db = db self.context = context.RequestContext('testuser', 'testproject', is_admin=False) def test_vpn_allocate_fixed_ip(self): self.mox.StubOutWithMock(db, 'fixed_ip_associate') self.mox.StubOutWithMock(db, 'fixed_ip_update') self.mox.StubOutWithMock(db, 'virtual_interface_get_by_instance_and_network') db.fixed_ip_associate(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg(), reserved=True).AndReturn('192.168.0.1') db.fixed_ip_update(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()) db.virtual_interface_get_by_instance_and_network(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'id': 0}) self.mox.ReplayAll() network = dict(networks[0]) network['vpn_private_address'] = '192.168.0.2' self.network.allocate_fixed_ip(None, 0, network, vpn=True) def test_allocate_fixed_ip(self): self.mox.StubOutWithMock(db, 'fixed_ip_associate_pool') self.mox.StubOutWithMock(db, 'fixed_ip_update') self.mox.StubOutWithMock(db, 'virtual_interface_get_by_instance_and_network') self.mox.StubOutWithMock(db, 'instance_get') db.instance_get(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'security_groups': [{'id': 0}]}) db.fixed_ip_associate_pool(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn('192.168.0.1') db.fixed_ip_update(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()) db.virtual_interface_get_by_instance_and_network(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'id': 0}) self.mox.ReplayAll() network = dict(networks[0]) network['vpn_private_address'] = '192.168.0.2' self.network.allocate_fixed_ip(self.context, 0, network) def test_create_networks_too_big(self): self.assertRaises(ValueError, self.network.create_networks, None, num_networks=4094, vlan_start=1) def test_create_networks_too_many(self): self.assertRaises(ValueError, self.network.create_networks, None, num_networks=100, vlan_start=1, cidr='192.168.0.1/24', network_size=100) def test_validate_networks(self): self.mox.StubOutWithMock(db, 'network_get_all_by_uuids') self.mox.StubOutWithMock(db, "fixed_ip_get_by_address") requested_networks = [("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", "192.168.1.100")] db.network_get_all_by_uuids(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(networks) fixed_ips[1]['network'] = FakeModel(**networks[1]) fixed_ips[1]['instance'] = None db.fixed_ip_get_by_address(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(fixed_ips[1]) self.mox.ReplayAll() self.network.validate_networks(self.context, requested_networks) def test_validate_networks_none_requested_networks(self): self.network.validate_networks(self.context, None) def test_validate_networks_empty_requested_networks(self): requested_networks = [] self.mox.ReplayAll() self.network.validate_networks(self.context, requested_networks) def test_validate_networks_invalid_fixed_ip(self): self.mox.StubOutWithMock(db, 'network_get_all_by_uuids') requested_networks = [(1, "192.168.0.100.1")] db.network_get_all_by_uuids(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(networks) self.mox.ReplayAll() self.assertRaises(exception.FixedIpInvalid, self.network.validate_networks, self.context, requested_networks) def test_validate_networks_empty_fixed_ip(self): self.mox.StubOutWithMock(db, 'network_get_all_by_uuids') requested_networks = [(1, "")] db.network_get_all_by_uuids(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(networks) self.mox.ReplayAll() self.assertRaises(exception.FixedIpInvalid, self.network.validate_networks, self.context, requested_networks) def test_validate_networks_none_fixed_ip(self): self.mox.StubOutWithMock(db, 'network_get_all_by_uuids') requested_networks = [(1, None)] db.network_get_all_by_uuids(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(networks) self.mox.ReplayAll() self.network.validate_networks(self.context, requested_networks) def test_cant_associate_associated_floating_ip(self): ctxt = context.RequestContext('testuser', 'testproject', is_admin=False) def fake_floating_ip_get_by_address(context, address): return {'address': '10.10.10.10', 'fixed_ip': {'address': '10.0.0.1'}} self.stubs.Set(self.network.db, 'floating_ip_get_by_address', fake_floating_ip_get_by_address) self.assertRaises(exception.FloatingIpAlreadyInUse, self.network.associate_floating_ip, ctxt, mox.IgnoreArg(), mox.IgnoreArg()) def test_add_fixed_ip_instance_without_vpn_requested_networks(self): self.mox.StubOutWithMock(db, 'network_get') self.mox.StubOutWithMock(db, 'fixed_ip_associate_pool') self.mox.StubOutWithMock(db, 'instance_get') self.mox.StubOutWithMock(db, 'virtual_interface_get_by_instance_and_network') self.mox.StubOutWithMock(db, 'fixed_ip_update') db.fixed_ip_update(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()) db.virtual_interface_get_by_instance_and_network(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'id': 0}) db.instance_get(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn({'security_groups': [{'id': 0}]}) db.fixed_ip_associate_pool(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()).AndReturn('192.168.0.101') db.network_get(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(networks[0]) self.mox.ReplayAll() self.network.add_fixed_ip_to_instance(self.context, 1, HOST, networks[0]['id']) class CommonNetworkTestCase(test.TestCase): def fake_create_fixed_ips(self, context, network_id): return None def test_remove_fixed_ip_from_instance(self): manager = fake_network.FakeNetworkManager() manager.remove_fixed_ip_from_instance(None, 99, '10.0.0.1') self.assertEquals(manager.deallocate_called, '10.0.0.1') def test_remove_fixed_ip_from_instance_bad_input(self): manager = fake_network.FakeNetworkManager() self.assertRaises(exception.FixedIpNotFoundForSpecificInstance, manager.remove_fixed_ip_from_instance, None, 99, 'bad input') def test_validate_cidrs(self): manager = fake_network.FakeNetworkManager() nets = manager.create_networks(None, 'fake', '192.168.0.0/24', False, 1, 256, None, None, None, None) self.assertEqual(1, len(nets)) cidrs = [str(net['cidr']) for net in nets] self.assertTrue('192.168.0.0/24' in cidrs) def test_validate_cidrs_split_exact_in_half(self): manager = fake_network.FakeNetworkManager() nets = manager.create_networks(None, 'fake', '192.168.0.0/24', False, 2, 128, None, None, None, None) self.assertEqual(2, len(nets)) cidrs = [str(net['cidr']) for net in nets] self.assertTrue('192.168.0.0/25' in cidrs) self.assertTrue('192.168.0.128/25' in cidrs) def test_validate_cidrs_split_cidr_in_use_middle_of_range(self): manager = fake_network.FakeNetworkManager() self.mox.StubOutWithMock(manager.db, 'network_get_all') ctxt = mox.IgnoreArg() manager.db.network_get_all(ctxt).AndReturn([{'id': 1, 'cidr': '192.168.2.0/24'}]) self.mox.ReplayAll() nets = manager.create_networks(None, 'fake', '192.168.0.0/16', False, 4, 256, None, None, None, None) self.assertEqual(4, len(nets)) cidrs = [str(net['cidr']) for net in nets] exp_cidrs = ['192.168.0.0/24', '192.168.1.0/24', '192.168.3.0/24', '192.168.4.0/24'] for exp_cidr in exp_cidrs: self.assertTrue(exp_cidr in cidrs) self.assertFalse('192.168.2.0/24' in cidrs) def test_validate_cidrs_smaller_subnet_in_use(self): manager = fake_network.FakeNetworkManager() self.mox.StubOutWithMock(manager.db, 'network_get_all') ctxt = mox.IgnoreArg() manager.db.network_get_all(ctxt).AndReturn([{'id': 1, 'cidr': '192.168.2.9/25'}]) self.mox.ReplayAll() # ValueError: requested cidr (192.168.2.0/24) conflicts with # existing smaller cidr args = (None, 'fake', '192.168.2.0/24', False, 1, 256, None, None, None, None) self.assertRaises(ValueError, manager.create_networks, *args) def test_validate_cidrs_split_smaller_cidr_in_use(self): manager = fake_network.FakeNetworkManager() self.mox.StubOutWithMock(manager.db, 'network_get_all') ctxt = mox.IgnoreArg() manager.db.network_get_all(ctxt).AndReturn([{'id': 1, 'cidr': '192.168.2.0/25'}]) self.mox.ReplayAll() nets = manager.create_networks(None, 'fake', '192.168.0.0/16', False, 4, 256, None, None, None, None) self.assertEqual(4, len(nets)) cidrs = [str(net['cidr']) for net in nets] exp_cidrs = ['192.168.0.0/24', '192.168.1.0/24', '192.168.3.0/24', '192.168.4.0/24'] for exp_cidr in exp_cidrs: self.assertTrue(exp_cidr in cidrs) self.assertFalse('192.168.2.0/24' in cidrs) def test_validate_cidrs_split_smaller_cidr_in_use2(self): manager = fake_network.FakeNetworkManager() self.mox.StubOutWithMock(manager.db, 'network_get_all') ctxt = mox.IgnoreArg() manager.db.network_get_all(ctxt).AndReturn([{'id': 1, 'cidr': '192.168.2.9/29'}]) self.mox.ReplayAll() nets = manager.create_networks(None, 'fake', '192.168.2.0/24', False, 3, 32, None, None, None, None) self.assertEqual(3, len(nets)) cidrs = [str(net['cidr']) for net in nets] exp_cidrs = ['192.168.2.32/27', '192.168.2.64/27', '192.168.2.96/27'] for exp_cidr in exp_cidrs: self.assertTrue(exp_cidr in cidrs) self.assertFalse('192.168.2.0/27' in cidrs) def test_validate_cidrs_split_all_in_use(self): manager = fake_network.FakeNetworkManager() self.mox.StubOutWithMock(manager.db, 'network_get_all') ctxt = mox.IgnoreArg() in_use = [{'id': 1, 'cidr': '192.168.2.9/29'}, {'id': 2, 'cidr': '192.168.2.64/26'}, {'id': 3, 'cidr': '192.168.2.128/26'}] manager.db.network_get_all(ctxt).AndReturn(in_use) self.mox.ReplayAll() args = (None, 'fake', '192.168.2.0/24', False, 3, 64, None, None, None, None) # ValueError: Not enough subnets avail to satisfy requested num_ # networks - some subnets in requested range already # in use self.assertRaises(ValueError, manager.create_networks, *args) def test_validate_cidrs_one_in_use(self): manager = fake_network.FakeNetworkManager() args = (None, 'fake', '192.168.0.0/24', False, 2, 256, None, None, None, None) # ValueError: network_size * num_networks exceeds cidr size self.assertRaises(ValueError, manager.create_networks, *args) def test_validate_cidrs_already_used(self): manager = fake_network.FakeNetworkManager() self.mox.StubOutWithMock(manager.db, 'network_get_all') ctxt = mox.IgnoreArg() manager.db.network_get_all(ctxt).AndReturn([{'id': 1, 'cidr': '192.168.0.0/24'}]) self.mox.ReplayAll() # ValueError: cidr already in use args = (None, 'fake', '192.168.0.0/24', False, 1, 256, None, None, None, None) self.assertRaises(ValueError, manager.create_networks, *args) def test_validate_cidrs_too_many(self): manager = fake_network.FakeNetworkManager() args = (None, 'fake', '192.168.0.0/24', False, 200, 256, None, None, None, None) # ValueError: Not enough subnets avail to satisfy requested # num_networks self.assertRaises(ValueError, manager.create_networks, *args) def test_validate_cidrs_split_partial(self): manager = fake_network.FakeNetworkManager() nets = manager.create_networks(None, 'fake', '192.168.0.0/16', False, 2, 256, None, None, None, None) returned_cidrs = [str(net['cidr']) for net in nets] self.assertTrue('192.168.0.0/24' in returned_cidrs) self.assertTrue('192.168.1.0/24' in returned_cidrs) def test_validate_cidrs_conflict_existing_supernet(self): manager = fake_network.FakeNetworkManager() self.mox.StubOutWithMock(manager.db, 'network_get_all') ctxt = mox.IgnoreArg() fakecidr = [{'id': 1, 'cidr': '192.168.0.0/8'}] manager.db.network_get_all(ctxt).AndReturn(fakecidr) self.mox.ReplayAll() args = (None, 'fake', '192.168.0.0/24', False, 1, 256, None, None, None, None) # ValueError: requested cidr (192.168.0.0/24) conflicts # with existing supernet self.assertRaises(ValueError, manager.create_networks, *args) def test_create_networks(self): cidr = '192.168.0.0/24' manager = fake_network.FakeNetworkManager() self.stubs.Set(manager, '_create_fixed_ips', self.fake_create_fixed_ips) args = [None, 'foo', cidr, None, 1, 256, 'fd00::/48', None, None, None] self.assertTrue(manager.create_networks(*args)) def test_create_networks_cidr_already_used(self): manager = fake_network.FakeNetworkManager() self.mox.StubOutWithMock(manager.db, 'network_get_all') ctxt = mox.IgnoreArg() fakecidr = [{'id': 1, 'cidr': '192.168.0.0/24'}] manager.db.network_get_all(ctxt).AndReturn(fakecidr) self.mox.ReplayAll() args = [None, 'foo', '192.168.0.0/24', None, 1, 256, 'fd00::/48', None, None, None] self.assertRaises(ValueError, manager.create_networks, *args) def test_create_networks_many(self): cidr = '192.168.0.0/16' manager = fake_network.FakeNetworkManager() self.stubs.Set(manager, '_create_fixed_ips', self.fake_create_fixed_ips) args = [None, 'foo', cidr, None, 10, 256, 'fd00::/48', None, None, None] self.assertTrue(manager.create_networks(*args)) def test_get_instance_uuids_by_ip_regex(self): manager = fake_network.FakeNetworkManager() _vifs = manager.db.virtual_interface_get_all(None) # Greedy get eveything res = manager.get_instance_uuids_by_ip_filter(None, {'ip': '.*'}) self.assertEqual(len(res), len(_vifs)) # Doesn't exist res = manager.get_instance_uuids_by_ip_filter(None, {'ip': '10.0.0.1'}) self.assertFalse(res) # Get instance 1 res = manager.get_instance_uuids_by_ip_filter(None, {'ip': '172.16.0.2'}) self.assertTrue(res) self.assertEqual(len(res), 1) self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id']) # Get instance 2 res = manager.get_instance_uuids_by_ip_filter(None, {'ip': '173.16.0.2'}) self.assertTrue(res) self.assertEqual(len(res), 1) self.assertEqual(res[0]['instance_id'], _vifs[2]['instance_id']) # Get instance 0 and 1 res = manager.get_instance_uuids_by_ip_filter(None, {'ip': '172.16.0.*'}) self.assertTrue(res) self.assertEqual(len(res), 2) self.assertEqual(res[0]['instance_id'], _vifs[0]['instance_id']) self.assertEqual(res[1]['instance_id'], _vifs[1]['instance_id']) # Get instance 1 and 2 res = manager.get_instance_uuids_by_ip_filter(None, {'ip': '17..16.0.2'}) self.assertTrue(res) self.assertEqual(len(res), 2) self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id']) self.assertEqual(res[1]['instance_id'], _vifs[2]['instance_id']) def test_get_instance_uuids_by_ipv6_regex(self): manager = fake_network.FakeNetworkManager() _vifs = manager.db.virtual_interface_get_all(None) # Greedy get eveything res = manager.get_instance_uuids_by_ip_filter(None, {'ip6': '.*'}) self.assertEqual(len(res), len(_vifs)) # Doesn't exist res = manager.get_instance_uuids_by_ip_filter(None, {'ip6': '.*1034.*'}) self.assertFalse(res) # Get instance 1 res = manager.get_instance_uuids_by_ip_filter(None, {'ip6': '2001:.*:2'}) self.assertTrue(res) self.assertEqual(len(res), 1) self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id']) # Get instance 2 ip6 = '2002:db8::dcad:beff:feef:2' res = manager.get_instance_uuids_by_ip_filter(None, {'ip6': ip6}) self.assertTrue(res) self.assertEqual(len(res), 1) self.assertEqual(res[0]['instance_id'], _vifs[2]['instance_id']) # Get instance 0 and 1 res = manager.get_instance_uuids_by_ip_filter(None, {'ip6': '2001:.*'}) self.assertTrue(res) self.assertEqual(len(res), 2) self.assertEqual(res[0]['instance_id'], _vifs[0]['instance_id']) self.assertEqual(res[1]['instance_id'], _vifs[1]['instance_id']) # Get instance 1 and 2 ip6 = '200.:db8::dcad:beff:feef:2' res = manager.get_instance_uuids_by_ip_filter(None, {'ip6': ip6}) self.assertTrue(res) self.assertEqual(len(res), 2) self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id']) self.assertEqual(res[1]['instance_id'], _vifs[2]['instance_id']) def test_get_instance_uuids_by_ip(self): manager = fake_network.FakeNetworkManager() _vifs = manager.db.virtual_interface_get_all(None) # No regex for you! res = manager.get_instance_uuids_by_ip_filter(None, {'fixed_ip': '.*'}) self.assertFalse(res) # Doesn't exist ip = '10.0.0.1' res = manager.get_instance_uuids_by_ip_filter(None, {'fixed_ip': ip}) self.assertFalse(res) # Get instance 1 ip = '172.16.0.2' res = manager.get_instance_uuids_by_ip_filter(None, {'fixed_ip': ip}) self.assertTrue(res) self.assertEqual(len(res), 1) self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id']) # Get instance 2 ip = '173.16.0.2' res = manager.get_instance_uuids_by_ip_filter(None, {'fixed_ip': ip}) self.assertTrue(res) self.assertEqual(len(res), 1) self.assertEqual(res[0]['instance_id'], _vifs[2]['instance_id'])