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




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)



    @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)




    @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)