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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
# Copyright (c) 2011-2014 OpenStack Foundation.
#
# 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 unittest
import traceback
from contextlib import contextmanager
import logging
from unittest import SkipTest
import os
import test.functional as tf
from test.functional.s3api.s3_test_client import (
Connection, get_boto3_conn, tear_down_s3)
try:
import boto
except ImportError:
boto = None
def setUpModule():
tf.setup_package()
def tearDownModule():
tf.teardown_package()
class S3ApiBase(unittest.TestCase):
def __init__(self, method_name):
super(S3ApiBase, self).__init__(method_name)
self.method_name = method_name
@contextmanager
def quiet_boto_logging(self):
original_level = logging.getLogger('boto').getEffectiveLevel()
try:
logging.getLogger('boto').setLevel(logging.INFO)
yield
finally:
logging.getLogger('boto').setLevel(original_level)
def setUp(self):
if not tf.config.get('s3_access_key'):
raise SkipTest('no s3api user configured')
if 's3api' not in tf.cluster_info:
raise SkipTest('s3api middleware is not enabled')
if boto is None:
raise SkipTest('boto 2.x library is not installed')
if tf.config.get('account'):
user_id = '%s:%s' % (tf.config['account'], tf.config['username'])
else:
user_id = tf.config['username']
try:
self.conn = Connection(
tf.config['s3_access_key'], tf.config['s3_secret_key'],
user_id=user_id)
self.conn.reset()
except Exception:
message = '%s got an error during initialize process.\n\n%s' % \
(self.method_name, traceback.format_exc())
# TODO: Find a way to make this go to FAIL instead of Error
self.fail(message)
def assertCommonResponseHeaders(self, headers, etag=None):
"""
asserting common response headers with args
:param headers: a dict of response headers
:param etag: a string of md5(content).hexdigest() if not given,
this won't assert anything about etag. (e.g. DELETE obj)
"""
self.assertTrue(headers['x-amz-id-2'] is not None)
self.assertTrue(headers['x-amz-request-id'] is not None)
self.assertTrue(headers['date'] is not None)
# TODO; requires consideration
# self.assertTrue(headers['server'] is not None)
if etag is not None:
self.assertTrue('etag' in headers) # sanity
self.assertEqual(etag, headers['etag'].strip('"'))
class S3ApiBaseBoto3(S3ApiBase):
def setUp(self):
if not tf.config.get('s3_access_key'):
raise SkipTest('no s3api user configured')
if 's3api' not in tf.cluster_info:
raise SkipTest('s3api middleware is not enabled')
try:
self.conn = get_boto3_conn(
tf.config['s3_access_key'], tf.config['s3_secret_key'])
self.endpoint_url = self.conn._endpoint.host
self.access_key = self.conn._request_signer._credentials.access_key
self.region = self.conn._client_config.region_name
tear_down_s3(self.conn)
except Exception:
message = '%s got an error during initialize process.\n\n%s' % \
(self.method_name, traceback.format_exc())
# TODO: Find a way to make this go to FAIL instead of Error
self.fail(message)
def tearDown(self):
tear_down_s3(self.conn)
def skip_boto2_sort_header_bug(m):
def wrapped(self, *args, **kwargs):
if os.environ.get('S3_USE_SIGV4') == "True":
# boto doesn't sort headers for v4 sigs properly; see
# https://github.com/boto/boto/pull/3032
# or https://github.com/boto/boto/pull/3176
# or https://github.com/boto/boto/pull/3751
# or https://github.com/boto/boto/pull/3824
self.skipTest('This stuff got the issue of boto<=2.x')
return m(self, *args, **kwargs)
return wrapped
class SigV4Mixin(object):
@classmethod
def setUpClass(cls):
os.environ['S3_USE_SIGV4'] = "True"
@classmethod
def tearDownClass(cls):
del os.environ['S3_USE_SIGV4']
def setUp(self):
super(SigV4Mixin, self).setUp()
|
