# Verify that keepalived config is properly updated.
verify_ip_in_keepalived_config(router, internal_iface)
- def test_ha_router_ipv6_radvd_status(self):
- router_info = self.generate_router_info(ip_version=6, enable_ha=True)
- router1 = self.manage_router(self.agent, router_info)
- utils.wait_until_true(lambda: router1.ha_state == 'master')
- utils.wait_until_true(lambda: router1.radvd.enabled)
-
- def _check_lla_status(router, expected):
- internal_devices = router.router[l3_constants.INTERFACE_KEY]
- for device in internal_devices:
- lladdr = ip_lib.get_ipv6_lladdr(device['mac_address'])
- exists = ip_lib.device_exists_with_ips_and_mac(
- router.get_internal_device_name(device['id']), [lladdr],
- device['mac_address'], router.ns_name)
- self.assertEqual(expected, exists)
-
- _check_lla_status(router1, True)
-
- device_name = router1.get_ha_device_name()
- ha_device = ip_lib.IPDevice(device_name, namespace=router1.ns_name)
- ha_device.link.set_down()
-
- utils.wait_until_true(lambda: router1.ha_state == 'backup')
- utils.wait_until_true(lambda: not router1.radvd.enabled, timeout=10)
- _check_lla_status(router1, False)
-
- def test_ha_router_process_ipv6_subnets_to_existing_port(self):
- router_info = self.generate_router_info(enable_ha=True, ip_version=6)
- router = self.manage_router(self.agent, router_info)
-
- def verify_ip_in_keepalived_config(router, iface):
- config = router.keepalived_manager.config.get_config_str()
- ip_cidrs = common_utils.fixed_ip_cidrs(iface['fixed_ips'])
- for ip_addr in ip_cidrs:
- self.assertIn(ip_addr, config)
-
- interface_id = router.router[l3_constants.INTERFACE_KEY][0]['id']
- slaac = l3_constants.IPV6_SLAAC
- slaac_mode = {'ra_mode': slaac, 'address_mode': slaac}
-
- # Add a second IPv6 subnet to the router internal interface.
- self._add_internal_interface_by_subnet(router.router, count=1,
- ip_version=6, ipv6_subnet_modes=[slaac_mode],
- interface_id=interface_id)
- router.process(self.agent)
- utils.wait_until_true(lambda: router.ha_state == 'master')
-
- # Verify that router internal interface is present and is configured
- # with IP address from both the subnets.
- internal_iface = router.router[l3_constants.INTERFACE_KEY][0]
- self.assertEqual(2, len(internal_iface['fixed_ips']))
- self._assert_internal_devices(router)
-
- # Verify that keepalived config is properly updated.
- verify_ip_in_keepalived_config(router, internal_iface)
-
- # Remove one subnet from the router internal iface
- interfaces = copy.deepcopy(router.router.get(
- l3_constants.INTERFACE_KEY, []))
- fixed_ips, subnets = [], []
- fixed_ips.append(interfaces[0]['fixed_ips'][0])
- subnets.append(interfaces[0]['subnets'][0])
- interfaces[0].update({'fixed_ips': fixed_ips, 'subnets': subnets})
- router.router[l3_constants.INTERFACE_KEY] = interfaces
- router.process(self.agent)
-
- # Verify that router internal interface has a single ipaddress
- internal_iface = router.router[l3_constants.INTERFACE_KEY][0]
- self.assertEqual(1, len(internal_iface['fixed_ips']))
- self._assert_internal_devices(router)
-
- # Verify that keepalived config is properly updated.
- verify_ip_in_keepalived_config(router, internal_iface)
-
class MetadataFakeProxyHandler(object):
self.dst_p.addr.wait_until_address_ready(self.dst_addr)
net_helpers.assert_ping(self.src_namespace, self.dst_addr, count=2)
- def test_arp_spoof_doesnt_block_ipv6(self):
- self.src_addr = '2000::1'
- self.dst_addr = '2000::2'
- self._setup_arp_spoof_for_port(self.src_p.name, [self.src_addr])
- self._setup_arp_spoof_for_port(self.dst_p.name, [self.dst_addr])
- self.src_p.addr.add('%s/64' % self.src_addr)
- self.dst_p.addr.add('%s/64' % self.dst_addr)
- # IPv6 addresses seem to take longer to initialize
- pinger = helpers.Pinger(self.src_ns, max_attempts=4)
- pinger.assert_ping(self.dst_addr)
-
def test_arp_spoof_blocks_response(self):
# this will prevent the destination from responding to the ARP
# request for it's own address
except RuntimeError:
pass
- def test_arp_spoof_blocks_request(self):
- # this will prevent the source from sending an ARP
- # request with its own address
- self._setup_arp_spoof_for_port(self.src_p.name, ['192.168.0.3'])
- self.src_p.addr.add('%s/24' % self.src_addr)
- self.dst_p.addr.add('%s/24' % self.dst_addr)
- ns_ip_wrapper = ip_lib.IPWrapper(self.src_ns)
- try:
- ns_ip_wrapper.netns.execute(['arping', '-I', self.src_p.name,
- '-c1', self.dst_addr])
- tools.fail("arping should have failed. The arp request should "
- "have been blocked.")
- except RuntimeError:
- pass
-
def test_arp_spoof_allowed_address_pairs(self):
self._setup_arp_spoof_for_port(self.dst_p.name, ['192.168.0.3',
self.dst_addr])
Code Review / openstack-build / neutron-build.git / commitdiff