# Copyright 2024 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. import re from neutron_lib import constants as lib_constants from neutron_tempest_plugin.common import ssh from neutron_tempest_plugin.common import utils as common_utils from neutron_tempest_plugin import config from oslo_log import log from tempest.common import utils from tempest.lib.common.utils import data_utils from tempest.lib import decorators from tempest.lib.exceptions import SSHExecCommandFailed from whitebox_neutron_tempest_plugin.tests.scenario import base CONF = config.CONF WB_CONF = CONF.whitebox_neutron_plugin_options LOG = log.getLogger(__name__) class InternalDNSBaseCommon(base.TrafficFlowTest): """Common base class of resources and functionalities for test classes.""" port_error_msg = ('Openstack command returned incorrect' ' hostname value in port.') ssh_error_msg = ('Remote shell command returned incorrect hostname value' " (command: 'cat /etc/hostname').") ssh_hostname_cmd = 'cat /etc/hostname' @staticmethod def _rand_name(name): """'data_utils.rand_name' wrapper, show name related to test suite.""" return data_utils.rand_name('internal-dns-test-{}'.format(name)) @classmethod def resource_setup(cls): super(InternalDNSBaseCommon, cls).resource_setup() # setup reusable resources for entire test suite cls.keypair = cls.create_keypair( name=cls._rand_name('shared-keypair')) cls.secgroup = cls.create_security_group( name=cls._rand_name('shared-secgroup')) cls.security_groups.append(cls.secgroup) cls.create_loginable_secgroup_rule( secgroup_id=cls.secgroup['id']) cls.create_pingable_secgroup_rule( secgroup_id=cls.secgroup['id']) cls.network = cls.create_network(name=cls._rand_name('shared-network')) cls.subnet = cls.create_subnet( cls.network, name=cls._rand_name('shared-subnet')) cls.router = cls.create_router_by_client() cls.create_router_interface(cls.router['id'], cls.subnet['id']) cls.vm_kwargs = { 'flavor_ref': cls.flavor_ref, 'image_ref': cls.image_ref, 'key_name': cls.keypair['name'], 'security_groups': [{'name': cls.secgroup['name']}] } def _create_ssh_client(self, ip_addr): return ssh.Client(ip_addr, self.username, pkey=self.keypair['private_key']) def _validate_port_dns_details(self, expected_hostname, checked_port, raise_exception=True): """Validates reused objects for correct dns values in tests.""" result = True dns_details = checked_port['dns_assignment'][0] try: self.assertEqual(expected_hostname, checked_port['dns_name'], self.port_error_msg) self.assertEqual(expected_hostname, dns_details['hostname'], self.port_error_msg) self.assertIn(expected_hostname, dns_details['fqdn'], self.port_error_msg) # returns boolean instead of raising assert exception when needed except AssertionError: if raise_exception: raise result = False return result def _validate_ssh_dns_details(self, expected_hostname, ssh_client, raise_exception=True): """Validates correct dns values returned from ssh command in tests.""" ssh_output = ssh_client.exec_command(self.ssh_hostname_cmd) result = expected_hostname in ssh_output if raise_exception and not result: self.fail(self.ssh_error_msg) return result def _dns_common_validations(self, vm_name, dns_port, vm_client): """Validate hostname (dns-name) using API, and guest VM.""" # retry to get ssh connection until VM is up and working (with timeout) try: common_utils.wait_until_true( lambda: self._validate_ssh_dns_details(vm_name, vm_client, raise_exception=False), timeout=120, sleep=10) except common_utils.WaitTimeout: self.fail(self.ssh_error_msg) # validate dns port hostname from API self._validate_port_dns_details(vm_name, dns_port) def _common_create_and_update_port_with_dns_name(self): """Helper function that creates and updates a port with correct internal dns-name (hostname), without any validations afterwards. """ # 1) Create a port with wrong dns-name (not as VM name). # 2) Verify that wrong port initial dns-name. # was queried from openstack API. # 3) Update the port with correct dns-name (as VM name). # 4) Boot a VM with corrected predefined port. # NOTE: VM's hostname has to be the same as VM's name # when a VM is created, it is a known limitation. # Therefore VM's dns-name/hostname is checked to be as VM's name. vm_correct_name = self._rand_name('vm') vm_wrong_name = self._rand_name('bazinga') # create port with wrong dns-name (not as VM name) dns_port = self.create_port(self.network, dns_name=vm_wrong_name, security_groups=[self.secgroup['id']], name=self._rand_name('port')) # validate dns port with wrong initial hostname from API self._validate_port_dns_details(vm_wrong_name, dns_port) # update port with correct dns-name (as VM name) dns_port = self.update_port(dns_port, dns_name=vm_correct_name) # create VM with correct predefined dns-name on port vm_1 = self.create_server(name=vm_correct_name, networks=[{'port': dns_port['id']}], **self.vm_kwargs) vm_1['fip'] = self.create_floatingip(port=dns_port) vm_1['ssh_client'] = self._create_ssh_client( vm_1['fip']['floating_ip_address']) # return parameters required for validations return (vm_correct_name, dns_port, vm_1) class InternalDNSBaseOvn(base.BaseTempestTestCaseOvn, InternalDNSBaseCommon): """Ovn base class of resources and functionalities for test class.""" ovn_db_hostname_cmd = '{} list dns' ovn_db_error_msg = ('Incorrect hostname/ip values in NBDB, ' 'or failed to reach OVN NBDB.') def _get_router_and_nodes_info(self): self.router_port = self.os_admin.network_client.list_ports( device_id=self.router['id'], device_owner=lib_constants.DEVICE_OWNER_ROUTER_GW)['ports'][0] self.router_gateway_chassis = self.get_router_gateway_chassis( self.router_port['id']) self.discover_nodes() def _validate_dns_ovn_nbdb(self, expected_hostname, local_ip): """Validates correct dns values exist in OVN NBDB, if so, then returns True, otherwise returns False. """ # optional quotation marks for OSP 13 dns_pattern = '.*"?{}"?="?{}"?.*'.format(expected_hostname, local_ip) try: db_dns_entries = self.run_on_master_controller( self.ovn_db_hostname_cmd.format(self.nbctl)).replace('\n', '') except SSHExecCommandFailed as err: LOG.warning(err) return False result = re.match(dns_pattern, db_dns_entries) if not result: err = "{}:\n'{}' regex not found in string '{}'".format( self.ovn_db_error_msg, dns_pattern, db_dns_entries) LOG.warning(err) return False return True def _dns_all_validations(self, vm_name, dns_port, vm_client): """Validate hostname (dns-name) using API, guest VM, and OVN NBDB.""" # validate dns port hostname using API and check on guest VM self._dns_common_validations(vm_name, dns_port, vm_client) # validate dns-name details in OVN NBDB try: common_utils.wait_until_true( lambda: self._validate_dns_ovn_nbdb( vm_name, dns_port['fixed_ips'][0]['ip_address']), timeout=120, sleep=10) except common_utils.WaitTimeout: self.fail(self.ovn_db_error_msg) class InternalDNSTestOvn(InternalDNSBaseOvn): """Tests internal DNS capabilities on OVN setups.""" @utils.requires_ext(extension="dns-integration", service="network") @decorators.idempotent_id('6349ce8c-bc10-485a-a21b-da073241420e') def test_ovn_create_and_update_port_with_dns_name(self): """Test creation of port with correct internal dns-name (hostname).""" # 1) Create resources: network, subnet, etc. # 2) Create a port with wrong dns-name (not as VM name). # 3) Verify that wrong port initial dns-name. # was queried from openstack API. # 4) Update the port with correct dns-name (as VM name). # 5) Boot a VM with corrected predefined port. # 6) Verify that correct port dns-name # was queried from openstack API. # 7) Validate hostname configured on VM is same as VM's name. # 8) Validate hostname configured correctly in OVN NBDB. # NOTE: VM's hostname has to be the same as VM's name # when a VM is created, it is a known limitation. # Therefore VM's dns-name/hostname is checked to be as VM's name. # all test steps 2 - 5 (inclusively) vm_name, dns_port, vm_1 = \ self._common_create_and_update_port_with_dns_name() # validate hostname (dns-name) using API, guest VM, and OVN NBDB self._dns_all_validations(vm_name, dns_port, vm_1['ssh_client']) class InternalDNSInterruptionsTestOvn(InternalDNSBaseOvn): """Tests internal DNS capabilities on OVN setups, with interruptions in overcloud. """ @utils.requires_ext(extension="dns-integration", service="network") @decorators.idempotent_id('bf11667e-34f8-4ac4-886b-45e099fdbffa') def test_dns_name_after_ovn_controller_restart(self): """Tests that OpenStack port, guest VM and OVN NB database have correct dns-name (hostname) set, after controller service restart on compute node. """ # 1) Create resources: network, subnet, etc. # 2) Create a port with dns-name. # 3) Boot a guest VM with predefined port. # 4) Restart ovn controller service on compute which runs guest VM. # 5) Validate hostname configured on VM is the same as VM's name. # 6) Verify that the correct port dns-name (as VM name) # was queried from openstack API. # 7) Validate dns-name details in OVN NB database. # NOTE: VM's hostname has to be the same as VM's name # when a VM is created, it is a known limitation. # Therefore VM's dns-name/hostname is checked to be as VM's name. vm_name = self._rand_name('vm') # create port with dns-name (as VM name) dns_port = self.create_port(self.network, dns_name=vm_name, security_groups=[self.secgroup['id']], name=self._rand_name('port')) # create VM with predefined dns-name on port vm_1 = self.create_server(name=vm_name, networks=[{'port': dns_port['id']}], **self.vm_kwargs) vm_1['fip'] = self.create_floatingip(port=dns_port) # restart controller service on compute which runs guest VM self.discover_nodes() compute_hostname = self.get_host_for_server( vm_1['server']['id']) compute_client = self.find_node_client(compute_hostname) self.reset_node_service('ovn controller', compute_client) # validate hostname configured on VM is same as VM's name vm_1['ssh_client'] = self._create_ssh_client( vm_1['fip']['floating_ip_address']) # validate hostname (dns-name) using API, guest VM, and OVN NBDB self._dns_all_validations(vm_name, dns_port, vm_1['ssh_client']) class InternalDNSInterruptionsAdvancedTestOvn( InternalDNSBaseOvn, base.BaseTempestTestCaseAdvanced, base.BaseDisruptiveTempestTestCase): """Tests internal DNS capabilities with interruptions in overcloud, on advanced image only. """ @classmethod def skip_checks(cls): super(InternalDNSInterruptionsAdvancedTestOvn, cls).skip_checks() if WB_CONF.openstack_type == 'devstack': raise cls.skipException( "Devstack doesn't support powering nodes on/off, " "skipping tests") if not WB_CONF.run_power_operations_tests: raise cls.skipException( "run_power_operations_tests config is not enabled, " "skipping tests") @classmethod def resource_setup(cls): super(InternalDNSInterruptionsAdvancedTestOvn, cls).resource_setup() for node in cls.nodes: if node['is_networker'] is True and node['is_compute'] is True: raise cls.skipException( "Not supported when environment allows OVN gateways on " "compute nodes.") @decorators.attr(type='slow') @utils.requires_ext(extension="dns-integration", service="network") @decorators.idempotent_id('e6c5dbea-d704-4cda-bb92-a5bfd0aa1bb2') def test_ovn_dns_name_after_networker_reboot(self): """Tests that OpenStack port, guest VM and OVN NB database have correct dns-name (hostname) when master networker node is turned off and on. """ # 1) Create resources: network, subnet, etc. # 2) Create a port with dns-name. # 3) Boot a VM with predefined port. # 4) Soft shutdown master networker node. # 5) Validate hostname (dns-name) using API, guest VM, # and OVN NBDB when networker node is off. # 6) Turn on previous master networker node, wait until it is working. # 7) Validate hostname (dns-name) using API, guest VM, # and OVN NBDB when networker node is on. # NOTE: VM's hostname has to be the same as VM's name # when a VM is created, it is a known limitation. # Therefore VM's dns-name/hostname is checked to be as VM's name. # ensures overcloud nodes are up for next tests self.addCleanup(self.ensure_overcloud_nodes_active) # create port with dns-name (as VM name) vm_name = self._rand_name('vm') dns_port = self.create_port(self.network, dns_name=vm_name, security_groups=[self.secgroup['id']], name=self._rand_name('port')) # create VM with predefined dns-name on port vm_1 = self.create_server(name=vm_name, networks=[{'port': dns_port['id']}], **self.vm_kwargs) vm_1['fip'] = self.create_floatingip(port=dns_port) vm_1['ssh_client'] = self._create_ssh_client( vm_1['fip']['floating_ip_address']) self._get_router_and_nodes_info() if self.get_node_setting(self.router_gateway_chassis, 'is_controller'): raise self.skipException( "The test currently does not support a required action " "when gateway chassis is on a node with OSP control plane " "services rather than on a standalone networker node.") # soft shutdown master networker node self.power_off_host(self.router_gateway_chassis) # validate hostname (dns-name) using API, guest VM, # and OVN NBDB when networker node is off and on self._dns_all_validations(vm_name, dns_port, vm_1['ssh_client']) # turn on networker node, wait until it is up and working self.power_on_host(self.router_gateway_chassis) self._dns_all_validations(vm_name, dns_port, vm_1['ssh_client'])