ex_gw_port = self._get_ex_gw_port(ri)
if ri.router['distributed'] and ex_gw_port:
snat_ports = self.get_snat_interfaces(ri)
- snat_ip = self._map_internal_interfaces(ri, port, snat_ports)
- if snat_ip:
- self._snat_redirect_add(ri, snat_ip['fixed_ips'][0]
+ sn_port = self._map_internal_interfaces(ri, port, snat_ports)
+ if sn_port:
+ self._snat_redirect_add(ri, sn_port['fixed_ips'][0]
['ip_address'], port, interface_name)
- if self.conf.agent_mode == 'dvr_snat' and (
+ if (self.conf.agent_mode == 'dvr_snat' and
ri.router['gw_port_host'] == self.host):
- ns_name = self.get_snat_ns_name(ri.router['id'])
- for port in snat_ports:
- self._set_subnet_info(port)
- interface_name = self.get_snat_int_device_name(port['id'])
- self._internal_network_added(ns_name, port['network_id'],
- port['id'], internal_cidr,
- port['mac_address'],
+ ns_name = self.get_snat_ns_name(ri.router['id'])
+ self._set_subnet_info(sn_port)
+ interface_name = (
+ self.get_snat_int_device_name(sn_port['id']))
+ self._internal_network_added(ns_name,
+ sn_port['network_id'],
+ sn_port['id'], internal_cidr,
+ sn_port['mac_address'],
interface_name,
SNAT_INT_DEV_PREFIX)
'neutron.openstack.common.loopingcall.FixedIntervalLoopingCall')
self.looping_call_p.start()
+ def _prepare_internal_network_data(self):
+ port_id = _uuid()
+ router_id = _uuid()
+ network_id = _uuid()
+ router = prepare_router_data(num_internal_ports=2)
+ router_id = router['id']
+ ri = l3_agent.RouterInfo(router_id, self.conf.root_helper,
+ self.conf.use_namespaces, router=router)
+ agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
+ cidr = '99.0.1.9/24'
+ mac = 'ca:fe:de:ad:be:ef'
+ port = {'network_id': network_id,
+ 'id': port_id, 'ip_cidr': cidr,
+ 'mac_address': mac}
+
+ return agent, ri, port
+
def test__sync_routers_task_raise_exception(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
self.plugin_api.get_routers.side_effect = Exception()
l3_agent.L3NATAgent(HOSTNAME, self.conf)
def _test_internal_network_action(self, action):
- port_id = _uuid()
- router_id = _uuid()
- network_id = _uuid()
- router = prepare_router_data(num_internal_ports=2)
- router_id = router['id']
- ri = l3_agent.RouterInfo(router_id, self.conf.root_helper,
- self.conf.use_namespaces, router=router)
- agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
- cidr = '99.0.1.9/24'
- mac = 'ca:fe:de:ad:be:ef'
- port = {'network_id': network_id,
- 'id': port_id, 'ip_cidr': cidr,
- 'mac_address': mac}
- interface_name = agent.get_internal_device_name(port_id)
+ agent, ri, port = self._prepare_internal_network_data()
+ interface_name = agent.get_internal_device_name(port['id'])
if action == 'add':
self.device_exists.return_value = False
else:
raise Exception("Invalid action %s" % action)
+ def _test_internal_network_action_dist(self, action):
+ agent, ri, port = self._prepare_internal_network_data()
+ ri.router['distributed'] = True
+ ri.router['gw_port_host'] = HOSTNAME
+ agent.host = HOSTNAME
+ agent.conf.agent_mode = 'dvr_snat'
+ sn_port = {'fixed_ips': [{'ip_address': '20.0.0.31',
+ 'subnet_id': _uuid()}],
+ 'subnet': {'gateway_ip': '20.0.0.1'},
+ 'extra_subnets': [{'cidr': '172.16.0.0/24'}],
+ 'id': _uuid(),
+ 'network_id': _uuid(),
+ 'mac_address': 'ca:fe:de:ad:be:ef',
+ 'ip_cidr': '20.0.0.31/24'}
+
+ if action == 'add':
+ self.device_exists.return_value = False
+
+ agent._map_internal_interfaces = mock.Mock(return_value=sn_port)
+ agent._snat_redirect_add = mock.Mock()
+ agent._set_subnet_info = mock.Mock()
+ agent._internal_network_added = mock.Mock()
+ agent.internal_network_added(ri, port)
+ self.assertEqual(agent._snat_redirect_add.call_count, 1)
+ self.assertEqual(agent._set_subnet_info.call_count, 1)
+ self.assertEqual(agent._internal_network_added.call_count, 2)
+
def test_agent_add_internal_network(self):
self._test_internal_network_action('add')
+ def test_agent_add_internal_network_dist(self):
+ self._test_internal_network_action_dist('add')
+
def test_agent_remove_internal_network(self):
self._test_internal_network_action('remove')