← Back to branch summary

~txawsteam/txaws/trunk

« back to all changes in this revision

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

Merged 416109-arbitrary-endpoints [r=therve,jkakar] [f=416109].

The primary change of this branch is support of arbitrary endpoints (needed for
the support of Eucalyptus). In addition, the following was also performed:
 * Added a parse utility function from Twisted
 * Created a testing subpackage for use by txAWS unit tests
 * Created a service module for abstracting regions and associated
   serices/credentials

Show diffs side-by-side

added added

removed removed

Lines of Context:
txaws/storage/tests/test_client.py
5
5
from twisted.internet.defer import succeed
6
6
 
7
7
from txaws.credentials import AWSCredentials
 
8
from txaws.service import AWSServiceEndpoint
8
9
from txaws.storage.client import S3, S3Request
9
 
from txaws.tests import TXAWSTestCase
 
10
from txaws.testing.base import TXAWSTestCase
10
11
from txaws.util import calculate_md5
11
12
 
12
13
 
18
19
        return succeed('')
19
20
 
20
21
 
21
 
class RequestTests(TXAWSTestCase):
22
 
 
23
 
    creds = AWSCredentials(access_key='0PN5J17HBGZHT7JJ3X82',
24
 
        secret_key='uV3F3YluFJax1cknvbcGwgjvx4QpvB+leU8dUj2o')
 
22
class RequestTestCase(TXAWSTestCase):
 
23
 
 
24
    creds = AWSCredentials(access_key='fookeyid', secret_key='barsecretkey')
 
25
    endpoint = AWSServiceEndpoint("https://s3.amazonaws.com/")
 
26
 
 
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:///")
 
45
 
 
46
    def test_get_path_with_bucket_and_object(self):
 
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(), "/")
25
53
 
26
54
    def test_objectRequest(self):
27
55
        """
31
59
        DIGEST = 'zhdB6gwvocWv/ourYUWMxA=='
32
60
 
33
61
        request = S3Request('PUT', 'somebucket', 'object/name/here', DATA,
34
 
                            content_type='text/plain', metadata={'foo': 'bar'})
 
62
                            content_type='text/plain', metadata={'foo': 'bar'},
 
63
                            creds=self.creds, endpoint=self.endpoint)
 
64
        request.get_signature = lambda headers: "TESTINGSIG="
35
65
        self.assertEqual(request.verb, 'PUT')
36
66
        self.assertEqual(
37
67
            request.get_uri(),
38
68
            'https://s3.amazonaws.com/somebucket/object/name/here')
39
69
        headers = request.get_headers()
40
70
        self.assertNotEqual(headers.pop('Date'), '')
41
 
        self.assertEqual(headers,
42
 
                         {'Content-Type': 'text/plain',
43
 
                          'Content-Length': len(DATA),
44
 
                          'Content-MD5': DIGEST,
45
 
                          'x-amz-meta-foo': 'bar'})
 
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'})
46
78
        self.assertEqual(request.data, 'objectData')
47
79
 
48
80
    def test_bucketRequest(self):
51
83
        """
52
84
        DIGEST = '1B2M2Y8AsgTpgAmY7PhCfg=='
53
85
 
54
 
        request = S3Request('GET', 'somebucket')
 
86
        request = S3Request('GET', 'somebucket', creds=self.creds,
 
87
                            endpoint=self.endpoint)
 
88
        request.get_signature = lambda headers: "TESTINGSIG="
55
89
        self.assertEqual(request.verb, 'GET')
56
90
        self.assertEqual(
57
91
            request.get_uri(), 'https://s3.amazonaws.com/somebucket')
58
92
        headers = request.get_headers()
59
93
        self.assertNotEqual(headers.pop('Date'), '')
60
 
        self.assertEqual(headers,
61
 
                         {'Content-Length': 0,
62
 
                          'Content-MD5': DIGEST})
 
94
        self.assertEqual(
 
95
            headers, {
 
96
            'Authorization': 'AWS fookeyid:TESTINGSIG=',
 
97
            'Content-Length': 0,
 
98
            'Content-MD5': DIGEST})
63
99
        self.assertEqual(request.data, '')
64
100
 
65
101
    def test_submit(self):
66
102
        """
67
103
        Submitting the request should invoke getPage correctly.
68
104
        """
69
 
        request = StubbedS3Request('GET', 'somebucket')
 
105
        request = StubbedS3Request('GET', 'somebucket', creds=self.creds,
 
106
                                   endpoint=self.endpoint)
70
107
 
71
108
        def _postCheck(result):
72
109
            self.assertEqual(result, '')
80
117
        return request.submit().addCallback(_postCheck)
81
118
 
82
119
    def test_authenticationTestCases(self):
83
 
        req = S3Request('GET', creds=self.creds)
84
 
        req.date = 'Wed, 28 Mar 2007 01:29:59 +0000'
 
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'
85
123
 
86
 
        headers = req.get_headers()
 
124
        headers = request.get_headers()
87
125
        self.assertEqual(
88
126
            headers['Authorization'], 
89
 
            'AWS 0PN5J17HBGZHT7JJ3X82:jF7L3z/FTV47vagZzhKupJ9oNig=')
 
127
            'AWS fookeyid:TESTINGSIG=')
90
128
 
91
129
 
92
130
class InertRequest(S3Request):
153
191
        TXAWSTestCase.setUp(self)
154
192
        self.creds = AWSCredentials(
155
193
            access_key='accessKey', secret_key='secretKey')
156
 
        self.s3 = TestableS3(creds=self.creds)
 
194
        self.endpoint = AWSServiceEndpoint()
 
195
        self.s3 = TestableS3(creds=self.creds, endpoint=self.endpoint)
157
196
 
158
197
    def test_make_request(self):
159
198
        """
160
 
        Test that make_request passes in the service credentials.
 
199
        Test that make_request passes in the credentials object.
161
200
        """
162
201
        marker = object()
163
202
 
164
203
        def _cb(*a, **kw):
165
204
            self.assertEqual(kw['creds'], self.creds)
 
205
            self.assertEqual(kw['endpoint'], self.endpoint)
166
206
            return marker
167
207
 
168
208
        self.s3.request_factory = _cb
243
283
 
244
284
 
245
285
class MiscellaneousTests(TXAWSTestCase):
 
286
 
246
287
    def test_contentMD5(self):
247
288
        self.assertEqual(calculate_md5('somedata'), 'rvr3UC1SmUw7AZV2NqPN0g==')