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