201 lines
8.1 KiB
Python
201 lines
8.1 KiB
Python
# Copyright 2020 Red Hat, Inc.
|
|
# All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
from neutron_lib import constants as lib_constants
|
|
from oslo_log import log
|
|
from paramiko import ssh_exception as ssh_exc
|
|
from tempest.common import utils as tempest_utils
|
|
from tempest.lib.common.utils import data_utils
|
|
from tempest.lib import decorators
|
|
from tempest.lib import exceptions as lib_exc
|
|
import testtools
|
|
|
|
from neutron_tempest_plugin.common import ip
|
|
from neutron_tempest_plugin.common import ssh
|
|
from neutron_tempest_plugin.common import utils
|
|
from neutron_tempest_plugin import config
|
|
from neutron_tempest_plugin.scenario import base
|
|
|
|
CONF = config.CONF
|
|
|
|
LOG = log.getLogger(__name__)
|
|
|
|
|
|
def turn_nic6_on(ssh, ipv6_port):
|
|
"""Turns the IPv6 vNIC on
|
|
|
|
Required because guest images usually set only the first vNIC on boot.
|
|
Searches for the IPv6 vNIC's MAC and brings it up.
|
|
|
|
@param ssh: RemoteClient ssh instance to server
|
|
@param ipv6_port: port from IPv6 network attached to the server
|
|
"""
|
|
ip_command = ip.IPCommand(ssh)
|
|
nic = ip_command.get_nic_name_by_mac(ipv6_port['mac_address'])
|
|
ip_command.set_link(nic, "up")
|
|
|
|
|
|
def configure_eth_connection_profile_NM(ssh):
|
|
"""Prepare a Network manager profile for ipv6 port
|
|
|
|
By default the NetworkManager uses IPv6 privacy
|
|
format it isn't supported by neutron then we create
|
|
a ether profile with eui64 supported format
|
|
|
|
@param ssh: RemoteClient ssh instance to server
|
|
"""
|
|
# NOTE(ccamposr): on RHEL based OS we need a ether profile with
|
|
# eui64 format
|
|
if nmcli_command_exists(ssh):
|
|
try:
|
|
ssh.execute_script('nmcli connection add type ethernet con-name '
|
|
'ether ifname "*"', become_root=True)
|
|
ssh.execute_script('nmcli con mod ether ipv6.addr-gen-mode eui64',
|
|
become_root=True)
|
|
|
|
except lib_exc.SSHExecCommandFailed as e:
|
|
# NOTE(slaweq): Sometimes it can happen that this SSH command
|
|
# will fail because of some error from network manager in
|
|
# guest os.
|
|
# But even then doing ip link set up below is fine and
|
|
# IP address should be configured properly.
|
|
LOG.debug("Error creating NetworkManager profile. "
|
|
"Error message: %(error)s",
|
|
{'error': e})
|
|
|
|
|
|
def nmcli_command_exists(ssh):
|
|
return "False" not in ssh.execute_script(
|
|
'if ! type nmcli > /dev/null ; then echo "False"; fi')
|
|
|
|
|
|
class IPv6Test(base.BaseTempestTestCase):
|
|
credentials = ['primary', 'admin']
|
|
|
|
ipv6_ra_mode = 'slaac'
|
|
ipv6_address_mode = 'slaac'
|
|
|
|
@classmethod
|
|
def skip_checks(cls):
|
|
super(IPv6Test, cls).skip_checks()
|
|
if not CONF.network_feature_enabled.ipv6:
|
|
raise cls.skipException("IPv6 is not enabled")
|
|
|
|
@classmethod
|
|
@tempest_utils.requires_ext(extension="router", service="network")
|
|
def resource_setup(cls):
|
|
super(IPv6Test, cls).resource_setup()
|
|
cls.reserve_external_subnet_cidrs()
|
|
cls._setup_basic_resources()
|
|
|
|
@classmethod
|
|
def _setup_basic_resources(cls):
|
|
cls.network = cls.create_network()
|
|
cls.subnet = cls.create_subnet(cls.network)
|
|
cls.router = cls.create_router_by_client()
|
|
cls.create_router_interface(cls.router['id'], cls.subnet['id'])
|
|
cls.keypair = cls.create_keypair()
|
|
cls.secgroup = cls.create_security_group(
|
|
name=data_utils.rand_name('secgroup'))
|
|
cls.create_loginable_secgroup_rule(secgroup_id=cls.secgroup['id'])
|
|
cls.create_pingable_secgroup_rule(secgroup_id=cls.secgroup['id'])
|
|
|
|
def _test_ipv6_address_configured(self, ssh_client, vm, ipv6_port):
|
|
ipv6_address = ipv6_port['fixed_ips'][0]['ip_address']
|
|
ip_command = ip.IPCommand(ssh_client)
|
|
|
|
def guest_has_address(expected_address):
|
|
ip_addresses = [a.address for a in ip_command.list_addresses()]
|
|
for ip_address in ip_addresses:
|
|
if expected_address in ip_address:
|
|
return True
|
|
return False
|
|
|
|
try:
|
|
# Set NIC with IPv6 to be UP and wait until IPv6 address will be
|
|
# configured on this NIC
|
|
turn_nic6_on(ssh_client, ipv6_port)
|
|
# And check if IPv6 address will be properly configured on this NIC
|
|
utils.wait_until_true(
|
|
lambda: guest_has_address(ipv6_address),
|
|
timeout=120,
|
|
exception=RuntimeError(
|
|
"Timed out waiting for IP address {!r} to be configured "
|
|
"in the VM {!r}.".format(ipv6_address, vm['id'])))
|
|
except (lib_exc.SSHTimeout, ssh_exc.AuthenticationException) as ssh_e:
|
|
LOG.debug(ssh_e)
|
|
self._log_console_output([vm])
|
|
self._log_local_network_status()
|
|
raise
|
|
|
|
def _test_ipv6_hotplug(self, ra_mode, address_mode):
|
|
ipv6_networks = [self.create_network() for _ in range(2)]
|
|
for net in ipv6_networks:
|
|
subnet = self.create_subnet(
|
|
network=net, ip_version=6,
|
|
ipv6_ra_mode=ra_mode, ipv6_address_mode=address_mode)
|
|
self.create_router_interface(self.router['id'], subnet['id'])
|
|
|
|
server_kwargs = {
|
|
'flavor_ref': CONF.compute.flavor_ref,
|
|
'image_ref': CONF.compute.image_ref,
|
|
'key_name': self.keypair['name'],
|
|
'networks': [
|
|
{'uuid': self.network['id']},
|
|
{'uuid': ipv6_networks[0]['id']}],
|
|
'security_groups': [{'name': self.secgroup['name']}],
|
|
}
|
|
vm = self.create_server(**server_kwargs)['server']
|
|
self.wait_for_server_active(vm)
|
|
self.wait_for_guest_os_ready(vm)
|
|
ipv4_port = self.client.list_ports(
|
|
network_id=self.network['id'],
|
|
device_id=vm['id'])['ports'][0]
|
|
fip = self.create_floatingip(port=ipv4_port)
|
|
ssh_client = ssh.Client(
|
|
fip['floating_ip_address'], CONF.validation.image_ssh_user,
|
|
pkey=self.keypair['private_key'])
|
|
|
|
ipv6_port = self.client.list_ports(
|
|
network_id=ipv6_networks[0]['id'],
|
|
device_id=vm['id'])['ports'][0]
|
|
self._test_ipv6_address_configured(ssh_client, vm, ipv6_port)
|
|
|
|
# Now remove this port IPv6 port from the VM and attach new one
|
|
self.delete_interface(vm['id'], ipv6_port['id'])
|
|
|
|
# And plug VM to the second IPv6 network
|
|
ipv6_port = self.create_port(ipv6_networks[1])
|
|
# Add NetworkManager profile with ipv6 eui64 format to guest OS
|
|
configure_eth_connection_profile_NM(ssh_client)
|
|
self.create_interface(vm['id'], ipv6_port['id'])
|
|
ip.wait_for_interface_status(
|
|
self.os_primary.interfaces_client, vm['id'],
|
|
ipv6_port['id'], lib_constants.PORT_STATUS_ACTIVE,
|
|
ssh_client=ssh_client, mac_address=ipv6_port['mac_address'])
|
|
self._test_ipv6_address_configured(ssh_client, vm, ipv6_port)
|
|
|
|
@testtools.skipUnless(CONF.network_feature_enabled.ipv6_subnet_attributes,
|
|
"DHCPv6 attributes are not enabled.")
|
|
@decorators.idempotent_id('b13e5408-5250-4a42-8e46-6996ce613e91')
|
|
def test_ipv6_hotplug_slaac(self):
|
|
self._test_ipv6_hotplug("slaac", "slaac")
|
|
|
|
@testtools.skipUnless(CONF.network_feature_enabled.ipv6_subnet_attributes,
|
|
"DHCPv6 attributes are not enabled.")
|
|
@decorators.idempotent_id('9aaedbc4-986d-42d5-9177-3e721728e7e0')
|
|
def test_ipv6_hotplug_dhcpv6stateless(self):
|
|
self._test_ipv6_hotplug("dhcpv6-stateless", "dhcpv6-stateless")
|