# Copyright 2020 Ericsson Software Technology # # 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 base64 import collections import textwrap import time from neutron_lib import constants as nlib_const from oslo_log import log as logging from tempest.common import utils from tempest.lib.common.utils import data_utils from tempest.lib import decorators from tempest.lib import exceptions import testtools 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 Server = collections.namedtuple( 'Server', ['floating_ip', 'server', 'ssh_client']) QUERY_MSG = 'Queried the metadata service over IPv6' class MetadataTest(base.BaseTempestTestCase): """Test metadata access over IPv6 tenant subnet. Please note that there is metadata over IPv4 test coverage in tempest: tempest.scenario.test_server_basic_ops\ .TestServerBasicOps.test_server_basic_ops """ credentials = ['primary', 'admin'] force_tenant_isolation = False @classmethod def skip_checks(cls): super(MetadataTest, cls).skip_checks() if not utils.is_network_feature_enabled('ipv6_metadata'): raise cls.skipException("Metadata over IPv6 is not enabled") @classmethod def resource_setup(cls): super(MetadataTest, cls).resource_setup() cls.rand_name = data_utils.rand_name( cls.__name__.rsplit('.', 1)[-1]) cls.reserve_external_subnet_cidrs() cls.network = cls.create_network(name=cls.rand_name) cls.subnet_v4 = cls.create_subnet( network=cls.network, name=cls.rand_name) cls.subnet_v6 = cls.create_subnet( network=cls.network, name=cls.rand_name, ip_version=6) cls.router = cls.create_router_by_client() cls.create_router_interface(cls.router['id'], cls.subnet_v4['id']) cls.create_router_interface(cls.router['id'], cls.subnet_v6['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']) def _create_server_with_network(self, network, use_advanced_image=False): port = self._create_server_port(network=network) floating_ip = self.create_floatingip(port=port) ssh_client = self._create_ssh_client( floating_ip=floating_ip, use_advanced_image=use_advanced_image) server = self._create_server(port=port, use_advanced_image=use_advanced_image) return Server( floating_ip=floating_ip, server=server, ssh_client=ssh_client) def _create_server_port(self, network=None, **params): network = network or self.network return self.create_port(network=network, name=self.rand_name, security_groups=[self.security_group['id']], **params) def _create_server(self, port=None, network_id=None, use_advanced_image=False, **params): if use_advanced_image: flavor_ref = CONF.neutron_plugin_options.advanced_image_flavor_ref image_ref = CONF.neutron_plugin_options.advanced_image_ref else: flavor_ref = CONF.compute.flavor_ref image_ref = CONF.compute.image_ref if port: networks = [{'port': port['id']}] else: networks = [{'uuid': network_id}] return self.create_server(flavor_ref=flavor_ref, image_ref=image_ref, key_name=self.keypair['name'], networks=networks, **params)['server'] def _get_metadata_query_script(self): sheebang_line = '\n#!/bin/bash' curl_cmd = '\ncurl http://[%(address)s' % {'address': nlib_const.METADATA_V6_IP} ip_cmd = ("%$(ip -6 -br address show scope link up | head -1 | " "cut -d ' ' -f1)]/openstack/") echo_cmd = '\necho %s' % QUERY_MSG script = '%s%s%s%s' % (sheebang_line, curl_cmd, ip_cmd, echo_cmd) script_clean = textwrap.dedent(script).lstrip().encode('utf8') script_b64 = base64.b64encode(script_clean) return {'user_data': script_b64} def _wait_for_metadata_query_msg(self, vm): timeout = 300 start_time = int(time.time()) while int(time.time()) - start_time < timeout: console_output = self.os_primary.servers_client.get_console_output( vm['id'])['output'] pos = console_output.find(QUERY_MSG) if pos > -1: return console_output, pos time.sleep(30) self.fail('Failed to find metadata query message in console log %s' % console_output) def _create_ssh_client(self, floating_ip, use_advanced_image=False): if use_advanced_image: username = CONF.neutron_plugin_options.advanced_image_ssh_user else: username = CONF.validation.image_ssh_user return ssh.Client(host=floating_ip['floating_ip_address'], username=username, pkey=self.keypair['private_key']) def _assert_has_ssh_connectivity(self, ssh_client): ssh_client.exec_command('true') def _get_primary_interface(self, ssh_client): out = ssh_client.exec_command( "ip -6 -br address show scope link up | head -1 | cut -d ' ' -f1") interface = out.strip() if not interface: self.fail( 'Could not find a single interface ' 'with an IPv6 link-local address.') return interface @testtools.skipUnless( CONF.neutron_plugin_options.advanced_image_ref or CONF.neutron_plugin_options.default_image_is_advanced, 'Advanced image is required to run this test.') @decorators.idempotent_id('e680949a-f1cc-11ea-b49a-cba39bbbe5ad') def test_metadata_routed(self): use_advanced_image = ( not CONF.neutron_plugin_options.default_image_is_advanced) vm = self._create_server_with_network( self.network, use_advanced_image=use_advanced_image) self.wait_for_server_active(server=vm.server) self.wait_for_guest_os_ready(vm.server) self.check_connectivity(ssh_client=vm.ssh_client) interface = self._get_primary_interface(vm.ssh_client) try: out = vm.ssh_client.exec_command( 'curl http://[%(address)s%%25%(interface)s]/' % { 'address': nlib_const.METADATA_V6_IP, 'interface': interface}) self.assertIn('latest', out) out = vm.ssh_client.exec_command( 'curl http://[%(address)s%%25%(interface)s]/openstack/' % { 'address': nlib_const.METADATA_V6_IP, 'interface': interface}) self.assertIn('latest', out) except exceptions.SSHExecCommandFailed: self._log_console_output() self._log_local_network_status() @testtools.skipUnless( CONF.neutron_plugin_options.advanced_image_ref or CONF.neutron_plugin_options.default_image_is_advanced, 'Advanced image is required to run this test.') @decorators.idempotent_id('7542892a-d132-471c-addb-172dcf888ff6') def test_metadata_ipv6_only_network(self): ipv6_network = self.create_network() ipv6_subnet = self.create_subnet(network=ipv6_network, ip_version=6, ipv6_ra_mode="slaac", ipv6_address_mode="slaac") if not CONF.neutron_plugin_options.firewall_driver == 'ovn': self.create_router_interface(self.router['id'], ipv6_subnet['id']) use_advanced_image = ( not CONF.neutron_plugin_options.default_image_is_advanced) params = self._get_metadata_query_script() params['config_drive'] = True vm = self._create_server( network_id=ipv6_network['id'], use_advanced_image=use_advanced_image, **params) self.wait_for_server_active(server=vm) self.wait_for_guest_os_ready(vm) console_output, pos = self._wait_for_metadata_query_msg(vm) self.assertIn('latest', console_output[pos - 100:])