txaws-dev team mailing list archive
-
txaws-dev team
-
Mailing list archive
-
Message #00008
[Merge] lp:~jkakar/txaws/discovery-tool into lp:txaws
Jamu Kakar has proposed merging lp:~jkakar/txaws/discovery-tool into lp:txaws.
Requested reviews:
txAWS Developers (txaws-dev)
Related bugs:
#589926 It would be nice if txAWS provided a tool to help learn how the EC2 API behaves
https://bugs.launchpad.net/bugs/589926
This branch introduces the following changes:
- A new bin/txaws-discover script provides a simple command-line
interface for invoking EC2 API methods on a cloud endpoint. It
prints the HTTP status code and response text to the screen. This
makes it possible to test assumptions about how a method behaves
in different conditions.
- A new txaws.client.discover package contains a Command object,
which can be configured to run a method and print the response to
an output stream. It also contains bootstrapping code that parses
command-line arguments, shows usage text if necessary, and creates
a Command instance and runs it.
--
https://code.launchpad.net/~jkakar/txaws/discovery-tool/+merge/26848
Your team txAWS Developers is requested to review the proposed merge of lp:~jkakar/txaws/discovery-tool into lp:txaws.
=== added file 'bin/txaws-discover'
--- bin/txaws-discover 1970-01-01 00:00:00 +0000
+++ bin/txaws-discover 2010-06-04 22:50:31 +0000
@@ -0,0 +1,15 @@
+#!/usr/bin/env python
+
+# Copyright (C) 2010 Jamu Kakar <jkakar@xxxxxxxx>
+# Licenced under the txaws licence available at /LICENSE in the txaws source.
+
+import os
+import sys
+
+if os.path.isdir("txaws"):
+ sys.path.insert(0, ".")
+
+from txaws.client.discover.entry_point import main
+
+
+sys.exit(main(sys.argv))
=== added directory 'txaws/client/discover'
=== added file 'txaws/client/discover/__init__.py'
=== added file 'txaws/client/discover/command.py'
--- txaws/client/discover/command.py 1970-01-01 00:00:00 +0000
+++ txaws/client/discover/command.py 2010-06-04 22:50:31 +0000
@@ -0,0 +1,72 @@
+# Copyright (C) 2010 Jamu Kakar <jkakar@xxxxxxxx>
+# Licenced under the txaws licence available at /LICENSE in the txaws source.
+
+"""
+A L{Command} object makes an arbitrary EC2 API method call and displays the
+response received from the backend cloud.
+"""
+
+import sys
+
+from txaws.ec2.client import Query
+from txaws.service import AWSServiceRegion
+
+
+class Command(object):
+ """
+ An EC2 API method call command that can make a request and display the
+ response received from the backend cloud.
+
+ @param key: The AWS access key ID to use when making the method call.
+ @param secret: The AWS secret key to sign the method call with.
+ @param endpoint: The URL of the cloud to invoke the method on.
+ @param parameters: A C{dict} with parameters to include with the method
+ call.
+ @param output: Optionally, a stream to write output to. Defaults to
+ C{sys.stdout}.
+ @param query_factory: Optionally, a factory to create the L{Query} object
+ used to invoke the method. Defaults to returning a L{Query} instance.
+ """
+
+ def __init__(self, key, secret, endpoint, action, parameters, output=None,
+ query_factory=None):
+ self.key = key
+ self.secret = secret
+ self.endpoint = endpoint
+ self.action = action
+ self.parameters = parameters
+ if output is None:
+ output = sys.stdout
+ self.output = output
+ if query_factory is None:
+ query_factory = Query
+ self.query_factory = query_factory
+
+ def run(self):
+ """
+ Run the configured method and write the HTTP response status and text
+ to the output stream.
+ """
+ region = AWSServiceRegion(access_key=self.key, secret_key=self.secret,
+ uri=self.endpoint)
+ query = self.query_factory(action=self.action, creds=region.creds,
+ endpoint=region.ec2_endpoint,
+ other_params=self.parameters)
+
+ def write_response(response):
+ print >>self.output, "HTTP status code: %s" % query.client.status
+ print >>self.output
+ print >>self.output, response
+
+ def write_error(failure):
+ message = failure.getErrorMessage()
+ if message.startswith("Error Message: "):
+ message = message[len("Error Message: "):]
+ print >>self.output, "HTTP status code: %s" % query.client.status
+ print >>self.output
+ print >>self.output, message
+
+ deferred = query.submit()
+ deferred.addCallback(write_response)
+ deferred.addErrback(write_error)
+ return deferred
=== added file 'txaws/client/discover/entry_point.py'
--- txaws/client/discover/entry_point.py 1970-01-01 00:00:00 +0000
+++ txaws/client/discover/entry_point.py 2010-06-04 22:50:31 +0000
@@ -0,0 +1,178 @@
+# Copyright (C) 2010 Jamu Kakar <jkakar@xxxxxxxx>
+# Licenced under the txaws licence available at /LICENSE in the txaws source.
+
+"""A command-line client for discovering how the EC2 API works."""
+
+import os
+import sys
+
+from txaws.client.discover.command import Command
+
+
+class OptionError(Exception):
+ """
+ Raised if insufficient command-line arguments are provided when creating a
+ L{Command}.
+ """
+
+
+class UsageError(Exception):
+ """Raised if the usage message should be shown."""
+
+
+USAGE_MESSAGE = """\
+Purpose: Invoke an EC2 API method with arbitrary parameters.
+Usage: txaws-discover [--key KEY] [--secret SECRET] [--endpoint ENDPOINT]
+ --action ACTION [PARAMETERS, ...]
+
+Options:
+ --key The AWS access key to use when making the API request.
+ --secret The AWS secret key to use when making the API request.
+ --endpoint The region endpoint to make the API request against.
+ --action The name of the EC2 API to invoke.
+ -h, --help Show help message.
+
+Description:
+ The purpose of this program is to aid discovery of the EC2 API. It can run
+ any EC2 API method, with arbitrary parameters. The response received from
+ the backend cloud is printed to the screen, to show exactly what happened in
+ response to the request. The --key, --secret, --endpoint and --action
+ command-line arguments are required. If AWS_ENDPOINT, AWS_ACCESS_KEY_ID or
+ AWS_SECRET_ACCESS_KEY environment variables are defined the corresponding
+ options can be omitted and the values defined in the environment variables
+ will be used.
+
+ Any additional parameters, beyond those defined above, will be included with
+ the request as method parameters.
+
+Examples:
+ The following examples omit the --key, --secret and --endpoint command-line
+ arguments for brevity. They must be included unless corresponding values
+ are available from the environment.
+
+ Run the DescribeRegions method, without any optional parameters:
+
+ txaws-discover --action DescribeRegions
+
+ Run the DescribeRegions method, with an optional RegionName.0 parameter:
+
+ txaws-discover --action DescribeRegions --RegionName.0 us-west-1
+"""
+
+
+def parse_options(arguments):
+ """Parse command line arguments.
+
+ The parsing logic is fairly simple. It can only parse long-style
+ parameters of the form::
+
+ --key value
+
+ Several parameters can be defined in the environment and will be used
+ unless explicitly overridden with command-line arguments. The access key,
+ secret and endpoint values will be loaded from C{AWS_ACCESS_KEY_ID},
+ C{AWS_SECRET_ACCESS_KEY} and C{AWS_ENDPOINT} environment variables.
+
+ @param arguments: A list of command-line arguments. The first item is
+ expected to be the name of the program being run.
+ @raises OptionError: Raised if incorrectly formed command-line arguments
+ are specified, or if required command-line arguments are not present.
+ @raises UsageError: Raised if C{--help} is present in command-line
+ arguments.
+ @return: A C{dict} with key/value pairs extracted from the argument list.
+ """
+ arguments = arguments[1:]
+ options = {}
+ while arguments:
+ key = arguments.pop(0)
+ if key in ("-h", "--help"):
+ raise UsageError("Help requested.")
+ if key.startswith("--"):
+ key = key[2:]
+ try:
+ value = arguments.pop(0)
+ except IndexError:
+ raise OptionError("'--%s' is missing a value." % key)
+ options[key] = value
+ else:
+ raise OptionError("Encountered unexpected value '%s'." % key)
+
+ default_key = os.environ.get("AWS_ACCESS_KEY_ID")
+ if "key" not in options and default_key:
+ options["key"] = default_key
+ default_secret = os.environ.get("AWS_SECRET_ACCESS_KEY")
+ if "secret" not in options and default_secret:
+ options["secret"] = default_secret
+ default_endpoint = os.environ.get("AWS_ENDPOINT")
+ if "endpoint" not in options and default_endpoint:
+ options["endpoint"] = default_endpoint
+ for name in ("key", "secret", "endpoint", "action"):
+ if name not in options:
+ raise OptionError(
+ "The '--%s' command-line argument is required." % name)
+
+ return options
+
+
+def get_command(arguments, output=None):
+ """Parse C{arguments} and configure a L{Command} instance.
+
+ An access key, secret key, endpoint and action are required. Additional
+ parameters included with the request are passed as parameters to the
+ method call. For example, the following command will create a L{Command}
+ object that can invoke the C{DescribeRegions} method with the optional
+ C{RegionName.0} parameter included in the request::
+
+ txaws-discover --key KEY --secret SECRET --endpoint URL \
+ --action DescribeRegions --RegionName.0 us-west-1
+
+ @param arguments: The command-line arguments to parse.
+ @raises OptionError: Raised if C{arguments} can't be used to create a
+ L{Command} object.
+ @return: A L{Command} instance configured to make an EC2 API method call.
+ """
+ options = parse_options(arguments)
+ key = options.pop("key")
+ secret = options.pop("secret")
+ endpoint = options.pop("endpoint")
+ action = options.pop("action")
+ return Command(key, secret, endpoint, action, options, output)
+
+
+def main(arguments, output=None, testing_mode=None):
+ """
+ Entry point parses command-line arguments, runs the specified EC2 API
+ method and prints the response to the screen.
+
+ @param arguments: Command-line arguments, typically retrieved from
+ C{sys.argv}.
+ @param output: Optionally, a stream to write output to.
+ @param testing_mode: Optionally, a condition that specifies whether or not
+ to run in test mode. When the value is true a reactor will not be run
+ or stopped, to prevent interfering with the test suite.
+ """
+
+ def run_command(arguments, output, reactor):
+ if output is None:
+ output = sys.stdout
+ try:
+ command = get_command(arguments, output)
+ except UsageError:
+ print >>output, USAGE_MESSAGE.strip()
+ if reactor:
+ reactor.callLater(0, reactor.stop)
+ except Exception, e:
+ print >>output, "ERROR:", str(e)
+ if reactor:
+ reactor.callLater(0, reactor.stop)
+ else:
+ deferred = command.run()
+ if reactor:
+ deferred.addCallback(lambda ignored: reactor.stop())
+
+ if not testing_mode:
+ from twisted.internet import reactor
+ reactor.callLater(0, run_command, arguments, output, reactor)
+ reactor.run()
+ else:
+ run_command(arguments, output, None)
=== added directory 'txaws/client/discover/tests'
=== added file 'txaws/client/discover/tests/__init__.py'
=== added file 'txaws/client/discover/tests/test_command.py'
--- txaws/client/discover/tests/test_command.py 1970-01-01 00:00:00 +0000
+++ txaws/client/discover/tests/test_command.py 2010-06-04 22:50:31 +0000
@@ -0,0 +1,164 @@
+# Copyright (C) 2010 Jamu Kakar <jkakar@xxxxxxxx>
+# 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 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):
+ self.status = status
+
+
+class CommandTest(TXAWSTestCase):
+
+ def prepare_command(self, response, status, action, parameters={},
+ get_page=None):
+ """Prepare a L{Command} for testing."""
+ self.url = None
+ self.method = None
+ 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):
+ """Fake C{get_page} method simulates a successful request."""
+ self.url = url
+ self.method = method
+ self.query.client = FakeHTTPClient(self.status)
+ return succeed(self.response)
+
+ def get_error_page(self, url, method=None):
+ """Fake C{get_page} method simulates an error."""
+ self.url = url
+ self.method = method
+ self.query.client = FakeHTTPClient(self.status)
+ return fail(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):
+ self.assertEqual("GET", self.method)
+ self.assertEqual(
+ "http://endpoint?AWSAccessKeyId=key&"
+ "Action=DescribeRegions&"
+ "Signature=uAlV2ALkp7qTxZrTNNuJhHl0i9xiTK5faZOhJTgGS1E%3D&"
+ "SignatureMethod=HmacSHA256&SignatureVersion=2&"
+ "Timestamp=2010-06-04T23%3A40%3A00Z&Version=2008-12-01",
+ self.url)
+ self.assertEqual("HTTP status code: 200\n"
+ "\n"
+ "The response\n",
+ 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):
+ self.assertEqual("GET", self.method)
+ self.assertEqual(
+ "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=2008-12-01",
+ self.url)
+ self.assertEqual("HTTP status code: 200\n"
+ "\n"
+ "The response\n",
+ 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):
+ self.assertEqual("GET", self.method)
+ self.assertEqual(
+ "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=2008-12-01",
+ self.url)
+ self.assertEqual("HTTP status code: 400\n"
+ "\n"
+ "The error response\n",
+ self.output.getvalue())
+
+ deferred = self.command.run()
+ deferred.addErrback(check)
+ return deferred
+
+ 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):
+ self.assertEqual("GET", self.method)
+ self.assertEqual(
+ "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=2008-12-01",
+ self.url)
+ self.assertEqual("HTTP status code: 400\n"
+ "\n"
+ "The error response\n",
+ self.output.getvalue())
+
+ deferred = self.command.run()
+ deferred.addErrback(check)
+ return deferred
=== added file 'txaws/client/discover/tests/test_entry_point.py'
--- txaws/client/discover/tests/test_entry_point.py 1970-01-01 00:00:00 +0000
+++ txaws/client/discover/tests/test_entry_point.py 2010-06-04 22:50:31 +0000
@@ -0,0 +1,246 @@
+# Copyright (C) 2010 Jamu Kakar <jkakar@xxxxxxxx>
+# Licenced under the txaws licence available at /LICENSE in the txaws source.
+
+"""Unit tests for L{get_command}, L{parse_options} and L{main} functions."""
+
+from cStringIO import StringIO
+import os
+import sys
+
+from txaws.client.discover.entry_point import (
+ OptionError, UsageError, get_command, main, parse_options, USAGE_MESSAGE)
+from txaws.testing.base import TXAWSTestCase
+
+
+class ParseOptionsTest(TXAWSTestCase):
+
+ def test_parse_options(self):
+ """
+ L{parse_options} returns a C{dict} contains options parsed from the
+ command-line.
+ """
+ options = parse_options([
+ "txaws-discover", "--key", "key", "--secret", "secret",
+ "--endpoint", "endpoint", "--action", "action",
+ "--something.else", "something.else"])
+ self.assertEqual({"key": "key", "secret": "secret",
+ "endpoint": "endpoint", "action": "action",
+ "something.else": "something.else"},
+ options)
+
+ def test_parse_options_without_options(self):
+ """An L{OptionError} is raised if no options are provided."""
+ self.assertRaises(OptionError, parse_options, ["txaws-discover"])
+
+ def test_parse_options_with_missing_value(self):
+ """
+ An L{OptionError} is raised if an option is specified without a value.
+ """
+ self.assertRaises(OptionError, parse_options,
+ ["txaws-discover", "--key"])
+
+ def test_parse_options_with_missing_option(self):
+ """
+ An L{OptionError} is raised if a value is specified without an option
+ name.
+ """
+ self.assertRaises(
+ OptionError, parse_options,
+ ["txaws-discover", "--key", "key", "--secret", "secret",
+ "--endpoint", "endpoint", "--action", "action",
+ "random-value"])
+
+ def test_parse_options_without_required_arguments(self):
+ """
+ An access key, access secret, endpoint and action can be specified as
+ command-line arguments. An L{OptionError} is raised if any one of
+ these is missing.
+ """
+ self.assertRaises(OptionError, parse_options,
+ ["txaws-discover", "--secret", "secret",
+ "--endpoint", "endpoint", "--action", "action"])
+ self.assertRaises(OptionError, parse_options,
+ ["txaws-discover", "--key", "key",
+ "--endpoint", "endpoint", "--action", "action"])
+ self.assertRaises(OptionError, parse_options,
+ ["txaws-discover", "--key", "key",
+ "--secret", "secret", "--action", "action"])
+ self.assertRaises(OptionError, parse_options,
+ ["txaws-discover", "--key", "key",
+ "--secret", "secret", "--endpoint", "endpoint"])
+
+ def test_parse_options_gets_key_from_environment(self):
+ """
+ If the C{AWS_ACCESS_KEY_ID} environment variable is present, it will
+ be used if the C{--key} command-line argument isn't specified.
+ """
+ os.environ["AWS_ACCESS_KEY_ID"] = "key"
+ options = parse_options([
+ "txaws-discover", "--secret", "secret", "--endpoint", "endpoint",
+ "--action", "action"])
+ self.assertEqual({"key": "key", "secret": "secret",
+ "endpoint": "endpoint", "action": "action"},
+ options)
+
+ def test_parse_options_prefers_explicit_key(self):
+ """
+ If an explicit C{--key} command-line argument is specified it will be
+ preferred over the value specified in the C{AWS_ACCESS_KEY_ID}
+ environment variable.
+ """
+ os.environ["AWS_ACCESS_KEY_ID"] = "fail"
+ options = parse_options([
+ "txaws-discover", "--key", "key", "--secret", "secret",
+ "--endpoint", "endpoint", "--action", "action"])
+ self.assertEqual({"key": "key", "secret": "secret",
+ "endpoint": "endpoint", "action": "action"},
+ options)
+
+ def test_parse_options_gets_secret_from_environment(self):
+ """
+ If the C{AWS_SECRET_ACCESS_KEY} environment variable is present, it
+ will be used if the C{--secret} command-line argument isn't specified.
+ """
+ os.environ["AWS_SECRET_ACCESS_KEY"] = "secret"
+ options = parse_options([
+ "txaws-discover", "--key", "key", "--endpoint", "endpoint",
+ "--action", "action"])
+ self.assertEqual({"key": "key", "secret": "secret",
+ "endpoint": "endpoint", "action": "action"},
+ options)
+
+ def test_parse_options_prefers_explicit_secret(self):
+ """
+ If an explicit C{--secret} command-line argument is specified it will
+ be preferred over the value specified in the C{AWS_SECRET_ACCESS_KEY}
+ environment variable.
+ """
+ os.environ["AWS_SECRET_ACCESS_KEY"] = "fail"
+ options = parse_options([
+ "txaws-discover", "--key", "key", "--secret", "secret",
+ "--endpoint", "endpoint", "--action", "action"])
+ self.assertEqual({"key": "key", "secret": "secret",
+ "endpoint": "endpoint", "action": "action"},
+ options)
+
+ def test_parse_options_gets_endpoint_from_environment(self):
+ """
+ If the C{AWS_ENDPOINT} environment variable is present, it will be
+ used if the C{--endpoint} command-line argument isn't specified.
+ """
+ os.environ["AWS_ENDPOINT"] = "endpoint"
+ options = parse_options([
+ "txaws-discover", "--key", "key", "--secret", "secret",
+ "--action", "action"])
+ self.assertEqual({"key": "key", "secret": "secret",
+ "endpoint": "endpoint", "action": "action"},
+ options)
+
+ def test_parse_options_prefers_explicit_endpoint(self):
+ """
+ If an explicit C{--endpoint} command-line argument is specified it
+ will be preferred over the value specified in the C{AWS_ENDPOINT}
+ environment variable.
+ """
+ os.environ["AWS_ENDPOINT"] = "fail"
+ options = parse_options([
+ "txaws-discover", "--key", "key", "--secret", "secret",
+ "--endpoint", "endpoint", "--action", "action"])
+ self.assertEqual({"key": "key", "secret": "secret",
+ "endpoint": "endpoint", "action": "action"},
+ options)
+
+ def test_parse_options_raises_usage_error_when_help_specified(self):
+ """
+ L{UsageError} is raised if C{-h} or C{--help} appears in command-line
+ arguments.
+ """
+ self.assertRaises(UsageError, parse_options,
+ ["txaws-discover", "-h"])
+ self.assertRaises(UsageError, parse_options,
+ ["txaws-discover", "--help"])
+ self.assertRaises(UsageError, parse_options,
+ ["txaws-discover", "--key", "key",
+ "--secret", "secret", "--endpoint", "endpoint",
+ "--action", "action", "--help"])
+
+
+class GetCommandTest(TXAWSTestCase):
+
+ def test_get_command_without_arguments(self):
+ """An L{OptionError} is raised if no arguments are provided."""
+ self.assertRaises(OptionError, get_command, ["txaws-discover"])
+
+ def test_get_command(self):
+ """
+ An access key, access secret, endpoint and action can be specified as
+ command-line arguments.
+ """
+ command = get_command([
+ "txaws-discover", "--key", "key", "--secret", "secret",
+ "--endpoint", "endpoint", "--action", "action"])
+ self.assertEqual("key", command.key)
+ self.assertEqual("secret", command.secret)
+ self.assertEqual("endpoint", command.endpoint)
+ self.assertEqual("action", command.action)
+ self.assertIdentical(sys.stdout, command.output)
+
+ def test_get_command_with_custom_output_stream(self):
+ output = StringIO()
+ command = get_command([
+ "txaws-discover", "--key", "key", "--secret", "secret",
+ "--endpoint", "endpoint", "--action", "action"], output)
+ self.assertIdentical(output, command.output)
+
+ def test_get_command_without_required_arguments(self):
+ """
+ An access key, access secret, endpoint and action can be specified as
+ command-line arguments. An L{OptionError} is raised if any one of
+ these is missing.
+ """
+ self.assertRaises(OptionError, get_command,
+ ["txaws-discover", "--secret", "secret",
+ "--endpoint", "endpoint", "--action", "action"])
+ self.assertRaises(OptionError, get_command,
+ ["txaws-discover", "--key", "key",
+ "--endpoint", "endpoint", "--action", "action"])
+ self.assertRaises(OptionError, get_command,
+ ["txaws-discover", "--key", "key",
+ "--secret", "secret", "--action", "action"])
+ self.assertRaises(OptionError, get_command,
+ ["txaws-discover", "--key", "key",
+ "--secret", "secret", "--endpoint", "endpoint"])
+
+ def test_get_command_passes_additional_parameters_to_command(self):
+ """
+ Command-line parameters beyond C{--key}, C{--secret}, C{--endpoint}
+ and C{--action} are passed to the L{Command} in a parameter C{dict}.
+ """
+ command = get_command([
+ "txaws-discover", "--key", "key", "--secret", "secret",
+ "--endpoint", "endpoint", "--action", "DescribeRegions",
+ "--Region.Name.0", "us-west-1"])
+ self.assertEqual({"Region.Name.0": "us-west-1"}, command.parameters)
+
+
+class MainTest(TXAWSTestCase):
+
+ def test_usage_message(self):
+ """
+ If a L{UsageError} is raised, the help screen is written to the output
+ stream.
+ """
+ output = StringIO()
+ main(["txaws-discover", "--help"], output, True)
+ self.assertEqual(USAGE_MESSAGE, output.getvalue())
+
+ def test_error_message(self):
+ """
+ If an exception is raised, its message is written to the output
+ stream.
+ """
+ output = StringIO()
+ main(["txaws-discover"], output, True)
+ self.assertEqual(
+ "ERROR: The '--key' command-line argument is required.\n",
+ output.getvalue())
=== modified file 'txaws/testing/base.py'
--- txaws/testing/base.py 2009-09-05 00:26:12 +0000
+++ txaws/testing/base.py 2010-06-04 22:50:31 +0000
@@ -17,6 +17,8 @@
del os.environ["AWS_ACCESS_KEY_ID"]
if "AWS_SECRET_ACCESS_KEY" in os.environ:
del os.environ["AWS_SECRET_ACCESS_KEY"]
+ if "AWS_ENDPOINT" in os.environ:
+ del os.environ["AWS_ENDPOINT"]
def _restore_environ(self):
for key in set(os.environ) - set(self.orig_environ):