17
17
def get_page(self, url, method, postdata, headers):
18
18
self.getPageArgs = (url, method, postdata, headers)
22
22
class RequestTestCase(TXAWSTestCase):
24
creds = AWSCredentials(access_key='fookeyid', secret_key='barsecretkey')
24
creds = AWSCredentials(access_key="fookeyid", secret_key="barsecretkey")
25
25
endpoint = AWSServiceEndpoint("https://s3.amazonaws.com/")
27
27
def test_get_uri_with_endpoint(self):
28
28
endpoint = AWSServiceEndpoint("http://localhost/")
29
request = S3Request('PUT', endpoint=endpoint)
29
request = S3Request("PUT", endpoint=endpoint)
30
30
self.assertEquals(request.endpoint.get_uri(), "http://localhost/")
31
31
self.assertEquals(request.get_uri(), "http://localhost/")
33
33
def test_get_uri_with_endpoint_bucket_and_object(self):
34
34
endpoint = AWSServiceEndpoint("http://localhost/")
35
request = S3Request('PUT', bucket="mybucket", object_name="myobject",
35
request = S3Request("PUT", bucket="mybucket", object_name="myobject",
39
39
"http://localhost/mybucket/myobject")
41
41
def test_get_uri_with_no_endpoint(self):
42
request = S3Request('PUT')
42
request = S3Request("PUT")
43
43
self.assertEquals(request.endpoint, None)
44
44
self.assertEquals(request.get_uri(), "http:///")
46
46
def test_get_path_with_bucket_and_object(self):
47
request = S3Request('PUT', bucket="mybucket", object_name="myobject")
47
request = S3Request("PUT", bucket="mybucket", object_name="myobject")
48
48
self.assertEquals(request.get_path(), "/mybucket/myobject")
50
50
def test_get_path_with_no_bucket_or_object(self):
51
request = S3Request('PUT')
51
request = S3Request("PUT")
52
52
self.assertEquals(request.get_path(), "/")
54
54
def test_objectRequest(self):
56
56
Test that a request addressing an object is created correctly.
59
DIGEST = 'zhdB6gwvocWv/ourYUWMxA=='
59
DIGEST = "zhdB6gwvocWv/ourYUWMxA=="
61
request = S3Request('PUT', 'somebucket', 'object/name/here', DATA,
62
content_type='text/plain', metadata={'foo': 'bar'},
61
request = S3Request("PUT", "somebucket", "object/name/here", DATA,
62
content_type="text/plain", metadata={"foo": "bar"},
63
63
creds=self.creds, endpoint=self.endpoint)
64
64
request.get_signature = lambda headers: "TESTINGSIG="
65
self.assertEqual(request.verb, 'PUT')
65
self.assertEqual(request.verb, "PUT")
68
'https://s3.amazonaws.com/somebucket/object/name/here')
68
"https://s3.amazonaws.com/somebucket/object/name/here")
69
69
headers = request.get_headers()
70
self.assertNotEqual(headers.pop('Date'), '')
70
self.assertNotEqual(headers.pop("Date"), "")
73
'Authorization': 'AWS fookeyid:TESTINGSIG=',
74
'Content-Type': 'text/plain',
75
'Content-Length': len(DATA),
76
'Content-MD5': DIGEST,
77
'x-amz-meta-foo': 'bar'})
78
self.assertEqual(request.data, 'objectData')
73
"Authorization": "AWS fookeyid:TESTINGSIG=",
74
"Content-Type": "text/plain",
75
"Content-Length": len(DATA),
76
"Content-MD5": DIGEST,
77
"x-amz-meta-foo": "bar"})
78
self.assertEqual(request.data, "objectData")
80
80
def test_bucketRequest(self):
82
82
Test that a request addressing a bucket is created correctly.
84
DIGEST = '1B2M2Y8AsgTpgAmY7PhCfg=='
84
DIGEST = "1B2M2Y8AsgTpgAmY7PhCfg=="
86
request = S3Request('GET', 'somebucket', creds=self.creds,
86
request = S3Request("GET", "somebucket", creds=self.creds,
87
87
endpoint=self.endpoint)
88
88
request.get_signature = lambda headers: "TESTINGSIG="
89
self.assertEqual(request.verb, 'GET')
89
self.assertEqual(request.verb, "GET")
91
request.get_uri(), 'https://s3.amazonaws.com/somebucket')
91
request.get_uri(), "https://s3.amazonaws.com/somebucket")
92
92
headers = request.get_headers()
93
self.assertNotEqual(headers.pop('Date'), '')
93
self.assertNotEqual(headers.pop("Date"), "")
96
'Authorization': 'AWS fookeyid:TESTINGSIG=',
98
'Content-MD5': DIGEST})
99
self.assertEqual(request.data, '')
96
"Authorization": "AWS fookeyid:TESTINGSIG=",
98
"Content-MD5": DIGEST})
99
self.assertEqual(request.data, "")
101
101
def test_submit(self):
103
103
Submitting the request should invoke getPage correctly.
105
request = StubbedS3Request('GET', 'somebucket', creds=self.creds,
105
request = StubbedS3Request("GET", "somebucket", creds=self.creds,
106
106
endpoint=self.endpoint)
108
108
def _postCheck(result):
109
self.assertEqual(result, '')
109
self.assertEqual(result, "")
111
111
url, method, postdata, headers = request.getPageArgs
112
112
self.assertEqual(url, request.get_uri())
201
201
marker = object()
203
203
def _cb(*a, **kw):
204
self.assertEqual(kw['creds'], self.creds)
205
self.assertEqual(kw['endpoint'], self.endpoint)
204
self.assertEqual(kw["creds"], self.creds)
205
self.assertEqual(kw["endpoint"], self.endpoint)
208
208
self.s3.request_factory = _cb
209
self.assertIdentical(self.s3.make_request('GET'), marker)
209
self.assertIdentical(self.s3.make_request("GET"), marker)
211
211
def test_list_buckets(self):
212
self.s3.response = samples['ListAllMyBucketsResult']
212
self.s3.response = samples["ListAllMyBucketsResult"]
213
213
d = self.s3.list_buckets()
215
215
req = self.s3._lastRequest
216
216
self.assertTrue(req.submitted)
217
self.assertEqual(req.verb, 'GET')
217
self.assertEqual(req.verb, "GET")
218
218
self.assertEqual(req.bucket, None)
219
219
self.assertEqual(req.object_name, None)
221
221
def _check_result(buckets):
222
222
self.assertEqual(
225
'created': Time.fromDatetime(
225
"created": Time.fromDatetime(
226
226
datetime(2006, 2, 3, 16, 45, 9))},
228
'created': Time.fromDatetime(
228
"created": Time.fromDatetime(
229
229
datetime(2006, 2, 3, 16, 41, 58))}])
230
230
return d.addCallback(_check_result)
232
232
def test_create_bucket(self):
233
self.s3.create_bucket('foo')
233
self.s3.create_bucket("foo")
234
234
req = self.s3._lastRequest
235
235
self.assertTrue(req.submitted)
236
self.assertEqual(req.verb, 'PUT')
237
self.assertEqual(req.bucket, 'foo')
236
self.assertEqual(req.verb, "PUT")
237
self.assertEqual(req.bucket, "foo")
238
238
self.assertEqual(req.object_name, None)
240
240
def test_delete_bucket(self):
241
self.s3.delete_bucket('foo')
241
self.s3.delete_bucket("foo")
242
242
req = self.s3._lastRequest
243
243
self.assertTrue(req.submitted)
244
self.assertEqual(req.verb, 'DELETE')
245
self.assertEqual(req.bucket, 'foo')
244
self.assertEqual(req.verb, "DELETE")
245
self.assertEqual(req.bucket, "foo")
246
246
self.assertEqual(req.object_name, None)
248
248
def test_put_object(self):
249
249
self.s3.put_object(
250
'foobucket', 'foo', 'data', 'text/plain', {'foo': 'bar'})
250
"foobucket", "foo", "data", "text/plain", {"foo": "bar"})
251
251
req = self.s3._lastRequest
252
252
self.assertTrue(req.submitted)
253
self.assertEqual(req.verb, 'PUT')
254
self.assertEqual(req.bucket, 'foobucket')
255
self.assertEqual(req.object_name, 'foo')
256
self.assertEqual(req.data, 'data')
257
self.assertEqual(req.content_type, 'text/plain')
258
self.assertEqual(req.metadata, {'foo': 'bar'})
253
self.assertEqual(req.verb, "PUT")
254
self.assertEqual(req.bucket, "foobucket")
255
self.assertEqual(req.object_name, "foo")
256
self.assertEqual(req.data, "data")
257
self.assertEqual(req.content_type, "text/plain")
258
self.assertEqual(req.metadata, {"foo": "bar"})
260
260
def test_get_object(self):
261
self.s3.get_object('foobucket', 'foo')
261
self.s3.get_object("foobucket", "foo")
262
262
req = self.s3._lastRequest
263
263
self.assertTrue(req.submitted)
264
self.assertEqual(req.verb, 'GET')
265
self.assertEqual(req.bucket, 'foobucket')
266
self.assertEqual(req.object_name, 'foo')
264
self.assertEqual(req.verb, "GET")
265
self.assertEqual(req.bucket, "foobucket")
266
self.assertEqual(req.object_name, "foo")
268
268
def test_head_object(self):
269
self.s3.head_object('foobucket', 'foo')
269
self.s3.head_object("foobucket", "foo")
270
270
req = self.s3._lastRequest
271
271
self.assertTrue(req.submitted)
272
self.assertEqual(req.verb, 'HEAD')
273
self.assertEqual(req.bucket, 'foobucket')
274
self.assertEqual(req.object_name, 'foo')
272
self.assertEqual(req.verb, "HEAD")
273
self.assertEqual(req.bucket, "foobucket")
274
self.assertEqual(req.object_name, "foo")
276
276
def test_delete_object(self):
277
self.s3.delete_object('foobucket', 'foo')
277
self.s3.delete_object("foobucket", "foo")
278
278
req = self.s3._lastRequest
279
279
self.assertTrue(req.submitted)
280
self.assertEqual(req.verb, 'DELETE')
281
self.assertEqual(req.bucket, 'foobucket')
282
self.assertEqual(req.object_name, 'foo')
280
self.assertEqual(req.verb, "DELETE")
281
self.assertEqual(req.bucket, "foobucket")
282
self.assertEqual(req.object_name, "foo")
285
285
class MiscellaneousTests(TXAWSTestCase):
287
287
def test_contentMD5(self):
288
self.assertEqual(calculate_md5('somedata'), 'rvr3UC1SmUw7AZV2NqPN0g==')
288
self.assertEqual(calculate_md5("somedata"), "rvr3UC1SmUw7AZV2NqPN0g==")