From 3e4925b36f11abc8cc746682118c3c58fb2bb88d Mon Sep 17 00:00:00 2001 From: Rick Harris Date: Thu, 17 May 2012 12:05:41 -0500 Subject: Moving network tests into the network folder. Change-Id: Ic8f20967fb999cc67f74bd5ef751c7bb23129604 --- nova/tests/network/test_linux_net.py | 501 +++++++++++++++++++++++++ nova/tests/network/test_network_info.py | 326 ++++++++++++++++ nova/tests/network/test_quantum.py | 647 ++++++++++++++++++++++++++++++++ nova/tests/test_linux_net.py | 501 ------------------------- nova/tests/test_network_info.py | 326 ---------------- nova/tests/test_quantum.py | 647 -------------------------------- 6 files changed, 1474 insertions(+), 1474 deletions(-) create mode 100644 nova/tests/network/test_linux_net.py create mode 100644 nova/tests/network/test_network_info.py create mode 100644 nova/tests/network/test_quantum.py delete mode 100644 nova/tests/test_linux_net.py delete mode 100644 nova/tests/test_network_info.py delete mode 100644 nova/tests/test_quantum.py diff --git a/nova/tests/network/test_linux_net.py b/nova/tests/network/test_linux_net.py new file mode 100644 index 000000000..1e7d3e0ff --- /dev/null +++ b/nova/tests/network/test_linux_net.py @@ -0,0 +1,501 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 NTT +# 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 os + +import mox + +from nova import context +from nova import db +from nova import flags +from nova import log as logging +from nova.openstack.common import importutils +from nova import test +from nova import utils +from nova.network import linux_net + + +FLAGS = flags.FLAGS + +LOG = logging.getLogger(__name__) + + +HOST = "testhost" + +instances = [{'id': 0, + 'host': 'fake_instance00', + 'created_at': 'fakedate', + 'updated_at': 'fakedate', + 'hostname': 'fake_instance00'}, + {'id': 1, + 'host': 'fake_instance01', + 'created_at': 'fakedate', + 'updated_at': 'fakedate', + 'hostname': 'fake_instance01'}] + + +addresses = [{"address": "10.0.0.1"}, + {"address": "10.0.0.2"}, + {"address": "10.0.0.3"}, + {"address": "10.0.0.4"}, + {"address": "10.0.0.5"}, + {"address": "10.0.0.6"}] + + +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', + 'dhcp_server': '0.0.0.0', + 'dhcp_start': '192.168.100.1', + 'vlan': None, + 'host': None, + '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': True, + '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', + 'dhcp_server': '0.0.0.0', + 'dhcp_start': '192.168.100.1', + 'vlan': None, + 'host': None, + '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': True, + 'virtual_interface_id': 0, + 'instance_id': 0, + 'floating_ips': []}, + {'id': 1, + 'network_id': 1, + 'address': '192.168.1.100', + 'instance_id': 0, + 'allocated': True, + 'virtual_interface_id': 1, + 'instance_id': 0, + 'floating_ips': []}, + {'id': 2, + 'network_id': 1, + 'address': '192.168.0.101', + 'instance_id': 1, + 'allocated': True, + 'virtual_interface_id': 2, + 'instance_id': 1, + 'floating_ips': []}, + {'id': 3, + 'network_id': 0, + 'address': '192.168.1.101', + 'instance_id': 1, + 'allocated': True, + 'virtual_interface_id': 3, + 'instance_id': 1, + 'floating_ips': []}, + {'id': 4, + 'network_id': 0, + 'address': '192.168.0.102', + 'instance_id': 0, + 'allocated': True, + 'virtual_interface_id': 4, + 'instance_id': 0, + 'floating_ips': []}, + {'id': 5, + 'network_id': 1, + 'address': '192.168.1.102', + 'instance_id': 1, + 'allocated': True, + 'virtual_interface_id': 5, + 'instance_id': 1, + 'floating_ips': []}] + + +vifs = [{'id': 0, + 'address': 'DE:AD:BE:EF:00:00', + 'uuid': '00000000-0000-0000-0000-0000000000000000', + 'network_id': 0, + 'instance_id': 0}, + {'id': 1, + 'address': 'DE:AD:BE:EF:00:01', + 'uuid': '00000000-0000-0000-0000-0000000000000001', + 'network_id': 1, + 'instance_id': 0}, + {'id': 2, + 'address': 'DE:AD:BE:EF:00:02', + 'uuid': '00000000-0000-0000-0000-0000000000000002', + 'network_id': 1, + 'instance_id': 1}, + {'id': 3, + 'address': 'DE:AD:BE:EF:00:03', + 'uuid': '00000000-0000-0000-0000-0000000000000003', + 'network_id': 0, + 'instance_id': 1}, + {'id': 4, + 'address': 'DE:AD:BE:EF:00:04', + 'uuid': '00000000-0000-0000-0000-0000000000000004', + 'network_id': 0, + 'instance_id': 0}, + {'id': 5, + 'address': 'DE:AD:BE:EF:00:05', + 'uuid': '00000000-0000-0000-0000-0000000000000005', + 'network_id': 1, + 'instance_id': 1}] + + +def get_associated(context, network_id, host=None): + result = [] + for datum in fixed_ips: + if (datum['network_id'] == network_id and datum['allocated'] + and datum['instance_id'] is not None + and datum['virtual_interface_id'] is not None): + instance = instances[datum['instance_id']] + if host and host != instance['host']: + continue + cleaned = {} + cleaned['address'] = datum['address'] + cleaned['instance_id'] = datum['instance_id'] + cleaned['network_id'] = datum['network_id'] + cleaned['vif_id'] = datum['virtual_interface_id'] + vif = vifs[datum['virtual_interface_id']] + cleaned['vif_address'] = vif['address'] + cleaned['instance_hostname'] = instance['hostname'] + cleaned['instance_updated'] = instance['updated_at'] + cleaned['instance_created'] = instance['created_at'] + result.append(cleaned) + return result + + +class LinuxNetworkTestCase(test.TestCase): + + def setUp(self): + super(LinuxNetworkTestCase, self).setUp() + network_driver = FLAGS.network_driver + self.driver = importutils.import_module(network_driver) + self.driver.db = db + self.context = context.RequestContext('testuser', 'testproject', + is_admin=True) + + def get_vifs(_context, instance_id): + return [vif for vif in vifs if vif['instance_id'] == instance_id] + + def get_instance(_context, instance_id): + return instances[instance_id] + + self.stubs.Set(db, 'virtual_interface_get_by_instance', get_vifs) + self.stubs.Set(db, 'instance_get', get_instance) + self.stubs.Set(db, 'network_get_associated_fixed_ips', get_associated) + + def test_update_dhcp_for_nw00(self): + self.flags(use_single_default_gateway=True) + + self.mox.StubOutWithMock(self.driver, 'write_to_file') + self.mox.StubOutWithMock(self.driver, 'ensure_path') + self.mox.StubOutWithMock(os, 'chmod') + + self.driver.write_to_file(mox.IgnoreArg(), mox.IgnoreArg()) + self.driver.write_to_file(mox.IgnoreArg(), mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + os.chmod(mox.IgnoreArg(), mox.IgnoreArg()) + os.chmod(mox.IgnoreArg(), mox.IgnoreArg()) + + self.mox.ReplayAll() + + self.driver.update_dhcp(self.context, "eth0", networks[0]) + + def test_update_dhcp_for_nw01(self): + self.flags(use_single_default_gateway=True) + + self.mox.StubOutWithMock(self.driver, 'write_to_file') + self.mox.StubOutWithMock(self.driver, 'ensure_path') + self.mox.StubOutWithMock(os, 'chmod') + + self.driver.write_to_file(mox.IgnoreArg(), mox.IgnoreArg()) + self.driver.write_to_file(mox.IgnoreArg(), mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + self.driver.ensure_path(mox.IgnoreArg()) + os.chmod(mox.IgnoreArg(), mox.IgnoreArg()) + os.chmod(mox.IgnoreArg(), mox.IgnoreArg()) + + self.mox.ReplayAll() + + self.driver.update_dhcp(self.context, "eth0", networks[0]) + + def test_get_dhcp_hosts_for_nw00(self): + self.flags(use_single_default_gateway=True) + + expected = ( + "DE:AD:BE:EF:00:00,fake_instance00.novalocal," + "192.168.0.100,net:NW-0\n" + "DE:AD:BE:EF:00:03,fake_instance01.novalocal," + "192.168.1.101,net:NW-3\n" + "DE:AD:BE:EF:00:04,fake_instance00.novalocal," + "192.168.0.102,net:NW-4" + ) + actual_hosts = self.driver.get_dhcp_hosts(self.context, networks[0]) + + self.assertEquals(actual_hosts, expected) + + def test_get_dhcp_hosts_for_nw01(self): + self.flags(use_single_default_gateway=True) + self.flags(host='fake_instance01') + + expected = ( + "DE:AD:BE:EF:00:02,fake_instance01.novalocal," + "192.168.0.101,net:NW-2\n" + "DE:AD:BE:EF:00:05,fake_instance01.novalocal," + "192.168.1.102,net:NW-5" + ) + actual_hosts = self.driver.get_dhcp_hosts(self.context, networks[1]) + + self.assertEquals(actual_hosts, expected) + + def test_get_dhcp_opts_for_nw00(self): + expected_opts = 'NW-0,3\nNW-3,3\nNW-4,3' + actual_opts = self.driver.get_dhcp_opts(self.context, networks[0]) + + self.assertEquals(actual_opts, expected_opts) + + def test_get_dhcp_opts_for_nw01(self): + self.flags(host='fake_instance01') + expected_opts = "NW-5,3" + actual_opts = self.driver.get_dhcp_opts(self.context, networks[1]) + + self.assertEquals(actual_opts, expected_opts) + + def test_dhcp_opts_not_default_gateway_network(self): + expected = "NW-0,3" + data = get_associated(self.context, 0)[0] + actual = self.driver._host_dhcp_opts(data) + self.assertEquals(actual, expected) + + def test_host_dhcp_without_default_gateway_network(self): + expected = ','.join(['DE:AD:BE:EF:00:00', + 'fake_instance00.novalocal', + '192.168.0.100']) + data = get_associated(self.context, 0)[0] + actual = self.driver._host_dhcp(data) + self.assertEquals(actual, expected) + + def test_linux_bridge_driver_plug(self): + """Makes sure plug doesn't drop FORWARD by default. + + Ensures bug 890195 doesn't reappear.""" + + def fake_execute(*args, **kwargs): + return "", "" + self.stubs.Set(utils, 'execute', fake_execute) + + def verify_add_rule(chain, rule): + self.assertEqual(chain, 'FORWARD') + self.assertIn('ACCEPT', rule) + self.stubs.Set(linux_net.iptables_manager.ipv4['filter'], + 'add_rule', verify_add_rule) + driver = linux_net.LinuxBridgeInterfaceDriver() + driver.plug({"bridge": "br100", "bridge_interface": "eth0"}, + "fakemac") + + def test_vlan_override(self): + """Makes sure vlan_interface flag overrides network bridge_interface. + + Allows heterogeneous networks a la bug 833426""" + + driver = linux_net.LinuxBridgeInterfaceDriver() + + info = {} + + @classmethod + def test_ensure(_self, vlan, bridge, interface, network, mac_address): + info['passed_interface'] = interface + + self.stubs.Set(linux_net.LinuxBridgeInterfaceDriver, + 'ensure_vlan_bridge', test_ensure) + + network = { + "bridge": "br100", + "bridge_interface": "base_interface", + "vlan": "fake" + } + driver.plug(network, "fakemac") + self.assertEqual(info['passed_interface'], "base_interface") + self.flags(vlan_interface="override_interface") + driver.plug(network, "fakemac") + self.assertEqual(info['passed_interface'], "override_interface") + driver.plug(network, "fakemac") + + def test_flat_override(self): + """Makes sure flat_interface flag overrides network bridge_interface. + + Allows heterogeneous networks a la bug 833426""" + + driver = linux_net.LinuxBridgeInterfaceDriver() + + info = {} + + @classmethod + def test_ensure(_self, bridge, interface, network, gateway): + info['passed_interface'] = interface + + self.stubs.Set(linux_net.LinuxBridgeInterfaceDriver, + 'ensure_bridge', test_ensure) + + network = { + "bridge": "br100", + "bridge_interface": "base_interface", + } + driver.plug(network, "fakemac") + self.assertEqual(info['passed_interface'], "base_interface") + self.flags(flat_interface="override_interface") + driver.plug(network, "fakemac") + self.assertEqual(info['passed_interface'], "override_interface") + + def _test_initialize_gateway(self, existing, expected, routes=''): + self.flags(fake_network=False) + executes = [] + + def fake_execute(*args, **kwargs): + executes.append(args) + if args[0] == 'ip' and args[1] == 'addr' and args[2] == 'show': + return existing, "" + if args[0] == 'route' and args[1] == '-n': + return routes, "" + self.stubs.Set(utils, 'execute', fake_execute) + network = {'dhcp_server': '192.168.1.1', + 'cidr': '192.168.1.0/24', + 'broadcast': '192.168.1.255', + 'cidr_v6': '2001:db8::/64'} + self.driver.initialize_gateway_device('eth0', network) + self.assertEqual(executes, expected) + + def test_initialize_gateway_moves_wrong_ip(self): + existing = ("2: eth0: " + " mtu 1500 qdisc pfifo_fast state UNKNOWN qlen 1000\n" + " link/ether de:ad:be:ef:be:ef brd ff:ff:ff:ff:ff:ff\n" + " inet 192.168.0.1/24 brd 192.168.0.255 scope global eth0\n" + " inet6 dead::beef:dead:beef:dead/64 scope link\n" + " valid_lft forever preferred_lft forever\n") + expected = [ + ('sysctl', '-w', 'net.ipv4.ip_forward=1'), + ('ip', 'addr', 'show', 'dev', 'eth0', 'scope', 'global'), + ('route', '-n'), + ('ip', 'addr', 'del', '192.168.0.1/24', + 'brd', '192.168.0.255', 'scope', 'global', 'dev', 'eth0'), + ('ip', 'addr', 'add', '192.168.1.1/24', + 'brd', '192.168.1.255', 'dev', 'eth0'), + ('ip', 'addr', 'add', '192.168.0.1/24', + 'brd', '192.168.0.255', 'scope', 'global', 'dev', 'eth0'), + ('ip', '-f', 'inet6', 'addr', 'change', + '2001:db8::/64', 'dev', 'eth0'), + ] + self._test_initialize_gateway(existing, expected) + + def test_initialize_gateway_resets_route(self): + routes = ("0.0.0.0 192.68.0.1 0.0.0.0 " + "UG 100 0 0 eth0") + existing = ("2: eth0: " + " mtu 1500 qdisc pfifo_fast state UNKNOWN qlen 1000\n" + " link/ether de:ad:be:ef:be:ef brd ff:ff:ff:ff:ff:ff\n" + " inet 192.168.0.1/24 brd 192.168.0.255 scope global eth0\n" + " inet6 dead::beef:dead:beef:dead/64 scope link\n" + " valid_lft forever preferred_lft forever\n") + expected = [ + ('sysctl', '-w', 'net.ipv4.ip_forward=1'), + ('ip', 'addr', 'show', 'dev', 'eth0', 'scope', 'global'), + ('route', '-n'), + ('route', 'del', 'default', 'gw', '192.68.0.1', 'dev', 'eth0'), + ('ip', 'addr', 'del', '192.168.0.1/24', + 'brd', '192.168.0.255', 'scope', 'global', 'dev', 'eth0'), + ('ip', 'addr', 'add', '192.168.1.1/24', + 'brd', '192.168.1.255', 'dev', 'eth0'), + ('ip', 'addr', 'add', '192.168.0.1/24', + 'brd', '192.168.0.255', 'scope', 'global', 'dev', 'eth0'), + ('route', 'add', 'default', 'gw', '192.68.0.1'), + ('ip', '-f', 'inet6', 'addr', 'change', + '2001:db8::/64', 'dev', 'eth0'), + ] + self._test_initialize_gateway(existing, expected, routes) + + def test_initialize_gateway_no_move_right_ip(self): + existing = ("2: eth0: " + " mtu 1500 qdisc pfifo_fast state UNKNOWN qlen 1000\n" + " link/ether de:ad:be:ef:be:ef brd ff:ff:ff:ff:ff:ff\n" + " inet 192.168.1.1/24 brd 192.168.1.255 scope global eth0\n" + " inet 192.168.0.1/24 brd 192.168.0.255 scope global eth0\n" + " inet6 dead::beef:dead:beef:dead/64 scope link\n" + " valid_lft forever preferred_lft forever\n") + expected = [ + ('sysctl', '-w', 'net.ipv4.ip_forward=1'), + ('ip', 'addr', 'show', 'dev', 'eth0', 'scope', 'global'), + ('ip', '-f', 'inet6', 'addr', 'change', + '2001:db8::/64', 'dev', 'eth0'), + ] + self._test_initialize_gateway(existing, expected) + + def test_initialize_gateway_add_if_blank(self): + existing = ("2: eth0: " + " mtu 1500 qdisc pfifo_fast state UNKNOWN qlen 1000\n" + " link/ether de:ad:be:ef:be:ef brd ff:ff:ff:ff:ff:ff\n" + " inet6 dead::beef:dead:beef:dead/64 scope link\n" + " valid_lft forever preferred_lft forever\n") + expected = [ + ('sysctl', '-w', 'net.ipv4.ip_forward=1'), + ('ip', 'addr', 'show', 'dev', 'eth0', 'scope', 'global'), + ('route', '-n'), + ('ip', 'addr', 'add', '192.168.1.1/24', + 'brd', '192.168.1.255', 'dev', 'eth0'), + ('ip', '-f', 'inet6', 'addr', 'change', + '2001:db8::/64', 'dev', 'eth0'), + ] + self._test_initialize_gateway(existing, expected) diff --git a/nova/tests/network/test_network_info.py b/nova/tests/network/test_network_info.py new file mode 100644 index 000000000..f1c908b2c --- /dev/null +++ b/nova/tests/network/test_network_info.py @@ -0,0 +1,326 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 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 exception +from nova import log as logging +from nova.network import model +from nova import test +from nova.tests import fake_network_cache_model + +LOG = logging.getLogger(__name__) + + +class RouteTests(test.TestCase): + def test_create_route_with_attrs(self): + route = fake_network_cache_model.new_route() + ip = fake_network_cache_model.new_ip(dict(address='192.168.1.1')) + self.assertEqual(route['cidr'], '0.0.0.0/24') + self.assertEqual(route['gateway']['address'], '192.168.1.1') + self.assertEqual(route['interface'], 'eth0') + + def test_routes_equal(self): + route1 = fake_network_cache_model.new_route() + route2 = fake_network_cache_model.new_route() + self.assertEqual(route1, route2) + + def test_routes_not_equal(self): + route1 = fake_network_cache_model.new_route() + route2 = fake_network_cache_model.new_route(dict(cidr='1.1.1.1/24')) + self.assertNotEqual(route1, route2) + + def test_hydrate(self): + route = model.Route.hydrate( + {'gateway': fake_network_cache_model.new_ip( + dict(address='192.168.1.1'))}) + self.assertEqual(route['cidr'], None) + self.assertEqual(route['gateway']['address'], '192.168.1.1') + self.assertEqual(route['interface'], None) + + +class FixedIPTests(test.TestCase): + def test_createnew_fixed_ip_with_attrs(self): + fixed_ip = model.FixedIP(address='192.168.1.100') + self.assertEqual(fixed_ip['address'], '192.168.1.100') + self.assertEqual(fixed_ip['floating_ips'], []) + self.assertEqual(fixed_ip['type'], 'fixed') + self.assertEqual(fixed_ip['version'], 4) + + def test_create_fixed_ipv6(self): + fixed_ip = model.FixedIP(address='::1') + self.assertEqual(fixed_ip['address'], '::1') + self.assertEqual(fixed_ip['floating_ips'], []) + self.assertEqual(fixed_ip['type'], 'fixed') + self.assertEqual(fixed_ip['version'], 6) + + def test_create_fixed_bad_ip_fails(self): + self.assertRaises(exception.InvalidIpAddressError, + model.FixedIP, + address='picklespicklespickles') + + def test_equate_two_fixed_ips(self): + fixed_ip = model.FixedIP(address='::1') + fixed_ip2 = model.FixedIP(address='::1') + self.assertEqual(fixed_ip, fixed_ip2) + + def test_equate_two_dissimilar_fixed_ips_fails(self): + fixed_ip = model.FixedIP(address='::1') + fixed_ip2 = model.FixedIP(address='::2') + self.assertNotEqual(fixed_ip, fixed_ip2) + + def test_hydrate(self): + fixed_ip = model.FixedIP.hydrate({}) + self.assertEqual(fixed_ip['floating_ips'], []) + self.assertEqual(fixed_ip['address'], None) + self.assertEqual(fixed_ip['type'], 'fixed') + self.assertEqual(fixed_ip['version'], None) + + def test_add_floating_ip(self): + fixed_ip = model.FixedIP(address='192.168.1.100') + fixed_ip.add_floating_ip('192.168.1.101') + self.assertEqual(fixed_ip['floating_ips'], ['192.168.1.101']) + + def test_add_floating_ip_repeatedly_only_one_instance(self): + fixed_ip = model.FixedIP(address='192.168.1.100') + for i in xrange(10): + fixed_ip.add_floating_ip('192.168.1.101') + self.assertEqual(fixed_ip['floating_ips'], ['192.168.1.101']) + + +class SubnetTests(test.TestCase): + def test_create_subnet_with_attrs(self): + subnet = fake_network_cache_model.new_subnet() + + route1 = fake_network_cache_model.new_route() + + self.assertEqual(subnet['cidr'], '10.10.0.0/24') + self.assertEqual(subnet['dns'], + [fake_network_cache_model.new_ip(dict(address='1.2.3.4')), + fake_network_cache_model.new_ip(dict(address='2.3.4.5'))]) + self.assertEqual(subnet['gateway']['address'], '10.10.0.1') + self.assertEqual(subnet['ips'], + [fake_network_cache_model.new_ip( + dict(address='10.10.0.2')), + fake_network_cache_model.new_ip( + dict(address='10.10.0.3'))]) + self.assertEqual(subnet['routes'], [route1]) + self.assertEqual(subnet['version'], 4) + + def test_add_route(self): + subnet = fake_network_cache_model.new_subnet() + route1 = fake_network_cache_model.new_route() + route2 = fake_network_cache_model.new_route({'cidr': '1.1.1.1/24'}) + subnet.add_route(route2) + self.assertEqual(subnet['routes'], [route1, route2]) + + def test_add_route_a_lot(self): + subnet = fake_network_cache_model.new_subnet() + route1 = fake_network_cache_model.new_route() + route2 = fake_network_cache_model.new_route({'cidr': '1.1.1.1/24'}) + for i in xrange(10): + subnet.add_route(route2) + self.assertEqual(subnet['routes'], [route1, route2]) + + def test_add_dns(self): + subnet = fake_network_cache_model.new_subnet() + dns = fake_network_cache_model.new_ip(dict(address='9.9.9.9')) + subnet.add_dns(dns) + self.assertEqual(subnet['dns'], + [fake_network_cache_model.new_ip(dict(address='1.2.3.4')), + fake_network_cache_model.new_ip(dict(address='2.3.4.5')), + fake_network_cache_model.new_ip(dict(address='9.9.9.9'))]) + + def test_add_dns_a_lot(self): + subnet = fake_network_cache_model.new_subnet() + for i in xrange(10): + subnet.add_dns(fake_network_cache_model.new_ip( + dict(address='9.9.9.9'))) + self.assertEqual(subnet['dns'], + [fake_network_cache_model.new_ip(dict(address='1.2.3.4')), + fake_network_cache_model.new_ip(dict(address='2.3.4.5')), + fake_network_cache_model.new_ip(dict(address='9.9.9.9'))]) + + def test_add_ip(self): + subnet = fake_network_cache_model.new_subnet() + subnet.add_ip(fake_network_cache_model.new_ip( + dict(address='192.168.1.102'))) + self.assertEqual(subnet['ips'], + [fake_network_cache_model.new_ip( + dict(address='10.10.0.2')), + fake_network_cache_model.new_ip( + dict(address='10.10.0.3')), + fake_network_cache_model.new_ip( + dict(address='192.168.1.102'))]) + + def test_add_ip_a_lot(self): + subnet = fake_network_cache_model.new_subnet() + for i in xrange(10): + subnet.add_ip(fake_network_cache_model.new_ip( + dict(address='192.168.1.102'))) + self.assertEqual(subnet['ips'], + [fake_network_cache_model.new_ip( + dict(address='10.10.0.2')), + fake_network_cache_model.new_ip( + dict(address='10.10.0.3')), + fake_network_cache_model.new_ip( + dict(address='192.168.1.102'))]) + + def test_hydrate(self): + subnet_dict = { + 'cidr': '255.255.255.0', + 'dns': [fake_network_cache_model.new_ip(dict(address='1.1.1.1'))], + 'ips': [fake_network_cache_model.new_ip(dict(address='2.2.2.2'))], + 'routes': [fake_network_cache_model.new_route()], + 'version': 4, + 'gateway': fake_network_cache_model.new_ip( + dict(address='3.3.3.3'))} + subnet = model.Subnet.hydrate(subnet_dict) + + self.assertEqual(subnet['cidr'], '255.255.255.0') + self.assertEqual(subnet['dns'], [fake_network_cache_model.new_ip( + dict(address='1.1.1.1'))]) + self.assertEqual(subnet['gateway']['address'], '3.3.3.3') + self.assertEqual(subnet['ips'], [fake_network_cache_model.new_ip( + dict(address='2.2.2.2'))]) + self.assertEqual(subnet['routes'], [ + fake_network_cache_model.new_route()]) + self.assertEqual(subnet['version'], 4) + + +class NetworkTests(test.TestCase): + def test_create_network(self): + network = fake_network_cache_model.new_network() + self.assertEqual(network['id'], 1) + self.assertEqual(network['bridge'], 'br0') + self.assertEqual(network['label'], 'public') + self.assertEqual(network['subnets'], + [fake_network_cache_model.new_subnet(), + fake_network_cache_model.new_subnet( + dict(cidr='255.255.255.255'))]) + + def test_add_subnet(self): + network = fake_network_cache_model.new_network() + network.add_subnet(fake_network_cache_model.new_subnet( + dict(cidr='0.0.0.0'))) + self.assertEqual(network['subnets'], + [fake_network_cache_model.new_subnet(), + fake_network_cache_model.new_subnet( + dict(cidr='255.255.255.255')), + fake_network_cache_model.new_subnet(dict(cidr='0.0.0.0'))]) + + def test_add_subnet_a_lot(self): + network = fake_network_cache_model.new_network() + for i in xrange(10): + network.add_subnet(fake_network_cache_model.new_subnet( + dict(cidr='0.0.0.0'))) + self.assertEqual(network['subnets'], + [fake_network_cache_model.new_subnet(), + fake_network_cache_model.new_subnet( + dict(cidr='255.255.255.255')), + fake_network_cache_model.new_subnet(dict(cidr='0.0.0.0'))]) + + def test_hydrate(self): + new_network = dict( + id=1, + bridge='br0', + label='public', + subnets=[fake_network_cache_model.new_subnet(), + fake_network_cache_model.new_subnet( + dict(cidr='255.255.255.255'))]) + network = model.Network.hydrate(fake_network_cache_model.new_network()) + + self.assertEqual(network['id'], 1) + self.assertEqual(network['bridge'], 'br0') + self.assertEqual(network['label'], 'public') + self.assertEqual(network['subnets'], + [fake_network_cache_model.new_subnet(), + fake_network_cache_model.new_subnet( + dict(cidr='255.255.255.255'))]) + + +class VIFTests(test.TestCase): + def test_create_vif(self): + vif = fake_network_cache_model.new_vif() + self.assertEqual(vif['id'], 1) + self.assertEqual(vif['address'], 'aa:aa:aa:aa:aa:aa') + self.assertEqual(vif['network'], + fake_network_cache_model.new_network()) + + def test_vif_get_fixed_ips(self): + vif = fake_network_cache_model.new_vif() + fixed_ips = vif.fixed_ips() + ips = [fake_network_cache_model.new_ip(dict(address='10.10.0.2')), + fake_network_cache_model.new_ip( + dict(address='10.10.0.3'))] * 2 + self.assertEqual(fixed_ips, ips) + + def test_vif_get_floating_ips(self): + vif = fake_network_cache_model.new_vif() + vif['network']['subnets'][0]['ips'][0].add_floating_ip('192.168.1.1') + floating_ips = vif.floating_ips() + self.assertEqual(floating_ips, ['192.168.1.1']) + + def test_vif_get_labeled_ips(self): + vif = fake_network_cache_model.new_vif() + labeled_ips = vif.labeled_ips() + ip_dict = { + 'network_id': 1, + 'ips': [fake_network_cache_model.new_ip( + {'address': '10.10.0.2'}), + fake_network_cache_model.new_ip( + {'address': '10.10.0.3'})] * 2, + 'network_label': 'public'} + self.assertEqual(labeled_ips, ip_dict) + + def test_hydrate(self): + new_vif = dict( + id=1, + address='127.0.0.1', + network=fake_network_cache_model.new_network()) + vif = model.VIF.hydrate(fake_network_cache_model.new_vif()) + self.assertEqual(vif['id'], 1) + self.assertEqual(vif['address'], 'aa:aa:aa:aa:aa:aa') + self.assertEqual(vif['network'], + fake_network_cache_model.new_network()) + + +class NetworkInfoTests(test.TestCase): + def test_create_model(self): + ninfo = model.NetworkInfo([fake_network_cache_model.new_vif(), + fake_network_cache_model.new_vif( + {'address':'bb:bb:bb:bb:bb:bb'})]) + self.assertEqual(ninfo.fixed_ips(), + [fake_network_cache_model.new_ip({'address': '10.10.0.2'}), + fake_network_cache_model.new_ip( + {'address': '10.10.0.3'})] * 4) + + def test_get_floating_ips(self): + vif = fake_network_cache_model.new_vif() + vif['network']['subnets'][0]['ips'][0].add_floating_ip('192.168.1.1') + ninfo = model.NetworkInfo([vif, + fake_network_cache_model.new_vif( + {'address':'bb:bb:bb:bb:bb:bb'})]) + self.assertEqual(ninfo.floating_ips(), ['192.168.1.1']) + + def test_hydrate(self): + ninfo = model.NetworkInfo([fake_network_cache_model.new_vif(), + fake_network_cache_model.new_vif( + {'address':'bb:bb:bb:bb:bb:bb'})]) + deserialized = model.NetworkInfo.hydrate(ninfo) + self.assertEqual(ninfo.fixed_ips(), + [fake_network_cache_model.new_ip({'address': '10.10.0.2'}), + fake_network_cache_model.new_ip( + {'address': '10.10.0.3'})] * 4) diff --git a/nova/tests/network/test_quantum.py b/nova/tests/network/test_quantum.py new file mode 100644 index 000000000..9f03c13f8 --- /dev/null +++ b/nova/tests/network/test_quantum.py @@ -0,0 +1,647 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011,2012 Nicira, Inc. +# 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.db.sqlalchemy import models +from nova.db.sqlalchemy import session as sql_session +from nova import exception +from nova import flags +from nova import log as logging +from nova.network.quantum import client as quantum_client +from nova.network.quantum import fake_client +from nova.network.quantum import manager as quantum_manager +from nova.network.quantum import quantum_connection +from nova.network.quantum import melange_connection +from nova.network.quantum import melange_ipam_lib + +from nova import test +from nova import utils + +LOG = logging.getLogger(__name__) +FLAGS = flags.FLAGS + + +networks = [{'label': 'project1-net1', + 'injected': False, + 'multi_host': False, + 'cidr': '100.168.0.0/24', + 'cidr_v6': '100:1db8::/64', + 'gateway_v6': '100:1db8::1', + 'netmask_v6': '64', + 'netmask': '255.255.255.0', + 'bridge': None, + 'bridge_interface': None, + 'gateway': '100.168.0.1', + 'broadcast': '100.168.0.255', + 'dns1': '8.8.8.8', + 'vlan': None, + 'host': None, + 'vpn_public_address': None, + 'project_id': 'fake_project1', + 'priority': 1}, + {'label': 'project2-net1', + 'injected': False, + 'multi_host': False, + 'cidr': '101.168.1.0/24', + 'cidr_v6': '101:1db9::/64', + 'gateway_v6': '101:1db9::1', + 'netmask_v6': '64', + 'netmask': '255.255.255.0', + 'bridge': None, + 'bridge_interface': None, + 'gateway': '101.168.1.1', + 'broadcast': '101.168.1.255', + 'dns1': '8.8.8.8', + 'vlan': None, + 'host': None, + 'project_id': 'fake_project2', + 'priority': 1}, + {'label': "public", + 'injected': False, + 'multi_host': False, + 'cidr': '102.0.0.0/24', + 'cidr_v6': '102:1dba::/64', + 'gateway_v6': '102:1dba::1', + 'netmask_v6': '64', + 'netmask': '255.255.255.0', + 'bridge': None, + 'bridge_interface': None, + 'gateway': '102.0.0.1', + 'broadcast': '102.0.0.255', + 'dns1': '8.8.8.8', + 'vlan': None, + 'host': None, + 'project_id': None, + 'priority': 0}, + {'label': "project2-net2", + 'injected': False, + 'multi_host': False, + 'cidr': '103.0.0.0/24', + 'cidr_v6': '103:1dbb::/64', + 'gateway_v6': '103:1dbb::1', + 'netmask_v6': '64', + 'netmask': '255.255.255.0', + 'bridge': None, + 'bridge_interface': None, + 'gateway': '103.0.0.1', + 'broadcast': '103.0.0.255', + 'dns1': '8.8.8.8', + 'vlan': None, + 'host': None, + 'project_id': "fake_project2", + 'priority': 2}] + + +class QuantumConnectionTestCase(test.TestCase): + + def test_connection(self): + fc = fake_client.FakeClient(LOG) + qc = quantum_connection.QuantumClientConnection(client=fc) + t = "tenant1" + net1_name = "net1" + net1_uuid = qc.create_network(t, net1_name) + self.assertEquals(net1_name, qc.get_network_name(t, net1_uuid)) + self.assertTrue(qc.network_exists(t, net1_uuid)) + self.assertFalse(qc.network_exists(t, "fake-uuid")) + self.assertFalse(qc.network_exists("fake-tenant", net1_uuid)) + + nets = qc.get_networks(t)['networks'] + self.assertEquals(len(nets), 1) + self.assertEquals(nets[0]['id'], net1_uuid) + + num_ports = 10 + for i in range(0, num_ports): + qc.create_and_attach_port(t, net1_uuid, + 'iface' + str(i), state='ACTIVE') + + self.assertEquals(len(qc.get_attached_ports(t, net1_uuid)), num_ports) + + for i in range(0, num_ports): + port_uuid = qc.get_port_by_attachment(t, net1_uuid, + 'iface' + str(i)) + self.assertTrue(port_uuid) + qc.detach_and_delete_port(t, net1_uuid, port_uuid) + + self.assertEquals(len(qc.get_attached_ports(t, net1_uuid)), 0) + + # test port not found + qc.create_and_attach_port(t, net1_uuid, 'foo', state='ACTIVE') + port_uuid = qc.get_port_by_attachment(t, net1_uuid, 'foo') + qc.detach_and_delete_port(t, net1_uuid, port_uuid) + self.assertRaises(quantum_client.QuantumNotFoundException, + qc.detach_and_delete_port, t, + net1_uuid, port_uuid) + + qc.delete_network(t, net1_uuid) + self.assertFalse(qc.network_exists(t, net1_uuid)) + self.assertEquals(len(qc.get_networks(t)['networks']), 0) + + self.assertRaises(quantum_client.QuantumNotFoundException, + qc.get_network_name, t, net1_uuid) + + +# this is a base class to be used by other QuantumManager Test classes +class QuantumNovaTestCase(test.TestCase): + + def setUp(self): + super(QuantumNovaTestCase, self).setUp() + + self.flags(quantum_use_dhcp=True) + self.flags(l3_lib="nova.network.l3.LinuxNetL3") + linuxdrv = "nova.network.linux_net.LinuxOVSInterfaceDriver" + self.flags(linuxnet_interface_driver=linuxdrv) + fc = fake_client.FakeClient(LOG) + qc = quantum_connection.QuantumClientConnection(client=fc) + + self.net_man = quantum_manager.QuantumManager( + ipam_lib="nova.network.quantum.nova_ipam_lib", + q_conn=qc) + + def func(arg1, arg2): + pass + + def func2(arg1, arg2, arg3): + pass + + def func1(arg1): + pass + + self.net_man.driver.update_dhcp_hostfile_with_text = func + self.net_man.driver.restart_dhcp = func2 + self.net_man.driver.kill_dhcp = func1 + + # Tests seem to create some networks by default, which + # we don't want. So we delete them. + + ctx = context.RequestContext('user1', 'fake_project1').elevated() + for n in db.network_get_all(ctx): + db.network_delete_safe(ctx, n['id']) + + # Other unit tests (e.g., test_compute.py) have a nasty + # habit of of creating fixed IPs and not cleaning up, which + # can confuse these tests, so we remove all existing fixed + # ips before starting. + session = sql_session.get_session() + result = session.query(models.FixedIp).all() + with session.begin(): + for fip_ref in result: + session.delete(fip_ref) + + self.net_man.init_host() + + def _create_network(self, n): + ctx = context.RequestContext('user1', n['project_id']) + nwks = self.net_man.create_networks( + ctx, + label=n['label'], cidr=n['cidr'], + multi_host=n['multi_host'], + num_networks=1, network_size=256, + cidr_v6=n['cidr_v6'], + gateway=n['gateway'], + gateway_v6=n['gateway_v6'], bridge=None, + bridge_interface=None, dns1=n['dns1'], + project_id=n['project_id'], + priority=n['priority']) + n['uuid'] = nwks[0]['uuid'] + + +class QuantumAllocationTestCase(QuantumNovaTestCase): + def test_get_network_in_db(self): + context = self.mox.CreateMockAnything() + context.elevated().AndReturn('elevated') + self.mox.StubOutWithMock(db, 'network_get_by_uuid') + self.net_man.context = context + db.network_get_by_uuid('elevated', 'quantum_net_id').AndReturn( + {'uuid': 1}) + + self.mox.ReplayAll() + + network = self.net_man.get_network(context, ('quantum_net_id', + 'net_tenant_id')) + self.assertEquals(network['quantum_net_id'], 'quantum_net_id') + self.assertEquals(network['uuid'], 1) + + def test_get_network_not_in_db(self): + context = self.mox.CreateMockAnything() + context.elevated().AndReturn('elevated') + self.mox.StubOutWithMock(db, 'network_get_by_uuid') + self.net_man.context = context + db.network_get_by_uuid('elevated', 'quantum_net_id').AndReturn(None) + + self.mox.ReplayAll() + + network = self.net_man.get_network(context, ('quantum_net_id', + 'net_tenant_id')) + self.assertEquals(network['quantum_net_id'], 'quantum_net_id') + self.assertEquals(network['uuid'], 'quantum_net_id') + + +class QuantumDeallocationTestCase(QuantumNovaTestCase): + def test_deallocate_port(self): + quantum = self.mox.CreateMock( + quantum_connection.QuantumClientConnection) + quantum.get_port_by_attachment('q_tenant_id', 'net_id', + 'interface_id').AndReturn('port_id') + quantum.detach_and_delete_port('q_tenant_id', 'net_id', 'port_id') + self.net_man.q_conn = quantum + + self.mox.ReplayAll() + + self.net_man.deallocate_port('interface_id', 'net_id', 'q_tenant_id', + 'instance_id') + + def test_deallocate_port_logs_error(self): + quantum = self.mox.CreateMock( + quantum_connection.QuantumClientConnection) + quantum.get_port_by_attachment('q_tenant_id', 'net_id', + 'interface_id').AndRaise(Exception) + self.net_man.q_conn = quantum + + self.mox.StubOutWithMock(quantum_manager.LOG, 'exception') + quantum_manager.LOG.exception(mox.Regex(r'port deallocation failed')) + + self.mox.ReplayAll() + + self.net_man.deallocate_port('interface_id', 'net_id', 'q_tenant_id', + 'instance_id') + + def test_deallocate_ip_address(self): + ipam = self.mox.CreateMock(melange_ipam_lib.QuantumMelangeIPAMLib) + ipam.get_tenant_id_by_net_id('context', 'net_id', {'uuid': 1}, + 'project_id').AndReturn('ipam_tenant_id') + self.net_man.ipam = ipam + self.mox.ReplayAll() + self.net_man.deallocate_ip_address('context', 'net_id', 'project_id', + {'uuid': 1}, 'instance_id') + + def test_deallocate_ip_address(self): + ipam = self.mox.CreateMock(melange_ipam_lib.QuantumMelangeIPAMLib) + ipam.get_tenant_id_by_net_id('context', 'net_id', {'uuid': 1}, + 'project_id').AndRaise(Exception()) + self.net_man.ipam = ipam + + self.mox.StubOutWithMock(quantum_manager.LOG, 'exception') + quantum_manager.LOG.exception(mox.Regex(r'ipam deallocation failed')) + + self.mox.ReplayAll() + self.net_man.deallocate_ip_address('context', 'net_id', 'project_id', + {'uuid': 1}, 'instance_id') + + +class QuantumManagerTestCase(QuantumNovaTestCase): + def test_create_and_delete_nets(self): + self._create_nets() + self._delete_nets() + + def _create_nets(self): + for n in networks: + self._create_network(n) + + def _delete_nets(self): + for n in networks: + ctx = context.RequestContext('user1', n['project_id']) + self.net_man.delete_network(ctx, None, n['uuid']) + self.assertRaises(exception.NoNetworksFound, + db.network_get_all, ctx.elevated()) + + def _validate_nw_info(self, nw_info, expected_net_labels): + + self.assertEquals(len(nw_info), len(expected_net_labels)) + + ctx = context.RequestContext('user1', 'foo').elevated() + all_net_map = {} + for n in db.network_get_all(ctx): + all_net_map[n['label']] = n + + for i in range(0, len(nw_info)): + vif = nw_info[i] + net = all_net_map[expected_net_labels[i]] + + # simple test assumes that each starting prefix is unique + expected_v4_cidr_start = net['cidr'].split(".")[0].lower() + expected_v6_cidr_start = net['cidr_v6'].split(":")[0].lower() + + for subnet in vif['network']['subnets']: + addr = subnet['ips'][0]['address'] + if subnet['version'] == 4: + address_start = addr.split(".")[0].lower() + self.assertTrue(expected_v4_cidr_start, address_start) + else: + address_start = addr.split(":")[0].lower() + self.assertTrue(expected_v6_cidr_start, address_start) + + # confirm that there is a DHCP device on corresponding net + for l in expected_net_labels: + n = all_net_map[l] + tenant_id = (n['project_id'] or + FLAGS.quantum_default_tenant_id) + ports = self.net_man.q_conn.get_attached_ports( + tenant_id, n['uuid']) + self.assertEquals(len(ports), 2) # gw + instance VIF + + # make sure we aren't allowed to delete network with + # active port + self.assertRaises(exception.NetworkBusy, + self.net_man.delete_network, + ctx, None, n['uuid']) + + def _check_vifs(self, expect_num_vifs): + ctx = context.RequestContext('user1', "").elevated() + self.assertEqual(len(db.virtual_interface_get_all(ctx)), + expect_num_vifs) + + def _allocate_and_deallocate_instance(self, project_id, requested_networks, + expected_labels): + + ctx = context.RequestContext('user1', project_id) + self._check_vifs(0) + + instance_ref = db.instance_create(ctx, + {"project_id": project_id}) + + nw_info = self.net_man.allocate_for_instance(ctx.elevated(), + instance_id=instance_ref['id'], host="", + rxtx_factor=3, + project_id=project_id, + requested_networks=requested_networks) + + self._check_vifs(len(nw_info)) + + self._validate_nw_info(nw_info, expected_labels) + + nw_info = self.net_man.get_instance_nw_info(ctx, instance_ref['id'], + instance_ref['uuid'], + instance_ref['instance_type_id'], "", + project_id=project_id) + + self._check_vifs(len(nw_info)) + self._validate_nw_info(nw_info, expected_labels) + + port_net_pairs = [] + for vif in nw_info: + nid = vif['network']['id'] + pid = self.net_man.q_conn.get_port_by_attachment( + project_id, nid, vif['id']) + if pid is None: + pid = self.net_man.q_conn.get_port_by_attachment( + FLAGS.quantum_default_tenant_id, + nid, vif['id']) + self.assertTrue(pid is not None) + port_net_pairs.append((pid, nid)) + + self.net_man.deallocate_for_instance(ctx, + instance_id=instance_ref['id'], + project_id=project_id) + + for pid, nid in port_net_pairs: + self.assertRaises(quantum_client.QuantumNotFoundException, + self.net_man.q_conn.detach_and_delete_port, + project_id, nid, pid) + self.assertRaises(quantum_client.QuantumNotFoundException, + self.net_man.q_conn.detach_and_delete_port, + FLAGS.quantum_default_tenant_id, nid, pid) + + self._check_vifs(0) + + def test_allocate_and_deallocate_instance_static(self): + self._create_nets() + self._allocate_and_deallocate_instance("fake_project1", None, + ['public', 'project1-net1']) + self._delete_nets() + + def test_allocate_and_deallocate_instance_dynamic(self): + + self._create_nets() + project_id = "fake_project2" + ctx = context.RequestContext('user1', project_id) + all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids( + ctx, project_id) + requested_networks = [(n[0], None) for n in all_valid_networks] + + self.net_man.validate_networks(ctx, requested_networks) + + label_map = {} + for n in db.network_get_all(ctx.elevated()): + label_map[n['uuid']] = n['label'] + expected_labels = [label_map[uid] for uid, _i in requested_networks] + + self._allocate_and_deallocate_instance(project_id, requested_networks, + expected_labels) + self._delete_nets() + + def test_validate_bad_network(self): + ctx = context.RequestContext('user1', 'fake_project1') + self.assertRaises(exception.NetworkNotFound, + self.net_man.validate_networks, ctx, [("", None)]) + + def test_create_net_external_uuid(self): + """Tests use case where network can be created directly via + Quantum API, then the UUID is passed in via nova-manage""" + project_id = "foo_project" + ctx = context.RequestContext('user1', project_id) + net_id = self.net_man.q_conn.create_network(project_id, 'net1') + self.net_man.create_networks( + ctx, + label='achtungbaby', + cidr="9.9.9.0/24", + multi_host=False, + num_networks=1, + network_size=256, + cidr_v6=None, + gateway="9.9.9.1", + gateway_v6=None, + bridge=None, + bridge_interface=None, + dns1="8.8.8.8", + project_id=project_id, + priority=9, + uuid=net_id) + net = db.network_get_by_uuid(ctx.elevated(), net_id) + self.assertTrue(net is not None) + self.assertEquals(net['uuid'], net_id) + + def test_create_net_external_uuid_and_host_is_set(self): + """Make sure network['host'] is set when creating a network via the + network manager""" + project_id = "foo_project" + ctx = context.RequestContext('user1', project_id) + net_id = self.net_man.q_conn.create_network(project_id, 'net2') + self.net_man.create_networks( + ctx, label='achtungbaby2', cidr="9.9.8.0/24", multi_host=False, + num_networks=1, network_size=256, cidr_v6=None, + gateway="9.9.8.1", gateway_v6=None, bridge=None, + bridge_interface=None, dns1="8.8.8.8", project_id=project_id, + priority=8, uuid=net_id) + net = db.network_get_by_uuid(ctx.elevated(), net_id) + self.assertTrue(net is not None) + self.assertEquals(net['uuid'], net_id) + self.assertTrue(net['host'] != None) + + +class QuantumNovaMACGenerationTestCase(QuantumNovaTestCase): + def test_local_mac_address_creation(self): + self.flags(use_melange_mac_generation=False) + fake_mac = "ab:cd:ef:ab:cd:ef" + self.stubs.Set(utils, "generate_mac_address", + lambda: fake_mac) + project_id = "fake_project1" + ctx = context.RequestContext('user1', project_id) + self._create_network(networks[0]) + + all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids( + ctx, project_id) + requested_networks = [(n[0], None) for n in all_valid_networks] + + instance_ref = db.api.instance_create(ctx, + {"project_id": project_id}) + nw_info = self.net_man.allocate_for_instance(ctx, + instance_id=instance_ref['id'], host="", + rxtx_factor=3, + project_id=project_id, + requested_networks=requested_networks) + self.assertEqual(nw_info[0]['address'], fake_mac) + + def test_melange_mac_address_creation(self): + self.flags(use_melange_mac_generation=True) + fake_mac = "ab:cd:ef:ab:cd:ef" + self.stubs.Set(melange_connection.MelangeConnection, "create_vif", + lambda w, x, y, z: fake_mac) + project_id = "fake_project1" + ctx = context.RequestContext('user1', project_id) + self._create_network(networks[0]) + + all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids( + ctx, project_id) + requested_networks = [(n[0], None) for n in all_valid_networks] + + instance_ref = db.api.instance_create(ctx, + {"project_id": project_id}) + nw_info = self.net_man.allocate_for_instance(ctx, + instance_id=instance_ref['id'], host="", + rxtx_factor=3, + project_id=project_id, + requested_networks=requested_networks) + self.assertEqual(nw_info[0]['address'], fake_mac) + + +class QuantumNovaPortSecurityTestCase(QuantumNovaTestCase): + def test_port_securty(self): + self.flags(use_melange_mac_generation=True) + self.flags(quantum_use_port_security=True) + fake_mac = "ab:cd:ef:ab:cd:ef" + self.stubs.Set(melange_connection.MelangeConnection, "create_vif", + lambda w, x, y, z: fake_mac) + project_id = "fake_project1" + ctx = context.RequestContext('user1', project_id) + self._create_network(networks[0]) + + all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids( + ctx, project_id) + requested_networks = [(n[0], None) for n in all_valid_networks] + + instance_ref = db.api.instance_create(ctx, + {"project_id": project_id}) + oldfunc = self.net_man.q_conn.create_and_attach_port + + # Make sure we get the appropriate mac set in allowed_address_pairs + # if port security is enabled. + def _instrumented_create_and_attach_port(tenant_id, net_id, + interface_id, **kwargs): + self.assertTrue('allowed_address_pairs' in kwargs.keys()) + pairs = kwargs['allowed_address_pairs'] + self.assertTrue(pairs[0]['mac_address'] == fake_mac) + self.net_man.q_conn.create_and_attach_port = oldfunc + return oldfunc(tenant_id, net_id, interface_id, **kwargs) + _port_attach = _instrumented_create_and_attach_port + self.net_man.q_conn.create_and_attach_port = _port_attach + nw_info = self.net_man.allocate_for_instance(ctx, + instance_id=instance_ref['id'], host="", + rxtx_factor=3, + project_id=project_id, + requested_networks=requested_networks) + self.assertEqual(nw_info[0]['address'], fake_mac) + + def test_port_securty_negative(self): + self.flags(use_melange_mac_generation=True) + self.flags(quantum_use_port_security=False) + fake_mac = "ab:cd:ef:ab:cd:ef" + self.stubs.Set(melange_connection.MelangeConnection, "create_vif", + lambda w, x, y, z: fake_mac) + project_id = "fake_project1" + ctx = context.RequestContext('user1', project_id) + self._create_network(networks[0]) + + all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids( + ctx, project_id) + requested_networks = [(n[0], None) for n in all_valid_networks] + + instance_ref = db.api.instance_create(ctx, + {"project_id": project_id}) + oldfunc = self.net_man.q_conn.create_and_attach_port + + # Make sure no pairs are passed in if port security is turned off + def _instrumented_create_and_attach_port(tenant_id, net_id, + interface_id, **kwargs): + self.assertTrue('allowed_address_pairs' in kwargs.keys()) + pairs = kwargs['allowed_address_pairs'] + self.assertTrue(len(pairs) == 0) + self.net_man.q_conn.create_and_attach_port = oldfunc + return oldfunc(tenant_id, net_id, interface_id, **kwargs) + _port_attach = _instrumented_create_and_attach_port + self.net_man.q_conn.create_and_attach_port = _port_attach + nw_info = self.net_man.allocate_for_instance(ctx, + instance_id=instance_ref['id'], host="", + rxtx_factor=3, + project_id=project_id, + requested_networks=requested_networks) + self.assertEqual(nw_info[0]['address'], fake_mac) + + +class QuantumMelangeTestCase(test.TestCase): + def setUp(self): + super(QuantumMelangeTestCase, self).setUp() + + fc = fake_client.FakeClient(LOG) + qc = quantum_connection.QuantumClientConnection(client=fc) + + self.net_man = quantum_manager.QuantumManager( + ipam_lib="nova.network.quantum.nova_ipam_lib", + q_conn=qc) + + def test_get_instance_uuids_by_ip_filter(self): + fake_context = context.RequestContext('user', 'project') + address = '1.2.3.4' + filters = {'ip': address} + + self.net_man.ipam = self.mox.CreateMockAnything() + self.net_man.ipam.get_instance_ids_by_ip_address(fake_context, + address).AndReturn(['instance_id']) + + instance = self.mox.CreateMockAnything() + instance.uuid = 'instance_uuid' + + self.mox.StubOutWithMock(db, 'instance_get') + db.instance_get(fake_context, 'instance_id').AndReturn(instance) + + self.mox.ReplayAll() + + uuids = self.net_man.get_instance_uuids_by_ip_filter(fake_context, + filters) + self.assertEquals(uuids, [{'instance_uuid':'instance_uuid'}]) diff --git a/nova/tests/test_linux_net.py b/nova/tests/test_linux_net.py deleted file mode 100644 index 1e7d3e0ff..000000000 --- a/nova/tests/test_linux_net.py +++ /dev/null @@ -1,501 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2011 NTT -# 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 os - -import mox - -from nova import context -from nova import db -from nova import flags -from nova import log as logging -from nova.openstack.common import importutils -from nova import test -from nova import utils -from nova.network import linux_net - - -FLAGS = flags.FLAGS - -LOG = logging.getLogger(__name__) - - -HOST = "testhost" - -instances = [{'id': 0, - 'host': 'fake_instance00', - 'created_at': 'fakedate', - 'updated_at': 'fakedate', - 'hostname': 'fake_instance00'}, - {'id': 1, - 'host': 'fake_instance01', - 'created_at': 'fakedate', - 'updated_at': 'fakedate', - 'hostname': 'fake_instance01'}] - - -addresses = [{"address": "10.0.0.1"}, - {"address": "10.0.0.2"}, - {"address": "10.0.0.3"}, - {"address": "10.0.0.4"}, - {"address": "10.0.0.5"}, - {"address": "10.0.0.6"}] - - -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', - 'dhcp_server': '0.0.0.0', - 'dhcp_start': '192.168.100.1', - 'vlan': None, - 'host': None, - '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': True, - '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', - 'dhcp_server': '0.0.0.0', - 'dhcp_start': '192.168.100.1', - 'vlan': None, - 'host': None, - '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': True, - 'virtual_interface_id': 0, - 'instance_id': 0, - 'floating_ips': []}, - {'id': 1, - 'network_id': 1, - 'address': '192.168.1.100', - 'instance_id': 0, - 'allocated': True, - 'virtual_interface_id': 1, - 'instance_id': 0, - 'floating_ips': []}, - {'id': 2, - 'network_id': 1, - 'address': '192.168.0.101', - 'instance_id': 1, - 'allocated': True, - 'virtual_interface_id': 2, - 'instance_id': 1, - 'floating_ips': []}, - {'id': 3, - 'network_id': 0, - 'address': '192.168.1.101', - 'instance_id': 1, - 'allocated': True, - 'virtual_interface_id': 3, - 'instance_id': 1, - 'floating_ips': []}, - {'id': 4, - 'network_id': 0, - 'address': '192.168.0.102', - 'instance_id': 0, - 'allocated': True, - 'virtual_interface_id': 4, - 'instance_id': 0, - 'floating_ips': []}, - {'id': 5, - 'network_id': 1, - 'address': '192.168.1.102', - 'instance_id': 1, - 'allocated': True, - 'virtual_interface_id': 5, - 'instance_id': 1, - 'floating_ips': []}] - - -vifs = [{'id': 0, - 'address': 'DE:AD:BE:EF:00:00', - 'uuid': '00000000-0000-0000-0000-0000000000000000', - 'network_id': 0, - 'instance_id': 0}, - {'id': 1, - 'address': 'DE:AD:BE:EF:00:01', - 'uuid': '00000000-0000-0000-0000-0000000000000001', - 'network_id': 1, - 'instance_id': 0}, - {'id': 2, - 'address': 'DE:AD:BE:EF:00:02', - 'uuid': '00000000-0000-0000-0000-0000000000000002', - 'network_id': 1, - 'instance_id': 1}, - {'id': 3, - 'address': 'DE:AD:BE:EF:00:03', - 'uuid': '00000000-0000-0000-0000-0000000000000003', - 'network_id': 0, - 'instance_id': 1}, - {'id': 4, - 'address': 'DE:AD:BE:EF:00:04', - 'uuid': '00000000-0000-0000-0000-0000000000000004', - 'network_id': 0, - 'instance_id': 0}, - {'id': 5, - 'address': 'DE:AD:BE:EF:00:05', - 'uuid': '00000000-0000-0000-0000-0000000000000005', - 'network_id': 1, - 'instance_id': 1}] - - -def get_associated(context, network_id, host=None): - result = [] - for datum in fixed_ips: - if (datum['network_id'] == network_id and datum['allocated'] - and datum['instance_id'] is not None - and datum['virtual_interface_id'] is not None): - instance = instances[datum['instance_id']] - if host and host != instance['host']: - continue - cleaned = {} - cleaned['address'] = datum['address'] - cleaned['instance_id'] = datum['instance_id'] - cleaned['network_id'] = datum['network_id'] - cleaned['vif_id'] = datum['virtual_interface_id'] - vif = vifs[datum['virtual_interface_id']] - cleaned['vif_address'] = vif['address'] - cleaned['instance_hostname'] = instance['hostname'] - cleaned['instance_updated'] = instance['updated_at'] - cleaned['instance_created'] = instance['created_at'] - result.append(cleaned) - return result - - -class LinuxNetworkTestCase(test.TestCase): - - def setUp(self): - super(LinuxNetworkTestCase, self).setUp() - network_driver = FLAGS.network_driver - self.driver = importutils.import_module(network_driver) - self.driver.db = db - self.context = context.RequestContext('testuser', 'testproject', - is_admin=True) - - def get_vifs(_context, instance_id): - return [vif for vif in vifs if vif['instance_id'] == instance_id] - - def get_instance(_context, instance_id): - return instances[instance_id] - - self.stubs.Set(db, 'virtual_interface_get_by_instance', get_vifs) - self.stubs.Set(db, 'instance_get', get_instance) - self.stubs.Set(db, 'network_get_associated_fixed_ips', get_associated) - - def test_update_dhcp_for_nw00(self): - self.flags(use_single_default_gateway=True) - - self.mox.StubOutWithMock(self.driver, 'write_to_file') - self.mox.StubOutWithMock(self.driver, 'ensure_path') - self.mox.StubOutWithMock(os, 'chmod') - - self.driver.write_to_file(mox.IgnoreArg(), mox.IgnoreArg()) - self.driver.write_to_file(mox.IgnoreArg(), mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - os.chmod(mox.IgnoreArg(), mox.IgnoreArg()) - os.chmod(mox.IgnoreArg(), mox.IgnoreArg()) - - self.mox.ReplayAll() - - self.driver.update_dhcp(self.context, "eth0", networks[0]) - - def test_update_dhcp_for_nw01(self): - self.flags(use_single_default_gateway=True) - - self.mox.StubOutWithMock(self.driver, 'write_to_file') - self.mox.StubOutWithMock(self.driver, 'ensure_path') - self.mox.StubOutWithMock(os, 'chmod') - - self.driver.write_to_file(mox.IgnoreArg(), mox.IgnoreArg()) - self.driver.write_to_file(mox.IgnoreArg(), mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - self.driver.ensure_path(mox.IgnoreArg()) - os.chmod(mox.IgnoreArg(), mox.IgnoreArg()) - os.chmod(mox.IgnoreArg(), mox.IgnoreArg()) - - self.mox.ReplayAll() - - self.driver.update_dhcp(self.context, "eth0", networks[0]) - - def test_get_dhcp_hosts_for_nw00(self): - self.flags(use_single_default_gateway=True) - - expected = ( - "DE:AD:BE:EF:00:00,fake_instance00.novalocal," - "192.168.0.100,net:NW-0\n" - "DE:AD:BE:EF:00:03,fake_instance01.novalocal," - "192.168.1.101,net:NW-3\n" - "DE:AD:BE:EF:00:04,fake_instance00.novalocal," - "192.168.0.102,net:NW-4" - ) - actual_hosts = self.driver.get_dhcp_hosts(self.context, networks[0]) - - self.assertEquals(actual_hosts, expected) - - def test_get_dhcp_hosts_for_nw01(self): - self.flags(use_single_default_gateway=True) - self.flags(host='fake_instance01') - - expected = ( - "DE:AD:BE:EF:00:02,fake_instance01.novalocal," - "192.168.0.101,net:NW-2\n" - "DE:AD:BE:EF:00:05,fake_instance01.novalocal," - "192.168.1.102,net:NW-5" - ) - actual_hosts = self.driver.get_dhcp_hosts(self.context, networks[1]) - - self.assertEquals(actual_hosts, expected) - - def test_get_dhcp_opts_for_nw00(self): - expected_opts = 'NW-0,3\nNW-3,3\nNW-4,3' - actual_opts = self.driver.get_dhcp_opts(self.context, networks[0]) - - self.assertEquals(actual_opts, expected_opts) - - def test_get_dhcp_opts_for_nw01(self): - self.flags(host='fake_instance01') - expected_opts = "NW-5,3" - actual_opts = self.driver.get_dhcp_opts(self.context, networks[1]) - - self.assertEquals(actual_opts, expected_opts) - - def test_dhcp_opts_not_default_gateway_network(self): - expected = "NW-0,3" - data = get_associated(self.context, 0)[0] - actual = self.driver._host_dhcp_opts(data) - self.assertEquals(actual, expected) - - def test_host_dhcp_without_default_gateway_network(self): - expected = ','.join(['DE:AD:BE:EF:00:00', - 'fake_instance00.novalocal', - '192.168.0.100']) - data = get_associated(self.context, 0)[0] - actual = self.driver._host_dhcp(data) - self.assertEquals(actual, expected) - - def test_linux_bridge_driver_plug(self): - """Makes sure plug doesn't drop FORWARD by default. - - Ensures bug 890195 doesn't reappear.""" - - def fake_execute(*args, **kwargs): - return "", "" - self.stubs.Set(utils, 'execute', fake_execute) - - def verify_add_rule(chain, rule): - self.assertEqual(chain, 'FORWARD') - self.assertIn('ACCEPT', rule) - self.stubs.Set(linux_net.iptables_manager.ipv4['filter'], - 'add_rule', verify_add_rule) - driver = linux_net.LinuxBridgeInterfaceDriver() - driver.plug({"bridge": "br100", "bridge_interface": "eth0"}, - "fakemac") - - def test_vlan_override(self): - """Makes sure vlan_interface flag overrides network bridge_interface. - - Allows heterogeneous networks a la bug 833426""" - - driver = linux_net.LinuxBridgeInterfaceDriver() - - info = {} - - @classmethod - def test_ensure(_self, vlan, bridge, interface, network, mac_address): - info['passed_interface'] = interface - - self.stubs.Set(linux_net.LinuxBridgeInterfaceDriver, - 'ensure_vlan_bridge', test_ensure) - - network = { - "bridge": "br100", - "bridge_interface": "base_interface", - "vlan": "fake" - } - driver.plug(network, "fakemac") - self.assertEqual(info['passed_interface'], "base_interface") - self.flags(vlan_interface="override_interface") - driver.plug(network, "fakemac") - self.assertEqual(info['passed_interface'], "override_interface") - driver.plug(network, "fakemac") - - def test_flat_override(self): - """Makes sure flat_interface flag overrides network bridge_interface. - - Allows heterogeneous networks a la bug 833426""" - - driver = linux_net.LinuxBridgeInterfaceDriver() - - info = {} - - @classmethod - def test_ensure(_self, bridge, interface, network, gateway): - info['passed_interface'] = interface - - self.stubs.Set(linux_net.LinuxBridgeInterfaceDriver, - 'ensure_bridge', test_ensure) - - network = { - "bridge": "br100", - "bridge_interface": "base_interface", - } - driver.plug(network, "fakemac") - self.assertEqual(info['passed_interface'], "base_interface") - self.flags(flat_interface="override_interface") - driver.plug(network, "fakemac") - self.assertEqual(info['passed_interface'], "override_interface") - - def _test_initialize_gateway(self, existing, expected, routes=''): - self.flags(fake_network=False) - executes = [] - - def fake_execute(*args, **kwargs): - executes.append(args) - if args[0] == 'ip' and args[1] == 'addr' and args[2] == 'show': - return existing, "" - if args[0] == 'route' and args[1] == '-n': - return routes, "" - self.stubs.Set(utils, 'execute', fake_execute) - network = {'dhcp_server': '192.168.1.1', - 'cidr': '192.168.1.0/24', - 'broadcast': '192.168.1.255', - 'cidr_v6': '2001:db8::/64'} - self.driver.initialize_gateway_device('eth0', network) - self.assertEqual(executes, expected) - - def test_initialize_gateway_moves_wrong_ip(self): - existing = ("2: eth0: " - " mtu 1500 qdisc pfifo_fast state UNKNOWN qlen 1000\n" - " link/ether de:ad:be:ef:be:ef brd ff:ff:ff:ff:ff:ff\n" - " inet 192.168.0.1/24 brd 192.168.0.255 scope global eth0\n" - " inet6 dead::beef:dead:beef:dead/64 scope link\n" - " valid_lft forever preferred_lft forever\n") - expected = [ - ('sysctl', '-w', 'net.ipv4.ip_forward=1'), - ('ip', 'addr', 'show', 'dev', 'eth0', 'scope', 'global'), - ('route', '-n'), - ('ip', 'addr', 'del', '192.168.0.1/24', - 'brd', '192.168.0.255', 'scope', 'global', 'dev', 'eth0'), - ('ip', 'addr', 'add', '192.168.1.1/24', - 'brd', '192.168.1.255', 'dev', 'eth0'), - ('ip', 'addr', 'add', '192.168.0.1/24', - 'brd', '192.168.0.255', 'scope', 'global', 'dev', 'eth0'), - ('ip', '-f', 'inet6', 'addr', 'change', - '2001:db8::/64', 'dev', 'eth0'), - ] - self._test_initialize_gateway(existing, expected) - - def test_initialize_gateway_resets_route(self): - routes = ("0.0.0.0 192.68.0.1 0.0.0.0 " - "UG 100 0 0 eth0") - existing = ("2: eth0: " - " mtu 1500 qdisc pfifo_fast state UNKNOWN qlen 1000\n" - " link/ether de:ad:be:ef:be:ef brd ff:ff:ff:ff:ff:ff\n" - " inet 192.168.0.1/24 brd 192.168.0.255 scope global eth0\n" - " inet6 dead::beef:dead:beef:dead/64 scope link\n" - " valid_lft forever preferred_lft forever\n") - expected = [ - ('sysctl', '-w', 'net.ipv4.ip_forward=1'), - ('ip', 'addr', 'show', 'dev', 'eth0', 'scope', 'global'), - ('route', '-n'), - ('route', 'del', 'default', 'gw', '192.68.0.1', 'dev', 'eth0'), - ('ip', 'addr', 'del', '192.168.0.1/24', - 'brd', '192.168.0.255', 'scope', 'global', 'dev', 'eth0'), - ('ip', 'addr', 'add', '192.168.1.1/24', - 'brd', '192.168.1.255', 'dev', 'eth0'), - ('ip', 'addr', 'add', '192.168.0.1/24', - 'brd', '192.168.0.255', 'scope', 'global', 'dev', 'eth0'), - ('route', 'add', 'default', 'gw', '192.68.0.1'), - ('ip', '-f', 'inet6', 'addr', 'change', - '2001:db8::/64', 'dev', 'eth0'), - ] - self._test_initialize_gateway(existing, expected, routes) - - def test_initialize_gateway_no_move_right_ip(self): - existing = ("2: eth0: " - " mtu 1500 qdisc pfifo_fast state UNKNOWN qlen 1000\n" - " link/ether de:ad:be:ef:be:ef brd ff:ff:ff:ff:ff:ff\n" - " inet 192.168.1.1/24 brd 192.168.1.255 scope global eth0\n" - " inet 192.168.0.1/24 brd 192.168.0.255 scope global eth0\n" - " inet6 dead::beef:dead:beef:dead/64 scope link\n" - " valid_lft forever preferred_lft forever\n") - expected = [ - ('sysctl', '-w', 'net.ipv4.ip_forward=1'), - ('ip', 'addr', 'show', 'dev', 'eth0', 'scope', 'global'), - ('ip', '-f', 'inet6', 'addr', 'change', - '2001:db8::/64', 'dev', 'eth0'), - ] - self._test_initialize_gateway(existing, expected) - - def test_initialize_gateway_add_if_blank(self): - existing = ("2: eth0: " - " mtu 1500 qdisc pfifo_fast state UNKNOWN qlen 1000\n" - " link/ether de:ad:be:ef:be:ef brd ff:ff:ff:ff:ff:ff\n" - " inet6 dead::beef:dead:beef:dead/64 scope link\n" - " valid_lft forever preferred_lft forever\n") - expected = [ - ('sysctl', '-w', 'net.ipv4.ip_forward=1'), - ('ip', 'addr', 'show', 'dev', 'eth0', 'scope', 'global'), - ('route', '-n'), - ('ip', 'addr', 'add', '192.168.1.1/24', - 'brd', '192.168.1.255', 'dev', 'eth0'), - ('ip', '-f', 'inet6', 'addr', 'change', - '2001:db8::/64', 'dev', 'eth0'), - ] - self._test_initialize_gateway(existing, expected) diff --git a/nova/tests/test_network_info.py b/nova/tests/test_network_info.py deleted file mode 100644 index f1c908b2c..000000000 --- a/nova/tests/test_network_info.py +++ /dev/null @@ -1,326 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2011 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 exception -from nova import log as logging -from nova.network import model -from nova import test -from nova.tests import fake_network_cache_model - -LOG = logging.getLogger(__name__) - - -class RouteTests(test.TestCase): - def test_create_route_with_attrs(self): - route = fake_network_cache_model.new_route() - ip = fake_network_cache_model.new_ip(dict(address='192.168.1.1')) - self.assertEqual(route['cidr'], '0.0.0.0/24') - self.assertEqual(route['gateway']['address'], '192.168.1.1') - self.assertEqual(route['interface'], 'eth0') - - def test_routes_equal(self): - route1 = fake_network_cache_model.new_route() - route2 = fake_network_cache_model.new_route() - self.assertEqual(route1, route2) - - def test_routes_not_equal(self): - route1 = fake_network_cache_model.new_route() - route2 = fake_network_cache_model.new_route(dict(cidr='1.1.1.1/24')) - self.assertNotEqual(route1, route2) - - def test_hydrate(self): - route = model.Route.hydrate( - {'gateway': fake_network_cache_model.new_ip( - dict(address='192.168.1.1'))}) - self.assertEqual(route['cidr'], None) - self.assertEqual(route['gateway']['address'], '192.168.1.1') - self.assertEqual(route['interface'], None) - - -class FixedIPTests(test.TestCase): - def test_createnew_fixed_ip_with_attrs(self): - fixed_ip = model.FixedIP(address='192.168.1.100') - self.assertEqual(fixed_ip['address'], '192.168.1.100') - self.assertEqual(fixed_ip['floating_ips'], []) - self.assertEqual(fixed_ip['type'], 'fixed') - self.assertEqual(fixed_ip['version'], 4) - - def test_create_fixed_ipv6(self): - fixed_ip = model.FixedIP(address='::1') - self.assertEqual(fixed_ip['address'], '::1') - self.assertEqual(fixed_ip['floating_ips'], []) - self.assertEqual(fixed_ip['type'], 'fixed') - self.assertEqual(fixed_ip['version'], 6) - - def test_create_fixed_bad_ip_fails(self): - self.assertRaises(exception.InvalidIpAddressError, - model.FixedIP, - address='picklespicklespickles') - - def test_equate_two_fixed_ips(self): - fixed_ip = model.FixedIP(address='::1') - fixed_ip2 = model.FixedIP(address='::1') - self.assertEqual(fixed_ip, fixed_ip2) - - def test_equate_two_dissimilar_fixed_ips_fails(self): - fixed_ip = model.FixedIP(address='::1') - fixed_ip2 = model.FixedIP(address='::2') - self.assertNotEqual(fixed_ip, fixed_ip2) - - def test_hydrate(self): - fixed_ip = model.FixedIP.hydrate({}) - self.assertEqual(fixed_ip['floating_ips'], []) - self.assertEqual(fixed_ip['address'], None) - self.assertEqual(fixed_ip['type'], 'fixed') - self.assertEqual(fixed_ip['version'], None) - - def test_add_floating_ip(self): - fixed_ip = model.FixedIP(address='192.168.1.100') - fixed_ip.add_floating_ip('192.168.1.101') - self.assertEqual(fixed_ip['floating_ips'], ['192.168.1.101']) - - def test_add_floating_ip_repeatedly_only_one_instance(self): - fixed_ip = model.FixedIP(address='192.168.1.100') - for i in xrange(10): - fixed_ip.add_floating_ip('192.168.1.101') - self.assertEqual(fixed_ip['floating_ips'], ['192.168.1.101']) - - -class SubnetTests(test.TestCase): - def test_create_subnet_with_attrs(self): - subnet = fake_network_cache_model.new_subnet() - - route1 = fake_network_cache_model.new_route() - - self.assertEqual(subnet['cidr'], '10.10.0.0/24') - self.assertEqual(subnet['dns'], - [fake_network_cache_model.new_ip(dict(address='1.2.3.4')), - fake_network_cache_model.new_ip(dict(address='2.3.4.5'))]) - self.assertEqual(subnet['gateway']['address'], '10.10.0.1') - self.assertEqual(subnet['ips'], - [fake_network_cache_model.new_ip( - dict(address='10.10.0.2')), - fake_network_cache_model.new_ip( - dict(address='10.10.0.3'))]) - self.assertEqual(subnet['routes'], [route1]) - self.assertEqual(subnet['version'], 4) - - def test_add_route(self): - subnet = fake_network_cache_model.new_subnet() - route1 = fake_network_cache_model.new_route() - route2 = fake_network_cache_model.new_route({'cidr': '1.1.1.1/24'}) - subnet.add_route(route2) - self.assertEqual(subnet['routes'], [route1, route2]) - - def test_add_route_a_lot(self): - subnet = fake_network_cache_model.new_subnet() - route1 = fake_network_cache_model.new_route() - route2 = fake_network_cache_model.new_route({'cidr': '1.1.1.1/24'}) - for i in xrange(10): - subnet.add_route(route2) - self.assertEqual(subnet['routes'], [route1, route2]) - - def test_add_dns(self): - subnet = fake_network_cache_model.new_subnet() - dns = fake_network_cache_model.new_ip(dict(address='9.9.9.9')) - subnet.add_dns(dns) - self.assertEqual(subnet['dns'], - [fake_network_cache_model.new_ip(dict(address='1.2.3.4')), - fake_network_cache_model.new_ip(dict(address='2.3.4.5')), - fake_network_cache_model.new_ip(dict(address='9.9.9.9'))]) - - def test_add_dns_a_lot(self): - subnet = fake_network_cache_model.new_subnet() - for i in xrange(10): - subnet.add_dns(fake_network_cache_model.new_ip( - dict(address='9.9.9.9'))) - self.assertEqual(subnet['dns'], - [fake_network_cache_model.new_ip(dict(address='1.2.3.4')), - fake_network_cache_model.new_ip(dict(address='2.3.4.5')), - fake_network_cache_model.new_ip(dict(address='9.9.9.9'))]) - - def test_add_ip(self): - subnet = fake_network_cache_model.new_subnet() - subnet.add_ip(fake_network_cache_model.new_ip( - dict(address='192.168.1.102'))) - self.assertEqual(subnet['ips'], - [fake_network_cache_model.new_ip( - dict(address='10.10.0.2')), - fake_network_cache_model.new_ip( - dict(address='10.10.0.3')), - fake_network_cache_model.new_ip( - dict(address='192.168.1.102'))]) - - def test_add_ip_a_lot(self): - subnet = fake_network_cache_model.new_subnet() - for i in xrange(10): - subnet.add_ip(fake_network_cache_model.new_ip( - dict(address='192.168.1.102'))) - self.assertEqual(subnet['ips'], - [fake_network_cache_model.new_ip( - dict(address='10.10.0.2')), - fake_network_cache_model.new_ip( - dict(address='10.10.0.3')), - fake_network_cache_model.new_ip( - dict(address='192.168.1.102'))]) - - def test_hydrate(self): - subnet_dict = { - 'cidr': '255.255.255.0', - 'dns': [fake_network_cache_model.new_ip(dict(address='1.1.1.1'))], - 'ips': [fake_network_cache_model.new_ip(dict(address='2.2.2.2'))], - 'routes': [fake_network_cache_model.new_route()], - 'version': 4, - 'gateway': fake_network_cache_model.new_ip( - dict(address='3.3.3.3'))} - subnet = model.Subnet.hydrate(subnet_dict) - - self.assertEqual(subnet['cidr'], '255.255.255.0') - self.assertEqual(subnet['dns'], [fake_network_cache_model.new_ip( - dict(address='1.1.1.1'))]) - self.assertEqual(subnet['gateway']['address'], '3.3.3.3') - self.assertEqual(subnet['ips'], [fake_network_cache_model.new_ip( - dict(address='2.2.2.2'))]) - self.assertEqual(subnet['routes'], [ - fake_network_cache_model.new_route()]) - self.assertEqual(subnet['version'], 4) - - -class NetworkTests(test.TestCase): - def test_create_network(self): - network = fake_network_cache_model.new_network() - self.assertEqual(network['id'], 1) - self.assertEqual(network['bridge'], 'br0') - self.assertEqual(network['label'], 'public') - self.assertEqual(network['subnets'], - [fake_network_cache_model.new_subnet(), - fake_network_cache_model.new_subnet( - dict(cidr='255.255.255.255'))]) - - def test_add_subnet(self): - network = fake_network_cache_model.new_network() - network.add_subnet(fake_network_cache_model.new_subnet( - dict(cidr='0.0.0.0'))) - self.assertEqual(network['subnets'], - [fake_network_cache_model.new_subnet(), - fake_network_cache_model.new_subnet( - dict(cidr='255.255.255.255')), - fake_network_cache_model.new_subnet(dict(cidr='0.0.0.0'))]) - - def test_add_subnet_a_lot(self): - network = fake_network_cache_model.new_network() - for i in xrange(10): - network.add_subnet(fake_network_cache_model.new_subnet( - dict(cidr='0.0.0.0'))) - self.assertEqual(network['subnets'], - [fake_network_cache_model.new_subnet(), - fake_network_cache_model.new_subnet( - dict(cidr='255.255.255.255')), - fake_network_cache_model.new_subnet(dict(cidr='0.0.0.0'))]) - - def test_hydrate(self): - new_network = dict( - id=1, - bridge='br0', - label='public', - subnets=[fake_network_cache_model.new_subnet(), - fake_network_cache_model.new_subnet( - dict(cidr='255.255.255.255'))]) - network = model.Network.hydrate(fake_network_cache_model.new_network()) - - self.assertEqual(network['id'], 1) - self.assertEqual(network['bridge'], 'br0') - self.assertEqual(network['label'], 'public') - self.assertEqual(network['subnets'], - [fake_network_cache_model.new_subnet(), - fake_network_cache_model.new_subnet( - dict(cidr='255.255.255.255'))]) - - -class VIFTests(test.TestCase): - def test_create_vif(self): - vif = fake_network_cache_model.new_vif() - self.assertEqual(vif['id'], 1) - self.assertEqual(vif['address'], 'aa:aa:aa:aa:aa:aa') - self.assertEqual(vif['network'], - fake_network_cache_model.new_network()) - - def test_vif_get_fixed_ips(self): - vif = fake_network_cache_model.new_vif() - fixed_ips = vif.fixed_ips() - ips = [fake_network_cache_model.new_ip(dict(address='10.10.0.2')), - fake_network_cache_model.new_ip( - dict(address='10.10.0.3'))] * 2 - self.assertEqual(fixed_ips, ips) - - def test_vif_get_floating_ips(self): - vif = fake_network_cache_model.new_vif() - vif['network']['subnets'][0]['ips'][0].add_floating_ip('192.168.1.1') - floating_ips = vif.floating_ips() - self.assertEqual(floating_ips, ['192.168.1.1']) - - def test_vif_get_labeled_ips(self): - vif = fake_network_cache_model.new_vif() - labeled_ips = vif.labeled_ips() - ip_dict = { - 'network_id': 1, - 'ips': [fake_network_cache_model.new_ip( - {'address': '10.10.0.2'}), - fake_network_cache_model.new_ip( - {'address': '10.10.0.3'})] * 2, - 'network_label': 'public'} - self.assertEqual(labeled_ips, ip_dict) - - def test_hydrate(self): - new_vif = dict( - id=1, - address='127.0.0.1', - network=fake_network_cache_model.new_network()) - vif = model.VIF.hydrate(fake_network_cache_model.new_vif()) - self.assertEqual(vif['id'], 1) - self.assertEqual(vif['address'], 'aa:aa:aa:aa:aa:aa') - self.assertEqual(vif['network'], - fake_network_cache_model.new_network()) - - -class NetworkInfoTests(test.TestCase): - def test_create_model(self): - ninfo = model.NetworkInfo([fake_network_cache_model.new_vif(), - fake_network_cache_model.new_vif( - {'address':'bb:bb:bb:bb:bb:bb'})]) - self.assertEqual(ninfo.fixed_ips(), - [fake_network_cache_model.new_ip({'address': '10.10.0.2'}), - fake_network_cache_model.new_ip( - {'address': '10.10.0.3'})] * 4) - - def test_get_floating_ips(self): - vif = fake_network_cache_model.new_vif() - vif['network']['subnets'][0]['ips'][0].add_floating_ip('192.168.1.1') - ninfo = model.NetworkInfo([vif, - fake_network_cache_model.new_vif( - {'address':'bb:bb:bb:bb:bb:bb'})]) - self.assertEqual(ninfo.floating_ips(), ['192.168.1.1']) - - def test_hydrate(self): - ninfo = model.NetworkInfo([fake_network_cache_model.new_vif(), - fake_network_cache_model.new_vif( - {'address':'bb:bb:bb:bb:bb:bb'})]) - deserialized = model.NetworkInfo.hydrate(ninfo) - self.assertEqual(ninfo.fixed_ips(), - [fake_network_cache_model.new_ip({'address': '10.10.0.2'}), - fake_network_cache_model.new_ip( - {'address': '10.10.0.3'})] * 4) diff --git a/nova/tests/test_quantum.py b/nova/tests/test_quantum.py deleted file mode 100644 index 9f03c13f8..000000000 --- a/nova/tests/test_quantum.py +++ /dev/null @@ -1,647 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2011,2012 Nicira, Inc. -# 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.db.sqlalchemy import models -from nova.db.sqlalchemy import session as sql_session -from nova import exception -from nova import flags -from nova import log as logging -from nova.network.quantum import client as quantum_client -from nova.network.quantum import fake_client -from nova.network.quantum import manager as quantum_manager -from nova.network.quantum import quantum_connection -from nova.network.quantum import melange_connection -from nova.network.quantum import melange_ipam_lib - -from nova import test -from nova import utils - -LOG = logging.getLogger(__name__) -FLAGS = flags.FLAGS - - -networks = [{'label': 'project1-net1', - 'injected': False, - 'multi_host': False, - 'cidr': '100.168.0.0/24', - 'cidr_v6': '100:1db8::/64', - 'gateway_v6': '100:1db8::1', - 'netmask_v6': '64', - 'netmask': '255.255.255.0', - 'bridge': None, - 'bridge_interface': None, - 'gateway': '100.168.0.1', - 'broadcast': '100.168.0.255', - 'dns1': '8.8.8.8', - 'vlan': None, - 'host': None, - 'vpn_public_address': None, - 'project_id': 'fake_project1', - 'priority': 1}, - {'label': 'project2-net1', - 'injected': False, - 'multi_host': False, - 'cidr': '101.168.1.0/24', - 'cidr_v6': '101:1db9::/64', - 'gateway_v6': '101:1db9::1', - 'netmask_v6': '64', - 'netmask': '255.255.255.0', - 'bridge': None, - 'bridge_interface': None, - 'gateway': '101.168.1.1', - 'broadcast': '101.168.1.255', - 'dns1': '8.8.8.8', - 'vlan': None, - 'host': None, - 'project_id': 'fake_project2', - 'priority': 1}, - {'label': "public", - 'injected': False, - 'multi_host': False, - 'cidr': '102.0.0.0/24', - 'cidr_v6': '102:1dba::/64', - 'gateway_v6': '102:1dba::1', - 'netmask_v6': '64', - 'netmask': '255.255.255.0', - 'bridge': None, - 'bridge_interface': None, - 'gateway': '102.0.0.1', - 'broadcast': '102.0.0.255', - 'dns1': '8.8.8.8', - 'vlan': None, - 'host': None, - 'project_id': None, - 'priority': 0}, - {'label': "project2-net2", - 'injected': False, - 'multi_host': False, - 'cidr': '103.0.0.0/24', - 'cidr_v6': '103:1dbb::/64', - 'gateway_v6': '103:1dbb::1', - 'netmask_v6': '64', - 'netmask': '255.255.255.0', - 'bridge': None, - 'bridge_interface': None, - 'gateway': '103.0.0.1', - 'broadcast': '103.0.0.255', - 'dns1': '8.8.8.8', - 'vlan': None, - 'host': None, - 'project_id': "fake_project2", - 'priority': 2}] - - -class QuantumConnectionTestCase(test.TestCase): - - def test_connection(self): - fc = fake_client.FakeClient(LOG) - qc = quantum_connection.QuantumClientConnection(client=fc) - t = "tenant1" - net1_name = "net1" - net1_uuid = qc.create_network(t, net1_name) - self.assertEquals(net1_name, qc.get_network_name(t, net1_uuid)) - self.assertTrue(qc.network_exists(t, net1_uuid)) - self.assertFalse(qc.network_exists(t, "fake-uuid")) - self.assertFalse(qc.network_exists("fake-tenant", net1_uuid)) - - nets = qc.get_networks(t)['networks'] - self.assertEquals(len(nets), 1) - self.assertEquals(nets[0]['id'], net1_uuid) - - num_ports = 10 - for i in range(0, num_ports): - qc.create_and_attach_port(t, net1_uuid, - 'iface' + str(i), state='ACTIVE') - - self.assertEquals(len(qc.get_attached_ports(t, net1_uuid)), num_ports) - - for i in range(0, num_ports): - port_uuid = qc.get_port_by_attachment(t, net1_uuid, - 'iface' + str(i)) - self.assertTrue(port_uuid) - qc.detach_and_delete_port(t, net1_uuid, port_uuid) - - self.assertEquals(len(qc.get_attached_ports(t, net1_uuid)), 0) - - # test port not found - qc.create_and_attach_port(t, net1_uuid, 'foo', state='ACTIVE') - port_uuid = qc.get_port_by_attachment(t, net1_uuid, 'foo') - qc.detach_and_delete_port(t, net1_uuid, port_uuid) - self.assertRaises(quantum_client.QuantumNotFoundException, - qc.detach_and_delete_port, t, - net1_uuid, port_uuid) - - qc.delete_network(t, net1_uuid) - self.assertFalse(qc.network_exists(t, net1_uuid)) - self.assertEquals(len(qc.get_networks(t)['networks']), 0) - - self.assertRaises(quantum_client.QuantumNotFoundException, - qc.get_network_name, t, net1_uuid) - - -# this is a base class to be used by other QuantumManager Test classes -class QuantumNovaTestCase(test.TestCase): - - def setUp(self): - super(QuantumNovaTestCase, self).setUp() - - self.flags(quantum_use_dhcp=True) - self.flags(l3_lib="nova.network.l3.LinuxNetL3") - linuxdrv = "nova.network.linux_net.LinuxOVSInterfaceDriver" - self.flags(linuxnet_interface_driver=linuxdrv) - fc = fake_client.FakeClient(LOG) - qc = quantum_connection.QuantumClientConnection(client=fc) - - self.net_man = quantum_manager.QuantumManager( - ipam_lib="nova.network.quantum.nova_ipam_lib", - q_conn=qc) - - def func(arg1, arg2): - pass - - def func2(arg1, arg2, arg3): - pass - - def func1(arg1): - pass - - self.net_man.driver.update_dhcp_hostfile_with_text = func - self.net_man.driver.restart_dhcp = func2 - self.net_man.driver.kill_dhcp = func1 - - # Tests seem to create some networks by default, which - # we don't want. So we delete them. - - ctx = context.RequestContext('user1', 'fake_project1').elevated() - for n in db.network_get_all(ctx): - db.network_delete_safe(ctx, n['id']) - - # Other unit tests (e.g., test_compute.py) have a nasty - # habit of of creating fixed IPs and not cleaning up, which - # can confuse these tests, so we remove all existing fixed - # ips before starting. - session = sql_session.get_session() - result = session.query(models.FixedIp).all() - with session.begin(): - for fip_ref in result: - session.delete(fip_ref) - - self.net_man.init_host() - - def _create_network(self, n): - ctx = context.RequestContext('user1', n['project_id']) - nwks = self.net_man.create_networks( - ctx, - label=n['label'], cidr=n['cidr'], - multi_host=n['multi_host'], - num_networks=1, network_size=256, - cidr_v6=n['cidr_v6'], - gateway=n['gateway'], - gateway_v6=n['gateway_v6'], bridge=None, - bridge_interface=None, dns1=n['dns1'], - project_id=n['project_id'], - priority=n['priority']) - n['uuid'] = nwks[0]['uuid'] - - -class QuantumAllocationTestCase(QuantumNovaTestCase): - def test_get_network_in_db(self): - context = self.mox.CreateMockAnything() - context.elevated().AndReturn('elevated') - self.mox.StubOutWithMock(db, 'network_get_by_uuid') - self.net_man.context = context - db.network_get_by_uuid('elevated', 'quantum_net_id').AndReturn( - {'uuid': 1}) - - self.mox.ReplayAll() - - network = self.net_man.get_network(context, ('quantum_net_id', - 'net_tenant_id')) - self.assertEquals(network['quantum_net_id'], 'quantum_net_id') - self.assertEquals(network['uuid'], 1) - - def test_get_network_not_in_db(self): - context = self.mox.CreateMockAnything() - context.elevated().AndReturn('elevated') - self.mox.StubOutWithMock(db, 'network_get_by_uuid') - self.net_man.context = context - db.network_get_by_uuid('elevated', 'quantum_net_id').AndReturn(None) - - self.mox.ReplayAll() - - network = self.net_man.get_network(context, ('quantum_net_id', - 'net_tenant_id')) - self.assertEquals(network['quantum_net_id'], 'quantum_net_id') - self.assertEquals(network['uuid'], 'quantum_net_id') - - -class QuantumDeallocationTestCase(QuantumNovaTestCase): - def test_deallocate_port(self): - quantum = self.mox.CreateMock( - quantum_connection.QuantumClientConnection) - quantum.get_port_by_attachment('q_tenant_id', 'net_id', - 'interface_id').AndReturn('port_id') - quantum.detach_and_delete_port('q_tenant_id', 'net_id', 'port_id') - self.net_man.q_conn = quantum - - self.mox.ReplayAll() - - self.net_man.deallocate_port('interface_id', 'net_id', 'q_tenant_id', - 'instance_id') - - def test_deallocate_port_logs_error(self): - quantum = self.mox.CreateMock( - quantum_connection.QuantumClientConnection) - quantum.get_port_by_attachment('q_tenant_id', 'net_id', - 'interface_id').AndRaise(Exception) - self.net_man.q_conn = quantum - - self.mox.StubOutWithMock(quantum_manager.LOG, 'exception') - quantum_manager.LOG.exception(mox.Regex(r'port deallocation failed')) - - self.mox.ReplayAll() - - self.net_man.deallocate_port('interface_id', 'net_id', 'q_tenant_id', - 'instance_id') - - def test_deallocate_ip_address(self): - ipam = self.mox.CreateMock(melange_ipam_lib.QuantumMelangeIPAMLib) - ipam.get_tenant_id_by_net_id('context', 'net_id', {'uuid': 1}, - 'project_id').AndReturn('ipam_tenant_id') - self.net_man.ipam = ipam - self.mox.ReplayAll() - self.net_man.deallocate_ip_address('context', 'net_id', 'project_id', - {'uuid': 1}, 'instance_id') - - def test_deallocate_ip_address(self): - ipam = self.mox.CreateMock(melange_ipam_lib.QuantumMelangeIPAMLib) - ipam.get_tenant_id_by_net_id('context', 'net_id', {'uuid': 1}, - 'project_id').AndRaise(Exception()) - self.net_man.ipam = ipam - - self.mox.StubOutWithMock(quantum_manager.LOG, 'exception') - quantum_manager.LOG.exception(mox.Regex(r'ipam deallocation failed')) - - self.mox.ReplayAll() - self.net_man.deallocate_ip_address('context', 'net_id', 'project_id', - {'uuid': 1}, 'instance_id') - - -class QuantumManagerTestCase(QuantumNovaTestCase): - def test_create_and_delete_nets(self): - self._create_nets() - self._delete_nets() - - def _create_nets(self): - for n in networks: - self._create_network(n) - - def _delete_nets(self): - for n in networks: - ctx = context.RequestContext('user1', n['project_id']) - self.net_man.delete_network(ctx, None, n['uuid']) - self.assertRaises(exception.NoNetworksFound, - db.network_get_all, ctx.elevated()) - - def _validate_nw_info(self, nw_info, expected_net_labels): - - self.assertEquals(len(nw_info), len(expected_net_labels)) - - ctx = context.RequestContext('user1', 'foo').elevated() - all_net_map = {} - for n in db.network_get_all(ctx): - all_net_map[n['label']] = n - - for i in range(0, len(nw_info)): - vif = nw_info[i] - net = all_net_map[expected_net_labels[i]] - - # simple test assumes that each starting prefix is unique - expected_v4_cidr_start = net['cidr'].split(".")[0].lower() - expected_v6_cidr_start = net['cidr_v6'].split(":")[0].lower() - - for subnet in vif['network']['subnets']: - addr = subnet['ips'][0]['address'] - if subnet['version'] == 4: - address_start = addr.split(".")[0].lower() - self.assertTrue(expected_v4_cidr_start, address_start) - else: - address_start = addr.split(":")[0].lower() - self.assertTrue(expected_v6_cidr_start, address_start) - - # confirm that there is a DHCP device on corresponding net - for l in expected_net_labels: - n = all_net_map[l] - tenant_id = (n['project_id'] or - FLAGS.quantum_default_tenant_id) - ports = self.net_man.q_conn.get_attached_ports( - tenant_id, n['uuid']) - self.assertEquals(len(ports), 2) # gw + instance VIF - - # make sure we aren't allowed to delete network with - # active port - self.assertRaises(exception.NetworkBusy, - self.net_man.delete_network, - ctx, None, n['uuid']) - - def _check_vifs(self, expect_num_vifs): - ctx = context.RequestContext('user1', "").elevated() - self.assertEqual(len(db.virtual_interface_get_all(ctx)), - expect_num_vifs) - - def _allocate_and_deallocate_instance(self, project_id, requested_networks, - expected_labels): - - ctx = context.RequestContext('user1', project_id) - self._check_vifs(0) - - instance_ref = db.instance_create(ctx, - {"project_id": project_id}) - - nw_info = self.net_man.allocate_for_instance(ctx.elevated(), - instance_id=instance_ref['id'], host="", - rxtx_factor=3, - project_id=project_id, - requested_networks=requested_networks) - - self._check_vifs(len(nw_info)) - - self._validate_nw_info(nw_info, expected_labels) - - nw_info = self.net_man.get_instance_nw_info(ctx, instance_ref['id'], - instance_ref['uuid'], - instance_ref['instance_type_id'], "", - project_id=project_id) - - self._check_vifs(len(nw_info)) - self._validate_nw_info(nw_info, expected_labels) - - port_net_pairs = [] - for vif in nw_info: - nid = vif['network']['id'] - pid = self.net_man.q_conn.get_port_by_attachment( - project_id, nid, vif['id']) - if pid is None: - pid = self.net_man.q_conn.get_port_by_attachment( - FLAGS.quantum_default_tenant_id, - nid, vif['id']) - self.assertTrue(pid is not None) - port_net_pairs.append((pid, nid)) - - self.net_man.deallocate_for_instance(ctx, - instance_id=instance_ref['id'], - project_id=project_id) - - for pid, nid in port_net_pairs: - self.assertRaises(quantum_client.QuantumNotFoundException, - self.net_man.q_conn.detach_and_delete_port, - project_id, nid, pid) - self.assertRaises(quantum_client.QuantumNotFoundException, - self.net_man.q_conn.detach_and_delete_port, - FLAGS.quantum_default_tenant_id, nid, pid) - - self._check_vifs(0) - - def test_allocate_and_deallocate_instance_static(self): - self._create_nets() - self._allocate_and_deallocate_instance("fake_project1", None, - ['public', 'project1-net1']) - self._delete_nets() - - def test_allocate_and_deallocate_instance_dynamic(self): - - self._create_nets() - project_id = "fake_project2" - ctx = context.RequestContext('user1', project_id) - all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids( - ctx, project_id) - requested_networks = [(n[0], None) for n in all_valid_networks] - - self.net_man.validate_networks(ctx, requested_networks) - - label_map = {} - for n in db.network_get_all(ctx.elevated()): - label_map[n['uuid']] = n['label'] - expected_labels = [label_map[uid] for uid, _i in requested_networks] - - self._allocate_and_deallocate_instance(project_id, requested_networks, - expected_labels) - self._delete_nets() - - def test_validate_bad_network(self): - ctx = context.RequestContext('user1', 'fake_project1') - self.assertRaises(exception.NetworkNotFound, - self.net_man.validate_networks, ctx, [("", None)]) - - def test_create_net_external_uuid(self): - """Tests use case where network can be created directly via - Quantum API, then the UUID is passed in via nova-manage""" - project_id = "foo_project" - ctx = context.RequestContext('user1', project_id) - net_id = self.net_man.q_conn.create_network(project_id, 'net1') - self.net_man.create_networks( - ctx, - label='achtungbaby', - cidr="9.9.9.0/24", - multi_host=False, - num_networks=1, - network_size=256, - cidr_v6=None, - gateway="9.9.9.1", - gateway_v6=None, - bridge=None, - bridge_interface=None, - dns1="8.8.8.8", - project_id=project_id, - priority=9, - uuid=net_id) - net = db.network_get_by_uuid(ctx.elevated(), net_id) - self.assertTrue(net is not None) - self.assertEquals(net['uuid'], net_id) - - def test_create_net_external_uuid_and_host_is_set(self): - """Make sure network['host'] is set when creating a network via the - network manager""" - project_id = "foo_project" - ctx = context.RequestContext('user1', project_id) - net_id = self.net_man.q_conn.create_network(project_id, 'net2') - self.net_man.create_networks( - ctx, label='achtungbaby2', cidr="9.9.8.0/24", multi_host=False, - num_networks=1, network_size=256, cidr_v6=None, - gateway="9.9.8.1", gateway_v6=None, bridge=None, - bridge_interface=None, dns1="8.8.8.8", project_id=project_id, - priority=8, uuid=net_id) - net = db.network_get_by_uuid(ctx.elevated(), net_id) - self.assertTrue(net is not None) - self.assertEquals(net['uuid'], net_id) - self.assertTrue(net['host'] != None) - - -class QuantumNovaMACGenerationTestCase(QuantumNovaTestCase): - def test_local_mac_address_creation(self): - self.flags(use_melange_mac_generation=False) - fake_mac = "ab:cd:ef:ab:cd:ef" - self.stubs.Set(utils, "generate_mac_address", - lambda: fake_mac) - project_id = "fake_project1" - ctx = context.RequestContext('user1', project_id) - self._create_network(networks[0]) - - all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids( - ctx, project_id) - requested_networks = [(n[0], None) for n in all_valid_networks] - - instance_ref = db.api.instance_create(ctx, - {"project_id": project_id}) - nw_info = self.net_man.allocate_for_instance(ctx, - instance_id=instance_ref['id'], host="", - rxtx_factor=3, - project_id=project_id, - requested_networks=requested_networks) - self.assertEqual(nw_info[0]['address'], fake_mac) - - def test_melange_mac_address_creation(self): - self.flags(use_melange_mac_generation=True) - fake_mac = "ab:cd:ef:ab:cd:ef" - self.stubs.Set(melange_connection.MelangeConnection, "create_vif", - lambda w, x, y, z: fake_mac) - project_id = "fake_project1" - ctx = context.RequestContext('user1', project_id) - self._create_network(networks[0]) - - all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids( - ctx, project_id) - requested_networks = [(n[0], None) for n in all_valid_networks] - - instance_ref = db.api.instance_create(ctx, - {"project_id": project_id}) - nw_info = self.net_man.allocate_for_instance(ctx, - instance_id=instance_ref['id'], host="", - rxtx_factor=3, - project_id=project_id, - requested_networks=requested_networks) - self.assertEqual(nw_info[0]['address'], fake_mac) - - -class QuantumNovaPortSecurityTestCase(QuantumNovaTestCase): - def test_port_securty(self): - self.flags(use_melange_mac_generation=True) - self.flags(quantum_use_port_security=True) - fake_mac = "ab:cd:ef:ab:cd:ef" - self.stubs.Set(melange_connection.MelangeConnection, "create_vif", - lambda w, x, y, z: fake_mac) - project_id = "fake_project1" - ctx = context.RequestContext('user1', project_id) - self._create_network(networks[0]) - - all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids( - ctx, project_id) - requested_networks = [(n[0], None) for n in all_valid_networks] - - instance_ref = db.api.instance_create(ctx, - {"project_id": project_id}) - oldfunc = self.net_man.q_conn.create_and_attach_port - - # Make sure we get the appropriate mac set in allowed_address_pairs - # if port security is enabled. - def _instrumented_create_and_attach_port(tenant_id, net_id, - interface_id, **kwargs): - self.assertTrue('allowed_address_pairs' in kwargs.keys()) - pairs = kwargs['allowed_address_pairs'] - self.assertTrue(pairs[0]['mac_address'] == fake_mac) - self.net_man.q_conn.create_and_attach_port = oldfunc - return oldfunc(tenant_id, net_id, interface_id, **kwargs) - _port_attach = _instrumented_create_and_attach_port - self.net_man.q_conn.create_and_attach_port = _port_attach - nw_info = self.net_man.allocate_for_instance(ctx, - instance_id=instance_ref['id'], host="", - rxtx_factor=3, - project_id=project_id, - requested_networks=requested_networks) - self.assertEqual(nw_info[0]['address'], fake_mac) - - def test_port_securty_negative(self): - self.flags(use_melange_mac_generation=True) - self.flags(quantum_use_port_security=False) - fake_mac = "ab:cd:ef:ab:cd:ef" - self.stubs.Set(melange_connection.MelangeConnection, "create_vif", - lambda w, x, y, z: fake_mac) - project_id = "fake_project1" - ctx = context.RequestContext('user1', project_id) - self._create_network(networks[0]) - - all_valid_networks = self.net_man.ipam.get_project_and_global_net_ids( - ctx, project_id) - requested_networks = [(n[0], None) for n in all_valid_networks] - - instance_ref = db.api.instance_create(ctx, - {"project_id": project_id}) - oldfunc = self.net_man.q_conn.create_and_attach_port - - # Make sure no pairs are passed in if port security is turned off - def _instrumented_create_and_attach_port(tenant_id, net_id, - interface_id, **kwargs): - self.assertTrue('allowed_address_pairs' in kwargs.keys()) - pairs = kwargs['allowed_address_pairs'] - self.assertTrue(len(pairs) == 0) - self.net_man.q_conn.create_and_attach_port = oldfunc - return oldfunc(tenant_id, net_id, interface_id, **kwargs) - _port_attach = _instrumented_create_and_attach_port - self.net_man.q_conn.create_and_attach_port = _port_attach - nw_info = self.net_man.allocate_for_instance(ctx, - instance_id=instance_ref['id'], host="", - rxtx_factor=3, - project_id=project_id, - requested_networks=requested_networks) - self.assertEqual(nw_info[0]['address'], fake_mac) - - -class QuantumMelangeTestCase(test.TestCase): - def setUp(self): - super(QuantumMelangeTestCase, self).setUp() - - fc = fake_client.FakeClient(LOG) - qc = quantum_connection.QuantumClientConnection(client=fc) - - self.net_man = quantum_manager.QuantumManager( - ipam_lib="nova.network.quantum.nova_ipam_lib", - q_conn=qc) - - def test_get_instance_uuids_by_ip_filter(self): - fake_context = context.RequestContext('user', 'project') - address = '1.2.3.4' - filters = {'ip': address} - - self.net_man.ipam = self.mox.CreateMockAnything() - self.net_man.ipam.get_instance_ids_by_ip_address(fake_context, - address).AndReturn(['instance_id']) - - instance = self.mox.CreateMockAnything() - instance.uuid = 'instance_uuid' - - self.mox.StubOutWithMock(db, 'instance_get') - db.instance_get(fake_context, 'instance_id').AndReturn(instance) - - self.mox.ReplayAll() - - uuids = self.net_man.get_instance_uuids_by_ip_filter(fake_context, - filters) - self.assertEquals(uuids, [{'instance_uuid':'instance_uuid'}]) -- cgit