Source code for object_storage.test_object_expiry

# 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
#    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.api.object_storage import base
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc

[docs]class ObjectExpiryTest(base.BaseObjectTest): """Test object expiry""" @classmethod def resource_setup(cls): super(ObjectExpiryTest, cls).resource_setup() cls.container_name = cls.create_container() def setUp(self): super(ObjectExpiryTest, self).setUp() # create object self.object_name, _ = self.create_object( self.container_name) @classmethod def resource_cleanup(cls): cls.delete_containers() super(ObjectExpiryTest, cls).resource_cleanup() def _test_object_expiry(self, metadata): # update object metadata resp, _ = \ self.object_client.create_or_update_object_metadata( self.container_name, self.object_name, headers=metadata) # verify object metadata resp, _ = \ self.object_client.list_object_metadata(self.container_name, self.object_name) self.assertHeaders(resp, 'Object', 'HEAD') self.assertIn('x-delete-at', resp) # we want to ensure that we will sleep long enough for things to # actually expire, so figure out how many secs in the future that is. sleepy_time = int(resp['x-delete-at']) - int(time.time()) sleepy_time = sleepy_time if sleepy_time > 0 else 0 resp, _ = self.object_client.get_object(self.container_name, self.object_name) self.assertHeaders(resp, 'Object', 'GET') self.assertIn('x-delete-at', resp) # add several seconds for safety. time.sleep(sleepy_time) # Checking whether object still exists for several seconds: # sometimes object is not deleted immediately, so we are making # get calls for an approximately 1 minute in a total. Get calls # can take 3s each sometimes so we are making the requests in # exponential periodicity for i in range(1, 6): time.sleep(2 ** i) try: self.object_client.get_object(self.container_name, self.object_name) except lib_exc.NotFound: break # object should not be there anymore self.assertRaises(lib_exc.NotFound, self.object_client.get_object, self.container_name, self.object_name)
[docs] @decorators.idempotent_id('fb024a42-37f3-4ba5-9684-4f40a7910b41') def test_get_object_after_expiry_time(self): """Test object is expired after x-delete-after time""" # the 10s is important, because the get calls can take 3s each # some times metadata = {'X-Delete-After': '10'} self._test_object_expiry(metadata)
[docs] @decorators.idempotent_id('e592f18d-679c-48fe-9e36-4be5f47102c5') def test_get_object_at_expiry_time(self): """Test object is expired at x-delete-at time""" metadata = {'X-Delete-At': str(int(time.time()) + 10)} self._test_object_expiry(metadata)