← Back to branch summary

~tribaal/txaws/xss-hardening

« back to all changes in this revision

Viewing changes to txaws/storage/tests/test_client.py

  • Committer: Duncan McGreggor
  • Date: 2009-11-22 02:20:42 UTC
  • mto: (44.3.2 484858-s3-scripts)
  • mto: This revision was merged to the branch mainline in revision 52.
  • Revision ID: duncan@canonical.com-20091122022042-4zi231hxni1z53xd
* Updated the LICENSE file with copyright information.
* Updated the README with license information.

Show diffs side-by-side

added added

removed removed

Lines of Context:
txaws/storage/tests/test_client.py
 
1
from datetime import datetime
 
2
 
 
3
from epsilon.extime import Time
 
4
 
1
5
from twisted.internet.defer import succeed
2
6
 
3
7
from txaws.credentials import AWSCredentials
4
 
try:
5
 
    from txaws.s3 import client
6
 
except ImportError:
7
 
    s3clientSkip = ("S3Client couldn't be imported (perhaps because dateutil, "
8
 
                    "on which it depends, isn't present)")
9
 
else:
10
 
    s3clientSkip = None
11
 
from txaws.s3.acls import AccessControlPolicy
12
 
from txaws.s3.model import (RequestPayment, MultipartInitiationResponse,
13
 
    MultipartCompletionResponse)
14
 
from txaws.testing.producers import StringBodyProducer
15
8
from txaws.service import AWSServiceEndpoint
16
 
from txaws.testing import payload
 
9
from txaws.storage.client import S3, S3Request
17
10
from txaws.testing.base import TXAWSTestCase
18
11
from txaws.util import calculate_md5
19
12
 
20
13
 
21
 
class URLContextTestCase(TXAWSTestCase):
22
 
 
 
14
 
 
15
class StubbedS3Request(S3Request):
 
16
 
 
17
    def get_page(self, url, method, postdata, headers):
 
18
        self.getPageArgs = (url, method, postdata, headers)
 
19
        return succeed("")
 
20
 
 
21
 
 
22
class RequestTestCase(TXAWSTestCase):
 
23
 
 
24
    creds = AWSCredentials(access_key="fookeyid", secret_key="barsecretkey")
23
25
    endpoint = AWSServiceEndpoint("https://s3.amazonaws.com/")
24
26
 
25
 
    def test_get_host_with_no_bucket(self):
26
 
        url_context = client.URLContext(self.endpoint)
27
 
        self.assertEquals(url_context.get_host(), "s3.amazonaws.com")
28
 
 
29
 
    def test_get_host_with_bucket(self):
30
 
        url_context = client.URLContext(self.endpoint, "mystuff")
31
 
        self.assertEquals(url_context.get_host(), "s3.amazonaws.com")
32
 
 
33
 
    def test_get_path_with_no_bucket(self):
34
 
        url_context = client.URLContext(self.endpoint)
35
 
        self.assertEquals(url_context.get_path(), "/")
36
 
 
37
 
    def test_get_path_with_bucket(self):
38
 
        url_context = client.URLContext(self.endpoint, bucket="mystuff")
39
 
        self.assertEquals(url_context.get_path(), "/mystuff/")
 
27
    def test_get_uri_with_endpoint(self):
 
28
        endpoint = AWSServiceEndpoint("http://localhost/")
 
29
        request = S3Request("PUT", endpoint=endpoint)
 
30
        self.assertEquals(request.endpoint.get_uri(), "http://localhost/")
 
31
        self.assertEquals(request.get_uri(), "http://localhost/")
 
32
 
 
33
    def test_get_uri_with_endpoint_bucket_and_object(self):
 
34
        endpoint = AWSServiceEndpoint("http://localhost/")
 
35
        request = S3Request("PUT", bucket="mybucket", object_name="myobject",
 
36
                            endpoint=endpoint)
 
37
        self.assertEquals(
 
38
            request.get_uri(),
 
39
            "http://localhost/mybucket/myobject")
 
40
 
 
41
    def test_get_uri_with_no_endpoint(self):
 
42
        request = S3Request("PUT")
 
43
        self.assertEquals(request.endpoint, None)
 
44
        self.assertEquals(request.get_uri(), "http:///")
40
45
 
41
46
    def test_get_path_with_bucket_and_object(self):
42
 
        url_context = client.URLContext(
43
 
            self.endpoint, bucket="mystuff", object_name="/images/thing.jpg")
44
 
        self.assertEquals(url_context.get_host(), "s3.amazonaws.com")
45
 
        self.assertEquals(url_context.get_path(), "/mystuff/images/thing.jpg")
46
 
 
47
 
    def test_get_path_with_bucket_and_object_without_slash(self):
48
 
        url_context = client.URLContext(
49
 
            self.endpoint, bucket="mystuff", object_name="images/thing.jpg")
50
 
        self.assertEquals(url_context.get_host(), "s3.amazonaws.com")
51
 
        self.assertEquals(url_context.get_path(), "/mystuff/images/thing.jpg")
52
 
 
53
 
    def test_get_url_with_custom_endpoint(self):
54
 
        endpoint = AWSServiceEndpoint("http://localhost/")
55
 
        url_context = client.URLContext(endpoint)
56
 
        self.assertEquals(url_context.endpoint.get_uri(), "http://localhost/")
57
 
        self.assertEquals(url_context.get_url(), "http://localhost/")
58
 
 
59
 
    def test_get_uri_with_endpoint_bucket_and_object(self):
60
 
        endpoint = AWSServiceEndpoint("http://localhost/")
61
 
        url_context = client.URLContext(
62
 
            endpoint, bucket="mydocs", object_name="notes.txt")
63
 
        self.assertEquals(
64
 
            url_context.get_url(),
65
 
            "http://localhost/mydocs/notes.txt")
66
 
 
67
 
    def test_custom_port_endpoint(self):
68
 
        test_uri = 'http://0.0.0.0:12345/'
69
 
        endpoint = AWSServiceEndpoint(uri=test_uri)
70
 
        self.assertEquals(endpoint.port, 12345)
71
 
        self.assertEquals(endpoint.scheme, 'http')
72
 
        context = client.URLContext(service_endpoint=endpoint,
73
 
                bucket="foo",
74
 
                object_name="bar")
75
 
        self.assertEquals(context.get_host(), '0.0.0.0')
76
 
        self.assertEquals(context.get_url(), test_uri + 'foo/bar')
77
 
 
78
 
    def test_custom_port_endpoint_https(self):
79
 
        test_uri = 'https://0.0.0.0:12345/'
80
 
        endpoint = AWSServiceEndpoint(uri=test_uri)
81
 
        self.assertEquals(endpoint.port, 12345)
82
 
        self.assertEquals(endpoint.scheme, 'https')
83
 
        context = client.URLContext(service_endpoint=endpoint,
84
 
                bucket="foo",
85
 
                object_name="bar")
86
 
        self.assertEquals(context.get_host(), '0.0.0.0')
87
 
        self.assertEquals(context.get_url(), test_uri + 'foo/bar')
88
 
 
89
 
 
90
 
URLContextTestCase.skip = s3clientSkip
91
 
 
92
 
 
93
 
class S3ClientTestCase(TXAWSTestCase):
 
47
        request = S3Request("PUT", bucket="mybucket", object_name="myobject")
 
48
        self.assertEquals(request.get_path(), "/mybucket/myobject")
 
49
 
 
50
    def test_get_path_with_no_bucket_or_object(self):
 
51
        request = S3Request("PUT")
 
52
        self.assertEquals(request.get_path(), "/")
 
53
 
 
54
    def test_objectRequest(self):
 
55
        """
 
56
        Test that a request addressing an object is created correctly.
 
57
        """
 
58
        DATA = "objectData"
 
59
        DIGEST = "zhdB6gwvocWv/ourYUWMxA=="
 
60
 
 
61
        request = S3Request("PUT", "somebucket", "object/name/here", DATA,
 
62
                            content_type="text/plain", metadata={"foo": "bar"},
 
63
                            creds=self.creds, endpoint=self.endpoint)
 
64
        request.get_signature = lambda headers: "TESTINGSIG="
 
65
        self.assertEqual(request.verb, "PUT")
 
66
        self.assertEqual(
 
67
            request.get_uri(),
 
68
            "https://s3.amazonaws.com/somebucket/object/name/here")
 
69
        headers = request.get_headers()
 
70
        self.assertNotEqual(headers.pop("Date"), "")
 
71
        self.assertEqual(
 
72
            headers, {
 
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")
 
79
 
 
80
    def test_bucketRequest(self):
 
81
        """
 
82
        Test that a request addressing a bucket is created correctly.
 
83
        """
 
84
        DIGEST = "1B2M2Y8AsgTpgAmY7PhCfg=="
 
85
 
 
86
        request = S3Request("GET", "somebucket", creds=self.creds,
 
87
                            endpoint=self.endpoint)
 
88
        request.get_signature = lambda headers: "TESTINGSIG="
 
89
        self.assertEqual(request.verb, "GET")
 
90
        self.assertEqual(
 
91
            request.get_uri(), "https://s3.amazonaws.com/somebucket")
 
92
        headers = request.get_headers()
 
93
        self.assertNotEqual(headers.pop("Date"), "")
 
94
        self.assertEqual(
 
95
            headers, {
 
96
            "Authorization": "AWS fookeyid:TESTINGSIG=",
 
97
            "Content-Length": 0,
 
98
            "Content-MD5": DIGEST})
 
99
        self.assertEqual(request.data, "")
 
100
 
 
101
    def test_submit(self):
 
102
        """
 
103
        Submitting the request should invoke getPage correctly.
 
104
        """
 
105
        request = StubbedS3Request("GET", "somebucket", creds=self.creds,
 
106
                                   endpoint=self.endpoint)
 
107
 
 
108
        def _postCheck(result):
 
109
            self.assertEqual(result, "")
 
110
 
 
111
            url, method, postdata, headers = request.getPageArgs
 
112
            self.assertEqual(url, request.get_uri())
 
113
            self.assertEqual(method, request.verb)
 
114
            self.assertEqual(postdata, request.data)
 
115
            self.assertEqual(headers, request.get_headers())
 
116
 
 
117
        return request.submit().addCallback(_postCheck)
 
118
 
 
119
    def test_authenticationTestCases(self):
 
120
        request = S3Request("GET", creds=self.creds, endpoint=self.endpoint)
 
121
        request.get_signature = lambda headers: "TESTINGSIG="
 
122
        request.date = "Wed, 28 Mar 2007 01:29:59 +0000"
 
123
 
 
124
        headers = request.get_headers()
 
125
        self.assertEqual(
 
126
            headers["Authorization"], 
 
127
            "AWS fookeyid:TESTINGSIG=")
 
128
 
 
129
 
 
130
class InertRequest(S3Request):
 
131
    """
 
132
    Inert version of S3Request.
 
133
 
 
134
    The submission action is stubbed out to return the provided response.
 
135
    """
 
136
    submitted = False
 
137
 
 
138
    def __init__(self, *a, **kw):
 
139
        self.response = kw.pop("response")
 
140
        super(InertRequest, self).__init__(*a, **kw)
 
141
 
 
142
    def submit(self):
 
143
        """
 
144
        Return the canned result instead of performing a network operation.
 
145
        """
 
146
        self.submitted = True
 
147
        return succeed(self.response)
 
148
 
 
149
 
 
150
class TestableS3(S3):
 
151
    """
 
152
    Testable version of S3.
 
153
 
 
154
    This subclass stubs request_factory to use InertRequest, making it easy to
 
155
    assert things about the requests that are created in response to various
 
156
    operations.
 
157
    """
 
158
    response = None
 
159
 
 
160
    def request_factory(self, *a, **kw):
 
161
        req = InertRequest(response=self.response, *a, **kw)
 
162
        self._lastRequest = req
 
163
        return req
 
164
 
 
165
 
 
166
samples = {
 
167
    "ListAllMyBucketsResult":
 
168
    """<?xml version="1.0" encoding="UTF-8"?>
 
169
<ListAllMyBucketsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
 
170
  <Owner>
 
171
    <ID>bcaf1ffd86f41caff1a493dc2ad8c2c281e37522a640e161ca5fb16fd081034f</ID>
 
172
    <DisplayName>webfile</DisplayName>
 
173
  </Owner>
 
174
  <Buckets>
 
175
    <Bucket>
 
176
      <Name>quotes</Name>
 
177
      <CreationDate>2006-02-03T16:45:09.000Z</CreationDate>
 
178
    </Bucket>
 
179
    <Bucket>
 
180
      <Name>samples</Name>
 
181
      <CreationDate>2006-02-03T16:41:58.000Z</CreationDate>
 
182
    </Bucket>
 
183
  </Buckets>
 
184
</ListAllMyBucketsResult>""",
 
185
    }
 
186
 
 
187
 
 
188
class WrapperTests(TXAWSTestCase):
94
189
 
95
190
    def setUp(self):
96
191
        TXAWSTestCase.setUp(self)
97
192
        self.creds = AWSCredentials(
98
193
            access_key="accessKey", secret_key="secretKey")
99
194
        self.endpoint = AWSServiceEndpoint()
 
195
        self.s3 = TestableS3(creds=self.creds, endpoint=self.endpoint)
 
196
 
 
197
    def test_make_request(self):
 
198
        """
 
199
        Test that make_request passes in the credentials object.
 
200
        """
 
201
        marker = object()
 
202
 
 
203
        def _cb(*a, **kw):
 
204
            self.assertEqual(kw["creds"], self.creds)
 
205
            self.assertEqual(kw["endpoint"], self.endpoint)
 
206
            return marker
 
207
 
 
208
        self.s3.request_factory = _cb
 
209
        self.assertIdentical(self.s3.make_request("GET"), marker)
100
210
 
101
211
    def test_list_buckets(self):
102
 
 
103
 
        class StubQuery(client.Query):
104
 
 
105
 
            def __init__(query, action, creds, endpoint,
106
 
                body_producer=None, receiver_factory=None):
107
 
                super(StubQuery, query).__init__(
108
 
                    action=action, creds=creds)
109
 
                self.assertEquals(action, "GET")
110
 
                self.assertEqual(creds.access_key, "foo")
111
 
                self.assertEqual(creds.secret_key, "bar")
112
 
                self.assertEqual(query.bucket, None)
113
 
                self.assertEqual(query.object_name, None)
114
 
                self.assertEqual(query.data, "")
115
 
                self.assertEqual(query.metadata, {})
116
 
 
117
 
            def submit(query):
118
 
                return succeed(payload.sample_list_buckets_result)
119
 
 
120
 
        def check_list_buckets(results):
121
 
            bucket1, bucket2 = results
122
 
            self.assertEquals(bucket1.name, "quotes")
123
 
            self.assertEquals(
124
 
                bucket1.creation_date.timetuple(),
125
 
                (2006, 2, 3, 16, 45, 9, 4, 34, 0))
126
 
            self.assertEquals(bucket2.name, "samples")
127
 
            self.assertEquals(
128
 
                bucket2.creation_date.timetuple(),
129
 
                (2006, 2, 3, 16, 41, 58, 4, 34, 0))
130
 
 
131
 
        creds = AWSCredentials("foo", "bar")
132
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
133
 
        d = s3.list_buckets()
134
 
        return d.addCallback(check_list_buckets)
 
212
        self.s3.response = samples["ListAllMyBucketsResult"]
 
213
        d = self.s3.list_buckets()
 
214
 
 
215
        req = self.s3._lastRequest
 
216
        self.assertTrue(req.submitted)
 
217
        self.assertEqual(req.verb, "GET")
 
218
        self.assertEqual(req.bucket, None)
 
219
        self.assertEqual(req.object_name, None)
 
220
 
 
221
        def _check_result(buckets):
 
222
            self.assertEqual(
 
223
                list(buckets),
 
224
                [{"name": u"quotes",
 
225
                  "created": Time.fromDatetime(
 
226
                    datetime(2006, 2, 3, 16, 45, 9))},
 
227
                 {"name": u"samples",
 
228
                  "created": Time.fromDatetime(
 
229
                    datetime(2006, 2, 3, 16, 41, 58))}])
 
230
        return d.addCallback(_check_result)
135
231
 
136
232
    def test_create_bucket(self):
137
 
 
138
 
        class StubQuery(client.Query):
139
 
 
140
 
            def __init__(query, action, creds, endpoint, bucket=None,
141
 
                body_producer=None, receiver_factory=None):
142
 
                super(StubQuery, query).__init__(
143
 
                    action=action, creds=creds, bucket=bucket)
144
 
                self.assertEquals(action, "PUT")
145
 
                self.assertEqual(creds.access_key, "foo")
146
 
                self.assertEqual(creds.secret_key, "bar")
147
 
                self.assertEqual(query.bucket, "mybucket")
148
 
                self.assertEqual(query.object_name, None)
149
 
                self.assertEqual(query.data, "")
150
 
                self.assertEqual(query.metadata, {})
151
 
 
152
 
            def submit(query, url_context=None):
153
 
                return succeed(None)
154
 
 
155
 
        creds = AWSCredentials("foo", "bar")
156
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
157
 
        return s3.create_bucket("mybucket")
158
 
 
159
 
    def test_get_bucket(self):
160
 
 
161
 
        class StubQuery(client.Query):
162
 
 
163
 
            def __init__(query, action, creds, endpoint, bucket=None,
164
 
                body_producer=None, receiver_factory=None):
165
 
                super(StubQuery, query).__init__(
166
 
                    action=action, creds=creds, bucket=bucket)
167
 
                self.assertEquals(action, "GET")
168
 
                self.assertEqual(creds.access_key, "foo")
169
 
                self.assertEqual(creds.secret_key, "bar")
170
 
                self.assertEqual(query.bucket, "mybucket")
171
 
                self.assertEqual(query.object_name, None)
172
 
                self.assertEqual(query.data, "")
173
 
                self.assertEqual(query.metadata, {})
174
 
 
175
 
            def submit(query, url_context=None):
176
 
                return succeed(payload.sample_get_bucket_result)
177
 
 
178
 
        def check_results(listing):
179
 
            self.assertEquals(listing.name, "mybucket")
180
 
            self.assertEquals(listing.prefix, "N")
181
 
            self.assertEquals(listing.marker, "Ned")
182
 
            self.assertEquals(listing.max_keys, "40")
183
 
            self.assertEquals(listing.is_truncated, "false")
184
 
            self.assertEquals(len(listing.contents), 2)
185
 
            content1 = listing.contents[0]
186
 
            self.assertEquals(content1.key, "Nelson")
187
 
            self.assertEquals(
188
 
                content1.modification_date.timetuple(),
189
 
                (2006, 1, 1, 12, 0, 0, 6, 1, 0))
190
 
            self.assertEquals(
191
 
                content1.etag, '"828ef3fdfa96f00ad9f27c383fc9ac7f"')
192
 
            self.assertEquals(content1.size, "5")
193
 
            self.assertEquals(content1.storage_class, "STANDARD")
194
 
            owner = content1.owner
195
 
            self.assertEquals(owner.id,
196
 
                              "bcaf1ffd86f41caff1a493dc2ad8c2c281e37522a640e16"
197
 
                              "1ca5fb16fd081034f")
198
 
            self.assertEquals(owner.display_name, "webfile")
199
 
 
200
 
        creds = AWSCredentials("foo", "bar")
201
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
202
 
        d = s3.get_bucket("mybucket")
203
 
        return d.addCallback(check_results)
204
 
 
205
 
    def test_get_bucket_location(self):
206
 
        """
207
 
        L{S3Client.get_bucket_location} creates a L{Query} to get a bucket's
208
 
        location.  It parses the returned C{LocationConstraint} XML document
209
 
        and returns a C{Deferred} that requests the bucket's location
210
 
        constraint.
211
 
        """
212
 
 
213
 
        class StubQuery(client.Query):
214
 
 
215
 
            def __init__(query, action, creds, endpoint, bucket=None,
216
 
                         object_name=None, body_producer=None,
217
 
                         receiver_factory=None):
218
 
                super(StubQuery, query).__init__(action=action, creds=creds,
219
 
                                                 bucket=bucket,
220
 
                                                 object_name=object_name)
221
 
                self.assertEquals(action, "GET")
222
 
                self.assertEqual(creds.access_key, "foo")
223
 
                self.assertEqual(creds.secret_key, "bar")
224
 
                self.assertEqual(query.bucket, "mybucket")
225
 
                self.assertEqual(query.object_name, "?location")
226
 
                self.assertEqual(query.data, "")
227
 
                self.assertEqual(query.metadata, {})
228
 
                self.assertEqual(query.amz_headers, {})
229
 
 
230
 
            def submit(query, url_context=None):
231
 
                return succeed(payload.sample_get_bucket_location_result)
232
 
 
233
 
        def check_results(location_constraint):
234
 
            self.assertEquals(location_constraint, "EU")
235
 
 
236
 
        creds = AWSCredentials("foo", "bar")
237
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
238
 
        d = s3.get_bucket_location("mybucket")
239
 
        return d.addCallback(check_results)
240
 
 
241
 
    def test_get_bucket_lifecycle_multiple_rules(self):
242
 
        """
243
 
        L{S3Client.get_bucket_lifecycle} creates a L{Query} to get a bucket's
244
 
        lifecycle.  It parses the returned C{LifecycleConfiguration} XML
245
 
        document and returns a C{Deferred} that requests the bucket's lifecycle
246
 
        configuration with multiple rules.
247
 
        """
248
 
 
249
 
        class StubQuery(client.Query):
250
 
 
251
 
            def __init__(query, action, creds, endpoint, bucket=None,
252
 
                         object_name=None, body_producer=None,
253
 
                         receiver_factory=None):
254
 
                super(StubQuery, query).__init__(action=action, creds=creds,
255
 
                                                 bucket=bucket,
256
 
                                                 object_name=object_name)
257
 
                self.assertEquals(action, "GET")
258
 
                self.assertEqual(creds.access_key, "foo")
259
 
                self.assertEqual(creds.secret_key, "bar")
260
 
                self.assertEqual(query.bucket, "mybucket")
261
 
                self.assertEqual(query.object_name, "?lifecycle")
262
 
                self.assertEqual(query.data, "")
263
 
                self.assertEqual(query.metadata, {})
264
 
                self.assertEqual(query.amz_headers, {})
265
 
 
266
 
            def submit(query, url_context=None):
267
 
                return succeed(payload.
268
 
                    sample_s3_get_bucket_lifecycle_multiple_rules_result)
269
 
 
270
 
        def check_results(lifecycle_config):
271
 
            self.assertTrue(len(lifecycle_config.rules) == 2)
272
 
            rule = lifecycle_config.rules[1]
273
 
            self.assertEquals(rule.id, 'another-id')
274
 
            self.assertEquals(rule.prefix, 'another-logs')
275
 
            self.assertEquals(rule.status, 'Disabled')
276
 
            self.assertEquals(rule.expiration, 37)
277
 
 
278
 
        creds = AWSCredentials("foo", "bar")
279
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
280
 
        d = s3.get_bucket_lifecycle("mybucket")
281
 
        return d.addCallback(check_results)
282
 
 
283
 
    def test_get_bucket_lifecycle(self):
284
 
        """
285
 
        L{S3Client.get_bucket_lifecycle} creates a L{Query} to get a bucket's
286
 
        lifecycle.  It parses the returned C{LifecycleConfiguration} XML
287
 
        document and returns a C{Deferred} that requests the bucket's lifecycle
288
 
        configuration.
289
 
        """
290
 
 
291
 
        class StubQuery(client.Query):
292
 
 
293
 
            def __init__(query, action, creds, endpoint, bucket=None,
294
 
                         object_name=None, body_producer=None,
295
 
                         receiver_factory=None):
296
 
                super(StubQuery, query).__init__(action=action, creds=creds,
297
 
                                                 bucket=bucket,
298
 
                                                 object_name=object_name)
299
 
                self.assertEquals(action, "GET")
300
 
                self.assertEqual(creds.access_key, "foo")
301
 
                self.assertEqual(creds.secret_key, "bar")
302
 
                self.assertEqual(query.bucket, "mybucket")
303
 
                self.assertEqual(query.object_name, "?lifecycle")
304
 
                self.assertEqual(query.data, "")
305
 
                self.assertEqual(query.metadata, {})
306
 
                self.assertEqual(query.amz_headers, {})
307
 
 
308
 
            def submit(query, url_context=None):
309
 
                return succeed(payload.sample_s3_get_bucket_lifecycle_result)
310
 
 
311
 
        def check_results(lifecycle_config):
312
 
            rule = lifecycle_config.rules[0]
313
 
            self.assertEquals(rule.id, '30-day-log-deletion-rule')
314
 
            self.assertEquals(rule.prefix, 'logs')
315
 
            self.assertEquals(rule.status, 'Enabled')
316
 
            self.assertEquals(rule.expiration, 30)
317
 
 
318
 
        creds = AWSCredentials("foo", "bar")
319
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
320
 
        d = s3.get_bucket_lifecycle("mybucket")
321
 
        return d.addCallback(check_results)
322
 
 
323
 
    def test_get_bucket_website_config(self):
324
 
        """
325
 
        L{S3Client.get_bucket_website_config} creates a L{Query} to get a
326
 
        bucket's website configurtion.  It parses the returned
327
 
        C{WebsiteConfiguration} XML document and returns a C{Deferred} that
328
 
        requests the bucket's website configuration.
329
 
        """
330
 
 
331
 
        class StubQuery(client.Query):
332
 
 
333
 
            def __init__(query, action, creds, endpoint, bucket=None,
334
 
                         object_name=None, body_producer=None,
335
 
                         receiver_factory=None):
336
 
                super(StubQuery, query).__init__(action=action, creds=creds,
337
 
                                                 bucket=bucket,
338
 
                                                 object_name=object_name)
339
 
                self.assertEquals(action, "GET")
340
 
                self.assertEqual(creds.access_key, "foo")
341
 
                self.assertEqual(creds.secret_key, "bar")
342
 
                self.assertEqual(query.bucket, "mybucket")
343
 
                self.assertEqual(query.object_name, "?website")
344
 
                self.assertEqual(query.data, "")
345
 
                self.assertEqual(query.metadata, {})
346
 
                self.assertEqual(query.amz_headers, {})
347
 
 
348
 
            def submit(query, url_context=None):
349
 
                return succeed(payload.
350
 
                    sample_s3_get_bucket_website_no_error_result)
351
 
 
352
 
        def check_results(website_config):
353
 
            self.assertEquals(website_config.index_suffix, "index.html")
354
 
            self.assertEquals(website_config.error_key, None)
355
 
 
356
 
        creds = AWSCredentials("foo", "bar")
357
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
358
 
        d = s3.get_bucket_website_config("mybucket")
359
 
        return d.addCallback(check_results)
360
 
 
361
 
    def test_get_bucket_website_config_with_error_doc(self):
362
 
        """
363
 
        L{S3Client.get_bucket_website_config} creates a L{Query} to get a
364
 
        bucket's website configurtion.  It parses the returned
365
 
        C{WebsiteConfiguration} XML document and returns a C{Deferred} that
366
 
        requests the bucket's website configuration with the error document.
367
 
        """
368
 
 
369
 
        class StubQuery(client.Query):
370
 
 
371
 
            def __init__(query, action, creds, endpoint, bucket=None,
372
 
                         object_name=None, body_producer=None,
373
 
                         receiver_factory=None):
374
 
                super(StubQuery, query).__init__(action=action, creds=creds,
375
 
                                                 bucket=bucket,
376
 
                                                 object_name=object_name)
377
 
                self.assertEquals(action, "GET")
378
 
                self.assertEqual(creds.access_key, "foo")
379
 
                self.assertEqual(creds.secret_key, "bar")
380
 
                self.assertEqual(query.bucket, "mybucket")
381
 
                self.assertEqual(query.object_name, "?website")
382
 
                self.assertEqual(query.data, "")
383
 
                self.assertEqual(query.metadata, {})
384
 
                self.assertEqual(query.amz_headers, {})
385
 
 
386
 
            def submit(query, url_context=None):
387
 
                return succeed(payload.sample_s3_get_bucket_website_result)
388
 
 
389
 
        def check_results(website_config):
390
 
            self.assertEquals(website_config.index_suffix, "index.html")
391
 
            self.assertEquals(website_config.error_key, "404.html")
392
 
 
393
 
        creds = AWSCredentials("foo", "bar")
394
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
395
 
        d = s3.get_bucket_website_config("mybucket")
396
 
        return d.addCallback(check_results)
397
 
 
398
 
    def test_get_bucket_notification_config(self):
399
 
        """
400
 
        L{S3Client.get_bucket_notification_config} creates a L{Query} to get a
401
 
        bucket's notification configuration.  It parses the returned
402
 
        C{NotificationConfiguration} XML document and returns a C{Deferred}
403
 
        that requests the bucket's notification configuration.
404
 
        """
405
 
 
406
 
        class StubQuery(client.Query):
407
 
 
408
 
            def __init__(query, action, creds, endpoint, bucket=None,
409
 
                         object_name=None, body_producer=None,
410
 
                         receiver_factory=None):
411
 
                super(StubQuery, query).__init__(action=action, creds=creds,
412
 
                                                 bucket=bucket,
413
 
                                                 object_name=object_name)
414
 
                self.assertEquals(action, "GET")
415
 
                self.assertEqual(creds.access_key, "foo")
416
 
                self.assertEqual(creds.secret_key, "bar")
417
 
                self.assertEqual(query.bucket, "mybucket")
418
 
                self.assertEqual(query.object_name, "?notification")
419
 
                self.assertEqual(query.data, "")
420
 
                self.assertEqual(query.metadata, {})
421
 
                self.assertEqual(query.amz_headers, {})
422
 
 
423
 
            def submit(query, url_context=None):
424
 
                return succeed(payload.
425
 
                               sample_s3_get_bucket_notification_result)
426
 
 
427
 
        def check_results(notification_config):
428
 
            self.assertEquals(notification_config.topic, None)
429
 
            self.assertEquals(notification_config.event, None)
430
 
 
431
 
        creds = AWSCredentials("foo", "bar")
432
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
433
 
        d = s3.get_bucket_notification_config("mybucket")
434
 
        return d.addCallback(check_results)
435
 
 
436
 
    def test_get_bucket_notification_config_with_topic(self):
437
 
        """
438
 
        L{S3Client.get_bucket_notification_config} creates a L{Query} to get a
439
 
        bucket's notification configuration.  It parses the returned
440
 
        C{NotificationConfiguration} XML document and returns a C{Deferred}
441
 
        that requests the bucket's notification configuration with a topic.
442
 
        """
443
 
 
444
 
        class StubQuery(client.Query):
445
 
 
446
 
            def __init__(query, action, creds, endpoint, bucket=None,
447
 
                         object_name=None, body_producer=None,
448
 
                         receiver_factory=None):
449
 
                super(StubQuery, query).__init__(action=action, creds=creds,
450
 
                                                 bucket=bucket,
451
 
                                                 object_name=object_name)
452
 
                self.assertEquals(action, "GET")
453
 
                self.assertEqual(creds.access_key, "foo")
454
 
                self.assertEqual(creds.secret_key, "bar")
455
 
                self.assertEqual(query.bucket, "mybucket")
456
 
                self.assertEqual(query.object_name, "?notification")
457
 
                self.assertEqual(query.data, "")
458
 
                self.assertEqual(query.metadata, {})
459
 
                self.assertEqual(query.amz_headers, {})
460
 
 
461
 
            def submit(query, url_context=None):
462
 
                return succeed(
463
 
                    payload.
464
 
                        sample_s3_get_bucket_notification_with_topic_result)
465
 
 
466
 
        def check_results(notification_config):
467
 
            self.assertEquals(notification_config.topic,
468
 
                              "arn:aws:sns:us-east-1:123456789012:myTopic")
469
 
            self.assertEquals(notification_config.event,
470
 
                              "s3:ReducedRedundancyLostObject")
471
 
 
472
 
        creds = AWSCredentials("foo", "bar")
473
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
474
 
        d = s3.get_bucket_notification_config("mybucket")
475
 
        return d.addCallback(check_results)
476
 
 
477
 
    def test_get_bucket_versioning_config(self):
478
 
        """
479
 
        L{S3Client.get_bucket_versioning_configuration} creates a L{Query} to
480
 
        get a bucket's versioning status.  It parses the returned
481
 
        C{VersioningConfiguration} XML document and returns a C{Deferred} that
482
 
        requests the bucket's versioning configuration.
483
 
        """
484
 
 
485
 
        class StubQuery(client.Query):
486
 
 
487
 
            def __init__(query, action, creds, endpoint, bucket=None,
488
 
                         object_name=None, body_producer=None,
489
 
                         receiver_factory=None):
490
 
                super(StubQuery, query).__init__(action=action, creds=creds,
491
 
                                                 bucket=bucket,
492
 
                                                 object_name=object_name)
493
 
                self.assertEquals(action, "GET")
494
 
                self.assertEqual(creds.access_key, "foo")
495
 
                self.assertEqual(creds.secret_key, "bar")
496
 
                self.assertEqual(query.bucket, "mybucket")
497
 
                self.assertEqual(query.object_name, "?versioning")
498
 
                self.assertEqual(query.data, "")
499
 
                self.assertEqual(query.metadata, {})
500
 
                self.assertEqual(query.amz_headers, {})
501
 
 
502
 
            def submit(query, url_context=None):
503
 
                return succeed(payload.sample_s3_get_bucket_versioning_result)
504
 
 
505
 
        def check_results(versioning_config):
506
 
            self.assertEquals(versioning_config.status, None)
507
 
 
508
 
        creds = AWSCredentials("foo", "bar")
509
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
510
 
        d = s3.get_bucket_versioning_config("mybucket")
511
 
        return d.addCallback(check_results)
512
 
 
513
 
    def test_get_bucket_versioning_config_enabled(self):
514
 
        """
515
 
        L{S3Client.get_bucket_versioning_config} creates a L{Query} to get a
516
 
        bucket's versioning configuration.  It parses the returned
517
 
        C{VersioningConfiguration} XML document and returns a C{Deferred} that
518
 
        requests the bucket's versioning configuration that has a enabled
519
 
        C{Status}.
520
 
        """
521
 
 
522
 
        class StubQuery(client.Query):
523
 
 
524
 
            def __init__(query, action, creds, endpoint, bucket=None,
525
 
                         object_name=None, body_producer=None,
526
 
                         receiver_factory=None):
527
 
                super(StubQuery, query).__init__(action=action, creds=creds,
528
 
                                                 bucket=bucket,
529
 
                                                 object_name=object_name)
530
 
                self.assertEquals(action, "GET")
531
 
                self.assertEqual(creds.access_key, "foo")
532
 
                self.assertEqual(creds.secret_key, "bar")
533
 
                self.assertEqual(query.bucket, "mybucket")
534
 
                self.assertEqual(query.object_name, "?versioning")
535
 
                self.assertEqual(query.data, "")
536
 
                self.assertEqual(query.metadata, {})
537
 
                self.assertEqual(query.amz_headers, {})
538
 
 
539
 
            def submit(query, url_context=None):
540
 
                return succeed(payload.
541
 
                               sample_s3_get_bucket_versioning_enabled_result)
542
 
 
543
 
        def check_results(versioning_config):
544
 
            self.assertEquals(versioning_config.status, 'Enabled')
545
 
 
546
 
        creds = AWSCredentials("foo", "bar")
547
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
548
 
        d = s3.get_bucket_versioning_config("mybucket")
549
 
        return d.addCallback(check_results)
550
 
 
551
 
    def test_get_bucket_versioning_config_mfa_disabled(self):
552
 
        """
553
 
        L{S3Client.get_bucket_versioning_config} creates a L{Query} to get a
554
 
        bucket's versioning configuration.  It parses the returned
555
 
        C{VersioningConfiguration} XML document and returns a C{Deferred} that
556
 
        requests the bucket's versioning configuration that has a disabled
557
 
        C{MfaDelete}.
558
 
        """
559
 
 
560
 
        class StubQuery(client.Query):
561
 
 
562
 
            def __init__(query, action, creds, endpoint, bucket=None,
563
 
                         object_name=None, body_producer=None,
564
 
                         receiver_factory=None):
565
 
                super(StubQuery, query).__init__(action=action, creds=creds,
566
 
                                                 bucket=bucket,
567
 
                                                 object_name=object_name)
568
 
                self.assertEquals(action, "GET")
569
 
                self.assertEqual(creds.access_key, "foo")
570
 
                self.assertEqual(creds.secret_key, "bar")
571
 
                self.assertEqual(query.bucket, "mybucket")
572
 
                self.assertEqual(query.object_name, "?versioning")
573
 
                self.assertEqual(query.data, "")
574
 
                self.assertEqual(query.metadata, {})
575
 
                self.assertEqual(query.amz_headers, {})
576
 
 
577
 
            def submit(query, url_context=None):
578
 
                return succeed(
579
 
                    payload.
580
 
                        sample_s3_get_bucket_versioning_mfa_disabled_result)
581
 
 
582
 
        def check_results(versioning_config):
583
 
            self.assertEquals(versioning_config.mfa_delete, 'Disabled')
584
 
 
585
 
        creds = AWSCredentials("foo", "bar")
586
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
587
 
        d = s3.get_bucket_versioning_config("mybucket")
588
 
        return d.addCallback(check_results)
 
233
        self.s3.create_bucket("foo")
 
234
        req = self.s3._lastRequest
 
235
        self.assertTrue(req.submitted)
 
236
        self.assertEqual(req.verb, "PUT")
 
237
        self.assertEqual(req.bucket, "foo")
 
238
        self.assertEqual(req.object_name, None)
589
239
 
590
240
    def test_delete_bucket(self):
591
 
 
592
 
        class StubQuery(client.Query):
593
 
 
594
 
            def __init__(query, action, creds, endpoint, bucket=None,
595
 
                body_producer=None, receiver_factory=None):
596
 
                super(StubQuery, query).__init__(
597
 
                    action=action, creds=creds, bucket=bucket)
598
 
                self.assertEquals(action, "DELETE")
599
 
                self.assertEqual(creds.access_key, "foo")
600
 
                self.assertEqual(creds.secret_key, "bar")
601
 
                self.assertEqual(query.bucket, "mybucket")
602
 
                self.assertEqual(query.object_name, None)
603
 
                self.assertEqual(query.data, "")
604
 
                self.assertEqual(query.metadata, {})
605
 
 
606
 
            def submit(query, url_context=None):
607
 
                return succeed(None)
608
 
 
609
 
        creds = AWSCredentials("foo", "bar")
610
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
611
 
        return s3.delete_bucket("mybucket")
612
 
 
613
 
    def test_put_bucket_acl(self):
614
 
 
615
 
        class StubQuery(client.Query):
616
 
 
617
 
            def __init__(query, action, creds, endpoint, bucket=None,
618
 
                         object_name=None, data="", body_producer=None,
619
 
                         receiver_factory=None):
620
 
                super(StubQuery, query).__init__(action=action, creds=creds,
621
 
                                                 bucket=bucket,
622
 
                                                 object_name=object_name,
623
 
                    data=data)
624
 
                self.assertEquals(action, "PUT")
625
 
                self.assertEqual(creds.access_key, "foo")
626
 
                self.assertEqual(creds.secret_key, "bar")
627
 
                self.assertEqual(query.bucket, "mybucket")
628
 
                self.assertEqual(query.object_name, "?acl")
629
 
                self.assertEqual(query.data,
630
 
                                 payload.sample_access_control_policy_result)
631
 
                self.assertEqual(query.metadata, {})
632
 
 
633
 
            def submit(query, url_context=None):
634
 
                return succeed(payload.sample_access_control_policy_result)
635
 
 
636
 
        def check_result(result):
637
 
            self.assert_(isinstance(result, AccessControlPolicy))
638
 
 
639
 
        creds = AWSCredentials("foo", "bar")
640
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
641
 
        policy = AccessControlPolicy.from_xml(
642
 
            payload.sample_access_control_policy_result)
643
 
        return s3.put_bucket_acl("mybucket", policy).addCallback(check_result)
644
 
 
645
 
    def test_get_bucket_acl(self):
646
 
 
647
 
        class StubQuery(client.Query):
648
 
 
649
 
            def __init__(query, action, creds, endpoint, bucket=None,
650
 
                         object_name=None, data="", receiver_factory=None,
651
 
                         body_producer=None):
652
 
                super(StubQuery, query).__init__(action=action, creds=creds,
653
 
                                                 bucket=bucket,
654
 
                                                 object_name=object_name,
655
 
                                                 data=data)
656
 
                self.assertEquals(action, "GET")
657
 
                self.assertEqual(creds.access_key, "foo")
658
 
                self.assertEqual(creds.secret_key, "bar")
659
 
                self.assertEqual(query.bucket, "mybucket")
660
 
                self.assertEqual(query.object_name, "?acl")
661
 
                self.assertEqual(query.data, "")
662
 
                self.assertEqual(query.metadata, {})
663
 
 
664
 
            def submit(query, url_context=None):
665
 
                return succeed(payload.sample_access_control_policy_result)
666
 
 
667
 
        def check_result(result):
668
 
            self.assert_(isinstance(result, AccessControlPolicy))
669
 
 
670
 
        creds = AWSCredentials("foo", "bar")
671
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
672
 
        return s3.get_bucket_acl("mybucket").addCallback(check_result)
673
 
 
674
 
    def test_put_request_payment(self):
675
 
        """
676
 
        L{S3Client.put_request_payment} creates a L{Query} to set payment
677
 
        information.  An C{RequestPaymentConfiguration} XML document is built
678
 
        and sent to the endpoint and a C{Deferred} is returned that fires with
679
 
        the results of the request.
680
 
        """
681
 
 
682
 
        class StubQuery(client.Query):
683
 
 
684
 
            def __init__(query, action, creds, endpoint, bucket=None,
685
 
                object_name=None, data=None, content_type=None,
686
 
                metadata=None, body_producer=None, receiver_factory=None):
687
 
                super(StubQuery, query).__init__(
688
 
                    action=action, creds=creds, bucket=bucket,
689
 
                    object_name=object_name, data=data,
690
 
                    content_type=content_type, metadata=metadata)
691
 
                self.assertEqual(action, "PUT")
692
 
                self.assertEqual(creds.access_key, "foo")
693
 
                self.assertEqual(creds.secret_key, "bar")
694
 
                self.assertEqual(query.bucket, "mybucket")
695
 
                self.assertEqual(query.object_name, "?requestPayment")
696
 
                xml = ("<RequestPaymentConfiguration "
697
 
                         'xmlns="http://s3.amazonaws.com/doc/2006-03-01/">\n'
698
 
                       "  <Payer>Requester</Payer>\n"
699
 
                       "</RequestPaymentConfiguration>")
700
 
                self.assertEqual(query.data, xml)
701
 
                self.assertEqual(query.metadata, None)
702
 
 
703
 
            def submit(query):
704
 
                return succeed(None)
705
 
 
706
 
        creds = AWSCredentials("foo", "bar")
707
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
708
 
        return s3.put_request_payment("mybucket", "Requester")
709
 
 
710
 
    def test_get_request_payment(self):
711
 
        """
712
 
        L{S3Client.get_request_payment} creates a L{Query} to get payment
713
 
        information.  It parses the returned C{RequestPaymentConfiguration}
714
 
        XML document and returns a C{Deferred} that fires with the payer's
715
 
        name.
716
 
        """
717
 
 
718
 
        class StubQuery(client.Query):
719
 
 
720
 
            def __init__(query, action, creds, endpoint, bucket=None,
721
 
                object_name=None, data=None, content_type=None,
722
 
                metadata=None, body_producer=None, receiver_factory=None):
723
 
                super(StubQuery, query).__init__(
724
 
                    action=action, creds=creds, bucket=bucket,
725
 
                    object_name=object_name, data=data,
726
 
                    content_type=content_type, metadata=metadata)
727
 
                self.assertEqual(action, "GET")
728
 
                self.assertEqual(creds.access_key, "foo")
729
 
                self.assertEqual(creds.secret_key, "bar")
730
 
                self.assertEqual(query.bucket, "mybucket")
731
 
                self.assertEqual(query.object_name, "?requestPayment")
732
 
                self.assertEqual(query.metadata, None)
733
 
 
734
 
            def submit(query):
735
 
                return succeed(payload.sample_request_payment)
736
 
 
737
 
        def check_request_payment(result):
738
 
            self.assertEquals(result, "Requester")
739
 
 
740
 
        creds = AWSCredentials("foo", "bar")
741
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
742
 
        deferred = s3.get_request_payment("mybucket")
743
 
        return deferred.addCallback(check_request_payment)
 
241
        self.s3.delete_bucket("foo")
 
242
        req = self.s3._lastRequest
 
243
        self.assertTrue(req.submitted)
 
244
        self.assertEqual(req.verb, "DELETE")
 
245
        self.assertEqual(req.bucket, "foo")
 
246
        self.assertEqual(req.object_name, None)
744
247
 
745
248
    def test_put_object(self):
746
 
 
747
 
        class StubQuery(client.Query):
748
 
 
749
 
            def __init__(query, action, creds, endpoint, bucket=None,
750
 
                object_name=None, data=None, content_type=None,
751
 
                metadata=None, amz_headers=None, body_producer=None,
752
 
                receiver_factory=None):
753
 
                super(StubQuery, query).__init__(
754
 
                    action=action, creds=creds, bucket=bucket,
755
 
                    object_name=object_name, data=data,
756
 
                    content_type=content_type, metadata=metadata,
757
 
                    amz_headers=amz_headers)
758
 
                self.assertEqual(action, "PUT")
759
 
                self.assertEqual(creds.access_key, "foo")
760
 
                self.assertEqual(creds.secret_key, "bar")
761
 
                self.assertEqual(query.bucket, "mybucket")
762
 
                self.assertEqual(query.object_name, "objectname")
763
 
                self.assertEqual(query.data, "some data")
764
 
                self.assertEqual(query.content_type, "text/plain")
765
 
                self.assertEqual(query.metadata, {"key": "some meta data"})
766
 
                self.assertEqual(query.amz_headers, {"acl": "public-read"})
767
 
 
768
 
            def submit(query):
769
 
                return succeed(None)
770
 
 
771
 
        creds = AWSCredentials("foo", "bar")
772
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
773
 
        return s3.put_object("mybucket", "objectname", "some data",
774
 
                             content_type="text/plain",
775
 
                             metadata={"key": "some meta data"},
776
 
                             amz_headers={"acl": "public-read"})
777
 
 
778
 
    def test_put_object_with_custom_body_producer(self):
779
 
 
780
 
        class StubQuery(client.Query):
781
 
 
782
 
            def __init__(query, action, creds, endpoint, bucket=None,
783
 
                object_name=None, data=None, content_type=None,
784
 
                metadata=None, amz_headers=None, body_producer=None,
785
 
                receiver_factory=None):
786
 
                super(StubQuery, query).__init__(
787
 
                    action=action, creds=creds, bucket=bucket,
788
 
                    object_name=object_name, data=data,
789
 
                    content_type=content_type, metadata=metadata,
790
 
                    amz_headers=amz_headers, body_producer=body_producer)
791
 
                self.assertEqual(action, "PUT")
792
 
                self.assertEqual(creds.access_key, "foo")
793
 
                self.assertEqual(creds.secret_key, "bar")
794
 
                self.assertEqual(query.bucket, "mybucket")
795
 
                self.assertEqual(query.object_name, "objectname")
796
 
                self.assertEqual(query.content_type, "text/plain")
797
 
                self.assertEqual(query.metadata, {"key": "some meta data"})
798
 
                self.assertEqual(query.amz_headers, {"acl": "public-read"})
799
 
                self.assertIdentical(body_producer, string_producer)
800
 
 
801
 
            def submit(query):
802
 
                return succeed(None)
803
 
 
804
 
 
805
 
        string_producer = StringBodyProducer("some data")
806
 
        creds = AWSCredentials("foo", "bar")
807
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
808
 
        return s3.put_object("mybucket", "objectname",
809
 
                             content_type="text/plain",
810
 
                             metadata={"key": "some meta data"},
811
 
                             amz_headers={"acl": "public-read"},
812
 
                             body_producer=string_producer)
813
 
 
814
 
    def test_copy_object(self):
815
 
        """
816
 
        L{S3Client.copy_object} creates a L{Query} to copy an object from one
817
 
        bucket to another.
818
 
        """
819
 
 
820
 
        class StubQuery(client.Query):
821
 
 
822
 
            def __init__(query, action, creds, endpoint, bucket=None,
823
 
                object_name=None, data=None, content_type=None,
824
 
                metadata=None, amz_headers=None, body_producer=None,
825
 
                receiver_factory=None):
826
 
                super(StubQuery, query).__init__(
827
 
                    action=action, creds=creds, bucket=bucket,
828
 
                    object_name=object_name, data=data,
829
 
                    content_type=content_type, metadata=metadata,
830
 
                    amz_headers=amz_headers)
831
 
                self.assertEqual(action, "PUT")
832
 
                self.assertEqual(creds.access_key, "foo")
833
 
                self.assertEqual(creds.secret_key, "bar")
834
 
                self.assertEqual(query.bucket, "newbucket")
835
 
                self.assertEqual(query.object_name, "newobjectname")
836
 
                self.assertEqual(query.data, None)
837
 
                self.assertEqual(query.content_type, None)
838
 
                self.assertEqual(query.metadata, {"key": "some meta data"})
839
 
                self.assertEqual(query.amz_headers,
840
 
                                 {"copy-source": "/mybucket/objectname"})
841
 
 
842
 
            def submit(query):
843
 
                return succeed(None)
844
 
 
845
 
        creds = AWSCredentials("foo", "bar")
846
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
847
 
        return s3.copy_object("mybucket", "objectname", "newbucket",
848
 
                              "newobjectname",
849
 
                              metadata={"key": "some meta data"})
 
249
        self.s3.put_object(
 
250
            "foobucket", "foo", "data", "text/plain", {"foo": "bar"})
 
251
        req = self.s3._lastRequest
 
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"})
850
259
 
851
260
    def test_get_object(self):
852
 
 
853
 
        class StubQuery(client.Query):
854
 
 
855
 
            def __init__(query, action, creds, endpoint, bucket=None,
856
 
                object_name=None, data=None, content_type=None,
857
 
                metadata=None, amz_headers=None, body_producer=None,
858
 
                receiver_factory=None):
859
 
                super(StubQuery, query).__init__(
860
 
                    action=action, creds=creds, bucket=bucket,
861
 
                    object_name=object_name, data=data,
862
 
                    content_type=content_type, metadata=metadata)
863
 
                self.assertEqual(action, "GET")
864
 
                self.assertEqual(creds.access_key, "foo")
865
 
                self.assertEqual(creds.secret_key, "bar")
866
 
                self.assertEqual(query.bucket, "mybucket")
867
 
                self.assertEqual(query.object_name, "objectname")
868
 
 
869
 
            def submit(query):
870
 
                return succeed(None)
871
 
 
872
 
        creds = AWSCredentials("foo", "bar")
873
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
874
 
        return s3.get_object("mybucket", "objectname")
 
261
        self.s3.get_object("foobucket", "foo")
 
262
        req = self.s3._lastRequest
 
263
        self.assertTrue(req.submitted)
 
264
        self.assertEqual(req.verb, "GET")
 
265
        self.assertEqual(req.bucket, "foobucket")
 
266
        self.assertEqual(req.object_name, "foo")
875
267
 
876
268
    def test_head_object(self):
877
 
 
878
 
        class StubQuery(client.Query):
879
 
 
880
 
            def __init__(query, action, creds, endpoint, bucket=None,
881
 
                object_name=None, data=None, content_type=None,
882
 
                metadata=None, body_producer=None, receiver_factory=None):
883
 
                super(StubQuery, query).__init__(
884
 
                    action=action, creds=creds, bucket=bucket,
885
 
                    object_name=object_name, data=data,
886
 
                    content_type=content_type, metadata=metadata)
887
 
                self.assertEqual(action, "HEAD")
888
 
                self.assertEqual(creds.access_key, "foo")
889
 
                self.assertEqual(creds.secret_key, "bar")
890
 
                self.assertEqual(query.bucket, "mybucket")
891
 
                self.assertEqual(query.object_name, "objectname")
892
 
 
893
 
            def submit(query):
894
 
                return succeed(None)
895
 
 
896
 
        creds = AWSCredentials("foo", "bar")
897
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
898
 
        return s3.head_object("mybucket", "objectname")
 
269
        self.s3.head_object("foobucket", "foo")
 
270
        req = self.s3._lastRequest
 
271
        self.assertTrue(req.submitted)
 
272
        self.assertEqual(req.verb, "HEAD")
 
273
        self.assertEqual(req.bucket, "foobucket")
 
274
        self.assertEqual(req.object_name, "foo")
899
275
 
900
276
    def test_delete_object(self):
901
 
 
902
 
        class StubQuery(client.Query):
903
 
 
904
 
            def __init__(query, action, creds, endpoint, bucket=None,
905
 
                object_name=None, data=None, content_type=None,
906
 
                metadata=None, body_producer=None, receiver_factory=None):
907
 
                super(StubQuery, query).__init__(
908
 
                    action=action, creds=creds, bucket=bucket,
909
 
                    object_name=object_name, data=data,
910
 
                    content_type=content_type, metadata=metadata)
911
 
                self.assertEqual(action, "DELETE")
912
 
                self.assertEqual(creds.access_key, "foo")
913
 
                self.assertEqual(creds.secret_key, "bar")
914
 
                self.assertEqual(query.bucket, "mybucket")
915
 
                self.assertEqual(query.object_name, "objectname")
916
 
 
917
 
            def submit(query):
918
 
                return succeed(None)
919
 
 
920
 
        creds = AWSCredentials("foo", "bar")
921
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
922
 
        return s3.delete_object("mybucket", "objectname")
923
 
 
924
 
    def test_put_object_acl(self):
925
 
 
926
 
        class StubQuery(client.Query):
927
 
 
928
 
            def __init__(query, action, creds, endpoint, bucket=None,
929
 
                         object_name=None, data="", body_producer=None,
930
 
                         receiver_factory=None):
931
 
                super(StubQuery, query).__init__(action=action, creds=creds,
932
 
                                                 bucket=bucket,
933
 
                                                 object_name=object_name,
934
 
                                                 data=data)
935
 
                self.assertEquals(action, "PUT")
936
 
                self.assertEqual(creds.access_key, "foo")
937
 
                self.assertEqual(creds.secret_key, "bar")
938
 
                self.assertEqual(query.bucket, "mybucket")
939
 
                self.assertEqual(query.object_name, "myobject?acl")
940
 
                self.assertEqual(query.data,
941
 
                                 payload.sample_access_control_policy_result)
942
 
                self.assertEqual(query.metadata, {})
943
 
                self.assertEqual(query.metadata, {})
944
 
 
945
 
            def submit(query, url_context=None):
946
 
                return succeed(payload.sample_access_control_policy_result)
947
 
 
948
 
        def check_result(result):
949
 
            self.assert_(isinstance(result, AccessControlPolicy))
950
 
 
951
 
        creds = AWSCredentials("foo", "bar")
952
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
953
 
        policy = AccessControlPolicy.from_xml(
954
 
            payload.sample_access_control_policy_result)
955
 
        deferred = s3.put_object_acl("mybucket", "myobject", policy)
956
 
        return deferred.addCallback(check_result)
957
 
 
958
 
    def test_get_object_acl(self):
959
 
 
960
 
        class StubQuery(client.Query):
961
 
 
962
 
            def __init__(query, action, creds, endpoint, bucket=None,
963
 
                         object_name=None, data="", body_producer=None,
964
 
                         receiver_factory=None):
965
 
                super(StubQuery, query).__init__(action=action, creds=creds,
966
 
                                                 bucket=bucket,
967
 
                                                 object_name=object_name,
968
 
                                                 data=data)
969
 
                self.assertEquals(action, "GET")
970
 
                self.assertEqual(creds.access_key, "foo")
971
 
                self.assertEqual(creds.secret_key, "bar")
972
 
                self.assertEqual(query.bucket, "mybucket")
973
 
                self.assertEqual(query.object_name, "myobject?acl")
974
 
                self.assertEqual(query.data, "")
975
 
                self.assertEqual(query.metadata, {})
976
 
 
977
 
            def submit(query, url_context=None):
978
 
                return succeed(payload.sample_access_control_policy_result)
979
 
 
980
 
        def check_result(result):
981
 
            self.assert_(isinstance(result, AccessControlPolicy))
982
 
 
983
 
        creds = AWSCredentials("foo", "bar")
984
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
985
 
        deferred = s3.get_object_acl("mybucket", "myobject")
986
 
        return deferred.addCallback(check_result)
987
 
 
988
 
    def test_init_multipart_upload(self):
989
 
 
990
 
        class StubQuery(client.Query):
991
 
 
992
 
            def __init__(query, action, creds, endpoint, bucket=None,
993
 
                         object_name=None, data="", body_producer=None,
994
 
                         content_type=None, receiver_factory=None, metadata={},
995
 
                         amz_headers={}):
996
 
                super(StubQuery, query).__init__(action=action, creds=creds,
997
 
                                                 bucket=bucket,
998
 
                                                 amz_headers=amz_headers,
999
 
                                                 object_name=object_name,
1000
 
                                                 data=data)
1001
 
                self.assertEquals(action, "POST")
1002
 
                self.assertEqual(creds.access_key, "foo")
1003
 
                self.assertEqual(creds.secret_key, "bar")
1004
 
                self.assertEqual(query.bucket, "example-bucket")
1005
 
                self.assertEqual(query.object_name, "example-object?uploads")
1006
 
                self.assertEqual(query.data, "")
1007
 
                self.assertEqual(query.metadata, {})
1008
 
                self.assertEqual(query.amz_headers, {"acl": "public"})
1009
 
 
1010
 
            def submit(query, url_context=None):
1011
 
                return succeed(payload.sample_s3_init_multipart_upload_result)
1012
 
 
1013
 
 
1014
 
        def check_result(result):
1015
 
            self.assert_(isinstance(result, MultipartInitiationResponse))
1016
 
            self.assertEqual(result.bucket, "example-bucket")
1017
 
            self.assertEqual(result.object_name, "example-object")
1018
 
            self.assertEqual(result.upload_id, "deadbeef")
1019
 
 
1020
 
        creds = AWSCredentials("foo", "bar")
1021
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
1022
 
        deferred = s3.init_multipart_upload("example-bucket", "example-object",
1023
 
            amz_headers={"acl": "public"})
1024
 
        return deferred.addCallback(check_result)
1025
 
 
1026
 
    def test_upload_part(self):
1027
 
 
1028
 
        class StubQuery(client.Query):
1029
 
 
1030
 
            def __init__(query, action, creds, endpoint, bucket=None,
1031
 
                         object_name=None, data="", body_producer=None,
1032
 
                         content_type=None, receiver_factory=None, metadata={}):
1033
 
                super(StubQuery, query).__init__(action=action, creds=creds,
1034
 
                                                 bucket=bucket,
1035
 
                                                 object_name=object_name,
1036
 
                                                 data=data)
1037
 
                self.assertEquals(action, "PUT")
1038
 
                self.assertEqual(creds.access_key, "foo")
1039
 
                self.assertEqual(creds.secret_key, "bar")
1040
 
                self.assertEqual(query.bucket, "example-bucket")
1041
 
                self.assertEqual(query.object_name,
1042
 
                    "example-object?partNumber=3&uploadId=testid")
1043
 
                self.assertEqual(query.data, "some data")
1044
 
                self.assertEqual(query.metadata, {})
1045
 
 
1046
 
            def submit(query, url_context=None):
1047
 
                return succeed(None)
1048
 
 
1049
 
        creds = AWSCredentials("foo", "bar")
1050
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
1051
 
        return s3.upload_part("example-bucket", "example-object", "testid", 3,
1052
 
                              "some data")
1053
 
 
1054
 
    def test_complete_multipart_upload(self):
1055
 
 
1056
 
        class StubQuery(client.Query):
1057
 
 
1058
 
            def __init__(query, action, creds, endpoint, bucket=None,
1059
 
                         object_name=None, data="", body_producer=None,
1060
 
                         content_type=None, receiver_factory=None, metadata={}):
1061
 
                super(StubQuery, query).__init__(action=action, creds=creds,
1062
 
                                                 bucket=bucket,
1063
 
                                                 object_name=object_name,
1064
 
                                                 data=data)
1065
 
                self.assertEquals(action, "POST")
1066
 
                self.assertEqual(creds.access_key, "foo")
1067
 
                self.assertEqual(creds.secret_key, "bar")
1068
 
                self.assertEqual(query.bucket, "example-bucket")
1069
 
                self.assertEqual(query.object_name,
1070
 
                    "example-object?uploadId=testid")
1071
 
                self.assertEqual(query.data, "<CompleteMultipartUpload>\n"
1072
 
                    "<Part>\n<PartNumber>1</PartNumber>\n<ETag>a</ETag>\n"
1073
 
                    "</Part>\n<Part>\n<PartNumber>2</PartNumber>\n"
1074
 
                    "<ETag>b</ETag>\n</Part>\n</CompleteMultipartUpload>")
1075
 
                self.assertEqual(query.metadata, {})
1076
 
 
1077
 
            def submit(query, url_context=None):
1078
 
                return succeed(
1079
 
                    payload.sample_s3_complete_multipart_upload_result)
1080
 
 
1081
 
 
1082
 
        def check_result(result):
1083
 
            self.assert_(isinstance(result, MultipartCompletionResponse))
1084
 
            self.assertEqual(result.bucket, "example-bucket")
1085
 
            self.assertEqual(result.object_name, "example-object")
1086
 
            self.assertEqual(result.location,
1087
 
                "http://example-bucket.s3.amazonaws.com/example-object")
1088
 
            self.assertEqual(result.etag,
1089
 
                '"3858f62230ac3c915f300c664312c11f-9"')
1090
 
 
1091
 
        creds = AWSCredentials("foo", "bar")
1092
 
        s3 = client.S3Client(creds, query_factory=StubQuery)
1093
 
        deferred = s3.complete_multipart_upload("example-bucket",
1094
 
                                                "example-object",
1095
 
                                                "testid", [(1, "a"), (2, "b")])
1096
 
        return deferred.addCallback(check_result)
1097
 
 
1098
 
 
1099
 
S3ClientTestCase.skip = s3clientSkip
1100
 
 
1101
 
 
1102
 
class QueryTestCase(TXAWSTestCase):
1103
 
 
1104
 
    creds = AWSCredentials(access_key="fookeyid", secret_key="barsecretkey")
1105
 
    endpoint = AWSServiceEndpoint("https://choopy.s3.amazonaws.com/")
1106
 
 
1107
 
    def test_default_creation(self):
1108
 
        query = client.Query(action="PUT")
1109
 
        self.assertEquals(query.bucket, None)
1110
 
        self.assertEquals(query.object_name, None)
1111
 
        self.assertEquals(query.data, "")
1112
 
        self.assertEquals(query.content_type, None)
1113
 
        self.assertEquals(query.metadata, {})
1114
 
 
1115
 
    def test_default_endpoint(self):
1116
 
        query = client.Query(action="PUT")
1117
 
        self.assertEquals(self.endpoint.host, "choopy.s3.amazonaws.com")
1118
 
        self.assertEquals(query.endpoint.host, "s3.amazonaws.com")
1119
 
        self.assertEquals(self.endpoint.method, "GET")
1120
 
        self.assertEquals(query.endpoint.method, "PUT")
1121
 
 
1122
 
    def test_set_content_type_no_object_name(self):
1123
 
        query = client.Query(action="PUT")
1124
 
        query.set_content_type()
1125
 
        self.assertEquals(query.content_type, None)
1126
 
 
1127
 
    def test_set_content_type(self):
1128
 
        query = client.Query(action="PUT", object_name="advicedog.jpg")
1129
 
        query.set_content_type()
1130
 
        self.assertEquals(query.content_type, "image/jpeg")
1131
 
 
1132
 
    def test_set_content_type_with_content_type_already_set(self):
1133
 
        query = client.Query(
1134
 
            action="PUT", object_name="data.txt", content_type="text/csv")
1135
 
        query.set_content_type()
1136
 
        self.assertNotEquals(query.content_type, "text/plain")
1137
 
        self.assertEquals(query.content_type, "text/csv")
1138
 
 
1139
 
    def test_get_headers(self):
1140
 
        query = client.Query(
1141
 
            action="GET", creds=self.creds, bucket="mystuff",
1142
 
            object_name="/images/thing.jpg")
1143
 
        headers = query.get_headers()
1144
 
        self.assertEquals(headers.get("Content-Type"), "image/jpeg")
1145
 
        self.assertEquals(headers.get("Content-Length"), 0)
1146
 
        self.assertEquals(
1147
 
            headers.get("Content-MD5"), "1B2M2Y8AsgTpgAmY7PhCfg==")
1148
 
        self.assertTrue(len(headers.get("Date")) > 25)
1149
 
        self.assertTrue(
1150
 
            headers.get("Authorization").startswith("AWS fookeyid:"))
1151
 
        self.assertTrue(len(headers.get("Authorization")) > 40)
1152
 
 
1153
 
    def test_get_headers_with_data(self):
1154
 
        query = client.Query(
1155
 
            action="GET", creds=self.creds, bucket="mystuff",
1156
 
            object_name="/images/thing.jpg", data="BINARY IMAGE DATA")
1157
 
        headers = query.get_headers()
1158
 
        self.assertEquals(headers.get("Content-Type"), "image/jpeg")
1159
 
        self.assertEquals(headers.get("Content-Length"), 17)
1160
 
        self.assertTrue(len(headers.get("Date")) > 25)
1161
 
        self.assertTrue(
1162
 
            headers.get("Authorization").startswith("AWS fookeyid:"))
1163
 
        self.assertTrue(len(headers.get("Authorization")) > 40)
1164
 
 
1165
 
    def test_get_canonicalized_amz_headers(self):
1166
 
        query = client.Query(
1167
 
            action="SomeThing", metadata={"a": 1, "b": 2, "c": 3})
1168
 
        headers = query.get_headers()
1169
 
        self.assertEquals(
1170
 
            sorted(headers.keys()),
1171
 
            ["Content-Length", "Content-MD5", "Date", "x-amz-meta-a",
1172
 
             "x-amz-meta-b", "x-amz-meta-c"])
1173
 
        amz_headers = query.get_canonicalized_amz_headers(headers)
1174
 
        self.assertEquals(
1175
 
            amz_headers,
1176
 
            "x-amz-meta-a:1\nx-amz-meta-b:2\nx-amz-meta-c:3\n")
1177
 
 
1178
 
    def test_get_canonicalized_resource(self):
1179
 
        query = client.Query(action="PUT", bucket="images")
1180
 
        result = query.get_canonicalized_resource()
1181
 
        self.assertEquals(result, "/images/")
1182
 
 
1183
 
    def test_get_canonicalized_resource_with_object_name(self):
1184
 
        query = client.Query(
1185
 
            action="PUT", bucket="images", object_name="advicedog.jpg")
1186
 
        result = query.get_canonicalized_resource()
1187
 
        self.assertEquals(result, "/images/advicedog.jpg")
1188
 
 
1189
 
    def test_get_canonicalized_resource_with_slashed_object_name(self):
1190
 
        query = client.Query(
1191
 
            action="PUT", bucket="images", object_name="/advicedog.jpg")
1192
 
        result = query.get_canonicalized_resource()
1193
 
        self.assertEquals(result, "/images/advicedog.jpg")
1194
 
 
1195
 
    def test_sign(self):
1196
 
        query = client.Query(action="PUT", creds=self.creds)
1197
 
        signed = query.sign({})
1198
 
        self.assertEquals(signed, "H6UJCNHizzXZCGPl7wM6nL6tQdo=")
1199
 
 
1200
 
    def test_object_query(self):
1201
 
        """
1202
 
        Test that a request addressing an object is created correctly.
1203
 
        """
1204
 
        DATA = "objectData"
1205
 
        DIGEST = "zhdB6gwvocWv/ourYUWMxA=="
1206
 
 
1207
 
        request = client.Query(
1208
 
            action="PUT", bucket="somebucket", object_name="object/name/here",
1209
 
            data=DATA, content_type="text/plain", metadata={"foo": "bar"},
1210
 
            amz_headers={"acl": "public-read"}, creds=self.creds,
1211
 
            endpoint=self.endpoint)
1212
 
        request.sign = lambda headers: "TESTINGSIG="
1213
 
        self.assertEqual(request.action, "PUT")
1214
 
        headers = request.get_headers()
1215
 
        self.assertNotEqual(headers.pop("Date"), "")
1216
 
        self.assertEqual(headers, {"Authorization": "AWS fookeyid:TESTINGSIG=",
1217
 
                                   "Content-Type": "text/plain",
1218
 
                                   "Content-Length": len(DATA),
1219
 
                                   "Content-MD5": DIGEST,
1220
 
                                   "x-amz-meta-foo": "bar",
1221
 
                                   "x-amz-acl": "public-read"})
1222
 
        self.assertEqual(request.data, "objectData")
1223
 
 
1224
 
    def test_bucket_query(self):
1225
 
        """
1226
 
        Test that a request addressing a bucket is created correctly.
1227
 
        """
1228
 
        DIGEST = "1B2M2Y8AsgTpgAmY7PhCfg=="
1229
 
 
1230
 
        query = client.Query(
1231
 
            action="GET", bucket="somebucket", creds=self.creds,
1232
 
            endpoint=self.endpoint)
1233
 
        query.sign = lambda headers: "TESTINGSIG="
1234
 
        self.assertEqual(query.action, "GET")
1235
 
        headers = query.get_headers()
1236
 
        self.assertNotEqual(headers.pop("Date"), "")
1237
 
        self.assertEqual(
1238
 
            headers, {
1239
 
            "Authorization": "AWS fookeyid:TESTINGSIG=",
1240
 
            "Content-Length": 0,
1241
 
            "Content-MD5": DIGEST})
1242
 
        self.assertEqual(query.data, "")
1243
 
 
1244
 
    def test_submit(self):
1245
 
        """
1246
 
        Submitting the request should invoke getPage correctly.
1247
 
        """
1248
 
        class StubQuery(client.Query):
1249
 
 
1250
 
            def __init__(query, action, creds, endpoint, bucket,
1251
 
                body_producer=None, receiver_factory=None):
1252
 
                super(StubQuery, query).__init__(
1253
 
                    action=action, creds=creds, bucket=bucket)
1254
 
                self.assertEquals(action, "GET")
1255
 
                self.assertEqual(creds.access_key, "fookeyid")
1256
 
                self.assertEqual(creds.secret_key, "barsecretkey")
1257
 
                self.assertEqual(query.bucket, "somebucket")
1258
 
                self.assertEqual(query.object_name, None)
1259
 
                self.assertEqual(query.data, "")
1260
 
                self.assertEqual(query.metadata, {})
1261
 
 
1262
 
            def submit(query):
1263
 
                return succeed("")
1264
 
 
1265
 
        query = StubQuery(action="GET", creds=self.creds,
1266
 
                          endpoint=self.endpoint, bucket="somebucket")
1267
 
        return query.submit()
1268
 
 
1269
 
    def test_authentication(self):
1270
 
        query = client.Query(
1271
 
            action="GET", creds=self.creds, endpoint=self.endpoint)
1272
 
        query.sign = lambda headers: "TESTINGSIG="
1273
 
        query.date = "Wed, 28 Mar 2007 01:29:59 +0000"
1274
 
 
1275
 
        headers = query.get_headers()
1276
 
        self.assertEqual(
1277
 
            headers["Authorization"],
1278
 
            "AWS fookeyid:TESTINGSIG=")
1279
 
 
1280
 
QueryTestCase.skip = s3clientSkip
1281
 
 
1282
 
 
1283
 
class MiscellaneousTestCase(TXAWSTestCase):
1284
 
 
1285
 
    def test_content_md5(self):
 
277
        self.s3.delete_object("foobucket", "foo")
 
278
        req = self.s3._lastRequest
 
279
        self.assertTrue(req.submitted)
 
280
        self.assertEqual(req.verb, "DELETE")
 
281
        self.assertEqual(req.bucket, "foobucket")
 
282
        self.assertEqual(req.object_name, "foo")
 
283
 
 
284
 
 
285
class MiscellaneousTests(TXAWSTestCase):
 
286
 
 
287
    def test_contentMD5(self):
1286
288
        self.assertEqual(calculate_md5("somedata"), "rvr3UC1SmUw7AZV2NqPN0g==")
1287
 
 
1288
 
    def test_request_payment_enum(self):
1289
 
        """
1290
 
        Only 'Requester' or 'BucketOwner' may be provided when a
1291
 
        L{RequestPayment} is instantiated.
1292
 
        """
1293
 
        RequestPayment("Requester")
1294
 
        RequestPayment("BucketOwner")
1295
 
        self.assertRaises(ValueError, RequestPayment, "Bob")