d4c707e68d
This patch adds 2 new scenario tests to test connectivity between VMs using vlan network with enabled QinQ and with vlan configured inside VMs. This is very similar to the "vlan_transparent" option which was already tested since some time. The only difference is in the ethtype of the outer vlan packets send through such network. Unfortunately we can't test ethtype of the outer vlan in the neutron-tempest-plugin tests as that would require access to the host where VMs runs and test packets on the physical NICs. So those new tests are actually the same as the existing tests for the network with "vlan_transparent" enabled to at least make sure that QinQ functionality works from the end user point of view. This patch also enables those new tests for the ML2/OVN job as this is currently only backend with implemented support for QinQ. Depends-On: https://review.opendev.org/c/openstack/neutron/+/937633 Related-Bug: #1915151 Change-Id: Icc2b175cf4cfe5c4ca73bb7ccc5a9089a27798c9
243 lines
9.8 KiB
Python
243 lines
9.8 KiB
Python
# Copyright (c) 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 oslo_log import log as logging
|
|
from tempest.lib.common.utils import data_utils
|
|
from tempest.lib import decorators
|
|
|
|
from neutron_tempest_plugin.common import ip
|
|
from neutron_tempest_plugin.common import ssh
|
|
from neutron_tempest_plugin import config
|
|
from neutron_tempest_plugin.scenario import base
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
CONF = config.CONF
|
|
MIN_VLAN_ID = 1
|
|
MAX_VLAN_ID = 4094
|
|
|
|
|
|
class BaseVlanTest(base.BaseAdminTempestTestCase):
|
|
|
|
"""Base class common for the tests for the "vlan_transparent" and "qinq".
|
|
|
|
This base class covers common things for the tests for networks with
|
|
enabled either `qinq` or `vlan_transparent` attributes. Those 2 attributes
|
|
are functionally the same even from the end user's point of view. The only
|
|
difference between them is ethtype used for the outer VLAN tag but this
|
|
can't really be tested in the neutron-tempest-plugin tests.
|
|
"""
|
|
|
|
network_kwargs = {}
|
|
|
|
@classmethod
|
|
def resource_setup(cls):
|
|
super(BaseVlanTest, cls).resource_setup()
|
|
# setup basic topology for servers we can log into
|
|
cls.rand_name = data_utils.rand_name(
|
|
cls.__name__.rsplit('.', 1)[-1])
|
|
cls.network = cls.create_network(name=cls.rand_name,
|
|
**cls.network_kwargs)
|
|
cls.subnet = cls.create_subnet(network=cls.network,
|
|
name=cls.rand_name)
|
|
cls.router = cls.create_router_by_client()
|
|
cls.create_router_interface(cls.router['id'], cls.subnet['id'])
|
|
cls.keypair = cls.create_keypair(name=cls.rand_name)
|
|
cls.vm_ports = []
|
|
cls.security_group = cls.create_security_group(name=cls.rand_name)
|
|
cls.create_loginable_secgroup_rule(cls.security_group['id'])
|
|
|
|
if CONF.neutron_plugin_options.default_image_is_advanced:
|
|
cls.flavor_ref = CONF.compute.flavor_ref
|
|
cls.image_ref = CONF.compute.image_ref
|
|
else:
|
|
cls.flavor_ref = \
|
|
CONF.neutron_plugin_options.advanced_image_flavor_ref
|
|
cls.image_ref = CONF.neutron_plugin_options.advanced_image_ref
|
|
|
|
@classmethod
|
|
def skip_checks(cls):
|
|
super().skip_checks()
|
|
if not (CONF.neutron_plugin_options.advanced_image_ref or
|
|
CONF.neutron_plugin_options.default_image_is_advanced):
|
|
raise cls.skipException(
|
|
'Advanced image is required to run these tests.')
|
|
|
|
def _create_port_and_server(self, index,
|
|
port_security=True,
|
|
allowed_address_pairs=None):
|
|
server_name = 'server-%s-%d' % (self.rand_name, index)
|
|
port_name = 'port-%s-%d' % (self.rand_name, index)
|
|
if port_security:
|
|
sec_groups = [self.security_group['id']]
|
|
else:
|
|
sec_groups = None
|
|
self.vm_ports.append(
|
|
self.create_port(network=self.network, name=port_name,
|
|
security_groups=sec_groups,
|
|
port_security_enabled=port_security,
|
|
allowed_address_pairs=allowed_address_pairs))
|
|
return self.create_server(flavor_ref=self.flavor_ref,
|
|
image_ref=self.image_ref,
|
|
key_name=self.keypair['name'],
|
|
networks=[{'port': self.vm_ports[-1]['id']}],
|
|
name=server_name)['server']
|
|
|
|
def _configure_inner_vlan(self, port, ssh_client, vlan_tag, vlan_ip):
|
|
ip_command = ip.IPCommand(ssh_client=ssh_client)
|
|
addresses = ip_command.list_addresses(port=port)
|
|
port_iface = ip.get_port_device_name(addresses, port)
|
|
subport_iface = ip_command.configure_inner_vlan(
|
|
port=port, vlan_tag=vlan_tag, ip_addresses=[vlan_ip])
|
|
|
|
for address in ip_command.list_addresses(ip_addresses=vlan_ip):
|
|
self.assertEqual(subport_iface, address.device.name)
|
|
self.assertEqual(port_iface, address.device.parent)
|
|
break
|
|
else:
|
|
self.fail("Sub-port fixed IP not found on server.")
|
|
|
|
def _create_ssh_client(self, floating_ip):
|
|
if CONF.neutron_plugin_options.default_image_is_advanced:
|
|
username = CONF.validation.image_ssh_user
|
|
else:
|
|
username = CONF.neutron_plugin_options.advanced_image_ssh_user
|
|
return ssh.Client(host=floating_ip['floating_ip_address'],
|
|
username=username,
|
|
pkey=self.keypair['private_key'])
|
|
|
|
def _test_basic_inner_vlan_connectivity(
|
|
self, port_security=True, use_allowed_address_pairs=False):
|
|
self._ensure_ethtype()
|
|
vlan_tag = data_utils.rand_int_id(start=MIN_VLAN_ID, end=MAX_VLAN_ID)
|
|
vlan_ipmask_template = '192.168.%d.{ip_last_byte}/24' % (vlan_tag %
|
|
256)
|
|
vms = []
|
|
vlan_ipmasks = []
|
|
floating_ips = []
|
|
ssh_clients = []
|
|
|
|
for i in range(2):
|
|
vlan_ipmasks.append(vlan_ipmask_template.format(
|
|
ip_last_byte=(i + 1) * 10))
|
|
if use_allowed_address_pairs:
|
|
allowed_address_pairs = [{'ip_address': vlan_ipmasks[i]}]
|
|
else:
|
|
allowed_address_pairs = None
|
|
vms.append(self._create_port_and_server(
|
|
index=i,
|
|
port_security=port_security,
|
|
allowed_address_pairs=allowed_address_pairs))
|
|
floating_ips.append(self.create_floatingip(port=self.vm_ports[-1]))
|
|
ssh_clients.append(
|
|
self._create_ssh_client(floating_ip=floating_ips[i]))
|
|
|
|
self.check_connectivity(ssh_client=ssh_clients[i])
|
|
self._configure_inner_vlan(port=self.vm_ports[-1],
|
|
ssh_client=ssh_clients[i],
|
|
vlan_tag=vlan_tag,
|
|
vlan_ip=vlan_ipmasks[i])
|
|
|
|
if port_security:
|
|
# Ping from vm0 to vm1 via VLAN interface should fail because
|
|
# we haven't allowed ICMP
|
|
self.check_remote_connectivity(
|
|
ssh_clients[0],
|
|
vlan_ipmasks[1].split('/')[0],
|
|
servers=vms,
|
|
should_succeed=False)
|
|
|
|
# allow intra-security-group traffic
|
|
sg_rule = self.create_pingable_secgroup_rule(
|
|
self.security_group['id'])
|
|
self.addCleanup(
|
|
self.os_primary.network_client.delete_security_group_rule,
|
|
sg_rule['id'])
|
|
|
|
# Ping from vm0 to vm1 via VLAN interface should pass because
|
|
# either port security is disabled or the ICMP sec group rule has been
|
|
# added
|
|
self.check_remote_connectivity(
|
|
ssh_clients[0],
|
|
vlan_ipmasks[1].split('/')[0],
|
|
servers=vms)
|
|
# Ping from vm1 to vm0 and check untagged packets are not dropped
|
|
self.check_remote_connectivity(
|
|
ssh_clients[1],
|
|
self.vm_ports[-2]['fixed_ips'][0]['ip_address'],
|
|
servers=vms)
|
|
|
|
|
|
class VlanTransparencyTest(BaseVlanTest):
|
|
credentials = ['primary', 'admin']
|
|
force_tenant_isolation = False
|
|
|
|
required_extensions = ['vlan-transparent', 'allowed-address-pairs']
|
|
|
|
network_kwargs = {'vlan_transparent': True}
|
|
|
|
def _ensure_ethtype(self):
|
|
self.assertTrue(self.network.get('vlan_transparent'))
|
|
self.assertFalse(self.network.get('qinq'))
|
|
|
|
@decorators.idempotent_id('a2694e3a-6d4d-4a23-9fcc-c3ed3ef37b16')
|
|
def test_vlan_transparent_port_sec_disabled(self):
|
|
self._test_basic_inner_vlan_connectivity(
|
|
port_security=False, use_allowed_address_pairs=False)
|
|
|
|
@decorators.idempotent_id('2dd03b4f-9c20-4cda-8c6a-40fa453ec69a')
|
|
def test_vlan_transparent_allowed_address_pairs(self):
|
|
self._test_basic_inner_vlan_connectivity(
|
|
port_security=True, use_allowed_address_pairs=True)
|
|
|
|
|
|
class VlanQinqTest(BaseVlanTest):
|
|
credentials = ['primary', 'admin']
|
|
force_tenant_isolation = False
|
|
|
|
required_extensions = ['qinq', 'allowed-address-pairs']
|
|
|
|
network_kwargs = {
|
|
'qinq': True,
|
|
'provider_network_type': 'vlan'}
|
|
|
|
@classmethod
|
|
def resource_setup(cls):
|
|
cls.network_kwargs['provider_physical_network'] = (
|
|
CONF.neutron_plugin_options.provider_vlans[0])
|
|
super().resource_setup()
|
|
|
|
@classmethod
|
|
def skip_checks(cls):
|
|
super().skip_checks()
|
|
if not CONF.neutron_plugin_options.provider_vlans:
|
|
raise cls.skipException(
|
|
'Physical network is required to run these tests.')
|
|
|
|
def _ensure_ethtype(self):
|
|
self.assertFalse(self.network.get('vlan_transparent'))
|
|
self.assertTrue(self.network.get('qinq'))
|
|
|
|
@decorators.idempotent_id('ae78398e-9242-46b4-a5fc-227581821fca')
|
|
def test_vlan_transparent_port_sec_disabled(self):
|
|
self._test_basic_inner_vlan_connectivity(
|
|
port_security=False, use_allowed_address_pairs=False)
|
|
|
|
@decorators.idempotent_id('6ca983cd-b1c5-4e2c-949e-4be8ffa22a9c')
|
|
def test_vlan_transparent_allowed_address_pairs(self):
|
|
self._test_basic_inner_vlan_connectivity(
|
|
port_security=True, use_allowed_address_pairs=True)
|