x/python-ganttclient
Code Issues Proposed changes

Moving network tests into the network folder.

Browse Source 
Change-Id: Ic8f20967fb999cc67f74bd5ef751c7bb23129604
This commit is contained in:
Rick Harris
2012-05-17 12:05:41 -05:00
parent c39322c74d
commit d0b96653ac
3 changed files with 0 additions and 1474 deletions
Download Patch File Download Diff File Expand all files Collapse all files

501
nova/tests/test_linux_net.py
View File

@@ -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: <BROADCAST,MULTICAST,UP,LOWER_UP> "
" 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: <BROADCAST,MULTICAST,UP,LOWER_UP> "
" 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: <BROADCAST,MULTICAST,UP,LOWER_UP> "
" 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: <BROADCAST,MULTICAST,UP,LOWER_UP> "
" 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)

326
nova/tests/test_network_info.py
View File

@@ -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)

647
nova/tests/test_quantum.py
View File

@@ -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'}])