Source code for object_storage.test_account_bulk

# Copyright 2013 NTT Corporation
#
#    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 tarfile
import tempfile

from tempest.api.object_storage import base
from tempest.common import utils
from tempest.lib import decorators


[docs]class BulkTest(base.BaseObjectTest): """Test bulk operation of archived file""" def setUp(self): super(BulkTest, self).setUp() self.containers = [] def tearDown(self): # NOTE(andreaf) BulkTests needs to cleanup containers after each # test is executed. base.delete_containers(self.containers, self.container_client, self.object_client) super(BulkTest, self).tearDown() def _create_archive(self): # Create an archived file for bulk upload testing. # Directory and files contained in the directory correspond to # container and subsidiary objects. tmp_dir = tempfile.mkdtemp() tmp_file = tempfile.mkstemp(dir=tmp_dir) # Extract a container name and an object name container_name = tmp_dir.split("/")[-1] object_name = tmp_file[1].split("/")[-1] # Create tar file tarpath = tempfile.NamedTemporaryFile(suffix=".tar") tar = tarfile.open(None, 'w', tarpath) tar.add(tmp_dir, arcname=container_name) tar.close() tarpath.flush() return tarpath.name, container_name, object_name def _upload_archive(self, filepath): # upload an archived file with open(filepath) as fh: mydata = fh.read() resp = self.bulk_client.upload_archive( upload_path='', data=mydata, archive_file_format='tar') return resp def _check_contents_deleted(self, container_name): param = {'format': 'txt'} resp, body = self.account_client.list_account_containers(param) self.assertHeaders(resp, 'Account', 'GET') self.assertNotIn(container_name, body)
[docs] @decorators.idempotent_id('a407de51-1983-47cc-9f14-47c2b059413c') @utils.requires_ext(extension='bulk_upload', service='object') def test_extract_archive(self): """Test bulk operation of file upload with an archived file""" filepath, container_name, object_name = self._create_archive() resp = self._upload_archive(filepath) self.containers.append(container_name) self.assertHeaders(resp.response, 'Account', 'PUT') param = {'format': 'json'} resp, body = self.account_client.list_account_containers(param) self.assertHeaders(resp, 'Account', 'GET') self.assertIn(container_name, [b['name'] for b in body]) param = {'format': 'json'} resp, contents_list = self.container_client.list_container_objects( container_name, param) self.assertHeaders(resp, 'Container', 'GET') self.assertIn(object_name, [c['name'] for c in contents_list])
[docs] @decorators.idempotent_id('c075e682-0d2a-43b2-808d-4116200d736d') @utils.requires_ext(extension='bulk_delete', service='object') def test_bulk_delete(self): """Test bulk operation of deleting multiple files""" filepath, container_name, object_name = self._create_archive() self._upload_archive(filepath) data = '%s/%s\n%s' % (container_name, object_name, container_name) resp = self.bulk_client.delete_bulk_data(data=data) self.assertHeaders(resp.response, 'Account', 'DELETE') # Check if uploaded contents are completely deleted self._check_contents_deleted(container_name)
[docs] @decorators.idempotent_id('dbea2bcb-efbb-4674-ac8a-a5a0e33d1d79') @utils.requires_ext(extension='bulk_delete', service='object') def test_bulk_delete_by_POST(self): """Test bulk operation of deleting multiple files by HTTP POST""" filepath, container_name, object_name = self._create_archive() self._upload_archive(filepath) data = '%s/%s\n%s' % (container_name, object_name, container_name) resp = self.bulk_client.delete_bulk_data_with_post(data=data) self.assertHeaders(resp.response, 'Account', 'POST') # Check if uploaded contents are completely deleted self._check_contents_deleted(container_name)