Merge "[Ipv6] Adding Allowed address pair tests"
This commit is contained in:
Zuul
Gerrit Code Review
commit
a90110a98c
@ -99,6 +99,18 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
|
||||
client.delete_subnet, subnet['id'])
|
||||
return subnet
|
||||
|
||||
def _create_subnet_v6(self, network, cidr, subnets_client=None, **kwargs):
|
||||
client = subnets_client or self.subnets_client
|
||||
body = client.create_subnet(
|
||||
name=data_utils.rand_name('ipv6_subnet-default1'),
|
||||
network_id=network['id'], tenant_id=network['tenant_id'],
|
||||
cidr=cidr, ip_version=6, ipv6_ra_mode='slaac',
|
||||
ipv6_address_mode='slaac', **kwargs)
|
||||
subnet = body.get('subnet', body)
|
||||
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
|
||||
client.delete_subnet, subnet['id'])
|
||||
return subnet
|
||||
|
||||
def _list_ports(self, *args, **kwargs):
|
||||
"""List ports using admin creds """
|
||||
ports_list = self.admin_manager.ports_client.list_ports(
|
||||
@ -117,6 +129,20 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
|
||||
self.assertIsNotNone(port_id, "Failed to find Instance's port id!!!")
|
||||
return port_id
|
||||
|
||||
def get_port_ipv4v6_id(self, network_id, subnet_id, instance):
|
||||
_, instance_addr = instance["addresses"].items()[0]
|
||||
for addr in instance_addr:
|
||||
if addr['version'] == 4:
|
||||
instance_fixed_ip = addr["addr"]
|
||||
for port in self._list_ports():
|
||||
port_fixed_ip = port["fixed_ips"][0]["ip_address"]
|
||||
if port["network_id"] == network_id and port["fixed_ips"][0][
|
||||
"subnet_id"] == subnet_id and instance["id"] == port[
|
||||
"device_id"] and port_fixed_ip == instance_fixed_ip:
|
||||
port_id = port["id"]
|
||||
self.assertIsNotNone(port_id, "Failed to find Instance's port id!!!")
|
||||
return port_id
|
||||
|
||||
def _create_server(self, name, network, image_id=None):
|
||||
keypair = self.create_keypair()
|
||||
self.keypairs[keypair['name']] = keypair
|
||||
@ -165,6 +191,49 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
|
||||
subnet=self.subnet, router=self.router)
|
||||
return networks
|
||||
|
||||
def create_ipv6_network_topo(self):
|
||||
self.security_group = self._create_security_group()
|
||||
self.network = self._create_network()
|
||||
rulesets = [
|
||||
dict(
|
||||
direction='ingress',
|
||||
protocol='icmpv6',
|
||||
ethertype='IPv6',
|
||||
),
|
||||
dict(
|
||||
direction='egress',
|
||||
protocol='icmpv6',
|
||||
ethertype='IPv6',
|
||||
)
|
||||
]
|
||||
rules = []
|
||||
for ruleset in rulesets:
|
||||
rules.append(
|
||||
self._create_security_group_rule(
|
||||
tenant_id=self.network['tenant_id'],
|
||||
secgroup=self.security_group,
|
||||
**ruleset))
|
||||
self.subnet = self._create_subnet(self.network,
|
||||
cidr='14.168.1.0/24')
|
||||
self.subnet_v6 = self._create_subnet_v6(self.network,
|
||||
cidr='3010::/64')
|
||||
self.router = self._create_router(
|
||||
router_name=data_utils.rand_name('router-default1'),
|
||||
external_network_id=CONF.network.public_network_id)
|
||||
self.routers_client.add_router_interface(
|
||||
self.router['id'], subnet_id=self.subnet['id'])
|
||||
self.routers_client.add_router_interface(
|
||||
self.router['id'], subnet_id=self.subnet_v6['id'])
|
||||
self.addCleanup(self.routers_client.remove_router_interface,
|
||||
self.router['id'], subnet_id=self.subnet['id'])
|
||||
self.addCleanup(self.routers_client.remove_router_interface,
|
||||
self.router['id'], subnet_id=self.subnet_v6['id'])
|
||||
networks = dict(security_group=self.security_group,
|
||||
network=self.network,
|
||||
subnet=self.subnet, subnet_v6=self.subnet_v6,
|
||||
router=self.router)
|
||||
return networks
|
||||
|
||||
def _check_server_connectivity(self, floating_ip,
|
||||
remote_ip, private_key,
|
||||
should_connect=True):
|
||||
@ -182,8 +251,12 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
|
||||
raise
|
||||
|
||||
def _assign_ip_address(self, ssh_source, interface_name, ip_address):
|
||||
ssh_source.exec_command("sudo ifconfig %s %s/24 up" % (interface_name,
|
||||
ip_address))
|
||||
ssh_source.exec_command("sudo ifconfig %s %s netmask 255.255.255.0 \
|
||||
up" % (interface_name, ip_address))
|
||||
|
||||
def _assign_ipv6_address(self, ssh_source, interface_name, ip_address):
|
||||
ssh_source.exec_command("sudo ip -6 addr add %s/64 dev %s" % (
|
||||
ip_address, interface_name))
|
||||
|
||||
def _assign_mac_address(self, ssh_source, interface_name, mac_address):
|
||||
ssh_source.exec_command("sudo ifconfig %s down" % interface_name)
|
||||
@ -239,6 +312,114 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
|
||||
(ssh_source, ip_address_vm2, 'True'),
|
||||
'Destination is reachable')
|
||||
|
||||
def _test_v6_connectivity_between_allowed_adddress_pair_ports(self,
|
||||
network_topo):
|
||||
server_name_default = data_utils.rand_name('server-default')
|
||||
network = network_topo['network']
|
||||
server_default = self._create_server(server_name_default, network)
|
||||
server_name_default1 = \
|
||||
data_utils.rand_name('server-default1-sec-group')
|
||||
server_default1 = self._create_server(server_name_default1, network)
|
||||
floating_ip_default = self.create_floating_ip(server_default)
|
||||
floating_ip_default1 = self.create_floating_ip(server_default1)
|
||||
ip_address_default_vm = floating_ip_default['floating_ip_address']
|
||||
ip_address_default1_vm = floating_ip_default1['floating_ip_address']
|
||||
private_key_default_vm = self._get_server_key(server_default)
|
||||
private_key_default1_vm = self._get_server_key(server_default1)
|
||||
port_client = self.ports_client
|
||||
# Allowed Address pair
|
||||
ip_address_vm1 = '3000:10:10::2'
|
||||
ip_address_vm2 = '3000:10:10::3'
|
||||
port_id = self.get_port_ipv4v6_id(network['id'],
|
||||
network_topo['subnet']['id'],
|
||||
server_default)
|
||||
# Update allowed address pair attribute of port
|
||||
allowed_address_pairs = [{'ip_address': ip_address_vm1}]
|
||||
port_client.update_port(
|
||||
port_id, allowed_address_pairs=allowed_address_pairs)
|
||||
port1_id = self.get_port_ipv4v6_id(network['id'],
|
||||
network_topo['subnet']['id'],
|
||||
server_default1)
|
||||
# Update allowed address pair attribute of port
|
||||
allowed_address_pairs = [{'ip_address': ip_address_vm2}]
|
||||
port_client.update_port(
|
||||
port1_id, allowed_address_pairs=allowed_address_pairs)
|
||||
ssh_source = self.get_remote_client(
|
||||
ip_address_default_vm,
|
||||
private_key=private_key_default_vm)
|
||||
ssh_source1 = self.get_remote_client(
|
||||
ip_address_default1_vm,
|
||||
private_key=private_key_default1_vm)
|
||||
# Assign Allowed pair ip to vm's
|
||||
self._assign_ipv6_address(ssh_source, 'eth0', ip_address_vm1)
|
||||
self._assign_ipv6_address(ssh_source1, 'eth0', ip_address_vm2)
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source1, ip_address_vm1, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source, ip_address_vm2, 'True'),
|
||||
'Destination is reachable')
|
||||
|
||||
def _test_v4_v6_connectivity_between_allowed_adddress_pair_ports(self,
|
||||
network_topo):
|
||||
server_name_default = data_utils.rand_name('server-default')
|
||||
network = network_topo['network']
|
||||
server_default = self._create_server(server_name_default, network)
|
||||
server_name_default1 = \
|
||||
data_utils.rand_name('server-default1-sec-group')
|
||||
server_default1 = self._create_server(server_name_default1, network)
|
||||
floating_ip_default = self.create_floating_ip(server_default)
|
||||
floating_ip_default1 = self.create_floating_ip(server_default1)
|
||||
ip_address_default_vm = floating_ip_default['floating_ip_address']
|
||||
ip_address_default1_vm = floating_ip_default1['floating_ip_address']
|
||||
private_key_default_vm = self._get_server_key(server_default)
|
||||
private_key_default1_vm = self._get_server_key(server_default1)
|
||||
port_client = self.ports_client
|
||||
# Allowed Address pair
|
||||
ipv4_address_vm1 = '87.0.0.3'
|
||||
ipv4_address_vm2 = '87.0.0.4'
|
||||
ipv6_address_vm1 = '3000:10:10::2'
|
||||
ipv6_address_vm2 = '3000:10:10::3'
|
||||
port_id = self.get_port_ipv4v6_id(network['id'],
|
||||
network_topo['subnet']['id'],
|
||||
server_default)
|
||||
# Update allowed address pair attribute of port
|
||||
allowed_address_pairs = [{'ip_address': ipv4_address_vm1},
|
||||
{'ip_address': ipv6_address_vm1}]
|
||||
port_client.update_port(
|
||||
port_id, allowed_address_pairs=allowed_address_pairs)
|
||||
port1_id = self.get_port_ipv4v6_id(network['id'],
|
||||
network_topo['subnet']['id'],
|
||||
server_default1)
|
||||
# Update allowed address pair attribute of port
|
||||
allowed_address_pairs = [{'ip_address': ipv4_address_vm2},
|
||||
{'ip_address': ipv6_address_vm2}]
|
||||
port_client.update_port(
|
||||
port1_id, allowed_address_pairs=allowed_address_pairs)
|
||||
ssh_source = self.get_remote_client(
|
||||
ip_address_default_vm,
|
||||
private_key=private_key_default_vm)
|
||||
ssh_source1 = self.get_remote_client(
|
||||
ip_address_default1_vm,
|
||||
private_key=private_key_default1_vm)
|
||||
# Assign Allowed pair ip to vm's
|
||||
self._assign_ipv6_address(ssh_source, 'eth0', ipv6_address_vm1)
|
||||
self._assign_ipv6_address(ssh_source1, 'eth0', ipv6_address_vm2)
|
||||
self._assign_ip_address(ssh_source, 'eth0:1', ipv4_address_vm1)
|
||||
self._assign_ip_address(ssh_source1, 'eth0:1', ipv4_address_vm2)
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source1, ipv4_address_vm1, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source, ipv4_address_vm2, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source1, ipv6_address_vm1, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source, ipv6_address_vm2, 'True'),
|
||||
'Destination is reachable')
|
||||
|
||||
def _test_allowed_adddress_pair_ports_attach_as_interface_on_vms(
|
||||
self, network_topo):
|
||||
server_name_default = data_utils.rand_name('server-default')
|
||||
@ -439,6 +620,148 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
|
||||
(ssh_source1, ip_address_vm1_2, 'True'),
|
||||
'Destination is reachable')
|
||||
|
||||
def _test_v6_allowed_address_pair_on_vms_with_multiple_ips(
|
||||
self, network_topo):
|
||||
server_name_default = data_utils.rand_name('server-default')
|
||||
network = network_topo['network']
|
||||
server_default = self._create_server(server_name_default, network)
|
||||
server_name_default1 = \
|
||||
data_utils.rand_name('server-default1-sec-group')
|
||||
server_default1 = self._create_server(server_name_default1, network)
|
||||
floating_ip_default = self.create_floating_ip(server_default)
|
||||
floating_ip_default1 = self.create_floating_ip(server_default1)
|
||||
ip_address_default_vm = floating_ip_default['floating_ip_address']
|
||||
ip_address_default1_vm = floating_ip_default1['floating_ip_address']
|
||||
private_key_default_vm = self._get_server_key(server_default)
|
||||
private_key_default1_vm = self._get_server_key(server_default1)
|
||||
port_client = self.ports_client
|
||||
# Allowed Address pair
|
||||
ip_address_vm1_1 = '3000:10:10::2'
|
||||
ip_address_vm1_2 = '4000:10:10::2'
|
||||
ip_address_vm2_1 = '3000:10:10::3'
|
||||
ip_address_vm2_2 = '4000:10:10::3'
|
||||
port_id = self.get_port_ipv4v6_id(network['id'],
|
||||
network_topo['subnet']['id'],
|
||||
server_default)
|
||||
# Update allowed address pair attribute of port
|
||||
allowed_address_pairs = [{'ip_address': ip_address_vm1_1},
|
||||
{'ip_address': ip_address_vm1_2}]
|
||||
port_client.update_port(
|
||||
port_id, allowed_address_pairs=allowed_address_pairs)
|
||||
port1_id = self.get_port_ipv4v6_id(network['id'],
|
||||
network_topo['subnet']['id'],
|
||||
server_default1)
|
||||
# Update allowed address pair attribute of port
|
||||
allowed_address_pairs = [{'ip_address': ip_address_vm2_1},
|
||||
{'ip_address': ip_address_vm2_2}]
|
||||
port_client.update_port(
|
||||
port1_id, allowed_address_pairs=allowed_address_pairs)
|
||||
ssh_source = self.get_remote_client(ip_address_default_vm,
|
||||
private_key=private_key_default_vm)
|
||||
ssh_source1 = self.get_remote_client(
|
||||
ip_address_default1_vm,
|
||||
private_key=private_key_default1_vm)
|
||||
# Attach allowed pair ip's to vm's
|
||||
self._assign_ipv6_address(ssh_source, 'eth0', ip_address_vm1_1)
|
||||
self._assign_ipv6_address(ssh_source, 'eth0', ip_address_vm1_2)
|
||||
self._assign_ipv6_address(ssh_source1, 'eth0', ip_address_vm2_1)
|
||||
self._assign_ipv6_address(ssh_source1, 'eth0', ip_address_vm2_2)
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source, ip_address_vm2_1, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source, ip_address_vm2_2, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source1, ip_address_vm1_1, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source1, ip_address_vm1_2, 'True'),
|
||||
'Destination is reachable')
|
||||
|
||||
def _test_v4v6_allowed_address_pair_on_vms_with_multiple_ips(
|
||||
self, network_topo):
|
||||
server_name_default = data_utils.rand_name('server-default')
|
||||
network = network_topo['network']
|
||||
server_default = self._create_server(server_name_default, network)
|
||||
server_name_default1 = \
|
||||
data_utils.rand_name('server-default1-sec-group')
|
||||
server_default1 = self._create_server(server_name_default1, network)
|
||||
floating_ip_default = self.create_floating_ip(server_default)
|
||||
floating_ip_default1 = self.create_floating_ip(server_default1)
|
||||
ip_address_default_vm = floating_ip_default['floating_ip_address']
|
||||
ip_address_default1_vm = floating_ip_default1['floating_ip_address']
|
||||
private_key_default_vm = self._get_server_key(server_default)
|
||||
private_key_default1_vm = self._get_server_key(server_default1)
|
||||
port_client = self.ports_client
|
||||
# Allowed Address pair
|
||||
ipv4_address_vm1_1 = '77.0.0.3'
|
||||
ipv4_address_vm1_2 = '78.0.0.3'
|
||||
ipv4_address_vm2_1 = '77.0.0.4'
|
||||
ipv4_address_vm2_2 = '78.0.0.4'
|
||||
ipv6_address_vm1_1 = '3000:10:10::2'
|
||||
ipv6_address_vm1_2 = '4000:10:10::2'
|
||||
ipv6_address_vm2_1 = '3000:10:10::3'
|
||||
ipv6_address_vm2_2 = '4000:10:10::3'
|
||||
port_id = self.get_port_ipv4v6_id(network['id'],
|
||||
network_topo['subnet']['id'],
|
||||
server_default)
|
||||
# Update allowed address pair attribute of port
|
||||
allowed_address_pairs = [{'ip_address': ipv4_address_vm1_1},
|
||||
{'ip_address': ipv4_address_vm1_2},
|
||||
{'ip_address': ipv6_address_vm1_1},
|
||||
{'ip_address': ipv6_address_vm1_2}]
|
||||
port_client.update_port(
|
||||
port_id, allowed_address_pairs=allowed_address_pairs)
|
||||
port1_id = self.get_port_ipv4v6_id(network['id'],
|
||||
network_topo['subnet']['id'],
|
||||
server_default1)
|
||||
# Update allowed address pair attribute of port
|
||||
allowed_address_pairs = [{'ip_address': ipv4_address_vm2_1},
|
||||
{'ip_address': ipv4_address_vm2_2},
|
||||
{'ip_address': ipv6_address_vm2_1},
|
||||
{'ip_address': ipv6_address_vm2_2}]
|
||||
port_client.update_port(
|
||||
port1_id, allowed_address_pairs=allowed_address_pairs)
|
||||
ssh_source = self.get_remote_client(ip_address_default_vm,
|
||||
private_key=private_key_default_vm)
|
||||
ssh_source1 = self.get_remote_client(
|
||||
ip_address_default1_vm,
|
||||
private_key=private_key_default1_vm)
|
||||
# Attach allowed pair ip's to vm's
|
||||
self._assign_ip_address(ssh_source, 'eth0:1', ipv4_address_vm1_1)
|
||||
self._assign_ip_address(ssh_source, 'eth0:2', ipv4_address_vm1_2)
|
||||
self._assign_ip_address(ssh_source1, 'eth0:1', ipv4_address_vm2_1)
|
||||
self._assign_ip_address(ssh_source1, 'eth0:2', ipv4_address_vm2_2)
|
||||
self._assign_ipv6_address(ssh_source, 'eth0', ipv6_address_vm1_1)
|
||||
self._assign_ipv6_address(ssh_source, 'eth0', ipv6_address_vm1_2)
|
||||
self._assign_ipv6_address(ssh_source1, 'eth0', ipv6_address_vm2_1)
|
||||
self._assign_ipv6_address(ssh_source1, 'eth0', ipv6_address_vm2_2)
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source, ipv4_address_vm1_1, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source, ipv4_address_vm1_2, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source, ipv6_address_vm1_1, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source, ipv6_address_vm1_2, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source1, ipv4_address_vm2_1, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source1, ipv4_address_vm2_2, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source1, ipv6_address_vm2_1, 'True'),
|
||||
'Destination is reachable')
|
||||
self.assertTrue(self._check_remote_connectivity
|
||||
(ssh_source1, ipv6_address_vm2_2, 'True'),
|
||||
'Destination is reachable')
|
||||
|
||||
def _test_vm_accessible_using_allowed_adddress_pair_port_fip(
|
||||
self, network_topo):
|
||||
server_name_default = data_utils.rand_name('server-default')
|
||||
@ -521,3 +844,31 @@ class TestAllowedAddressPair(manager.NetworkScenarioTest):
|
||||
self.network_topo = self.create_network_topo()
|
||||
self._test_vm_accessible_using_allowed_adddress_pair_port_fip(
|
||||
self.network_topo)
|
||||
|
||||
@decorators.attr(type='nsxv3')
|
||||
@decorators.idempotent_id('9f192ed4-eac1-4935-9d12-d0c7bb8b52c1')
|
||||
def test_ipv6_allowed_adddress_pair_on_vms_with_single_ip(self):
|
||||
self.network_topo = self.create_ipv6_network_topo()
|
||||
self._test_v6_connectivity_between_allowed_adddress_pair_ports(
|
||||
self.network_topo)
|
||||
|
||||
@decorators.attr(type='nsxv3')
|
||||
@decorators.idempotent_id('046f503a-4b0f-4335-afe3-05f03d0733b3')
|
||||
def test_ipv4_v6_allowed_adddress_pair_on_vms_with_single_ip(self):
|
||||
self.network_topo = self.create_ipv6_network_topo()
|
||||
self._test_v4_v6_connectivity_between_allowed_adddress_pair_ports(
|
||||
self.network_topo)
|
||||
|
||||
@decorators.attr(type='nsxv3')
|
||||
@decorators.idempotent_id('513ea6d5-4ec6-430e-a501-a6ffbab624f7')
|
||||
def test_ipv6_allowed_address_pair_on_vms_with_multiple_ips(self):
|
||||
self.network_topo = self.create_ipv6_network_topo()
|
||||
self._test_v6_allowed_address_pair_on_vms_with_multiple_ips(
|
||||
self.network_topo)
|
||||
|
||||
@decorators.attr(type='nsxv3')
|
||||
@decorators.idempotent_id('6374aeb5-20e7-4f1e-a73b-7080875a843e')
|
||||
def test_ipv4_v6_allowed_address_pair_on_vms_with_multiple_ips(self):
|
||||
self.network_topo = self.create_ipv6_network_topo()
|
||||
self._test_v4v6_allowed_address_pair_on_vms_with_multiple_ips(
|
||||
self.network_topo)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user