# 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.
import netaddr
from neutron_lib import constants
from oslo_log import log
from paramiko import ssh_exception as ssh_exc
from tempest.common import 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 ssh
from neutron_tempest_plugin.common import utils as neutron_utils
from neutron_tempest_plugin import config
from neutron_tempest_plugin.scenario import base
CONF = config.CONF
LOG = log.getLogger(__name__)
class DHCPTest(base.BaseTempestTestCase):
credentials = ['primary', 'admin']
force_tenant_isolation = False
@classmethod
def resource_setup(cls):
super(DHCPTest, cls).resource_setup()
cls.rand_name = data_utils.rand_name(
cls.__name__.rsplit('.', 1)[-1])
cls.network = cls.create_network(name=cls.rand_name)
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.security_group = cls.create_security_group(name=cls.rand_name)
cls.create_loginable_secgroup_rule(cls.security_group['id'])
@utils.requires_ext(extension='extra_dhcp_opt', service='network')
@decorators.idempotent_id('58f7c094-1980-4e03-b0d3-6c4dd27217b1')
def test_extra_dhcp_opts(self):
"""This test case tests DHCP extra options configured for Neutron port.
Test is checking just extra option "15" which is domain-name
according to the RFC 2132:
https://tools.ietf.org/html/rfc2132#section-5.3
To test that option, there is spawned VM connected to the port with
configured extra_dhcp_opts and test asserts that search domain name is
configured inside VM in /etc/resolv.conf file
"""
test_domain = "test.domain"
extra_dhcp_opts = [
{'opt_name': 'domain-name',
'opt_value': '"%s"' % test_domain}]
port = self.create_port(
network=self.network, name=self.rand_name,
security_groups=[self.security_group['id']],
extra_dhcp_opts=extra_dhcp_opts)
floating_ip = self.create_floatingip(port=port)
server = self.create_server(
flavor_ref=CONF.compute.flavor_ref,
image_ref=CONF.compute.image_ref,
key_name=self.keypair['name'],
networks=[{'port': port['id']}])
self.wait_for_server_active(server['server'])
self.wait_for_guest_os_ready(server['server'])
try:
ssh_client = ssh.Client(
floating_ip['floating_ip_address'],
CONF.validation.image_ssh_user,
pkey=self.keypair['private_key'])
vm_resolv_conf = ssh_client.exec_command(
"cat /etc/resolv.conf")
self.assertIn(test_domain, vm_resolv_conf)
except (lib_exc.SSHTimeout,
ssh_exc.AuthenticationException,
AssertionError) as error:
LOG.debug(error)
self._log_console_output([server])
self._log_local_network_status()
raise
class DHCPPortUpdateTest(base.BaseTempestTestCase):
credentials = ['primary', 'admin']
@classmethod
def resource_setup(cls):
super(DHCPPortUpdateTest, cls).resource_setup()
cls.rand_name = data_utils.rand_name(
cls.__name__.rsplit('.', 1)[-1])
cls.network = cls.create_network(name=cls.rand_name)
cls.router = cls.create_router_by_client()
cls.keypair = cls.create_keypair(name=cls.rand_name)
cls.security_group = cls.create_security_group(name=cls.rand_name)
cls.create_loginable_secgroup_rule(cls.security_group['id'])
cls.create_pingable_secgroup_rule(cls.security_group['id'])
@testtools.skipUnless(
CONF.neutron_plugin_options.firewall_driver == 'ovn',
"OVN driver is required to run this test - "
"LP#1942794 solution only applied to OVN")
@decorators.idempotent_id('8171cc68-9dbb-46ca-b065-17b5b2e26094')
def test_modify_dhcp_port_ip_address(self):
"""Test Scenario
1) Create a network and a subnet with DHCP enabled
2) Modify the default IP address from the subnet DHCP port
3) Create a server in this network and check ssh connectivity
For the step 3), the server needs to obtain ssh keys from the metadata
Related bug: LP#1942794
"""
# create subnet (dhcp is enabled by default)
subnet = self.create_subnet(network=self.network, name=self.rand_name)
def _get_dhcp_ports():
# in some cases, like ML2/OVS, the subnet port associated to DHCP
# is created with device_owner='network:dhcp'
dhcp_ports = self.client.list_ports(
network_id=self.network['id'],
device_owner=constants.DEVICE_OWNER_DHCP)['ports']
# in other cases, like ML2/OVN, the subnet port used for metadata
# is created with device_owner='network:distributed'
distributed_ports = self.client.list_ports(
network_id=self.network['id'],
device_owner=constants.DEVICE_OWNER_DISTRIBUTED)['ports']
self.dhcp_ports = dhcp_ports + distributed_ports
self.assertLessEqual(
len(self.dhcp_ports), 1, msg='Only one port was expected')
return len(self.dhcp_ports) == 1
# obtain the dhcp port
# in some cases this port is not created together with the subnet, but
# immediately after it, so some delay may be needed and that is the
# reason why a waiter function is used here
self.dhcp_ports = []
neutron_utils.wait_until_true(
lambda: _get_dhcp_ports(),
timeout=10)
dhcp_port = self.dhcp_ports[0]
# modify DHCP port IP address
old_dhcp_port_ip = netaddr.IPAddress(
dhcp_port['fixed_ips'][0]['ip_address'])
if str(old_dhcp_port_ip) != subnet['allocation_pools'][0]['end']:
new_dhcp_port_ip = str(old_dhcp_port_ip + 1)
else:
new_dhcp_port_ip = str(old_dhcp_port_ip - 1)
self.update_port(port=dhcp_port,
fixed_ips=[{'subnet_id': subnet['id'],
'ip_address': new_dhcp_port_ip}])
# create server
server = self.create_server(
flavor_ref=CONF.compute.flavor_ref,
image_ref=CONF.compute.image_ref,
key_name=self.keypair['name'],
security_groups=[{'name': self.security_group['name']}],
networks=[{'uuid': self.network['id']}])
# attach fip to the server
self.create_router_interface(self.router['id'], subnet['id'])
server_port = self.client.list_ports(
network_id=self.network['id'],
device_id=server['server']['id'])['ports'][0]
fip = self.create_floatingip(port_id=server_port['id'])
# check connectivity
self.check_connectivity(fip['floating_ip_address'],
CONF.validation.image_ssh_user,
self.keypair['private_key'])