prefix=prefix)
ip_cidrs = common_utils.fixed_ip_cidrs(fixed_ips)
- self.driver.init_l3(interface_name, ip_cidrs, namespace=ns_name)
+ self.driver.init_router_port(
+ interface_name, ip_cidrs, namespace=ns_name)
for fixed_ip in fixed_ips:
ip_lib.send_ip_addr_adv_notif(ns_name,
interface_name,
ip_cidrs = common_utils.fixed_ip_cidrs(ex_gw_port['fixed_ips'])
gateway_ips, enable_ra_on_gw = self._get_external_gw_ips(ex_gw_port)
- self.driver.init_l3(interface_name,
- ip_cidrs,
- namespace=ns_name,
- gateway_ips=gateway_ips,
- extra_subnets=ex_gw_port.get('extra_subnets', []),
- preserve_ips=preserve_ips,
- enable_ra_on_gw=enable_ra_on_gw,
- clean_connections=True)
+ self.driver.init_router_port(
+ interface_name,
+ ip_cidrs,
+ namespace=ns_name,
+ gateway_ips=gateway_ips,
+ extra_subnets=ex_gw_port.get('extra_subnets', []),
+ preserve_ips=preserve_ips,
+ enable_ra_on_gw=enable_ra_on_gw,
+ clean_connections=True)
for fixed_ip in ex_gw_port['fixed_ips']:
ip_lib.send_ip_addr_adv_notif(ns_name,
interface_name,
self.conf = conf
def init_l3(self, device_name, ip_cidrs, namespace=None,
- preserve_ips=[], gateway_ips=None, extra_subnets=[],
- enable_ra_on_gw=False, clean_connections=False):
+ preserve_ips=[], gateway_ips=None,
+ clean_connections=False):
"""Set the L3 settings for the interface using data from the port.
ip_cidrs: list of 'X.X.X.X/YY' strings
preserve_ips: list of ip cidrs that should not be removed from device
gateway_ips: For gateway ports, list of external gateway ip addresses
- enable_ra_on_gw: Boolean to indicate configuring acceptance of IPv6 RA
clean_connections: Boolean to indicate if we should cleanup connections
associated to removed ips
"""
for gateway_ip in gateway_ips or []:
device.route.add_gateway(gateway_ip)
+ def init_router_port(self,
+ device_name,
+ ip_cidrs,
+ namespace,
+ preserve_ips=None,
+ gateway_ips=None,
+ extra_subnets=None,
+ enable_ra_on_gw=False,
+ clean_connections=False):
+ """Set the L3 settings for a router interface using data from the port.
+
+ ip_cidrs: list of 'X.X.X.X/YY' strings
+ preserve_ips: list of ip cidrs that should not be removed from device
+ gateway_ips: For gateway ports, list of external gateway ip addresses
+ enable_ra_on_gw: Boolean to indicate configuring acceptance of IPv6 RA
+ clean_connections: Boolean to indicate if we should cleanup connections
+ associated to removed ips
+ extra_subnets: An iterable of cidrs to add as routes without address
+ """
+ self.init_l3(device_name=device_name,
+ ip_cidrs=ip_cidrs,
+ namespace=namespace,
+ preserve_ips=preserve_ips or [],
+ gateway_ips=gateway_ips,
+ clean_connections=clean_connections)
+
if enable_ra_on_gw:
self.configure_ipv6_ra(namespace, device_name)
- new_onlink_routes = set(s['cidr'] for s in extra_subnets)
+ device = ip_lib.IPDevice(device_name, namespace=namespace)
+
+ # Manage on-link routes (routes without an associated address)
+ new_onlink_routes = set(s['cidr'] for s in extra_subnets or [])
existing_onlink_routes = set(
device.route.list_onlink_routes(n_const.IP_VERSION_4) +
device.route.list_onlink_routes(n_const.IP_VERSION_6))
self.device_exists.return_value = False
ri.internal_network_added(port)
self.assertEqual(self.mock_driver.plug.call_count, 1)
- self.assertEqual(self.mock_driver.init_l3.call_count, 1)
+ self.assertEqual(self.mock_driver.init_router_port.call_count, 1)
self.send_adv_notif.assert_called_once_with(ri.ns_name,
interface_name,
'99.0.1.9', mock.ANY)
ri.external_gateway_added(ex_gw_port, interface_name)
if not router.get('distributed'):
self.assertEqual(self.mock_driver.plug.call_count, 1)
- self.assertEqual(self.mock_driver.init_l3.call_count, 1)
+ self.assertEqual(self.mock_driver.init_router_port.call_count, 1)
if no_subnet and not dual_stack:
self.assertEqual(self.send_adv_notif.call_count, 0)
ip_cidrs = []
'extra_subnets': [{'cidr': '172.16.0.0/24'}],
'enable_ra_on_gw': enable_ra_on_gw,
'clean_connections': True}
- self.mock_driver.init_l3.assert_called_with(interface_name,
- ip_cidrs,
- **kwargs)
+ self.mock_driver.init_router_port.assert_called_with(
+ interface_name, ip_cidrs, **kwargs)
else:
ri._create_dvr_gateway.assert_called_once_with(
ex_gw_port, interface_name,
router[l3_constants.FLOATINGIP_KEY] = fake_fip['floatingips']
ri.external_gateway_updated(ex_gw_port, interface_name)
self.assertEqual(1, self.mock_driver.plug.call_count)
- self.assertEqual(self.mock_driver.init_l3.call_count, 1)
+ self.assertEqual(self.mock_driver.init_router_port.call_count, 1)
exp_arp_calls = [mock.call(ri.ns_name, interface_name,
'20.0.0.30', mock.ANY)]
if dual_stack:
'extra_subnets': [{'cidr': '172.16.0.0/24'}],
'enable_ra_on_gw': False,
'clean_connections': True}
- self.mock_driver.init_l3.assert_called_with(interface_name,
- ip_cidrs,
- **kwargs)
+ self.mock_driver.init_router_port.assert_called_with(interface_name,
+ ip_cidrs,
+ **kwargs)
def test_external_gateway_updated(self):
self._test_external_gateway_updated()
# check 2 internal ports are plugged
# check 1 ext-gw-port is plugged
self.assertEqual(self.mock_driver.plug.call_count, 3)
- self.assertEqual(self.mock_driver.init_l3.call_count, 3)
+ self.assertEqual(self.mock_driver.init_router_port.call_count, 3)
def test_get_service_plugin_list(self):
service_plugins = [p_const.L3_ROUTER_NAT]
device_name = bc.get_device_name(FakePort())
self.assertEqual('tapabcdef01-12', device_name)
- def test_l3_init(self):
+ def test_init_router_port(self):
addresses = [dict(scope='global',
dynamic=False, cidr='172.16.77.240/24')]
self.ip_dev().addr.list = mock.Mock(return_value=addresses)
bc = BaseChild(self.conf)
ns = '12345678-1234-5678-90ab-ba0987654321'
- bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns,
- extra_subnets=[{'cidr': '172.20.0.0/24'}])
+ bc.init_router_port('tap0', ['192.168.1.2/24'], namespace=ns,
+ extra_subnets=[{'cidr': '172.20.0.0/24'}])
self.ip_dev.assert_has_calls(
[mock.call('tap0', namespace=ns),
mock.call().addr.list(filters=['permanent']),
mock.call().addr.add('192.168.1.2/24'),
mock.call().addr.delete('172.16.77.240/24'),
+ mock.call('tap0', namespace=ns),
mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
mock.call().route.add_onlink_route('172.20.0.0/24')])
- def test_l3_init_delete_onlink_routes(self):
+ def test_init_router_port_delete_onlink_routes(self):
addresses = [dict(scope='global',
dynamic=False, cidr='172.16.77.240/24')]
self.ip_dev().addr.list = mock.Mock(return_value=addresses)
bc = BaseChild(self.conf)
ns = '12345678-1234-5678-90ab-ba0987654321'
- bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns)
+ bc.init_router_port('tap0', ['192.168.1.2/24'], namespace=ns)
self.ip_dev.assert_has_calls(
[mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
def test_l3_init_without_clean_connections(self):
self._test_l3_init_clean_connections(False)
- def _test_l3_init_with_ipv6(self, include_gw_ip):
+ def _test_init_router_port_with_ipv6(self, include_gw_ip):
addresses = [dict(scope='global',
dynamic=False,
cidr='2001:db8:a::123/64')]
'extra_subnets': [{'cidr': '2001:db8:b::/64'}]}
if include_gw_ip:
kwargs['gateway_ips'] = ['2001:db8:a::1']
- bc.init_l3('tap0', [new_cidr], **kwargs)
+ bc.init_router_port('tap0', [new_cidr], **kwargs)
expected_calls = (
[mock.call('tap0', namespace=ns),
mock.call().addr.list(filters=['permanent']),
expected_calls += (
[mock.call().route.add_gateway('2001:db8:a::1')])
expected_calls += (
- [mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
+ [mock.call('tap0', namespace=ns),
+ mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
mock.call().route.add_onlink_route('2001:db8:b::/64')])
self.ip_dev.assert_has_calls(expected_calls)
- def test_l3_init_ipv6_with_gw_ip(self):
- self._test_l3_init_with_ipv6(include_gw_ip=True)
+ def test_init_router_port_ipv6_with_gw_ip(self):
+ self._test_init_router_port_with_ipv6(include_gw_ip=True)
- def test_l3_init_ipv6_without_gw_ip(self):
- self._test_l3_init_with_ipv6(include_gw_ip=False)
+ def test_init_router_port_ipv6_without_gw_ip(self):
+ self._test_init_router_port_with_ipv6(include_gw_ip=False)
- def test_l3_init_ext_gw_with_dual_stack(self):
+ def test_init_router_port_ext_gw_with_dual_stack(self):
old_addrs = [dict(ip_version=4, scope='global',
dynamic=False, cidr='172.16.77.240/24'),
dict(ip_version=6, scope='global',
bc = BaseChild(self.conf)
ns = '12345678-1234-5678-90ab-ba0987654321'
new_cidrs = ['192.168.1.2/24', '2001:db8:a::124/64']
- bc.init_l3('tap0', new_cidrs, namespace=ns,
- extra_subnets=[{'cidr': '172.20.0.0/24'}])
+ bc.init_router_port('tap0', new_cidrs, namespace=ns,
+ extra_subnets=[{'cidr': '172.20.0.0/24'}])
self.ip_dev.assert_has_calls(
[mock.call('tap0', namespace=ns),
mock.call().addr.list(filters=['permanent']),
mock.call().route.add_onlink_route('172.20.0.0/24')],
any_order=True)
- def test_l3_init_with_ipv6_delete_onlink_routes(self):
+ def test_init_router_port_with_ipv6_delete_onlink_routes(self):
addresses = [dict(scope='global',
dynamic=False, cidr='2001:db8:a::123/64')]
route = '2001:db8:a::/64'
bc = BaseChild(self.conf)
ns = '12345678-1234-5678-90ab-ba0987654321'
- bc.init_l3('tap0', ['2001:db8:a::124/64'], namespace=ns)
+ bc.init_router_port('tap0', ['2001:db8:a::124/64'], namespace=ns)
self.ip_dev.assert_has_calls(
[mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
Code Review / openstack-build / neutron-build.git / commitdiff