1
# Copyright (C) 2009 Robert Collins <robertc@robertcollins.net>
2
# Licenced under the txaws licence available at /LICENSE in the txaws source.
6
from twisted.internet.defer import succeed
8
from txaws.credentials import AWSCredentials
9
from txaws.ec2 import client
10
from txaws.tests import TXAWSTestCase
12
sample_describe_instances_result = """<?xml version="1.0"?>
13
<DescribeInstancesResponse xmlns="http://ec2.amazonaws.com/doc/2008-12-01/">
14
<requestId>52b4c730-f29f-498d-94c1-91efb75994cc</requestId>
17
<reservationId>r-cf24b1a6</reservationId>
18
<ownerId>123456789012</ownerId>
21
<groupId>default</groupId>
26
<instanceId>i-abcdef01</instanceId>
27
<imageId>ami-12345678</imageId>
32
<privateDnsName>domU-12-31-39-03-15-11.compute-1.internal</privateDnsName>
33
<dnsName>ec2-75-101-245-65.compute-1.amazonaws.com</dnsName>
35
<keyName>keyname</keyName>
36
<amiLaunchIndex>0</amiLaunchIndex>
38
<instanceType>c1.xlarge</instanceType>
39
<launchTime>2009-04-27T02:23:18.000Z</launchTime>
41
<availabilityZone>us-east-1c</availabilityZone>
43
<kernelId>aki-b51cf9dc</kernelId>
44
<ramdiskId>ari-b31cf9da</ramdiskId>
49
</DescribeInstancesResponse>
53
class TestEC2Client(TXAWSTestCase):
55
def test_init_no_creds(self):
56
os.environ['AWS_SECRET_ACCESS_KEY'] = 'foo'
57
os.environ['AWS_ACCESS_KEY_ID'] = 'bar'
58
ec2 = client.EC2Client()
59
self.assertNotEqual(None, ec2.creds)
61
def test_init_no_creds_non_available_errors(self):
62
self.assertRaises(ValueError, client.EC2Client)
64
def test_init_explicit_creds(self):
66
ec2 = client.EC2Client(creds=creds)
67
self.assertEqual(creds, ec2.creds)
69
def test_describe_instances(self):
70
class StubQuery(object):
71
def __init__(stub, action, creds):
72
self.assertEqual(action, 'DescribeInstances')
73
self.assertEqual('foo', creds)
75
return succeed(sample_describe_instances_result)
76
ec2 = client.EC2Client(creds='foo', query_factory=StubQuery)
77
d = ec2.describe_instances()
78
def check_instances(reservation):
79
self.assertEqual(1, len(reservation))
80
self.assertEqual('i-abcdef01', reservation[0].instanceId)
81
self.assertEqual('running', reservation[0].instanceState)
82
d.addCallback(check_instances)
86
class TestQuery(TXAWSTestCase):
89
TXAWSTestCase.setUp(self)
90
self.creds = AWSCredentials('foo', 'bar')
92
def test_init_minimum(self):
93
query = client.Query('DescribeInstances', self.creds)
94
self.assertTrue('Timestamp' in query.params)
95
del query.params['Timestamp']
97
{'AWSAccessKeyId': 'foo',
98
'Action': 'DescribeInstances',
99
'SignatureMethod': 'HmacSHA1',
100
'SignatureVersion': '2',
101
'Version': '2008-12-01'},
104
def test_init_requires_action(self):
105
self.assertRaises(TypeError, client.Query)
107
def test_init_requires_creds(self):
108
self.assertRaises(TypeError, client.Query, None)
110
def test_init_other_args_are_params(self):
111
query = client.Query('DescribeInstances', self.creds,
112
{'InstanceId.0': '12345'},
113
time_tuple=(2007,11,12,13,14,15,0,0,0))
115
{'AWSAccessKeyId': 'foo',
116
'Action': 'DescribeInstances',
117
'InstanceId.0': '12345',
118
'SignatureMethod': 'HmacSHA1',
119
'SignatureVersion': '2',
120
'Timestamp': '2007-11-12T13:14:15Z',
121
'Version': '2008-12-01'},
124
def test_sorted_params(self):
125
query = client.Query('DescribeInstances', self.creds,
127
time_tuple=(2007,11,12,13,14,15,0,0,0))
129
('AWSAccessKeyId', 'foo'),
130
('Action', 'DescribeInstances'),
131
('SignatureMethod', 'HmacSHA1'),
132
('SignatureVersion', '2'),
133
('Timestamp', '2007-11-12T13:14:15Z'),
134
('Version', '2008-12-01'),
136
], query.sorted_params())
138
def test_encode_unreserved(self):
139
all_unreserved = ('ABCDEFGHIJKLMNOPQRSTUVWXYZ'
140
'abcdefghijklmnopqrstuvwxyz0123456789-_.~')
141
query = client.Query('DescribeInstances', self.creds)
142
self.assertEqual(all_unreserved, query.encode(all_unreserved))
144
def test_encode_space(self):
145
"""This may be just 'url encode', but the AWS manual isn't clear."""
146
query = client.Query('DescribeInstances', self.creds)
147
self.assertEqual('a%20space', query.encode('a space'))
149
def test_canonical_query(self):
150
query = client.Query('DescribeInstances', self.creds,
151
{'fu n': 'g/ames', 'argwithnovalue':''},
152
time_tuple=(2007,11,12,13,14,15,0,0,0))
153
expected_query = ('AWSAccessKeyId=foo&Action=DescribeInstances'
154
'&SignatureMethod=HmacSHA1&SignatureVersion=2&'
155
'Timestamp=2007-11-12T13%3A14%3A15Z&Version=2008-12-01&'
156
'argwithnovalue=&fu%20n=g%2Fames')
157
self.assertEqual(expected_query, query.canonical_query_params())
159
def test_signing_text(self):
160
query = client.Query('DescribeInstances', self.creds,
161
time_tuple=(2007,11,12,13,14,15,0,0,0))
162
signing_text = ('GET\nec2.amazonaws.com\n/\n'
163
'AWSAccessKeyId=foo&Action=DescribeInstances&'
164
'SignatureMethod=HmacSHA1&SignatureVersion=2&'
165
'Timestamp=2007-11-12T13%3A14%3A15Z&Version=2008-12-01')
166
self.assertEqual(signing_text, query.signing_text())
169
query = client.Query('DescribeInstances', self.creds,
170
time_tuple=(2007,11,12,13,14,15,0,0,0))
172
self.assertEqual('4hEtLuZo9i6kuG3TOXvRQNOrE/U=',
173
query.params['Signature'])