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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
# Copyright 2012 OpenStack Foundation
# 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 time
from tempest.common import custom_matchers
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import exceptions as lib_exc
import tempest.test
CONF = config.CONF
def delete_containers(containers, container_client, object_client):
"""Remove containers and all objects in them.
The containers should be visible from the container_client given.
Will not throw any error if the containers don't exist.
Will not check that object and container deletions succeed.
After delete all the objects from a container, it will wait 2
seconds before delete the container itself, in order to deployments
using HA proxy sync the deletion properly, otherwise, the container
might fail to be deleted because it's not empty.
:param containers: List of containers to be deleted
:param container_client: Client to be used to delete containers
:param object_client: Client to be used to delete objects
"""
for cont in containers:
try:
params = {'limit': 9999, 'format': 'json'}
resp, objlist = container_client.list_container_contents(
cont, params)
# delete every object in the container
for obj in objlist:
test_utils.call_and_ignore_notfound_exc(
object_client.delete_object, cont, obj['name'])
# sleep 2 seconds to sync the deletion of the objects
# in HA deployment
time.sleep(2)
container_client.delete_container(cont)
except lib_exc.NotFound:
pass
class BaseObjectTest(tempest.test.BaseTestCase):
credentials = [['operator', CONF.object_storage.operator_role]]
@classmethod
def skip_checks(cls):
super(BaseObjectTest, cls).skip_checks()
if not CONF.service_available.swift:
skip_msg = ("%s skipped as swift is not available" % cls.__name__)
raise cls.skipException(skip_msg)
@classmethod
def setup_credentials(cls):
cls.set_network_resources()
super(BaseObjectTest, cls).setup_credentials()
# credentials may be overwritten by children classes
if hasattr(cls, 'os_roles_operator'):
cls.os = cls.os_roles_operator
@classmethod
def setup_clients(cls):
super(BaseObjectTest, cls).setup_clients()
cls.object_client = cls.os.object_client
cls.bulk_client = cls.os.bulk_client
cls.container_client = cls.os.container_client
cls.account_client = cls.os.account_client
cls.capabilities_client = cls.os.capabilities_client
@classmethod
def resource_setup(cls):
super(BaseObjectTest, cls).resource_setup()
# Make sure we get fresh auth data after assigning swift role
cls.object_client.auth_provider.clear_auth()
cls.container_client.auth_provider.clear_auth()
cls.account_client.auth_provider.clear_auth()
# make sure that discoverability is enabled and that the sections
# have not been disallowed by Swift
cls.policies = None
if CONF.object_storage_feature_enabled.discoverability:
_, body = cls.capabilities_client.list_capabilities()
if 'swift' in body and 'policies' in body['swift']:
cls.policies = body['swift']['policies']
cls.containers = []
@classmethod
def create_container(cls):
# wrapper that returns a test container
container_name = data_utils.rand_name(name='TestContainer')
cls.container_client.create_container(container_name)
cls.containers.append(container_name)
return container_name
@classmethod
def create_object(cls, container_name, object_name=None,
data=None, metadata=None):
# wrapper that returns a test object
if object_name is None:
object_name = data_utils.rand_name(name='TestObject')
if data is None:
data = data_utils.random_bytes()
cls.object_client.create_object(container_name,
object_name,
data,
metadata=metadata)
return object_name, data
@classmethod
def delete_containers(cls, container_client=None, object_client=None):
if container_client is None:
container_client = cls.container_client
if object_client is None:
object_client = cls.object_client
delete_containers(cls.containers, container_client, object_client)
def assertHeaders(self, resp, target, method):
"""Check the existence and the format of response headers"""
self.assertThat(resp, custom_matchers.ExistsAllResponseHeaders(
target, method, self.policies))
self.assertThat(resp, custom_matchers.AreAllWellFormatted())
|
