neutron-tempest-plugin/neutron_tempest_plugin/scenario/test_dhcp.py

# 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'])