← Back to branch summary

~tribaal/txaws/xss-hardening

« back to all changes in this revision

Viewing changes to txaws/tests/test_service.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/tests/test_service.py
 
1
# Copyright (C) 2009 Duncan McGreggor <duncan@canonical.com>
 
2
# Licenced under the txaws licence available at /LICENSE in the txaws source.
 
3
 
 
4
from txaws.credentials import AWSCredentials
 
5
from txaws.ec2.client import EC2Client
 
6
from txaws.service import (AWSServiceEndpoint, AWSServiceRegion,
 
7
                           EC2_ENDPOINT_EU, EC2_ENDPOINT_US, REGION_EU)
 
8
from txaws.testing.base import TXAWSTestCase
 
9
 
 
10
 
 
11
class AWSServiceEndpointTestCase(TXAWSTestCase):
 
12
 
 
13
    def setUp(self):
 
14
        self.endpoint = AWSServiceEndpoint(uri="http://my.service/da_endpoint")
 
15
 
 
16
    def test_simple_creation(self):
 
17
        endpoint = AWSServiceEndpoint()
 
18
        self.assertEquals(endpoint.scheme, "http")
 
19
        self.assertEquals(endpoint.host, "")
 
20
        self.assertEquals(endpoint.port, 80)
 
21
        self.assertEquals(endpoint.path, "/")
 
22
        self.assertEquals(endpoint.method, "GET")
 
23
 
 
24
    def test_parse_uri(self):
 
25
        self.assertEquals(self.endpoint.scheme, "http")
 
26
        self.assertEquals(self.endpoint.host, "my.service")
 
27
        self.assertEquals(self.endpoint.port, 80)
 
28
        self.assertEquals(self.endpoint.path, "/da_endpoint")
 
29
 
 
30
    def test_parse_uri_https_and_custom_port(self):
 
31
        endpoint = AWSServiceEndpoint(uri="https://my.service:8080/endpoint")
 
32
        self.assertEquals(endpoint.scheme, "https")
 
33
        self.assertEquals(endpoint.host, "my.service")
 
34
        self.assertEquals(endpoint.port, 8080)
 
35
        self.assertEquals(endpoint.path, "/endpoint")
 
36
 
 
37
    def test_custom_method(self):
 
38
        endpoint = AWSServiceEndpoint(uri="http://service/endpoint",
 
39
                                      method="PUT")
 
40
        self.assertEquals(endpoint.method, "PUT")
 
41
 
 
42
    def test_get_uri(self):
 
43
        uri = self.endpoint.get_uri()
 
44
        self.assertEquals(uri, "http://my.service/da_endpoint")
 
45
 
 
46
    def test_get_uri_custom_port(self):
 
47
        uri = "https://my.service:8080/endpoint"
 
48
        endpoint = AWSServiceEndpoint(uri=uri)
 
49
        new_uri = endpoint.get_uri()
 
50
        self.assertEquals(new_uri, uri)
 
51
 
 
52
    def test_set_path(self):
 
53
        original_path = self.endpoint.path
 
54
        self.endpoint.set_path("/newpath")
 
55
        self.assertEquals(
 
56
            self.endpoint.get_uri(),
 
57
            "http://my.service/newpath")
 
58
 
 
59
 
 
60
class AWSServiceRegionTestCase(TXAWSTestCase):
 
61
 
 
62
    def setUp(self):
 
63
        self.creds = AWSCredentials("foo", "bar")
 
64
        self.region = AWSServiceRegion(creds=self.creds)
 
65
 
 
66
    def test_simple_creation(self):
 
67
        self.assertEquals(self.creds, self.region.creds)
 
68
        self.assertEquals(self.region._clients, {})
 
69
        self.assertEquals(self.region.ec2_endpoint.get_uri(), EC2_ENDPOINT_US)
 
70
 
 
71
    def test_creation_with_keys(self):
 
72
        region = AWSServiceRegion(access_key="baz", secret_key="quux")
 
73
        self.assertEquals(region.creds.access_key, "baz")
 
74
        self.assertEquals(region.creds.secret_key, "quux")
 
75
 
 
76
    def test_creation_with_keys_and_creds(self):
 
77
        region = AWSServiceRegion(self.creds, access_key="baz",
 
78
                                  secret_key="quux")
 
79
        self.assertEquals(self.creds.access_key, "foo")
 
80
        self.assertEquals(self.creds.secret_key, "bar")
 
81
 
 
82
    def test_creation_with_uri(self):
 
83
        region = AWSServiceRegion(creds=self.creds, uri="http://foo/bar")
 
84
        self.assertEquals(region.ec2_endpoint.get_uri(), "http://foo/bar")
 
85
 
 
86
    def test_creation_with_uri_and_region(self):
 
87
        region = AWSServiceRegion(creds=self.creds, region=REGION_EU,
 
88
                                  uri="http://foo/bar")
 
89
        self.assertEquals(region.ec2_endpoint.get_uri(), "http://foo/bar")
 
90
 
 
91
    def test_creation_with_region_override(self):
 
92
        region = AWSServiceRegion(creds=self.creds, region=REGION_EU)
 
93
        self.assertEquals(region.ec2_endpoint.get_uri(), EC2_ENDPOINT_EU)
 
94
 
 
95
    def test_get_client_with_empty_cache(self):
 
96
        key = str(EC2Client) + str(self.creds) + str(self.region.ec2_endpoint)
 
97
        original_client = self.region._clients.get(key)
 
98
        new_client = self.region.get_client(
 
99
            EC2Client, creds=self.creds, endpoint=self.region.ec2_endpoint)
 
100
        self.assertEquals(original_client, None)
 
101
        self.assertTrue(isinstance(new_client, EC2Client))
 
102
        self.assertNotEquals(original_client, new_client)
 
103
 
 
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)
 
109
        self.assertTrue(isinstance(client1, EC2Client))
 
110
        self.assertTrue(isinstance(client2, EC2Client))
 
111
        self.assertEquals(client1, client2)
 
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
 
 
124
    def test_get_ec2_client_from_cache(self):
 
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)
 
128
        self.assertTrue(isinstance(client1, EC2Client))
 
129
        self.assertTrue(isinstance(client2, EC2Client))
 
130
        self.assertEquals(client1, client2)