1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
# Copyright 2020 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.
from tempest.api.compute import base
from tempest.common import waiters
from tempest import config
from tempest.lib import decorators
CONF = config.CONF
class BaseAttachSCSIVolumeTest(base.BaseV2ComputeAdminTest):
"""Base class for the admin volume tests in this module."""
create_default_network = True
@classmethod
def skip_checks(cls):
super(BaseAttachSCSIVolumeTest, cls).skip_checks()
if not CONF.service_available.cinder:
skip_msg = ("%s skipped as Cinder is not available" % cls.__name__)
raise cls.skipException(skip_msg)
@classmethod
def setup_credentials(cls):
cls.prepare_instance_network()
super(BaseAttachSCSIVolumeTest, cls).setup_credentials()
def _create_image_with_custom_property(self, **kwargs):
"""Wrapper utility that returns the custom image.
Creates a new image by downloading the default image's bits and
uploading them to a new image. Any kwargs are set as image properties
on the new image.
:param return image_id: The UUID of the newly created image.
"""
image = self.admin_image_client.show_image(CONF.compute.image_ref)
# NOTE(danms): We need to stream this, so chunked=True means we get
# back a urllib3.HTTPResponse and have to carefully pass it to
# store_image_file() to upload it in pieces.
image_data_resp = self.admin_image_client.show_image_file(
CONF.compute.image_ref, chunked=True)
create_dict = {
'container_format': image['container_format'],
'disk_format': image['disk_format'],
'min_disk': image['min_disk'],
'min_ram': image['min_ram'],
'visibility': 'public',
}
create_dict.update(kwargs)
try:
new_image = self.admin_image_client.create_image(**create_dict)
self.addCleanup(self.admin_image_client.wait_for_resource_deletion,
new_image['id'])
self.addCleanup(
self.admin_image_client.delete_image, new_image['id'])
self.admin_image_client.store_image_file(new_image['id'],
image_data_resp)
finally:
image_data_resp.release_conn()
return new_image['id']
class AttachSCSIVolumeTestJSON(BaseAttachSCSIVolumeTest):
"""Test attaching scsi volume to server"""
@decorators.idempotent_id('777e468f-17ca-4da4-b93d-b7dbf56c0494')
def test_attach_scsi_disk_with_config_drive(self):
"""Test the attach/detach volume with config drive/scsi disk
Enable the config drive, followed by booting an instance
from an image with meta properties hw_cdrom: scsi and use
virtio-scsi mode with further asserting list volume attachments
in instance after attach and detach of the volume.
"""
custom_img = self._create_image_with_custom_property(
hw_scsi_model='virtio-scsi',
hw_disk_bus='scsi',
hw_cdrom_bus='scsi')
validation_resources = self.get_test_validation_resources(
self.os_primary)
server = self.create_test_server(
image_id=custom_img,
config_drive=True,
validatable=True,
validation_resources=validation_resources,
wait_until="SSHABLE")
# NOTE(lyarwood): self.create_test_server delete the server
# at class level cleanup so add server cleanup to ensure that
# the instance is deleted first before created image. This
# avoids failures when using the rbd backend is used for both
# Glance and Nova ephemeral storage. Also wait until server is
# deleted otherwise image deletion can start before server is
# deleted.
self.addCleanup(waiters.wait_for_server_termination,
self.servers_client, server['id'])
self.addCleanup(self.servers_client.delete_server, server['id'])
volume = self.create_volume()
attachment = self.attach_volume(server, volume)
waiters.wait_for_volume_resource_status(
self.volumes_client, attachment['volumeId'], 'in-use')
volume_after_attach = self.servers_client.list_volume_attachments(
server['id'])['volumeAttachments']
self.assertEqual(1, len(volume_after_attach),
"Failed to attach volume")
self.servers_client.detach_volume(
server['id'], attachment['volumeId'])
waiters.wait_for_volume_resource_status(
self.volumes_client, attachment['volumeId'], 'available')
waiters.wait_for_volume_attachment_remove_from_server(
self.servers_client, server['id'], attachment['volumeId'])
|
