ctx,
fake_firewall_list[0]['id'])
- def _prepare_router_data(self, use_namespaces):
+ def _prepare_router_data(self):
router = {'id': str(uuid.uuid4()), 'tenant_id': str(uuid.uuid4())}
ns = "ns-" + router['id']
return l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- use_namespaces, router=router, ns_name=ns)
+ router=router, ns_name=ns)
- def _get_router_info_list_with_namespace_helper(self,
- router_use_namespaces):
- self.conf.set_override('use_namespaces', True)
- ri = self._prepare_router_data(
- use_namespaces=router_use_namespaces)
+ def _get_router_info_list_helper(self, use_namespaces):
+ self.conf.set_override('use_namespaces', use_namespaces)
+ ri = self._prepare_router_data()
routers = [ri.router]
self.api.router_info = {ri.router_id: ri}
with mock.patch.object(ip_lib.IPWrapper,
'get_namespaces') as mock_get_namespaces:
- mock_get_namespaces.return_value = ri.ns_name
+ mock_get_namespaces.return_value = []
router_info_list = self.api._get_router_info_list_for_tenant(
routers,
ri.router['tenant_id'])
- self.assertEqual([ri], router_info_list)
- mock_get_namespaces.assert_called_once_with(
- self.conf.root_helper)
-
- def _get_router_info_list_without_namespace_helper(self,
- router_use_namespaces):
- self.conf.set_override('use_namespaces', False)
- ri = self._prepare_router_data(
- use_namespaces=router_use_namespaces)
- routers = [ri.router]
- self.api.router_info = {ri.router_id: ri}
- router_info_list = self.api._get_router_info_list_for_tenant(
- routers,
- ri.router['tenant_id'])
- if router_use_namespaces:
+ if use_namespaces:
+ mock_get_namespaces.assert_called_once_with(self.conf.root_helper)
self.assertFalse(router_info_list)
else:
self.assertEqual([ri], router_info_list)
- def test_get_router_info_list_for_tenant_for_namespaces_enabled(self):
- self._get_router_info_list_with_namespace_helper(
- router_use_namespaces=True)
-
def test_get_router_info_list_for_tenant_for_namespaces_disabled(self):
- self._get_router_info_list_without_namespace_helper(
- router_use_namespaces=False)
-
- def test_get_router_info_list_tenant_with_namespace_router_without(self):
- self._get_router_info_list_with_namespace_helper(
- router_use_namespaces=False)
+ self._get_router_info_list_helper(use_namespaces=False)
- def test_get_router_info_list_tenant_without_namespace_router_with(self):
- self._get_router_info_list_without_namespace_helper(
- router_use_namespaces=True)
+ def test_get_router_info_list_for_tenant(self):
+ self._get_router_info_list_helper(use_namespaces=True)
def _get_router_info_list_router_without_router_info_helper(self,
rtr_with_ri):
self.conf.set_override('use_namespaces', True)
# ri.router with associated router_info (ri)
# rtr2 has no router_info
- ri = self._prepare_router_data(use_namespaces=True)
+ ri = self._prepare_router_data()
rtr2 = {'id': str(uuid.uuid4()), 'tenant_id': ri.router['tenant_id']}
routers = [rtr2]
self.api.router_info = {}
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)
+ router=router)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
cidr = '99.0.1.9/24'
mac = 'ca:fe:de:ad:be:ef'
def test_router_info_create(self):
id = _uuid()
ns = "ns-" + id
- ri = l3_agent.RouterInfo(id, self.conf.root_helper,
- self.conf.use_namespaces, {}, ns_name=ns)
+ ri = l3_agent.RouterInfo(id, self.conf.root_helper, {}, ns_name=ns)
self.assertTrue(ri.ns_name.endswith(id))
'routes': [],
'gw_port': ex_gw_port}
ns = "ns-" + id
- ri = l3_agent.RouterInfo(id, self.conf.root_helper,
- self.conf.use_namespaces, router, ns_name=ns)
+ ri = l3_agent.RouterInfo(id, self.conf.root_helper, router, ns_name=ns)
self.assertTrue(ri.ns_name.endswith(id))
self.assertEqual(ri.router, router)
def _test_external_gateway_action(self, action, router):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router,
+ router=router,
ns_name=agent.get_ns_name(router['id']))
# Special setup for dvr routers
if router.get('distributed'):
router = prepare_router_data(num_internal_ports=2)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router,
+ router=router,
ns_name=agent.get_ns_name(router['id']))
interface_name, ex_gw_port = self._prepare_ext_gw_test(agent)
agent_mode, expected_call_count):
router = prepare_router_data(num_internal_ports=2)
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
interface_name, ex_gw_port = self._prepare_ext_gw_test(agent)
agent._external_gateway_added = mock.Mock()
self.conf.set_override('use_namespaces', False)
router_id = _uuid()
- ri = l3_agent.RouterInfo(router_id, self.conf.root_helper,
- self.conf.use_namespaces, {})
+ ri = l3_agent.RouterInfo(router_id, self.conf.root_helper, {})
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
floating_ip = '20.0.0.101'
interface_name = agent.get_external_device_name(router_id)
self.conf.set_override('use_namespaces', False)
router_id = _uuid()
- ri = l3_agent.RouterInfo(router_id, self.conf.root_helper,
- self.conf.use_namespaces,
- {})
+ ri = l3_agent.RouterInfo(router_id, self.conf.root_helper, {})
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
fake_route1 = {'destination': '135.207.0.0/16',
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router_id = _uuid()
- ri = l3_agent.RouterInfo(router_id, self.conf.root_helper,
- self.conf.use_namespaces,
- {})
+ ri = l3_agent.RouterInfo(router_id, self.conf.root_helper, {})
ri.router = {}
fake_old_routes = []
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data(num_internal_ports=4)
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
test_port = {
'mac_address': '00:12:23:34:45:56',
'fixed_ips': [{'subnet_id': _get_subnet_id(
subnet_ids = [_get_subnet_id(port) for port in
router[l3_constants.INTERFACE_KEY]]
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
# Test Basic cases
port = agent.get_internal_port(ri, subnet_ids[0])
router = prepare_router_data(num_internal_ports=2)
router['distributed'] = True
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
ports = ri.router.get(l3_constants.INTERFACE_KEY, [])
test_ports = [{'mac_address': '00:11:22:33:44:55',
'device_owner': 'network:dhcp',
def test__update_arp_entry_with_no_subnet(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
ri = l3_agent.RouterInfo(
- 'foo_router_id', mock.ANY, True,
+ 'foo_router_id', mock.ANY,
{'distributed': True, 'gw_port_host': HOSTNAME})
with mock.patch.object(l3_agent.ip_lib, 'IPDevice') as f:
agent._update_arp_entry(ri, mock.ANY, mock.ANY,
def test_process_cent_router(self):
router = prepare_router_data()
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
self._test_process_router(ri)
def test_process_dist_router(self):
router = prepare_router_data()
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
subnet_id = _get_subnet_id(router[l3_constants.INTERFACE_KEY][0])
ri.router['distributed'] = True
ri.router['_snat_router_interfaces'] = [{
{'destination': '8.8.8.8/32', 'nexthop': '35.4.0.10'},
{'destination': '8.8.4.4/32', 'nexthop': '35.4.0.11'}]
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
ri.router = router
with contextlib.nested(mock.patch.object(agent,
'_spawn_metadata_proxy'),
router = prepare_router_data(enable_snat=True)
router[l3_constants.FLOATINGIP_KEY] = fake_floatingips['floatingips']
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
ri.iptables_manager.ipv4['nat'] = mock.MagicMock()
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
self._test_process_router_floating_ip_addresses_add(ri, agent)
router[l3_constants.FLOATINGIP_KEY] = fake_floatingips['floatingips']
router['distributed'] = True
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
ri.iptables_manager.ipv4['nat'] = mock.MagicMock()
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
agent.host = HOSTNAME
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data(enable_snat=True)
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
# Process with NAT
agent.process_router(ri)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data(enable_snat=False)
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
# Process without NAT
agent.process_router(ri)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data()
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
# Process with NAT
agent.process_router(ri)
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)
+ 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)
+ router=router)
with mock.patch.object(
agent,
'external_gateway_nat_rules') as external_gateway_nat_rules:
self, router, ra_mode=None, addr_mode=None):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
# Process with NAT
agent.process_router(ri)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data()
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
# Process with NAT
agent.process_router(ri)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data(num_internal_ports=2)
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
# Process with NAT
agent.process_router(ri)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data()
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
ri.router = router
agent.process_router(ri)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data()
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
with mock.patch.object(
l3_agent.L3NATAgent,
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data()
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
# add an internal port
agent.process_router(ri)
'port_id': router[l3_constants.INTERFACE_KEY][0]['id']}]
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
agent.process_router(ri)
# Assess the call for putting the floating IP up was performed
'port_id': router[l3_constants.INTERFACE_KEY][0]['id']}]
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent.external_gateway_added = mock.Mock()
agent.process_router(ri)
# Assess the call for putting the floating IP into Error
def test_handle_router_snat_rules_distributed_without_snat_manager(self):
ri = l3_agent.RouterInfo(
- 'foo_router_id', mock.ANY, True, {'distributed': True})
+ 'foo_router_id', mock.ANY, {'distributed': True})
ri.iptables_manager = mock.Mock()
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
def test_handle_router_snat_rules_add_rules(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
- ri = l3_agent.RouterInfo(_uuid(), self.conf.root_helper,
- self.conf.use_namespaces, {})
+ ri = l3_agent.RouterInfo(_uuid(), self.conf.root_helper, {})
ex_gw_port = {'fixed_ips': [{'ip_address': '192.168.1.4'}]}
internal_cidrs = ['10.0.0.0/24']
ri.router = {'distributed': False}
router = prepare_router_data(enable_snat=True, num_internal_ports=1)
ri = l3_agent.RouterInfo(router['id'],
self.conf.root_helper,
- self.conf.use_namespaces,
router=router)
internal_ports = ri.router.get(l3_constants.INTERFACE_KEY, [])
del router['gw_port']
ri = l3_agent.RouterInfo(router['id'],
self.conf.root_helper,
- self.conf.use_namespaces,
router=router)
self.mock_ip.get_devices.return_value = stale_devlist
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data()
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
port_id = _uuid()
dvr_gw_port = {'fixed_ips': [{'ip_address': '20.0.0.30',
'port_id': _uuid()}
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
rtr_2_fip_name = agent.get_rtr_int_device_name(ri.router_id)
fip_2_rtr_name = agent.get_fip_int_device_name(ri.router_id)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router = prepare_router_data()
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
agent_gw_port = {'fixed_ips': [{'ip_address': '20.0.0.30',
'subnet_id': _uuid()}],
'subnet': {'gateway_ip': '20.0.0.1'},
fip_cidr = '11.22.33.44/24'
ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
- self.conf.use_namespaces, router=router)
+ router=router)
ri.dist_fip_count = 2
ri.floating_ips_dict['11.22.33.44'] = FIP_PRI
ri.fip_2_rtr = '11.22.33.42'
cfg.CONF.set_override('debug', True)
self.external_process_p.stop()
- ri = l3_agent.RouterInfo(router_id, None, True, None)
+ ri = l3_agent.RouterInfo(router_id, None, None)
try:
with mock.patch(ip_class_path) as ip_mock:
self.agent._spawn_metadata_proxy(ri.router_id, ri.ns_name)