1
# Copyright (C) 2010 Jamu Kakar <jkakar@kakar.ca>
2
# Licenced under the txaws licence available at /LICENSE in the txaws source.
4
"""Unit tests for L{Command}."""
6
from cStringIO import StringIO
8
from twisted.internet.defer import succeed, fail
9
from twisted.web.error import Error
11
from txaws.client.discover.command import Command
12
from txaws.ec2.client import Query
13
from txaws.testing.base import TXAWSTestCase
16
class FakeHTTPClient(object):
18
def __init__(self, status, url):
23
class CommandTestCase(TXAWSTestCase):
25
def prepare_command(self, response, status, action, parameters={},
26
get_page=None, error=None):
27
"""Prepare a L{Command} for testing."""
31
self.response = response
33
self.output = StringIO()
36
get_page = self.get_page
37
self.get_page_function = get_page
38
self.command = Command("key", "secret", "endpoint", action, parameters,
39
self.output, self.query_factory)
41
def query_factory(self, other_params=None, time_tuple=None,
42
api_version=None, *args, **kwargs):
44
Create a query with a hard-coded time to generate a fake response.
46
time_tuple = (2010, 6, 4, 23, 40, 0, 0, 0, 0)
47
self.query = Query(other_params, time_tuple, api_version,
49
self.query.get_page = self.get_page_function
52
def get_page(self, url, method=None, timeout=0):
53
"""Fake C{get_page} method simulates a successful request."""
56
self.query.client = FakeHTTPClient(self.status, url)
57
return succeed(self.response)
59
def get_error_page(self, url, method=None, timeout=0):
60
"""Fake C{get_page} method simulates an error."""
63
self.query.client = FakeHTTPClient(self.status, url)
64
return fail(self.error or Exception(self.response))
68
When a method is invoked its HTTP status code and response text is
69
written to the output stream.
71
self.prepare_command("The response", 200, "DescribeRegions")
75
"http://endpoint?AWSAccessKeyId=key&"
76
"Action=DescribeRegions&"
77
"Signature=7fyxNidMkL%2B85udGOxqm%2BgM2o1gLyeLG2a0UOmfBOXQ%3D&"
78
"SignatureMethod=HmacSHA256&SignatureVersion=2&"
79
"Timestamp=2010-06-04T23%3A40%3A00Z&Version=2012-08-15")
80
self.assertEqual("GET", self.method)
81
self.assertEqual(url, self.url)
82
self.assertEqual("URL: %s\n"
84
"HTTP status code: 200\n"
86
"The response\n" % url,
87
self.output.getvalue())
89
deferred = self.command.run()
90
deferred.addCallback(check)
93
def test_run_with_parameters(self):
94
"""Extra method parameters are included in the request."""
95
self.prepare_command("The response", 200, "DescribeRegions",
96
{"RegionName.0": "us-west-1"})
100
"http://endpoint?AWSAccessKeyId=key&"
101
"Action=DescribeRegions&RegionName.0=us-west-1&"
102
"Signature=FL4JjDKbWdg531q1KKUPild%2BvyqspA5wxSmOeWXWsJI%3D&"
103
"SignatureMethod=HmacSHA256&SignatureVersion=2&"
104
"Timestamp=2010-06-04T23%3A40%3A00Z&Version=2012-08-15")
105
self.assertEqual("GET", self.method)
106
self.assertEqual(url, self.url)
107
self.assertEqual("URL: %s\n"
109
"HTTP status code: 200\n"
111
"The response\n" % url,
112
self.output.getvalue())
114
deferred = self.command.run()
115
deferred.addCallback(check)
118
def test_run_with_error(self):
120
If an error message is returned by the backend cloud, it will be
121
written to the output stream.
123
self.prepare_command("The error response", 400, "DescribeRegions",
124
{"RegionName.0": "us-west-1"},
129
"http://endpoint?AWSAccessKeyId=key&"
130
"Action=DescribeRegions&RegionName.0=us-west-1&"
131
"Signature=FL4JjDKbWdg531q1KKUPild%2BvyqspA5wxSmOeWXWsJI%3D&"
132
"SignatureMethod=HmacSHA256&SignatureVersion=2&"
133
"Timestamp=2010-06-04T23%3A40%3A00Z&Version=2012-08-15")
134
self.assertEqual("GET", self.method)
135
self.assertEqual(url, self.url)
136
self.assertEqual("URL: %s\n"
138
"HTTP status code: 400\n"
140
"The error response\n" % url,
141
self.output.getvalue())
143
deferred = self.command.run()
144
return self.assertFailure(deferred, Exception).addErrback(check)
146
def test_run_with_error_strips_non_response_text(self):
148
The builtin L{AWSError} exception adds 'Error message: ' to beginning
149
of the text retuned by the backend cloud. This is stripped when the
150
message is written to the output stream.
152
self.prepare_command("Error Message: The error response", 400,
153
"DescribeRegions", {"RegionName.0": "us-west-1"},
158
"http://endpoint?AWSAccessKeyId=key&"
159
"Action=DescribeRegions&RegionName.0=us-west-1&"
160
"Signature=P6C7cQJ7j93uIJyv2dTbpQG3EI7ArGBJT%2FzVH%2BDFhyY%3D&"
161
"SignatureMethod=HmacSHA256&SignatureVersion=2&"
162
"Timestamp=2010-06-04T23%3A40%3A00Z&Version=2009-11-30")
163
self.assertEqual("GET", self.method)
164
self.assertEqual(url, self.url)
165
self.assertEqual("URL: %s\n"
167
"HTTP status code: 400\n"
169
"The error response\n" % url,
170
self.output.getvalue())
172
deferred = self.command.run()
173
deferred.addErrback(check)
176
def test_run_with_error_payload(self):
178
If the returned HTTP error contains a payload, it's printed out.
180
self.prepare_command("Bad Request", 400,
181
"DescribeRegions", {"RegionName.0": "us-west-1"},
182
self.get_error_page, Error(400, None, "bar"))
186
"http://endpoint?AWSAccessKeyId=key&"
187
"Action=DescribeRegions&RegionName.0=us-west-1&"
188
"Signature=FL4JjDKbWdg531q1KKUPild%2BvyqspA5wxSmOeWXWsJI%3D&"
189
"SignatureMethod=HmacSHA256&SignatureVersion=2&"
190
"Timestamp=2010-06-04T23%3A40%3A00Z&Version=2012-08-15")
191
self.assertEqual("GET", self.method)
192
self.assertEqual(url, self.url)
193
self.assertEqual("URL: %s\n"
195
"HTTP status code: 400\n"
200
self.output.getvalue())
202
deferred = self.command.run()
203
deferred.addCallback(check)
added
removed
txaws/client/discover/tests/test_command.py
