← Back to branch summary

~tribaal/txaws/xss-hardening

« back to all changes in this revision

Viewing changes to txaws/tests/test_service.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/tests/test_service.py
3
3
 
4
4
from txaws.credentials import AWSCredentials
5
5
from txaws.ec2.client import EC2Client
6
 
try:
7
 
    from txaws.s3.client import S3Client
8
 
except ImportError:
9
 
    s3clientSkip = ("S3Client couldn't be imported (perhaps because dateutil, "
10
 
                    "on which it depends, isn't present)")
11
 
else:
12
 
    s3clientSkip = None
13
 
 
14
6
from txaws.service import (AWSServiceEndpoint, AWSServiceRegion,
15
7
                           EC2_ENDPOINT_EU, EC2_ENDPOINT_US, REGION_EU)
16
8
from txaws.testing.base import TXAWSTestCase
25
17
        endpoint = AWSServiceEndpoint()
26
18
        self.assertEquals(endpoint.scheme, "http")
27
19
        self.assertEquals(endpoint.host, "")
28
 
        self.assertEquals(endpoint.port, None)
 
20
        self.assertEquals(endpoint.port, 80)
29
21
        self.assertEquals(endpoint.path, "/")
30
22
        self.assertEquals(endpoint.method, "GET")
31
23
 
32
 
    def test_custom_method(self):
33
 
        endpoint = AWSServiceEndpoint(
34
 
            uri="http://service/endpoint", method="PUT")
35
 
        self.assertEquals(endpoint.method, "PUT")
36
 
 
37
24
    def test_parse_uri(self):
38
25
        self.assertEquals(self.endpoint.scheme, "http")
39
26
        self.assertEquals(self.endpoint.host, "my.service")
40
 
        self.assertIdentical(self.endpoint.port, None)
 
27
        self.assertEquals(self.endpoint.port, 80)
41
28
        self.assertEquals(self.endpoint.path, "/da_endpoint")
42
29
 
43
30
    def test_parse_uri_https_and_custom_port(self):
47
34
        self.assertEquals(endpoint.port, 8080)
48
35
        self.assertEquals(endpoint.path, "/endpoint")
49
36
 
 
37
    def test_custom_method(self):
 
38
        endpoint = AWSServiceEndpoint(uri="http://service/endpoint",
 
39
                                      method="PUT")
 
40
        self.assertEquals(endpoint.method, "PUT")
 
41
 
50
42
    def test_get_uri(self):
51
43
        uri = self.endpoint.get_uri()
52
44
        self.assertEquals(uri, "http://my.service/da_endpoint")
57
49
        new_uri = endpoint.get_uri()
58
50
        self.assertEquals(new_uri, uri)
59
51
 
60
 
    def test_set_host(self):
61
 
        self.assertEquals(self.endpoint.host, "my.service")
62
 
        self.endpoint.set_host("newhost.com")
63
 
        self.assertEquals(self.endpoint.host, "newhost.com")
64
 
 
65
 
    def test_get_host(self):
66
 
        self.assertEquals(self.endpoint.host, self.endpoint.get_host())
67
 
 
68
 
    def test_get_canonical_host(self):
69
 
        """
70
 
        If the port is not specified the canonical host is the same as
71
 
        the host.
72
 
        """
73
 
        uri = "http://my.service/endpoint"
74
 
        endpoint = AWSServiceEndpoint(uri=uri)
75
 
        self.assertEquals("my.service", endpoint.get_canonical_host())
76
 
 
77
 
    def test_get_canonical_host_with_non_default_port(self):
78
 
        """
79
 
        If the port is not the default, the canonical host includes it.
80
 
        """
81
 
        uri = "http://my.service:99/endpoint"
82
 
        endpoint = AWSServiceEndpoint(uri=uri)
83
 
        self.assertEquals("my.service:99", endpoint.get_canonical_host())
84
 
 
85
 
    def test_get_canonical_host_is_lower_case(self):
86
 
        """
87
 
        The canonical host is guaranteed to be lower case.
88
 
        """
89
 
        uri = "http://MY.SerVice:99/endpoint"
90
 
        endpoint = AWSServiceEndpoint(uri=uri)
91
 
        self.assertEquals("my.service:99", endpoint.get_canonical_host())
92
 
 
93
 
    def test_set_canonical_host(self):
94
 
        """
95
 
        The canonical host is converted to lower case.
96
 
        """
97
 
        endpoint = AWSServiceEndpoint()
98
 
        endpoint.set_canonical_host("My.Service")
99
 
        self.assertEquals("my.service", endpoint.host)
100
 
        self.assertIdentical(None, endpoint.port)
101
 
 
102
 
    def test_set_canonical_host_with_port(self):
103
 
        """
104
 
        The canonical host can optionally have a port.
105
 
        """
106
 
        endpoint = AWSServiceEndpoint()
107
 
        endpoint.set_canonical_host("my.service:99")
108
 
        self.assertEquals("my.service", endpoint.host)
109
 
        self.assertEquals(99, endpoint.port)
110
 
 
111
 
    def test_set_canonical_host_with_empty_port(self):
112
 
        """
113
 
        The canonical host can also have no port.
114
 
        """
115
 
        endpoint = AWSServiceEndpoint()
116
 
        endpoint.set_canonical_host("my.service:")
117
 
        self.assertEquals("my.service", endpoint.host)
118
 
        self.assertIdentical(None, endpoint.port)
119
 
 
120
52
    def test_set_path(self):
 
53
        original_path = self.endpoint.path
121
54
        self.endpoint.set_path("/newpath")
122
55
        self.assertEquals(
123
56
            self.endpoint.get_uri(),
124
57
            "http://my.service/newpath")
125
58
 
126
 
    def test_set_method(self):
127
 
        self.assertEquals(self.endpoint.method, "GET")
128
 
        self.endpoint.set_method("PUT")
129
 
        self.assertEquals(self.endpoint.method, "PUT")
130
 
 
131
59
 
132
60
class AWSServiceRegionTestCase(TXAWSTestCase):
133
61
 
146
74
        self.assertEquals(region.creds.secret_key, "quux")
147
75
 
148
76
    def test_creation_with_keys_and_creds(self):
149
 
        """
150
 
        creds take precedence over individual access key/secret key pairs.
151
 
        """
152
77
        region = AWSServiceRegion(self.creds, access_key="baz",
153
78
                                  secret_key="quux")
154
 
        self.assertEquals(region.creds.access_key, "foo")
155
 
        self.assertEquals(region.creds.secret_key, "bar")
 
79
        self.assertEquals(self.creds.access_key, "foo")
 
80
        self.assertEquals(self.creds.secret_key, "bar")
156
81
 
157
82
    def test_creation_with_uri(self):
158
 
        region = AWSServiceRegion(
159
 
            creds=self.creds, ec2_uri="http://foo/bar")
160
 
        self.assertEquals(region.ec2_endpoint.get_uri(), "http://foo/bar")
161
 
 
162
 
    def test_creation_with_uri_backwards_compatible(self):
163
 
        region = AWSServiceRegion(
164
 
            creds=self.creds, uri="http://foo/bar")
 
83
        region = AWSServiceRegion(creds=self.creds, uri="http://foo/bar")
165
84
        self.assertEquals(region.ec2_endpoint.get_uri(), "http://foo/bar")
166
85
 
167
86
    def test_creation_with_uri_and_region(self):
168
 
        region = AWSServiceRegion(
169
 
            creds=self.creds, region=REGION_EU, ec2_uri="http://foo/bar")
 
87
        region = AWSServiceRegion(creds=self.creds, region=REGION_EU,
 
88
                                  uri="http://foo/bar")
170
89
        self.assertEquals(region.ec2_endpoint.get_uri(), "http://foo/bar")
171
90
 
172
91
    def test_creation_with_region_override(self):
173
92
        region = AWSServiceRegion(creds=self.creds, region=REGION_EU)
174
93
        self.assertEquals(region.ec2_endpoint.get_uri(), EC2_ENDPOINT_EU)
175
94
 
176
 
    def test_get_ec2_client_with_empty_cache(self):
 
95
    def test_get_client_with_empty_cache(self):
177
96
        key = str(EC2Client) + str(self.creds) + str(self.region.ec2_endpoint)
178
97
        original_client = self.region._clients.get(key)
179
98
        new_client = self.region.get_client(
182
101
        self.assertTrue(isinstance(new_client, EC2Client))
183
102
        self.assertNotEquals(original_client, new_client)
184
103
 
185
 
    def test_get_ec2_client_from_cache_default(self):
186
 
        client1 = self.region.get_ec2_client()
187
 
        client2 = self.region.get_ec2_client()
 
104
    def test_get_client_from_cache(self):
 
105
        client1 = self.region.get_client(
 
106
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
 
107
        client2 = self.region.get_client(
 
108
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
188
109
        self.assertTrue(isinstance(client1, EC2Client))
189
110
        self.assertTrue(isinstance(client2, EC2Client))
190
111
        self.assertEquals(client1, client2)
191
112
 
 
113
    def test_get_client_from_cache_with_purge(self):
 
114
        client1 = self.region.get_client(
 
115
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint,
 
116
            purge_cache=True)
 
117
        client2 = self.region.get_client(
 
118
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint,
 
119
            purge_cache=True)
 
120
        self.assertTrue(isinstance(client1, EC2Client))
 
121
        self.assertTrue(isinstance(client2, EC2Client))
 
122
        self.assertNotEquals(client1, client2)
 
123
 
192
124
    def test_get_ec2_client_from_cache(self):
193
 
        client1 = self.region.get_client(
194
 
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
195
 
        client2 = self.region.get_client(
196
 
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
 
125
        client1 = self.region.get_ec2_client(self.creds)
 
126
        client2 = self.region.get_ec2_client(self.creds)
 
127
        self.assertEquals(self.creds, self.region.creds)
197
128
        self.assertTrue(isinstance(client1, EC2Client))
198
129
        self.assertTrue(isinstance(client2, EC2Client))
199
130
        self.assertEquals(client1, client2)
200
 
 
201
 
    def test_get_ec2_client_from_cache_with_purge(self):
202
 
        client1 = self.region.get_client(
203
 
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint,
204
 
            purge_cache=True)
205
 
        client2 = self.region.get_client(
206
 
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint,
207
 
            purge_cache=True)
208
 
        self.assertTrue(isinstance(client1, EC2Client))
209
 
        self.assertTrue(isinstance(client2, EC2Client))
210
 
        self.assertNotEquals(client1, client2)
211
 
 
212
 
    def test_get_s3_client_with_empty_cache(self):
213
 
        key = str(S3Client) + str(self.creds) + str(self.region.s3_endpoint)
214
 
        original_client = self.region._clients.get(key)
215
 
        new_client = self.region.get_client(
216
 
            S3Client, creds=self.creds, endpoint=self.region.s3_endpoint)
217
 
        self.assertEquals(original_client, None)
218
 
        self.assertTrue(isinstance(new_client, S3Client))
219
 
        self.assertNotEquals(original_client, new_client)
220
 
    test_get_s3_client_with_empty_cache.skip = s3clientSkip