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
|
import json
import boto
def append_attr_value(d, attr, attrv):
if attrv and len(str(attrv)) > 0:
d[attr] = attrv
def append_attr(d, k, attr):
try:
attrv = getattr(k, attr)
except:
return
append_attr_value(d, attr, attrv)
def get_attrs(k, attrs):
d = {}
for a in attrs:
append_attr(d, k, a)
return d
def append_query_arg(s, n, v):
if not v:
return s
nv = '{n}={v}'.format(n=n, v=v)
if not s:
return nv
return '{s}&{nv}'.format(s=s, nv=nv)
class KeyJSONEncoder(boto.s3.key.Key):
@staticmethod
def default(k, versioned=False):
attrs = ['bucket', 'name', 'size', 'last_modified', 'metadata', 'cache_control',
'content_type', 'content_disposition', 'content_language',
'owner', 'storage_class', 'md5', 'version_id', 'encrypted',
'delete_marker', 'expiry_date', 'VersionedEpoch', 'RgwxTag']
d = get_attrs(k, attrs)
d['etag'] = k.etag[1:-1]
if versioned:
d['is_latest'] = k.is_latest
return d
class DeleteMarkerJSONEncoder(boto.s3.key.Key):
@staticmethod
def default(k):
attrs = ['name', 'version_id', 'last_modified', 'owner']
d = get_attrs(k, attrs)
d['delete_marker'] = True
d['is_latest'] = k.is_latest
return d
class UserJSONEncoder(boto.s3.user.User):
@staticmethod
def default(k):
attrs = ['id', 'display_name']
return get_attrs(k, attrs)
class BucketJSONEncoder(boto.s3.bucket.Bucket):
@staticmethod
def default(k):
attrs = ['name', 'creation_date']
return get_attrs(k, attrs)
class BotoJSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, boto.s3.key.Key):
return KeyJSONEncoder.default(obj)
if isinstance(obj, boto.s3.deletemarker.DeleteMarker):
return DeleteMarkerJSONEncoder.default(obj)
if isinstance(obj, boto.s3.user.User):
return UserJSONEncoder.default(obj)
if isinstance(obj, boto.s3.prefix.Prefix):
return (lambda x: {'prefix': x.name})(obj)
if isinstance(obj, boto.s3.bucket.Bucket):
return BucketJSONEncoder.default(obj)
return json.JSONEncoder.default(self, obj)
def dump_json(o, cls=BotoJSONEncoder):
return json.dumps(o, cls=cls, indent=4)
def assert_raises(excClass, callableObj, *args, **kwargs):
"""
Like unittest.TestCase.assertRaises, but returns the exception.
"""
try:
callableObj(*args, **kwargs)
except excClass as e:
return e
else:
if hasattr(excClass, '__name__'):
excName = excClass.__name__
else:
excName = str(excClass)
raise AssertionError("%s not raised" % excName)
|
