if r not in orig_nat_rules]
self.assertEqual(len(nat_rules_delta), 1)
self._verify_snat_rules(nat_rules_delta, router)
- self.send_arp.assert_called_once()
+ # send_arp is called both times process_router is called
+ self.assertEqual(self.send_arp.call_count, 2)
+ def test_process_ipv6_only_gw(self):
+ agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
+ router = self._prepare_router_data(ip_version=6)
+ # Get NAT rules without the gw_port
+ gw_port = router['gw_port']
+ router['gw_port'] = None
+ ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
+ self.conf.use_namespaces, router=router)
+ agent.external_gateway_added = mock.Mock()
+ agent.process_router(ri)
+ orig_nat_rules = ri.iptables_manager.ipv4['nat'].rules[:]
+
+ # Get NAT rules with the gw_port
+ router['gw_port'] = gw_port
+ ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
+ self.conf.use_namespaces, router=router)
+ with mock.patch.object(
+ agent,
+ 'external_gateway_nat_rules') as external_gateway_nat_rules:
+ agent.process_router(ri)
+ new_nat_rules = ri.iptables_manager.ipv4['nat'].rules[:]
+
+ # There should be no change with the NAT rules
+ self.assertFalse(external_gateway_nat_rules.called)
+ self.assertEqual(orig_nat_rules, new_nat_rules)
+
+ def test_process_router_ipv6_interface_added(self):
+ agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
+ router = self._prepare_router_data()
+ ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
+ self.conf.use_namespaces, router=router)
+ agent.external_gateway_added = mock.Mock()
+ # Process with NAT
+ agent.process_router(ri)
+ orig_nat_rules = ri.iptables_manager.ipv4['nat'].rules[:]
+ # Add an IPv6 interface and reprocess
+ router[l3_constants.INTERFACE_KEY].append(
+ {'id': _uuid(),
+ 'network_id': _uuid(),
+ 'admin_state_up': True,
+ 'fixed_ips': [{'ip_address': 'fd00::2',
+ 'subnet_id': _uuid()}],
+ 'mac_address': 'ca:fe:de:ad:be:ef',
+ 'subnet': {'cidr': 'fd00::/64',
+ 'gateway_ip': 'fd00::1'}})
+ # Reassign the router object to RouterInfo
+ ri.router = router
+ agent.process_router(ri)
+ # For some reason set logic does not work well with
+ # IpTablesRule instances
+ nat_rules_delta = [r for r in ri.iptables_manager.ipv4['nat'].rules
+ if r not in orig_nat_rules]
+ self.assertFalse(nat_rules_delta)
+
+ def test_process_router_ipv6v4_interface_added(self):
+ agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
+ router = self._prepare_router_data()
+ ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
+ self.conf.use_namespaces, router=router)
+ agent.external_gateway_added = mock.Mock()
+ # Process with NAT
+ agent.process_router(ri)
+ orig_nat_rules = ri.iptables_manager.ipv4['nat'].rules[:]
+ # Add an IPv4 and IPv6 interface and reprocess
+ router[l3_constants.INTERFACE_KEY].append(
+ {'id': _uuid(),
+ 'network_id': _uuid(),
+ 'admin_state_up': True,
+ 'fixed_ips': [{'ip_address': '35.4.1.4',
+ 'subnet_id': _uuid()}],
+ 'mac_address': 'ca:fe:de:ad:be:ef',
+ 'subnet': {'cidr': '35.4.1.0/24',
+ 'gateway_ip': '35.4.1.1'}})
+
+ router[l3_constants.INTERFACE_KEY].append(
+ {'id': _uuid(),
+ 'network_id': _uuid(),
+ 'admin_state_up': True,
+ 'fixed_ips': [{'ip_address': 'fd00::2',
+ 'subnet_id': _uuid()}],
+ 'mac_address': 'ca:fe:de:ad:be:ef',
+ 'subnet': {'cidr': 'fd00::/64',
+ 'gateway_ip': 'fd00::1'}})
+ # Reassign the router object to RouterInfo
+ ri.router = router
+ agent.process_router(ri)
+ # For some reason set logic does not work well with
+ # IpTablesRule instances
+ nat_rules_delta = [r for r in ri.iptables_manager.ipv4['nat'].rules
+ if r not in orig_nat_rules]
+ self.assertEqual(1, len(nat_rules_delta))
+ self._verify_snat_rules(nat_rules_delta, router)
+
def test_process_router_interface_removed(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = self._prepare_router_data(num_internal_ports=2)