openstack/neutron
Code Issues Proposed changes

Fix for adding gateway with IP outside subnet

Browse Source 
Currently 'force_gateway_on_subnet' configuration is set to True
by default and enforces the subnet on to the gateway. With this
fix 'force_gateway_on_subnet' can be changed to False, and
gateway outside the subnet can be added.
Before adding the default route, a route to the gateway IP is
added. This applies to both external and internal networks.

This configuration option is deprecated, and should be removed
in a future release. It should always allow gateway outside the
subnet. This is done as a separate patch
https://review.openstack.org/#/c/277303/

Change-Id: I3a942cf98d263681802729cf09527f06c80fab2b
Closes-Bug: #1335023
Closes-Bug: #1398768
This commit is contained in:
Sreekumar S 2015-10-10 03:18:00 +05:30
parent 557da957b8
commit b6126bc0f1
12 changed files with 400 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
neutron/agent/l3/dvr_fip_ns.py
View File

@ -22,6 +22,7 @@ from neutron.agent.l3 import namespaces
from neutron.agent.linux import ip_lib
from neutron.agent.linux import iptables_manager
from neutron.common import utils as common_utils
from neutron.ipam import utils as ipam_utils
LOG = logging.getLogger(__name__)
@ -120,8 +121,12 @@ class FipNamespace(namespaces.Namespace):
for subnet in ex_gw_port['subnets']:
gw_ip = subnet.get('gateway_ip')
if gw_ip:
is_gateway_not_in_subnet = not ipam_utils.check_subnet_ip(
subnet.get('cidr'), gw_ip)
ipd = ip_lib.IPDevice(interface_name,
namespace=ns_name)
if is_gateway_not_in_subnet:
ipd.route.add_route(gw_ip, scope='link')
ipd.route.add_gateway(gw_ip)
cmd = ['sysctl', '-w', 'net.ipv4.conf.%s.proxy_arp=1' % interface_name]

17
neutron/agent/l3/router_info.py
View File

@ -25,6 +25,7 @@ from neutron.common import constants as l3_constants
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron.common import utils as common_utils
from neutron.ipam import utils as ipam_utils
LOG = logging.getLogger(__name__)
INTERNAL_DEV_PREFIX = namespaces.INTERNAL_DEV_PREFIX
@ -571,6 +572,20 @@ class RouterInfo(object):
gateway_ips.append(self.agent_conf.ipv6_gateway)
return gateway_ips
def _add_route_to_gw(self, ex_gw_port, device_name,
namespace, preserve_ips):
# Note: ipv6_gateway is an ipv6 LLA
# and so doesn't need a special route
for subnet in ex_gw_port.get('subnets', []):
is_gateway_not_in_subnet = (subnet['gateway_ip'] and
not ipam_utils.check_subnet_ip(
subnet['cidr'],
subnet['gateway_ip']))
if is_gateway_not_in_subnet:
preserve_ips.append(subnet['gateway_ip'])
device = ip_lib.IPDevice(device_name, namespace=namespace)
device.route.add_route(subnet['gateway_ip'], scope='link')
def _external_gateway_added(self, ex_gw_port, interface_name,
ns_name, preserve_ips):
LOG.debug("External gateway added: port(%s), interface(%s), ns(%s)",
@ -587,6 +602,8 @@ class RouterInfo(object):
# There is no IPv6 gw_ip, use RouterAdvt for default route.
enable_ra_on_gw = True
self._add_route_to_gw(ex_gw_port, device_name=interface_name,
namespace=ns_name, preserve_ips=preserve_ips)
self.driver.init_router_port(
interface_name,
ip_cidrs,

22
neutron/agent/linux/dhcp.py
View File

@ -38,6 +38,7 @@ from neutron.common import exceptions
from neutron.common import ipv6_utils
from neutron.common import utils as common_utils
from neutron.extensions import extra_dhcp_opt as edo_ext
from neutron.ipam import utils as ipam_utils
LOG = logging.getLogger(__name__)
@ -1028,6 +1029,27 @@ class DeviceManager(object):
'%(ip)s',
{'n': network.id, 'ip': subnet.gateway_ip})
# Check for and remove the on-link route for the old
# gateway being replaced, if it is outside the subnet
is_old_gateway_not_in_subnet = (gateway and
not ipam_utils.check_subnet_ip(
subnet.cidr, gateway))
if is_old_gateway_not_in_subnet:
v4_onlink = device.route.list_onlink_routes(
constants.IP_VERSION_4)
v6_onlink = device.route.list_onlink_routes(
constants.IP_VERSION_6)
existing_onlink_routes = set(
r['cidr'] for r in v4_onlink + v6_onlink)
if gateway in existing_onlink_routes:
device.route.delete_route(gateway, scope='link')
is_new_gateway_not_in_subnet = (subnet.gateway_ip and
not ipam_utils.check_subnet_ip(
subnet.cidr,
subnet.gateway_ip))
if is_new_gateway_not_in_subnet:
device.route.add_route(subnet.gateway_ip, scope='link')
device.route.add_gateway(subnet.gateway_ip)
return

3
neutron/agent/linux/interface.py
View File

@ -183,7 +183,8 @@ class LinuxInterfaceDriver(object):
for route in new_onlink_cidrs - existing_onlink_cidrs:
LOG.debug("adding onlink route(%s)", route)
device.route.add_onlink_route(route)
for route in existing_onlink_cidrs - new_onlink_cidrs:
for route in (existing_onlink_cidrs - new_onlink_cidrs -
set(preserve_ips or [])):
LOG.debug("deleting onlink route(%s)", route)
device.route.delete_onlink_route(route)

14
neutron/db/db_base_plugin_v2.py
View File

@ -455,9 +455,17 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
if attributes.is_attr_set(s.get('gateway_ip')):
self._validate_ip_version(ip_ver, s['gateway_ip'], 'gateway_ip')
if (cfg.CONF.force_gateway_on_subnet and
not ipam.utils.check_gateway_in_subnet(
s['cidr'], s['gateway_ip'])):
if cfg.CONF.force_gateway_on_subnet:
# TODO(sreesiv) check_gateway_in_subnet() will be
# obsolete and should be removed when the option
# 'force_gateway_on_subnet' is removed.
is_gateway_not_valid = not ipam.utils.check_gateway_in_subnet(
s['cidr'], s['gateway_ip'])
else:
is_gateway_not_valid = (
ipam.utils.check_gateway_invalid_in_subnet(
s['cidr'], s['gateway_ip']))
if is_gateway_not_valid:
error_message = _("Gateway is not valid on subnet")
raise n_exc.InvalidInput(error_message=error_message)
# Ensure the gateway IP is not assigned to any port

15
neutron/ipam/utils.py
View File

@ -14,6 +14,7 @@
# under the License.
import netaddr
from neutron.common import constants
def check_subnet_ip(cidr, ip_address):
@ -27,6 +28,20 @@ def check_subnet_ip(cidr, ip_address):
and net.netmask & ip == net.network)
def check_gateway_invalid_in_subnet(cidr, gateway):
"""Check whether the gw IP address is invalid on the subnet."""
ip = netaddr.IPAddress(gateway)
net = netaddr.IPNetwork(cidr)
# Check whether the gw IP is in-valid on subnet.
# If gateway is in the subnet, it cannot be the
# 'network' or the 'broadcast address (only in IPv4)'.
# If gateway is out of subnet, there is no way to
# check since we don't have gateway's subnet cidr.
return (ip in net and
(ip == net.network or
(net.version == constants.IP_VERSION_4 and ip == net[-1])))
def check_gateway_in_subnet(cidr, gateway):
"""Validate that the gateway is on the subnet."""
ip = netaddr.IPAddress(gateway)

76
neutron/tests/unit/agent/dhcp/test_agent.py
View File

@ -1196,6 +1196,10 @@ class FakeV4Subnet(object):
enable_dhcp = True
class FakeV4SubnetOutsideGateway(FakeV4Subnet):
gateway_ip = '192.168.1.1'
class FakeV4SubnetNoGateway(object):
id = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
ip_version = 4
@ -1211,6 +1215,10 @@ class FakeV4Network(object):
namespace = 'qdhcp-aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
class FakeV4NetworkOutsideGateway(FakeV4Network):
subnets = [FakeV4SubnetOutsideGateway()]
class FakeV4NetworkNoSubnet(object):
id = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
subnets = []
@ -1320,21 +1328,25 @@ class TestDeviceManager(base.BaseTestCase):
self.mangle_inst.assert_has_calls(expected)
def test_setup_create_dhcp_port(self):
plugin = mock.Mock()
net = copy.deepcopy(fake_network)
plugin.create_dhcp_port.return_value = fake_dhcp_port
dh = dhcp.DeviceManager(cfg.CONF, plugin)
dh.setup(net)
with mock.patch.object(dhcp.ip_lib, 'IPDevice') as mock_IPDevice:
plugin = mock.Mock()
device = mock.Mock()
mock_IPDevice.return_value = device
device.route.get_gateway.return_value = None
net = copy.deepcopy(fake_network)
plugin.create_dhcp_port.return_value = fake_dhcp_port
dh = dhcp.DeviceManager(cfg.CONF, plugin)
dh.setup(net)
plugin.assert_has_calls([
mock.call.create_dhcp_port(
{'port': {'name': '', 'admin_state_up': True,
'network_id': net.id,
'tenant_id': net.tenant_id,
'fixed_ips': [{'subnet_id':
fake_dhcp_port.fixed_ips[0].subnet_id}],
'device_id': mock.ANY}})])
self.assertIn(fake_dhcp_port, net.ports)
plugin.assert_has_calls([
mock.call.create_dhcp_port(
{'port': {'name': '', 'admin_state_up': True,
'network_id': net.id,
'tenant_id': net.tenant_id,
'fixed_ips': [{'subnet_id':
fake_dhcp_port.fixed_ips[0].subnet_id}],
'device_id': mock.ANY}})])
self.assertIn(fake_dhcp_port, net.ports)
def test_setup_plug_exception(self):
plugin = mock.Mock()
@ -1543,6 +1555,22 @@ class TestDeviceManager(base.BaseTestCase):
self.assertFalse(device.route.delete_gateway.called)
device.route.add_gateway.assert_called_once_with('192.168.0.1')
def test_set_default_route_outside_subnet(self):
dh = dhcp.DeviceManager(cfg.CONF, None)
with mock.patch.object(dhcp.ip_lib, 'IPDevice') as mock_IPDevice:
device = mock.Mock()
mock_IPDevice.return_value = device
device.route.get_gateway.return_value = None
# Basic one subnet with gateway outside the subnet.
network = FakeV4NetworkOutsideGateway()
dh._set_default_route(network, 'tap-name')
self.assertEqual(device.route.get_gateway.call_count, 1)
self.assertFalse(device.route.delete_gateway.called)
device.route.add_route.assert_called_once_with('192.168.1.1',
scope='link')
device.route.add_gateway.assert_called_once_with('192.168.1.1')
def test_set_default_route_no_subnet(self):
dh = dhcp.DeviceManager(cfg.CONF, None)
with mock.patch.object(dhcp.ip_lib, 'IPDevice') as mock_IPDevice:
@ -1611,6 +1639,26 @@ class TestDeviceManager(base.BaseTestCase):
self.assertFalse(device.route.delete_gateway.called)
device.route.add_gateway.assert_called_once_with('192.168.0.1')
def test_set_default_route_change_gateway_outside_subnet(self):
dh = dhcp.DeviceManager(cfg.CONF, None)
with mock.patch.object(dhcp.ip_lib, 'IPDevice') as mock_IPDevice:
device = mock.Mock()
mock_IPDevice.return_value = device
device.route.list_onlink_routes.return_value = (
[{'cidr': '192.168.2.1'}])
device.route.get_gateway.return_value = dict(gateway='192.168.2.1')
network = FakeV4NetworkOutsideGateway()
dh._set_default_route(network, 'tap-name')
self.assertEqual(device.route.get_gateway.call_count, 1)
self.assertEqual(device.route.list_onlink_routes.call_count, 2)
self.assertFalse(device.route.delete_gateway.called)
device.route.delete_route.assert_called_once_with('192.168.2.1',
scope='link')
device.route.add_route.assert_called_once_with('192.168.1.1',
scope='link')
device.route.add_gateway.assert_called_once_with('192.168.1.1')
def test_set_default_route_two_subnets(self):
# Try two subnets. Should set gateway from the first.
dh = dhcp.DeviceManager(cfg.CONF, None)

4
neutron/tests/unit/agent/l3/test_agent.py
View File

@ -914,6 +914,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
'subnet_id': subnet_id}],
'subnets': [
{'id': subnet_id,
'cidr': '20.0.0.0/24',
'gateway_ip': '20.0.0.1'}],
'id': _uuid(),
'network_id': fake_network_id,
@ -980,6 +981,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
'subnet_id': fake_subnet_id}],
'subnets': [
{'id': fake_subnet_id,
'cidr': '20.0.0.0/24',
'gateway_ip': '20.0.0.1'}],
'id': _uuid(),
'network_id': fake_network_id,
@ -1024,6 +1026,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
'subnet_id': subnet_id}],
'subnets': [
{'id': subnet_id,
'cidr': '20.0.0.0/24',
'gateway_ip': '20.0.0.1'}],
'id': _uuid(),
'network_id': fake_network_id,
@ -1075,6 +1078,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
'subnet_id': 'subnet_id'}],
'subnets': [
{'id': 'subnet_id',
'cidr': '20.0.0.0/24',
'gateway_ip': '20.0.0.1'}],
'id': _uuid(),
'network_id': 'fake_network_id',

33
neutron/tests/unit/agent/l3/test_dvr_fip_ns.py
View File

@ -96,6 +96,39 @@ class TestDvrFipNs(base.BaseTestCase):
'20.0.0.30',
mock.ANY)
@mock.patch.object(ip_lib, 'IPWrapper')
@mock.patch.object(ip_lib, 'IPDevice')
@mock.patch.object(ip_lib, 'send_ip_addr_adv_notif')
@mock.patch.object(ip_lib, 'device_exists')
def test_gateway_outside_subnet_added(self, device_exists, send_adv_notif,
IPDevice, IPWrapper):
device = mock.Mock()
IPDevice.return_value = device
subnet_id = _uuid()
agent_gw_port = {'fixed_ips': [{'ip_address': '20.0.0.30',
'prefixlen': 24,
'subnet_id': subnet_id}],
'subnets': [{'id': subnet_id,
'cidr': '20.0.0.0/24',
'gateway_ip': '20.0.1.1'}],
'id': _uuid(),
'network_id': self.net_id,
'mac_address': 'ca:fe:de:ad:be:ef'}
device_exists.return_value = False
self.fip_ns._gateway_added(agent_gw_port,
mock.sentinel.interface_name)
self.assertEqual(1, self.driver.plug.call_count)
self.assertEqual(1, self.driver.init_l3.call_count)
send_adv_notif.assert_called_once_with(self.fip_ns.get_name(),
mock.sentinel.interface_name,
'20.0.0.30',
mock.ANY)
device.route.add_route.assert_called_once_with('20.0.1.1',
scope='link')
device.route.add_gateway.assert_called_once_with('20.0.1.1')
@mock.patch.object(iptables_manager, 'IptablesManager')
@mock.patch.object(utils, 'execute')
@mock.patch.object(ip_lib.IpNetnsCommand, 'exists')

225
neutron/tests/unit/agent/linux/test_dhcp.py
View File

@ -1991,66 +1991,70 @@ class TestDeviceManager(TestConfBase):
self.mock_load_interface_driver = load_interface_driver_patcher.start()
def _test_setup(self, load_interface_driver, ip_lib, use_gateway_ips):
# Create DeviceManager.
self.conf.register_opt(cfg.BoolOpt('enable_isolated_metadata',
default=False))
plugin = mock.Mock()
mgr = dhcp.DeviceManager(self.conf, plugin)
load_interface_driver.assert_called_with(self.conf)
with mock.patch.object(dhcp.ip_lib, 'IPDevice') as mock_IPDevice:
# Create DeviceManager.
self.conf.register_opt(cfg.BoolOpt('enable_isolated_metadata',
default=False))
plugin = mock.Mock()
device = mock.Mock()
mock_IPDevice.return_value = device
device.route.get_gateway.return_value = None
mgr = dhcp.DeviceManager(self.conf, plugin)
load_interface_driver.assert_called_with(self.conf)
# Setup with no existing DHCP port - expect a new DHCP port to
# be created.
network = FakeDeviceManagerNetwork()
network.tenant_id = 'Tenant A'
# Setup with no existing DHCP port - expect a new DHCP port to
# be created.
network = FakeDeviceManagerNetwork()
network.tenant_id = 'Tenant A'
def mock_create(dict):
port = dhcp.DictModel(dict['port'])
port.id = 'abcd-123456789'
port.mac_address = '00-12-34-56-78-90'
port.fixed_ips = [
dhcp.DictModel({'subnet_id': ip['subnet_id'],
'ip_address': 'unique-IP-address'})
for ip in port.fixed_ips
]
return port
def mock_create(dict):
port = dhcp.DictModel(dict['port'])
port.id = 'abcd-123456789'
port.mac_address = '00-12-34-56-78-90'
port.fixed_ips = [
dhcp.DictModel({'subnet_id': ip['subnet_id'],
'ip_address': 'unique-IP-address'})
for ip in port.fixed_ips
]
return port
plugin.create_dhcp_port.side_effect = mock_create
mgr.driver.get_device_name.return_value = 'ns-XXX'
mgr.driver.use_gateway_ips = use_gateway_ips
ip_lib.ensure_device_is_ready.return_value = True
mgr.setup(network)
plugin.create_dhcp_port.assert_called_with(mock.ANY)
plugin.create_dhcp_port.side_effect = mock_create
mgr.driver.get_device_name.return_value = 'ns-XXX'
mgr.driver.use_gateway_ips = use_gateway_ips
ip_lib.ensure_device_is_ready.return_value = True
mgr.setup(network)
plugin.create_dhcp_port.assert_called_with(mock.ANY)
mgr.driver.init_l3.assert_called_with('ns-XXX',
mock.ANY,
namespace='qdhcp-ns')
cidrs = set(mgr.driver.init_l3.call_args[0][1])
if use_gateway_ips:
self.assertEqual(cidrs, set(['%s/%s' % (s.gateway_ip,
s.cidr.split('/')[1])
for s in network.subnets]))
else:
self.assertEqual(cidrs, set(['unique-IP-address/24',
mgr.driver.init_l3.assert_called_with('ns-XXX',
mock.ANY,
namespace='qdhcp-ns')
cidrs = set(mgr.driver.init_l3.call_args[0][1])
if use_gateway_ips:
self.assertEqual(cidrs, set(['%s/%s' % (s.gateway_ip,
s.cidr.split('/')[1])
for s in network.subnets]))
else:
self.assertEqual(cidrs, set(['unique-IP-address/24',
'unique-IP-address/64']))
# Now call setup again. This time we go through the existing
# port code path, and the driver's init_l3 method is called
# again.
plugin.create_dhcp_port.reset_mock()
mgr.driver.init_l3.reset_mock()
mgr.setup(network)
mgr.driver.init_l3.assert_called_with('ns-XXX',
mock.ANY,
namespace='qdhcp-ns')
cidrs = set(mgr.driver.init_l3.call_args[0][1])
if use_gateway_ips:
self.assertEqual(cidrs, set(['%s/%s' % (s.gateway_ip,
s.cidr.split('/')[1])
for s in network.subnets]))
else:
self.assertEqual(cidrs, set(['unique-IP-address/24',
'unique-IP-address/64']))
self.assertFalse(plugin.create_dhcp_port.called)
# Now call setup again. This time we go through the existing
# port code path, and the driver's init_l3 method is called
# again.
plugin.create_dhcp_port.reset_mock()
mgr.driver.init_l3.reset_mock()
mgr.setup(network)
mgr.driver.init_l3.assert_called_with('ns-XXX',
mock.ANY,
namespace='qdhcp-ns')
cidrs = set(mgr.driver.init_l3.call_args[0][1])
if use_gateway_ips:
self.assertEqual(cidrs, set(['%s/%s' % (s.gateway_ip,
s.cidr.split('/')[1])
for s in network.subnets]))
else:
self.assertEqual(cidrs, set(['unique-IP-address/24',
'unique-IP-address/64']))
self.assertFalse(plugin.create_dhcp_port.called)
def test_setup_device_manager_dhcp_port_without_gateway_ips(self):
self._test_setup(self.mock_load_interface_driver,
@ -2065,73 +2069,82 @@ class TestDeviceManager(TestConfBase):
logic.
"""
# Create DeviceManager.
self.conf.register_opt(cfg.BoolOpt('enable_isolated_metadata',
default=False))
plugin = mock.Mock()
mgr = dhcp.DeviceManager(self.conf, plugin)
self.mock_load_interface_driver.assert_called_with(self.conf)
with mock.patch.object(dhcp.ip_lib, 'IPDevice') as mock_IPDevice:
# Create DeviceManager.
self.conf.register_opt(cfg.BoolOpt('enable_isolated_metadata',
default=False))
plugin = mock.Mock()
device = mock.Mock()
mock_IPDevice.return_value = device
device.route.get_gateway.return_value = None
mgr = dhcp.DeviceManager(self.conf, plugin)
self.mock_load_interface_driver.assert_called_with(self.conf)
# Setup with a reserved DHCP port.
network = FakeDualNetworkReserved()
network.tenant_id = 'Tenant A'
reserved_port = network.ports[-1]
# Setup with a reserved DHCP port.
network = FakeDualNetworkReserved()
network.tenant_id = 'Tenant A'
reserved_port = network.ports[-1]
def mock_update(port_id, dict):
port = reserved_port
port.network_id = dict['port']['network_id']
port.device_id = dict['port']['device_id']
return port
def mock_update(port_id, dict):
port = reserved_port
port.network_id = dict['port']['network_id']
port.device_id = dict['port']['device_id']
return port
plugin.update_dhcp_port.side_effect = mock_update
mgr.driver.get_device_name.return_value = 'ns-XXX'
mgr.driver.use_gateway_ips = False
self.mock_ip_lib.ensure_device_is_ready.return_value = True
mgr.setup(network)
plugin.update_dhcp_port.assert_called_with(reserved_port.id, mock.ANY)
plugin.update_dhcp_port.side_effect = mock_update
mgr.driver.get_device_name.return_value = 'ns-XXX'
mgr.driver.use_gateway_ips = False
self.mock_ip_lib.ensure_device_is_ready.return_value = True
mgr.setup(network)
plugin.update_dhcp_port.assert_called_with(reserved_port.id,
mock.ANY)
mgr.driver.init_l3.assert_called_with('ns-XXX',
['192.168.0.6/24'],
namespace='qdhcp-ns')
mgr.driver.init_l3.assert_called_with('ns-XXX',
['192.168.0.6/24'],
namespace='qdhcp-ns')
def test_setup_reserved_2(self):
"""Test scenario where a network has two reserved ports, and
update_dhcp_port fails for the first of those.
"""
# Create DeviceManager.
self.conf.register_opt(cfg.BoolOpt('enable_isolated_metadata',
default=False))
plugin = mock.Mock()
mgr = dhcp.DeviceManager(self.conf, plugin)
self.mock_load_interface_driver.assert_called_with(self.conf)
with mock.patch.object(dhcp.ip_lib, 'IPDevice') as mock_IPDevice:
# Create DeviceManager.
self.conf.register_opt(cfg.BoolOpt('enable_isolated_metadata',
default=False))
plugin = mock.Mock()
device = mock.Mock()
mock_IPDevice.return_value = device
device.route.get_gateway.return_value = None
mgr = dhcp.DeviceManager(self.conf, plugin)
self.mock_load_interface_driver.assert_called_with(self.conf)
# Setup with a reserved DHCP port.
network = FakeDualNetworkReserved2()
network.tenant_id = 'Tenant A'
reserved_port_1 = network.ports[-2]
reserved_port_2 = network.ports[-1]
# Setup with a reserved DHCP port.
network = FakeDualNetworkReserved2()
network.tenant_id = 'Tenant A'
reserved_port_1 = network.ports[-2]
reserved_port_2 = network.ports[-1]
def mock_update(port_id, dict):
if port_id == reserved_port_1.id:
return None
def mock_update(port_id, dict):
if port_id == reserved_port_1.id:
return None
port = reserved_port_2
port.network_id = dict['port']['network_id']
port.device_id = dict['port']['device_id']
return port
port = reserved_port_2
port.network_id = dict['port']['network_id']
port.device_id = dict['port']['device_id']
return port
plugin.update_dhcp_port.side_effect = mock_update
mgr.driver.get_device_name.return_value = 'ns-XXX'
mgr.driver.use_gateway_ips = False
self.mock_ip_lib.ensure_device_is_ready.return_value = True
mgr.setup(network)
plugin.update_dhcp_port.assert_called_with(reserved_port_2.id,
mock.ANY)
plugin.update_dhcp_port.side_effect = mock_update
mgr.driver.get_device_name.return_value = 'ns-XXX'
mgr.driver.use_gateway_ips = False
self.mock_ip_lib.ensure_device_is_ready.return_value = True
mgr.setup(network)
plugin.update_dhcp_port.assert_called_with(reserved_port_2.id,
mock.ANY)
mgr.driver.init_l3.assert_called_with('ns-XXX',
['192.168.0.6/24'],
namespace='qdhcp-ns')
mgr.driver.init_l3.assert_called_with('ns-XXX',
['192.168.0.6/24'],
namespace='qdhcp-ns')
class TestDictModel(base.BaseTestCase):

93
neutron/tests/unit/db/test_db_base_plugin_v2.py
View File

@ -3597,6 +3597,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
ipv6_address_mode=constants.IPV6_SLAAC)
def test_create_subnet_gw_outside_cidr_returns_400(self):
cfg.CONF.set_override('force_gateway_on_subnet', True)
with self.network() as network:
self._create_subnet(self.fmt,
network['network']['id'],
@ -3604,6 +3605,33 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
webob.exc.HTTPClientError.code,
gateway_ip='100.0.0.1')
def test_create_subnet_gw_outside_cidr_returns_201(self):
cfg.CONF.set_override('force_gateway_on_subnet', False)
with self.network() as network:
self._create_subnet(self.fmt,
network['network']['id'],
'10.0.0.0/24',
webob.exc.HTTPCreated.code,
gateway_ip='100.0.0.1')
def test_create_subnet_gw_is_nw_addr_returns_400(self):
cfg.CONF.set_override('force_gateway_on_subnet', False)
with self.network() as network:
self._create_subnet(self.fmt,
network['network']['id'],
'10.0.0.0/24',
webob.exc.HTTPClientError.code,
gateway_ip='10.0.0.0')
def test_create_subnet_gw_is_broadcast_addr_returns_400(self):
cfg.CONF.set_override('force_gateway_on_subnet', False)
with self.network() as network:
self._create_subnet(self.fmt,
network['network']['id'],
'10.0.0.0/24',
webob.exc.HTTPClientError.code,
gateway_ip='10.0.0.255')
def test_create_subnet_gw_of_network_returns_400(self):
with self.network() as network:
self._create_subnet(self.fmt,
@ -3953,19 +3981,68 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
ipv6_ra_mode=ra_mode,
ipv6_address_mode=addr_mode)
def test_create_subnet_ipv6_out_of_cidr_global(self):
def test_create_subnet_ipv6_out_of_cidr_global_returns_400(self):
cfg.CONF.set_override('force_gateway_on_subnet', True)
gateway_ip = '2000::1'
cidr = '2001::/64'
with testlib_api.ExpectedException(
webob.exc.HTTPClientError) as ctx_manager:
self._test_create_subnet(
gateway_ip=gateway_ip, cidr=cidr, ip_version=6,
gateway_ip=gateway_ip, cidr=cidr,
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.DHCPV6_STATEFUL,
ipv6_address_mode=constants.DHCPV6_STATEFUL)
self.assertEqual(webob.exc.HTTPClientError.code,
ctx_manager.exception.code)
def test_create_subnet_ipv6_out_of_cidr_global(self):
cfg.CONF.set_override('force_gateway_on_subnet', False)
gateway_ip = '2000::1'
cidr = '2001::/64'
subnet = self._test_create_subnet(
gateway_ip=gateway_ip, cidr=cidr,
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.DHCPV6_STATEFUL,
ipv6_address_mode=constants.DHCPV6_STATEFUL)
self.assertEqual(constants.IP_VERSION_6,
subnet['subnet']['ip_version'])
self.assertEqual(gateway_ip,
subnet['subnet']['gateway_ip'])
self.assertEqual(cidr,
subnet['subnet']['cidr'])
def test_create_subnet_ipv6_gw_is_nw_addr_returns_400(self):
cfg.CONF.set_override('force_gateway_on_subnet', False)
gateway_ip = '2001::0'
cidr = '2001::/64'
with testlib_api.ExpectedException(
webob.exc.HTTPClientError) as ctx_manager:
self._test_create_subnet(
gateway_ip=gateway_ip, cidr=cidr,
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.DHCPV6_STATEFUL,
ipv6_address_mode=constants.DHCPV6_STATEFUL)
self.assertEqual(webob.exc.HTTPClientError.code,
ctx_manager.exception.code)
def test_create_subnet_ipv6_gw_is_nw_end_addr_returns_201(self):
cfg.CONF.set_override('force_gateway_on_subnet', False)
gateway_ip = '2001::ffff'
cidr = '2001::/112'
subnet = self._test_create_subnet(
gateway_ip=gateway_ip, cidr=cidr,
ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.DHCPV6_STATEFUL,
ipv6_address_mode=constants.DHCPV6_STATEFUL)
self.assertEqual(constants.IP_VERSION_6,
subnet['subnet']['ip_version'])
self.assertEqual(gateway_ip,
subnet['subnet']['gateway_ip'])
self.assertEqual(cidr,
subnet['subnet']['cidr'])
def test_create_subnet_ipv6_out_of_cidr_lla(self):
gateway_ip = 'fe80::1'
cidr = '2001::/64'
@ -4173,6 +4250,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
res.status_int)
def test_update_subnet_gw_outside_cidr_returns_400(self):
cfg.CONF.set_override('force_gateway_on_subnet', True)
with self.network() as network:
with self.subnet(network=network) as subnet:
data = {'subnet': {'gateway_ip': '100.0.0.1'}}
@ -4182,6 +4260,17 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
self.assertEqual(webob.exc.HTTPClientError.code,
res.status_int)
def test_update_subnet_gw_outside_cidr_returns_200(self):
cfg.CONF.set_override('force_gateway_on_subnet', False)
with self.network() as network:
with self.subnet(network=network) as subnet:
data = {'subnet': {'gateway_ip': '100.0.0.1'}}
req = self.new_update_request('subnets', data,
subnet['subnet']['id'])
res = req.get_response(self.api)
self.assertEqual(webob.exc.HTTPOk.code,
res.status_int)
def test_update_subnet_gw_ip_in_use_by_router_returns_409(self):
with self.network() as network:
with self.subnet(network=network,

19
neutron/tests/unit/ipam/test_requests.py
View File

@ -133,6 +133,7 @@ class TestIpamAnySubnetRequest(IpamSubnetRequestTestCase):
129)
def test_subnet_request_bad_gateway(self):
cfg.CONF.set_override('force_gateway_on_subnet', True)
self.assertRaises(ValueError,
ipam_req.AnySubnetRequest,
self.tenant_id,
@ -141,6 +142,15 @@ class TestIpamAnySubnetRequest(IpamSubnetRequestTestCase):
64,
gateway_ip='2000::1')
def test_subnet_request_good_gateway(self):
cfg.CONF.set_override('force_gateway_on_subnet', False)
request = ipam_req.AnySubnetRequest(self.tenant_id,
self.subnet_id,
constants.IPv6,
64,
gateway_ip='2000::1')
self.assertEqual(netaddr.IPAddress('2000::1'), request.gateway_ip)
def test_subnet_request_allocation_pool_wrong_version(self):
pools = [netaddr.IPRange('0.0.0.4', '0.0.0.5')]
self.assertRaises(ValueError,
@ -174,6 +184,7 @@ class TestIpamSpecificSubnetRequest(IpamSubnetRequestTestCase):
self.assertEqual(netaddr.IPNetwork('1.2.3.0/24'), request.subnet_cidr)
def test_subnet_request_bad_gateway(self):
cfg.CONF.set_override('force_gateway_on_subnet', True)
self.assertRaises(ValueError,
ipam_req.SpecificSubnetRequest,
self.tenant_id,
@ -181,6 +192,14 @@ class TestIpamSpecificSubnetRequest(IpamSubnetRequestTestCase):
'2001::1',
gateway_ip='2000::1')
def test_subnet_request_good_gateway(self):
cfg.CONF.set_override('force_gateway_on_subnet', False)
request = ipam_req.SpecificSubnetRequest(self.tenant_id,
self.subnet_id,
'2001::1',
gateway_ip='2000::1')
self.assertEqual(netaddr.IPAddress('2000::1'), request.gateway_ip)
class TestAddressRequest(base.BaseTestCase):