Source code for object_storage.test_container_services
# 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.
from tempest.api.object_storage import base
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
CONF = config.CONF
[docs]
class ContainerTest(base.BaseObjectTest):
"""Test containers"""
@classmethod
def resource_cleanup(cls):
cls.delete_containers()
super(ContainerTest, cls).resource_cleanup()
[docs]
@decorators.attr(type='smoke')
@decorators.idempotent_id('92139d73-7819-4db1-85f8-3f2f22a8d91f')
def test_create_container(self):
"""Test creating container"""
container_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestContainer')
resp, _ = self.container_client.update_container(container_name)
self.containers.append(container_name)
self.assertHeaders(resp, 'Container', 'PUT')
[docs]
@decorators.idempotent_id('49f866ed-d6af-4395-93e7-4187eb56d322')
def test_create_container_overwrite(self):
"""Test overwriting container with the same name"""
container_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestContainer')
self.container_client.update_container(container_name)
self.containers.append(container_name)
resp, _ = self.container_client.update_container(container_name)
self.assertHeaders(resp, 'Container', 'PUT')
[docs]
@decorators.idempotent_id('c2ac4d59-d0f5-40d5-ba19-0635056d48cd')
def test_create_container_with_metadata_key(self):
"""Test creating container with the blank value of metadata"""
container_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestContainer')
headers = {'X-Container-Meta-test-container-meta': ''}
resp, _ = self.container_client.update_container(
container_name,
**headers)
self.containers.append(container_name)
self.assertHeaders(resp, 'Container', 'PUT')
resp, _ = self.container_client.list_container_metadata(
container_name)
# if the value of metadata is blank, metadata is not registered
# in the server
self.assertNotIn('x-container-meta-test-container-meta', resp)
[docs]
@decorators.idempotent_id('e1e8df32-7b22-44e1-aa08-ccfd8d446b58')
def test_create_container_with_metadata_value(self):
"""Test creating container with metadata value"""
container_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestContainer')
# metadata name using underscores should be converted to hyphens
headers = {'X-Container-Meta-test_container_meta': 'Meta1'}
resp, _ = self.container_client.update_container(
container_name,
**headers)
self.containers.append(container_name)
self.assertHeaders(resp, 'Container', 'PUT')
resp, _ = self.container_client.list_container_metadata(
container_name)
self.assertIn('x-container-meta-test-container-meta', resp)
self.assertEqual(resp['x-container-meta-test-container-meta'],
headers['X-Container-Meta-test_container_meta'])
[docs]
@decorators.idempotent_id('24d16451-1c0c-4e4f-b59c-9840a3aba40e')
def test_create_container_with_remove_metadata_key(self):
"""Test creating container with the blank value of remove metadata"""
container_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestContainer')
headers = {'X-Container-Meta-test-container-meta': 'Meta1'}
self.container_client.update_container(container_name, **headers)
self.containers.append(container_name)
headers = {'X-Remove-Container-Meta-test-container-meta': ''}
resp, _ = self.container_client.update_container(
container_name,
**headers)
self.assertHeaders(resp, 'Container', 'PUT')
resp, _ = self.container_client.list_container_metadata(
container_name)
self.assertNotIn('x-container-meta-test-container-meta', resp)
[docs]
@decorators.idempotent_id('8a21ebad-a5c7-4e29-b428-384edc8cd156')
def test_create_container_with_remove_metadata_value(self):
"""Test creating container with remove metadata"""
container_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestContainer')
headers = {'X-Container-Meta-test-container-meta': 'Meta1'}
self.container_client.update_container(container_name, **headers)
self.containers.append(container_name)
headers = {'X-Remove-Container-Meta-test-container-meta': 'Meta1'}
resp, _ = self.container_client.update_container(
container_name,
**headers)
self.assertHeaders(resp, 'Container', 'PUT')
resp, _ = self.container_client.list_container_metadata(
container_name)
self.assertNotIn('x-container-meta-test-container-meta', resp)
[docs]
@decorators.idempotent_id('95d3a249-b702-4082-a2c4-14bb860cf06a')
def test_delete_container(self):
"""Test deleting container"""
# create a container
container_name = self.create_container()
# delete container, success asserted within
resp, _ = self.container_client.delete_container(container_name)
self.assertHeaders(resp, 'Container', 'DELETE')
[docs]
@decorators.attr(type='smoke')
@decorators.idempotent_id('312ff6bd-5290-497f-bda1-7c5fec6697ab')
def test_list_container_contents(self):
"""Test getting container contents list"""
container_name = self.create_container()
object_name, _ = self.create_object(container_name)
resp, object_list = self.container_client.list_container_objects(
container_name)
self.assertHeaders(resp, 'Container', 'GET')
self.assertEqual([object_name], object_list)
[docs]
@decorators.idempotent_id('4646ac2d-9bfb-4c7d-a3c5-0f527402b3df')
def test_list_container_contents_with_no_object(self):
"""Test getting empty container contents list"""
container_name = self.create_container()
resp, object_list = self.container_client.list_container_objects(
container_name)
self.assertHeaders(resp, 'Container', 'GET')
self.assertEmpty(object_list)
[docs]
@decorators.idempotent_id('fe323a32-57b9-4704-a996-2e68f83b09bc')
def test_list_container_contents_with_delimiter(self):
"""Test getting container contents list using delimiter param"""
container_name = self.create_container()
object_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestObject/')
self.create_object(container_name, object_name)
params = {'delimiter': '/'}
resp, object_list = self.container_client.list_container_objects(
container_name,
params=params)
self.assertHeaders(resp, 'Container', 'GET')
self.assertEqual([object_name.split('/')[0] + '/'], object_list)
[docs]
@decorators.idempotent_id('55b4fa5c-e12e-4ca9-8fcf-a79afe118522')
def test_list_container_contents_with_end_marker(self):
"""Test getting container contents list using end_marker param"""
container_name = self.create_container()
object_name, _ = self.create_object(container_name)
params = {'end_marker': object_name + 'zzzz'}
resp, object_list = self.container_client.list_container_objects(
container_name,
params=params)
self.assertHeaders(resp, 'Container', 'GET')
self.assertEqual([object_name], object_list)
[docs]
@decorators.idempotent_id('196f5034-6ab0-4032-9da9-a937bbb9fba9')
def test_list_container_contents_with_format_json(self):
"""Test getting container contents list using format_json param"""
container_name = self.create_container()
self.create_object(container_name)
params = {'format': 'json'}
resp, object_list = self.container_client.list_container_objects(
container_name,
params=params)
self.assertHeaders(resp, 'Container', 'GET')
self.assertIsNotNone(object_list)
self.assertTrue([c['name'] for c in object_list])
self.assertTrue([c['hash'] for c in object_list])
self.assertTrue([c['bytes'] for c in object_list])
self.assertTrue([c['content_type'] for c in object_list])
self.assertTrue([c['last_modified'] for c in object_list])
[docs]
@decorators.idempotent_id('655a53ca-4d15-408c-a377-f4c6dbd0a1fa')
def test_list_container_contents_with_format_xml(self):
"""Test getting container contents list using format_xml param"""
container_name = self.create_container()
self.create_object(container_name)
params = {'format': 'xml'}
resp, object_list = self.container_client.list_container_objects(
container_name,
params=params)
self.assertHeaders(resp, 'Container', 'GET')
self.assertIsNotNone(object_list)
self.assertEqual(object_list.tag, 'container')
self.assertIn('name', object_list.keys())
self.assertEqual(object_list.find(".//object").tag, 'object')
self.assertEqual(object_list.find(".//name").tag, 'name')
self.assertEqual(object_list.find(".//hash").tag, 'hash')
self.assertEqual(object_list.find(".//bytes").tag, 'bytes')
self.assertEqual(object_list.find(".//content_type").tag,
'content_type')
self.assertEqual(object_list.find(".//last_modified").tag,
'last_modified')
[docs]
@decorators.idempotent_id('297ec38b-2b61-4ff4-bcd1-7fa055e97b61')
def test_list_container_contents_with_limit(self):
"""Test getting container contents list using limit param"""
container_name = self.create_container()
object_name, _ = self.create_object(container_name)
params = {'limit': data_utils.rand_int_id(1, 10000)}
resp, object_list = self.container_client.list_container_objects(
container_name,
params=params)
self.assertHeaders(resp, 'Container', 'GET')
self.assertEqual([object_name], object_list)
[docs]
@decorators.idempotent_id('c31ddc63-2a58-4f6b-b25c-94d2937e6867')
def test_list_container_contents_with_marker(self):
"""Test getting container contents list using marker param"""
container_name = self.create_container()
object_name, _ = self.create_object(container_name)
params = {'marker': 'AaaaObject1234567890'}
resp, object_list = self.container_client.list_container_objects(
container_name,
params=params)
self.assertHeaders(resp, 'Container', 'GET')
self.assertEqual([object_name], object_list)
[docs]
@decorators.idempotent_id('58ca6cc9-6af0-408d-aaec-2a6a7b2f0df9')
def test_list_container_contents_with_path(self):
"""Test getting container contents list using path param"""
container_name = self.create_container()
object_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestObject')
object_name = 'Swift/' + object_name
self.create_object(container_name, object_name)
params = {'path': 'Swift'}
resp, object_list = self.container_client.list_container_objects(
container_name,
params=params)
self.assertHeaders(resp, 'Container', 'GET')
self.assertEqual([object_name], object_list)
[docs]
@decorators.idempotent_id('77e742c7-caf2-4ec9-8aa4-f7d509a3344c')
def test_list_container_contents_with_prefix(self):
"""Test getting container contents list using prefix param"""
container_name = self.create_container()
object_name, _ = self.create_object(container_name)
prefix_key = object_name[0:8]
params = {'prefix': prefix_key}
resp, object_list = self.container_client.list_container_objects(
container_name,
params=params)
self.assertHeaders(resp, 'Container', 'GET')
self.assertEqual([object_name], object_list)
[docs]
@decorators.attr(type='smoke')
@decorators.idempotent_id('96e68f0e-19ec-4aa2-86f3-adc6a45e14dd')
def test_list_container_metadata(self):
"""Test listing container metadata"""
container_name = self.create_container()
metadata = {'name': 'Pictures'}
self.container_client.create_update_or_delete_container_metadata(
container_name,
create_update_metadata=metadata)
resp, _ = self.container_client.list_container_metadata(
container_name)
self.assertHeaders(resp, 'Container', 'HEAD')
self.assertIn('x-container-meta-name', resp)
self.assertEqual(resp['x-container-meta-name'], metadata['name'])
[docs]
@decorators.idempotent_id('a2faf936-6b13-4f8d-92a2-c2278355821e')
def test_list_no_container_metadata(self):
"""Test listing container without metadata"""
container_name = self.create_container()
resp, _ = self.container_client.list_container_metadata(
container_name)
self.assertHeaders(resp, 'Container', 'HEAD')
self.assertNotIn('x-container-meta-', str(resp))
[docs]
@decorators.idempotent_id('cf19bc0b-7e16-4a5a-aaed-cb0c2fe8deef')
def test_update_container_metadata_with_create_and_delete_metadata(self):
"""Test updating container with adding and deleting metadata
Send one request of adding and deleting metadata.
"""
container_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestContainer')
metadata_1 = {'X-Container-Meta-test-container-meta1': 'Meta1'}
self.container_client.update_container(container_name, **metadata_1)
self.containers.append(container_name)
metadata_2 = {'test-container-meta2': 'Meta2'}
resp, _ = (
self.container_client.create_update_or_delete_container_metadata(
container_name,
create_update_metadata=metadata_2,
delete_metadata={'test-container-meta1': 'Meta1'}))
self.assertHeaders(resp, 'Container', 'POST')
resp, _ = self.container_client.list_container_metadata(
container_name)
self.assertNotIn('x-container-meta-test-container-meta1', resp)
self.assertIn('x-container-meta-test-container-meta2', resp)
self.assertEqual(resp['x-container-meta-test-container-meta2'],
metadata_2['test-container-meta2'])
[docs]
@decorators.idempotent_id('2ae5f295-4bf1-4e04-bfad-21e54b62cec5')
def test_update_container_metadata_with_create_metadata(self):
"""Test updating container metadata using add metadata"""
container_name = self.create_container()
metadata = {'test-container-meta1': 'Meta1'}
resp, _ = (
self.container_client.create_update_or_delete_container_metadata(
container_name,
create_update_metadata=metadata))
self.assertHeaders(resp, 'Container', 'POST')
resp, _ = self.container_client.list_container_metadata(
container_name)
self.assertIn('x-container-meta-test-container-meta1', resp)
self.assertEqual(resp['x-container-meta-test-container-meta1'],
metadata['test-container-meta1'])
[docs]
@decorators.idempotent_id('3a5ce7d4-6e4b-47d0-9d87-7cd42c325094')
def test_update_container_metadata_with_delete_metadata(self):
"""Test updating container metadata using delete metadata"""
container_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestContainer')
metadata = {'X-Container-Meta-test-container-meta1': 'Meta1'}
self.container_client.update_container(container_name, **metadata)
self.containers.append(container_name)
resp, _ = (
self.container_client.create_update_or_delete_container_metadata(
container_name,
delete_metadata={'test-container-meta1': 'Meta1'}))
self.assertHeaders(resp, 'Container', 'POST')
resp, _ = self.container_client.list_container_metadata(
container_name)
self.assertNotIn('x-container-meta-test-container-meta1', resp)
[docs]
@decorators.idempotent_id('31f40a5f-6a52-4314-8794-cd89baed3040')
def test_update_container_metadata_with_create_metadata_key(self):
"""Test updating container metadata with a blank value of metadata"""
container_name = self.create_container()
metadata = {'test-container-meta1': ''}
resp, _ = (
self.container_client.create_update_or_delete_container_metadata(
container_name,
create_update_metadata=metadata))
self.assertHeaders(resp, 'Container', 'POST')
resp, _ = self.container_client.list_container_metadata(
container_name)
self.assertNotIn('x-container-meta-test-container-meta1', resp)
[docs]
@decorators.idempotent_id('a2e36378-6f1f-43f4-840a-ffd9cfd61914')
def test_update_container_metadata_with_delete_metadata_key(self):
"""Test updating container metadata with a blank value of metadata"""
container_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='TestContainer')
headers = {'X-Container-Meta-test-container-meta1': 'Meta1'}
self.container_client.update_container(container_name, **headers)
self.containers.append(container_name)
metadata = {'test-container-meta1': ''}
resp, _ = (
self.container_client.create_update_or_delete_container_metadata(
container_name,
delete_metadata=metadata))
self.assertHeaders(resp, 'Container', 'POST')
resp, _ = self.container_client.list_container_metadata(container_name)
self.assertNotIn('x-container-meta-test-container-meta1', resp)
