/usr/share/pyshared/txaws/client/discover/tests/test_command.py is in python-txaws 0.2.3-1ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 | # Copyright (C) 2010 Jamu Kakar <jkakar@kakar.ca>
# Licenced under the txaws licence available at /LICENSE in the txaws source.
"""Unit tests for L{Command}."""
from cStringIO import StringIO
from twisted.internet.defer import succeed, fail
from twisted.web.error import Error
from txaws.client.discover.command import Command
from txaws.ec2.client import Query
from txaws.testing.base import TXAWSTestCase
class FakeHTTPClient(object):
def __init__(self, status, url):
self.status = status
self.url = url
class CommandTestCase(TXAWSTestCase):
def prepare_command(self, response, status, action, parameters={},
get_page=None, error=None):
"""Prepare a L{Command} for testing."""
self.url = None
self.method = None
self.error = error
self.response = response
self.status = status
self.output = StringIO()
self.query = None
if get_page is None:
get_page = self.get_page
self.get_page_function = get_page
self.command = Command("key", "secret", "endpoint", action, parameters,
self.output, self.query_factory)
def query_factory(self, other_params=None, time_tuple=None,
api_version=None, *args, **kwargs):
"""
Create a query with a hard-coded time to generate a fake response.
"""
time_tuple = (2010, 6, 4, 23, 40, 0, 0, 0, 0)
self.query = Query(other_params, time_tuple, api_version,
*args, **kwargs)
self.query.get_page = self.get_page_function
return self.query
def get_page(self, url, method=None, timeout=0):
"""Fake C{get_page} method simulates a successful request."""
self.url = url
self.method = method
self.query.client = FakeHTTPClient(self.status, url)
return succeed(self.response)
def get_error_page(self, url, method=None, timeout=0):
"""Fake C{get_page} method simulates an error."""
self.url = url
self.method = method
self.query.client = FakeHTTPClient(self.status, url)
return fail(self.error or Exception(self.response))
def test_run(self):
"""
When a method is invoked its HTTP status code and response text is
written to the output stream.
"""
self.prepare_command("The response", 200, "DescribeRegions")
def check(result):
url = (
"http://endpoint?AWSAccessKeyId=key&"
"Action=DescribeRegions&"
"Signature=3%2BHSkQQosF1Sr9AL3kdY31tEfTWQ2whjJOUSc3kvc2c%3D&"
"SignatureMethod=HmacSHA256&SignatureVersion=2&"
"Timestamp=2010-06-04T23%3A40%3A00Z&Version=2009-11-30")
self.assertEqual("GET", self.method)
self.assertEqual(url, self.url)
self.assertEqual("URL: %s\n"
"\n"
"HTTP status code: 200\n"
"\n"
"The response\n" % url,
self.output.getvalue())
deferred = self.command.run()
deferred.addCallback(check)
return deferred
def test_run_with_parameters(self):
"""Extra method parameters are included in the request."""
self.prepare_command("The response", 200, "DescribeRegions",
{"RegionName.0": "us-west-1"})
def check(result):
url = (
"http://endpoint?AWSAccessKeyId=key&"
"Action=DescribeRegions&RegionName.0=us-west-1&"
"Signature=6D8aCgSPQOYixowRHy26aRFzK2Vwgixl9uwegYX9nLA%3D&"
"SignatureMethod=HmacSHA256&SignatureVersion=2&"
"Timestamp=2010-06-04T23%3A40%3A00Z&Version=2009-11-30")
self.assertEqual("GET", self.method)
self.assertEqual(url, self.url)
self.assertEqual("URL: %s\n"
"\n"
"HTTP status code: 200\n"
"\n"
"The response\n" % url,
self.output.getvalue())
deferred = self.command.run()
deferred.addCallback(check)
return deferred
def test_run_with_error(self):
"""
If an error message is returned by the backend cloud, it will be
written to the output stream.
"""
self.prepare_command("The error response", 400, "DescribeRegions",
{"RegionName.0": "us-west-1"},
self.get_error_page)
def check(result):
url = (
"http://endpoint?AWSAccessKeyId=key&"
"Action=DescribeRegions&RegionName.0=us-west-1&"
"Signature=6D8aCgSPQOYixowRHy26aRFzK2Vwgixl9uwegYX9nLA%3D&"
"SignatureMethod=HmacSHA256&SignatureVersion=2&"
"Timestamp=2010-06-04T23%3A40%3A00Z&Version=2009-11-30")
self.assertEqual("GET", self.method)
self.assertEqual(url, self.url)
self.assertEqual("URL: %s\n"
"\n"
"HTTP status code: 400\n"
"\n"
"The error response\n" % url,
self.output.getvalue())
deferred = self.command.run()
return self.assertFailure(deferred, Exception).addErrback(check)
def test_run_with_error_strips_non_response_text(self):
"""
The builtin L{AWSError} exception adds 'Error message: ' to beginning
of the text retuned by the backend cloud. This is stripped when the
message is written to the output stream.
"""
self.prepare_command("Error Message: The error response", 400,
"DescribeRegions", {"RegionName.0": "us-west-1"},
self.get_error_page)
def check(result):
url = (
"http://endpoint?AWSAccessKeyId=key&"
"Action=DescribeRegions&RegionName.0=us-west-1&"
"Signature=P6C7cQJ7j93uIJyv2dTbpQG3EI7ArGBJT%2FzVH%2BDFhyY%3D&"
"SignatureMethod=HmacSHA256&SignatureVersion=2&"
"Timestamp=2010-06-04T23%3A40%3A00Z&Version=2009-11-30")
self.assertEqual("GET", self.method)
self.assertEqual(url, self.url)
self.assertEqual("URL: %s\n"
"\n"
"HTTP status code: 400\n"
"\n"
"The error response\n" % url,
self.output.getvalue())
deferred = self.command.run()
deferred.addErrback(check)
return deferred
def test_run_with_error_payload(self):
"""
If the returned HTTP error contains a payload, it's printed out.
"""
self.prepare_command("Bad Request", 400,
"DescribeRegions", {"RegionName.0": "us-west-1"},
self.get_error_page, Error(400, None, "bar"))
def check(result):
url = (
"http://endpoint?AWSAccessKeyId=key&"
"Action=DescribeRegions&RegionName.0=us-west-1&"
"Signature=6D8aCgSPQOYixowRHy26aRFzK2Vwgixl9uwegYX9nLA%3D&"
"SignatureMethod=HmacSHA256&SignatureVersion=2&"
"Timestamp=2010-06-04T23%3A40%3A00Z&Version=2009-11-30")
self.assertEqual("GET", self.method)
self.assertEqual(url, self.url)
self.assertEqual("URL: %s\n"
"\n"
"HTTP status code: 400\n"
"\n"
"400 Bad Request\n"
"\n"
"bar\n" % url,
self.output.getvalue())
deferred = self.command.run()
deferred.addCallback(check)
return deferred
|