openstack/tempest
Code Issues Proposed changes

Merge "Integration tests for device role tagging"

Browse Source
This commit is contained in:
Jenkins 2016-12-02 12:54:37 +00:00 committed by Gerrit Code Review
commit 4541ebebcb
3 changed files with 279 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
doc/source/microversion_testing.rst
View File

@ -236,6 +236,10 @@ Microversion tests implemented in Tempest
.. _2.25: http://docs.openstack.org/developer/nova/api_microversion_history.html#maximum-in-mitaka
* `2.32`_
.. _2.32: http://docs.openstack.org/developer/nova/api_microversion_history.html#id29
* `2.37`_
.. _2.37: http://docs.openstack.org/developer/nova/api_microversion_history.html#id34

10
tempest/api/compute/base.py
View File

@ -373,14 +373,18 @@ class BaseV2ComputeTest(api_version_utils.BaseMicroversionTest,
self.request_microversion))
@classmethod
def create_volume(cls):
def create_volume(cls, image_ref=None):
"""Create a volume and wait for it to become 'available'.
:param image_ref: Specify an image id to create a bootable volume.
:returns: The available volume.
"""
vol_name = data_utils.rand_name(cls.__name__ + '-volume')
volume = cls.volumes_client.create_volume(
size=CONF.volume.volume_size, display_name=vol_name)['volume']
create_params = dict(size=CONF.volume.volume_size,
display_name=vol_name)
if image_ref is not None:
create_params['imageRef'] = image_ref
volume = cls.volumes_client.create_volume(**create_params)['volume']
cls.volumes.append(volume)
waiters.wait_for_volume_status(cls.volumes_client,
volume['id'], 'available')

268
tempest/api/compute/servers/test_device_tagging.py Normal file
View File

@ -0,0 +1,268 @@
# Copyright (C) 2016, Red Hat, Inc.
#
# 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 json
from oslo_log import log as logging
from tempest.api.compute import base
from tempest.common.utils import data_utils
from tempest.common.utils.linux import remote_client
from tempest.common import waiters
from tempest import config
from tempest import exceptions
from tempest import test
CONF = config.CONF
LOG = logging.getLogger(__name__)
class DeviceTaggingTest(base.BaseV2ComputeTest):
min_microversion = '2.32'
max_microversion = 'latest'
@classmethod
def skip_checks(cls):
super(DeviceTaggingTest, cls).skip_checks()
if not CONF.service_available.neutron:
raise cls.skipException('Neutron is required')
if not CONF.validation.run_validation:
raise cls.skipException('Validation must be enabled')
if (not CONF.compute_feature_enabled.config_drive
and not CONF.compute_feature_enabled.metadata_service):
raise cls.skipException('One of metadata or config drive must be '
'enabled')
@classmethod
def setup_clients(cls):
super(DeviceTaggingTest, cls).setup_clients()
cls.networks_client = cls.os.networks_client
cls.ports_client = cls.os.ports_client
cls.volumes_client = cls.os.volumes_client
cls.subnets_client = cls.os.subnets_client
cls.routers_client = cls.os.routers_client
cls.interfaces_client = cls.os.interfaces_client
@classmethod
def setup_credentials(cls):
cls.set_network_resources(network=True, subnet=True, router=True,
dhcp=True)
super(DeviceTaggingTest, cls).setup_credentials()
@classmethod
def resource_setup(cls):
cls.set_validation_resources()
super(DeviceTaggingTest, cls).resource_setup()
def verify_device_metadata(self, md_json):
md_dict = json.loads(md_json)
for d in md_dict['devices']:
if d['type'] == 'nic':
if d['mac'] == self.port1['mac_address']:
self.assertEqual(d['tags'], ['port-1'])
if d['mac'] == self.port2['mac_address']:
self.assertEqual(d['tags'], ['port-2'])
if d['mac'] == self.net_2_100_mac:
self.assertEqual(d['tags'], ['net-2-100'])
if d['mac'] == self.net_2_200_mac:
self.assertEqual(d['tags'], ['net-2-200'])
found_devices = [d['tags'][0] for d in md_dict['devices']]
self.assertItemsEqual(found_devices, ['port-1', 'port-2', 'net-1',
'net-2-100', 'net-2-200',
'boot', 'other'])
@test.idempotent_id('a2e65a6c-66f1-4442-aaa8-498c31778d96')
@test.services('network', 'volume', 'image')
def test_device_tagging(self):
# Create volumes
# The create_volume methods waits for the volumes to be available and
# the base class will clean them up on tearDown.
boot_volume = self.create_volume(CONF.compute.image_ref)
other_volume = self.create_volume()
untagged_volume = self.create_volume()
# Create networks
net1 = self.networks_client.create_network(
name=data_utils.rand_name('device-tagging-net1'))['network']
self.addCleanup(self.networks_client.delete_network, net1['id'])
net2 = self.networks_client.create_network(
name=data_utils.rand_name('device-tagging-net2'))['network']
self.addCleanup(self.networks_client.delete_network, net2['id'])
# Create subnets
subnet1 = self.subnets_client.create_subnet(
network_id=net1['id'],
cidr='10.1.1.0/24',
ip_version=4)['subnet']
self.addCleanup(self.subnets_client.delete_subnet, subnet1['id'])
subnet2 = self.subnets_client.create_subnet(
network_id=net2['id'],
cidr='10.2.2.0/24',
ip_version=4)['subnet']
self.addCleanup(self.subnets_client.delete_subnet, subnet2['id'])
# Create ports
self.port1 = self.ports_client.create_port(
network_id=net1['id'],
fixed_ips=[{'subnet_id': subnet1['id']}])['port']
self.addCleanup(self.ports_client.delete_port, self.port1['id'])
self.port2 = self.ports_client.create_port(
network_id=net1['id'],
fixed_ips=[{'subnet_id': subnet1['id']}])['port']
self.addCleanup(self.ports_client.delete_port, self.port2['id'])
# Create server
admin_pass = data_utils.rand_password()
config_drive_enabled = CONF.compute_feature_enabled.config_drive
server = self.create_test_server(
validatable=True,
config_drive=config_drive_enabled,
adminPass=admin_pass,
name=data_utils.rand_name('device-tagging-server'),
networks=[
# Validation network for ssh
{
'uuid': self.get_tenant_network()['id']
},
# Different tags for different ports
{
'port': self.port1['id'],
'tag': 'port-1'
},
{
'port': self.port2['id'],
'tag': 'port-2'
},
# Two nics on same net, one tagged one not
{
'uuid': net1['id'],
'tag': 'net-1'
},
{
'uuid': net1['id']
},
# Two nics on same net, different IP
{
'uuid': net2['id'],
'fixed_ip': '10.2.2.100',
'tag': 'net-2-100'
},
{
'uuid': net2['id'],
'fixed_ip': '10.2.2.200',
'tag': 'net-2-200'
}
],
block_device_mapping_v2=[
# Boot volume
{
'uuid': boot_volume['id'],
'source_type': 'volume',
'destination_type': 'volume',
'boot_index': 0,
'tag': 'boot'
},
# Other volume
{
'uuid': other_volume['id'],
'source_type': 'volume',
'destination_type': 'volume',
'boot_index': 1,
'tag': 'other'
},
# Untagged volume
{
'uuid': untagged_volume['id'],
'source_type': 'volume',
'destination_type': 'volume',
'boot_index': 2
}
])
self.addCleanup(waiters.wait_for_server_termination,
self.servers_client, server['id'])
self.addCleanup(self.servers_client.delete_server, server['id'])
self.ssh_client = remote_client.RemoteClient(
self.get_server_ip(server),
CONF.validation.image_ssh_user,
admin_pass,
self.validation_resources['keypair']['private_key'],
server=server,
servers_client=self.servers_client)
# Find the MAC addresses of our fixed IPs
self.net_2_100_mac = None
self.net_2_200_mac = None
ifaces = self.interfaces_client.list_interfaces(server['id'])
for iface in ifaces['interfaceAttachments']:
if 'fixed_ips' in iface:
for ip in iface['fixed_ips']:
if ip['ip_address'] == '10.2.2.100':
self.net_2_100_mac = iface['mac_addr']
if ip['ip_address'] == '10.2.2.200':
self.net_2_200_mac = iface['mac_addr']
# Make sure we have the MACs we need, there's no reason for some to be
# missing
self.assertTrue(self.net_2_100_mac)
self.assertTrue(self.net_2_200_mac)
# Verify metadata from metadata service
if CONF.compute_feature_enabled.metadata_service:
md_url = 'http://169.254.169.254/openstack/latest/meta_data.json'
LOG.info('Attempting to verify tagged devices in server %s via '
'the metadata service: %s', server['id'], md_url)
def get_and_verify_metadata():
try:
self.ssh_client.exec_command('curl -V')
except exceptions.SSHExecCommandFailed:
if not CONF.compute_feature_enabled.config_drive:
raise self.skipException('curl not found in guest '
'and config drive is '
'disabled')
LOG.warning('curl was not found in the guest, device '
'tagging metadata was not checked in the '
'metadata API')
return True
cmd = 'curl %s' % md_url
md_json = self.ssh_client.exec_command(cmd)
self.verify_device_metadata(md_json)
return True
if not test.call_until_true(get_and_verify_metadata,
CONF.compute.build_timeout,
CONF.compute.build_interval):
raise exceptions.TimeoutException('Timeout while verifying '
'metadata on server.')
# Verify metadata on config drive
if CONF.compute_feature_enabled.config_drive:
cmd_blkid = 'blkid -t LABEL=config-2 -o device'
LOG.info('Attempting to verify tagged devices in server %s via '
'the config drive.', server['id'])
dev_name = self.ssh_client.exec_command(cmd_blkid)
dev_name = dev_name.rstrip()
self.ssh_client.exec_command('sudo mount %s /mnt' % dev_name)
cmd_md = 'sudo cat /mnt/openstack/latest/meta_data.json'
md_json = self.ssh_client.exec_command(cmd_md)
self.verify_device_metadata(md_json)