Browse files

Refactor boto test organization

The boto tests have been split into two main directories,
unit and integration.  The unit test is intended for test cases
that do not talk to any remote services, while the integration
tests are for any testcases that talk to remote services and
require valid AWS credentials.

I've also updated the tests.py runner to be a lightweight wrapper
to nosetests.  This makes it easier for people to contribute
new tests and ensure they will be run as well as choosing which
tests you want to run.

While splitting the tests and running the tests, there are small
number of minor bug fixes I made (with appropriate tests included).
  • Loading branch information...
1 parent 6496de1 commit e62525b0b4cdab1c23cd97dfbbf72810d9d37810 @jamesls jamesls committed May 17, 2012
Showing with 226 additions and 274 deletions.
  1. +44 −30 boto/utils.py
  2. 0 tests/{swf → integration}/__init__.py
  3. 0 tests/{ → integration}/autoscale/__init__.py
  4. 0 tests/{ → integration}/autoscale/test_connection.py
  5. 0 tests/{sts → integration/dynamodb}/__init__.py
  6. 0 tests/{ → integration}/dynamodb/test_layer1.py
  7. 0 tests/{ → integration}/dynamodb/test_layer2.py
  8. 0 tests/{sqs → integration/ec2}/__init__.py
  9. 0 tests/{sdb → integration/ec2/cloudwatch}/__init__.py
  10. 0 tests/{ → integration}/ec2/cloudwatch/test_connection.py
  11. 0 tests/{s3 → integration/ec2/elb}/__init__.py
  12. +4 −5 tests/{ → integration}/ec2/elb/test_connection.py
  13. +0 −1 tests/{ → integration}/ec2/test_connection.py
  14. 0 tests/{ → integration}/mws/__init__.py
  15. 0 tests/{ → integration}/mws/test.py
  16. 0 tests/{ec2/elb → integration/s3}/__init__.py
  17. 0 tests/{ → integration}/s3/cb_test_harnass.py
  18. 0 tests/{ → integration}/s3/mock_storage_service.py
  19. 0 tests/{ → integration}/s3/other_cacerts.txt
  20. 0 tests/{ → integration}/s3/test_bucket.py
  21. +3 −2 tests/{ → integration}/s3/test_connection.py
  22. 0 tests/{ → integration}/s3/test_encryption.py
  23. +21 −0 tests/{ → integration}/s3/test_gsconnection.py
  24. +5 −0 tests/{ → integration}/s3/test_https_cert_validation.py
  25. 0 tests/{ → integration}/s3/test_key.py
  26. +4 −0 tests/{ → integration}/s3/test_mfa.py
  27. 0 tests/{ → integration}/s3/test_multidelete.py
  28. 0 tests/{ → integration}/s3/test_multipart.py
  29. 0 tests/{ → integration}/s3/test_pool.py
  30. 0 tests/{ → integration}/s3/test_resumable_downloads.py
  31. 0 tests/{ → integration}/s3/test_resumable_uploads.py
  32. +2 −0 tests/{ → integration}/s3/test_versioning.py
  33. 0 tests/{ec2/cloudwatch → integration/sdb}/__init__.py
  34. 0 tests/{ → integration}/sdb/test_connection.py
  35. 0 tests/{ec2 → integration/sqs}/__init__.py
  36. 0 tests/{ → integration}/sqs/test_connection.py
  37. 0 tests/{dynamodb → integration/sts}/__init__.py
  38. 0 tests/{ → integration}/sts/test_session_token.py
  39. 0 tests/{cloudfront → integration/swf}/__init__.py
  40. 0 tests/{ → integration}/swf/test_layer1.py
  41. 0 tests/{ → integration}/swf/test_layer1_workflow_execution.py
  42. +6 −118 tests/test.py
  43. +62 −0 tests/unit/__init__.py
  44. 0 tests/{ → unit}/cloudformation/__init__.py
  45. +27 −78 tests/{ → unit}/cloudformation/test_connection.py
  46. 0 tests/unit/cloudfront/__init__.py
  47. 0 tests/{ → unit}/cloudfront/test_signed_urls.py
  48. 0 tests/{ → unit}/emr/test_emr_responses.py
  49. +48 −40 tests/{utils/test_password.py → unit/utils/test_utils.py}
View
74 boto/utils.py
@@ -16,7 +16,7 @@
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
-# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
@@ -50,6 +50,7 @@
import tempfile
import smtplib
import datetime
+import re
from email.MIMEMultipart import MIMEMultipart
from email.MIMEBase import MIMEBase
from email.MIMEText import MIMEText
@@ -71,14 +72,20 @@
_hashfn = md5.md5
# List of Query String Arguments of Interest
-qsa_of_interest = ['acl', 'cors', 'defaultObjectAcl', 'location', 'logging',
- 'partNumber', 'policy', 'requestPayment', 'torrent',
- 'versioning', 'versionId', 'versions', 'website',
- 'uploads', 'uploadId', 'response-content-type',
- 'response-content-language', 'response-expires',
+qsa_of_interest = ['acl', 'cors', 'defaultObjectAcl', 'location', 'logging',
+ 'partNumber', 'policy', 'requestPayment', 'torrent',
+ 'versioning', 'versionId', 'versions', 'website',
+ 'uploads', 'uploadId', 'response-content-type',
+ 'response-content-language', 'response-expires',
'response-cache-control', 'response-content-disposition',
'response-content-encoding', 'delete', 'lifecycle']
+
+_first_cap_regex = re.compile('(.)([A-Z][a-z]+)')
+_number_cap_regex = re.compile('([a-z])([0-9]+)')
+_end_cap_regex = re.compile('([a-z0-9])([A-Z])')
+
+
def unquote_v(nv):
if len(nv) == 1:
return nv
@@ -237,7 +244,7 @@ def get_instance_userdata(version='latest', sep=None,
ISO8601 = '%Y-%m-%dT%H:%M:%SZ'
ISO8601_MS = '%Y-%m-%dT%H:%M:%S.%fZ'
-
+
def get_ts(ts=None):
if not ts:
ts = time.gmtime()
@@ -267,7 +274,7 @@ def find_class(module_name, class_name=None):
return c
except:
return None
-
+
def update_dme(username, password, dme_id, ip_address):
"""
Update your Dynamic DNS record with DNSMadeEasy.com
@@ -280,7 +287,7 @@ def update_dme(username, password, dme_id, ip_address):
def fetch_file(uri, file=None, username=None, password=None):
"""
Fetch a file based on the URI provided. If you do not pass in a file pointer
- a tempfile.NamedTemporaryFile, or None if the file could not be
+ a tempfile.NamedTemporaryFile, or None if the file could not be
retrieved is returned.
The URI can be either an HTTP url, or "s3://bucket_name/key_name"
"""
@@ -358,7 +365,7 @@ class AuthSMTPHandler(logging.handlers.SMTPHandler):
to accept a username and password on the constructor and to then use those
credentials to authenticate with the SMTP server. To use this, you could
add something like this in your boto config file:
-
+
[handler_hand07]
class=boto.utils.AuthSMTPHandler
level=WARN
@@ -376,7 +383,7 @@ def __init__(self, mailhost, username, password, fromaddr, toaddrs, subject):
logging.handlers.SMTPHandler.__init__(self, mailhost, fromaddr, toaddrs, subject)
self.username = username
self.password = password
-
+
def emit(self, record):
"""
Emit a record.
@@ -407,29 +414,29 @@ def emit(self, record):
class LRUCache(dict):
"""A dictionary-like object that stores only a certain number of items, and
discards its least recently used item when full.
-
+
>>> cache = LRUCache(3)
>>> cache['A'] = 0
>>> cache['B'] = 1
>>> cache['C'] = 2
>>> len(cache)
3
-
+
>>> cache['A']
0
-
+
Adding new items to the cache does not increase its size. Instead, the least
recently used item is dropped:
-
+
>>> cache['D'] = 3
>>> len(cache)
3
>>> 'B' in cache
False
-
+
Iterating over the cache returns the keys, starting with the most recently
used:
-
+
>>> for key in cache:
... print key
D
@@ -539,10 +546,10 @@ def __init__(self, str=None, hashfunc=None):
def set(self, value):
self.str = self.hashfunc(value).hexdigest()
-
+
def __str__(self):
return str(self.str)
-
+
def __eq__(self, other):
if other == None:
return False
@@ -569,7 +576,7 @@ def notify(subject, body=None, html_body=None, to_string=None, attachments=None,
msg['To'] = to_string
msg['Date'] = formatdate(localtime=True)
msg['Subject'] = subject
-
+
if body:
msg.attach(MIMEText(body))
@@ -620,16 +627,23 @@ def mklist(value):
value = [value]
return value
-def pythonize_name(name, sep='_'):
- s = ''
- if name[0].isupper:
- s = name[0].lower()
- for c in name[1:]:
- if c.isupper():
- s += sep + c.lower()
- else:
- s += c
- return s
+def pythonize_name(name):
+ """Convert camel case to a "pythonic" name.
+
+ Examples::
+
+ pythonize_name('CamelCase') -> 'camel_case'
+ pythonize_name('already_pythonized') -> 'already_pythonized'
+ pythonize_name('HTTPRequest') -> 'http_request'
+ pythonize_name('HTTPStatus200Ok') -> 'http_status_200_ok'
+ pythonize_name('UPPER') -> 'upper'
+ pythonize_name('') -> ''
+
+ """
+ s1 = _first_cap_regex.sub(r'\1_\2', name)
+ s2 = _number_cap_regex.sub(r'\1_\2', s1)
+ return _end_cap_regex.sub(r'\1_\2', s2).lower()
+
def write_mime_multipart(content, compress=False, deftype='text/plain', delimiter=':'):
"""Description:
View
0 tests/swf/__init__.py → tests/integration/__init__.py
File renamed without changes.
View
0 tests/autoscale/__init__.py → tests/integration/autoscale/__init__.py
File renamed without changes.
View
0 tests/autoscale/test_connection.py → .../integration/autoscale/test_connection.py
File renamed without changes.
View
0 tests/sts/__init__.py → tests/integration/dynamodb/__init__.py
File renamed without changes.
View
0 tests/dynamodb/test_layer1.py → tests/integration/dynamodb/test_layer1.py
File renamed without changes.
View
0 tests/dynamodb/test_layer2.py → tests/integration/dynamodb/test_layer2.py
File renamed without changes.
View
0 tests/sqs/__init__.py → tests/integration/ec2/__init__.py
File renamed without changes.
View
0 tests/sdb/__init__.py → tests/integration/ec2/cloudwatch/__init__.py
File renamed without changes.
View
0 tests/ec2/cloudwatch/test_connection.py → ...gration/ec2/cloudwatch/test_connection.py
File renamed without changes.
View
0 tests/s3/__init__.py → tests/integration/ec2/elb/__init__.py
File renamed without changes.
View
9 tests/ec2/elb/test_connection.py → tests/integration/ec2/elb/test_connection.py
@@ -89,15 +89,14 @@ def test_delete_load_balancer_listeners(self):
balancers = c.get_all_load_balancers()
self.assertEqual([lb.name for lb in balancers], [name])
self.assertEqual(
- [l.get_tuple() for l in balancers[0].listeners], listeners)
+ sorted([l.get_tuple() for l in balancers[0].listeners]),
+ sorted(listeners))
c.delete_load_balancer_listeners(name, [443])
balancers = c.get_all_load_balancers()
self.assertEqual([lb.name for lb in balancers], [name])
- self.assertEqual(
- [l.get_tuple() for l in balancers[0].listeners],
- listeners[:1]
- )
+ self.assertEqual([l.get_tuple() for l in balancers[0].listeners],
+ listeners[:1])
def test_create_load_balancer_listeners_with_policies(self):
c = ELBConnection()
View
1 tests/ec2/test_connection.py → tests/integration/ec2/test_connection.py
@@ -36,7 +36,6 @@ class EC2ConnectionTest (unittest.TestCase):
def test_1_basic(self):
# this is my user_id, if you want to run these tests you should
# replace this with yours or they won't work
- user_id = '963068290131'
print '--- running EC2Connection tests ---'
c = EC2Connection()
# get list of private AMI's
View
0 tests/mws/__init__.py → tests/integration/mws/__init__.py
File renamed without changes.
View
0 tests/mws/test.py → tests/integration/mws/test.py 100755 → 100644
File renamed without changes.
View
0 tests/ec2/elb/__init__.py → tests/integration/s3/__init__.py
File renamed without changes.
View
0 tests/s3/cb_test_harnass.py → tests/integration/s3/cb_test_harnass.py
File renamed without changes.
View
0 tests/s3/mock_storage_service.py → tests/integration/s3/mock_storage_service.py
File renamed without changes.
View
0 tests/s3/other_cacerts.txt → tests/integration/s3/other_cacerts.txt
File renamed without changes.
View
0 tests/s3/test_bucket.py → tests/integration/s3/test_bucket.py
File renamed without changes.
View
5 tests/s3/test_connection.py → tests/integration/s3/test_connection.py
@@ -228,8 +228,9 @@ def test_basic_anon(self):
try:
iter(anon_bucket.list()).next()
self.fail("not expecting contents")
- except S3ResponseError:
- self.fail("we should have public-read access.")
+ except S3ResponseError, e:
+ self.fail("We should have public-read access, but received "
+ "an error: %s" % e)
except StopIteration:
pass
View
0 tests/s3/test_encryption.py → tests/integration/s3/test_encryption.py
File renamed without changes.
View
21 tests/s3/test_gsconnection.py → tests/integration/s3/test_gsconnection.py
@@ -38,7 +38,28 @@
from boto.gs.cors import Cors
from boto import handler
from boto import storage_uri
+from boto.provider import Provider
+
+_HAS_GOOGLE_CREDENTIALS = None
+
+
+def has_google_credentials():
+ global _HAS_GOOGLE_CREDENTIALS
+ if _HAS_GOOGLE_CREDENTIALS is None:
+ provider = Provider('google')
+ if provider.access_key is None or provider.secret_key is None:
+ _HAS_GOOGLE_CREDENTIALS = False
+ else:
+ _HAS_GOOGLE_CREDENTIALS = True
+ return _HAS_GOOGLE_CREDENTIALS
+
+
+
+@unittest.skipUnless(has_google_credentials(),
+ "Google credentials are required to run the Google "
+ "Cloud Storage tests. Update your boto.cfg to run "
+ "these tests.")
class GSConnectionTest (unittest.TestCase):
def test_1_basic(self):
View
5 tests/s3/test_https_cert_validation.py → ...egration/s3/test_https_cert_validation.py
@@ -39,11 +39,14 @@
import ssl
import unittest
+from nose.plugins.attrib import attr
+
import boto
from boto import exception, https_connection
from boto.gs.connection import GSConnection
from boto.s3.connection import S3Connection
+
# File 'other_cacerts.txt' contains a valid CA certificate of a CA that is used
# by neither S3 nor Google Cloud Storage. Validation against this CA cert should
# result in a certificate error.
@@ -59,6 +62,8 @@
# the server should return a certificate with CN 'www.<somedomain>.com').
INVALID_HOSTNAME_HOST = os.environ.get('INVALID_HOSTNAME_HOST', 'www')
+
+@attr('notdefault')
class CertValidationTest (unittest.TestCase):
def setUp(self):
View
0 tests/s3/test_key.py → tests/integration/s3/test_key.py
File renamed without changes.
View
4 tests/s3/test_mfa.py → tests/integration/s3/test_mfa.py
@@ -27,10 +27,14 @@
import unittest
import time
+from nose.plugins.attrib import attr
+
from boto.s3.connection import S3Connection
from boto.exception import S3ResponseError
from boto.s3.deletemarker import DeleteMarker
+
+@attr('notdefault', 's3mfa')
class S3MFATest (unittest.TestCase):
def setUp(self):
View
0 tests/s3/test_multidelete.py → tests/integration/s3/test_multidelete.py
File renamed without changes.
View
0 tests/s3/test_multipart.py → tests/integration/s3/test_multipart.py
File renamed without changes.
View
0 tests/s3/test_pool.py → tests/integration/s3/test_pool.py
File renamed without changes.
View
0 tests/s3/test_resumable_downloads.py → ...ntegration/s3/test_resumable_downloads.py
File renamed without changes.
View
0 tests/s3/test_resumable_uploads.py → .../integration/s3/test_resumable_uploads.py
File renamed without changes.
View
2 tests/s3/test_versioning.py → tests/integration/s3/test_versioning.py
@@ -112,6 +112,8 @@ def test_1_versions(self):
# Now suspend Versioning on the bucket
self.bucket.configure_versioning(False)
+ # Allow time for the change to fully propagate.
+ time.sleep(3)
d = self.bucket.get_versioning_status()
self.assertEqual('Suspended', d['Versioning'])
View
0 tests/ec2/cloudwatch/__init__.py → tests/integration/sdb/__init__.py
File renamed without changes.
View
0 tests/sdb/test_connection.py → tests/integration/sdb/test_connection.py
File renamed without changes.
View
0 tests/ec2/__init__.py → tests/integration/sqs/__init__.py
File renamed without changes.
View
0 tests/sqs/test_connection.py → tests/integration/sqs/test_connection.py
File renamed without changes.
View
0 tests/dynamodb/__init__.py → tests/integration/sts/__init__.py
File renamed without changes.
View
0 tests/sts/test_session_token.py → tests/integration/sts/test_session_token.py
File renamed without changes.
View
0 tests/cloudfront/__init__.py → tests/integration/swf/__init__.py
File renamed without changes.
View
0 tests/swf/test_layer1.py → tests/integration/swf/test_layer1.py
File renamed without changes.
View
0 tests/swf/test_layer1_workflow_execution.py → ...ion/swf/test_layer1_workflow_execution.py
File renamed without changes.
View
124 tests/test.py
@@ -20,132 +20,20 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
-"""
-do the unit tests!
-"""
-
import logging
import sys
import unittest
-import getopt
+import argparse
-from sqs.test_connection import SQSConnectionTest
-from s3.test_connection import S3ConnectionTest
-from s3.test_versioning import S3VersionTest
-from s3.test_mfa import S3MFATest
-from s3.test_encryption import S3EncryptionTest
-from s3.test_bucket import S3BucketTest
-from s3.test_key import S3KeyTest
-from s3.test_multidelete import S3MultiDeleteTest
-from s3.test_multipart import S3MultiPartUploadTest
-from s3.test_gsconnection import GSConnectionTest
-from s3.test_https_cert_validation import CertValidationTest
-from s3.test_resumable_downloads import ResumableDownloadTests
-from s3.test_resumable_uploads import ResumableUploadTests
-from ec2.test_connection import EC2ConnectionTest
-from ec2.elb.test_connection import ELBConnectionTest
-from ec2.cloudwatch.test_connection import CloudWatchConnectionTest
-from autoscale.test_connection import AutoscaleConnectionTest
-from sdb.test_connection import SDBConnectionTest
-from cloudfront.test_signed_urls import CloudfrontSignedUrlsTest
-from dynamodb.test_layer1 import DynamoDBLayer1Test
-from dynamodb.test_layer2 import DynamoDBLayer2Test
-from sts.test_session_token import SessionTokenTest
-from swf.test_layer1 import SimpleWorkflowLayer1Test
-from swf.test_layer1_workflow_execution import SwfL1WorkflowExecutionTest
+from nose.core import run
-def usage():
- print "test.py [-t testsuite] [-v verbosity]"
- print " -t run specific testsuite (s3|ssl|s3mfa|gs|sqs|ec2|sdb|dynamodb|dynamodbL1|dynamodbL2|sts|all)"
- print " -v verbosity (0|1|2)"
def main():
- try:
- opts, args = getopt.getopt(sys.argv[1:], "ht:v:",
- ["help", "testsuite", "verbosity"])
- except:
- usage()
- sys.exit(2)
- testsuite = "all"
- verbosity = 1
- for o, a in opts:
- if o in ("-h", "--help"):
- usage()
- sys.exit()
- if o in ("-t", "--testsuite"):
- testsuite = a
- if o in ("-v", "--verbosity"):
- verbosity = int(a)
- if len(args) != 0:
- usage()
- sys.exit()
- try:
- tests = suite(testsuite)
- except ValueError:
- usage()
- sys.exit()
- if verbosity > 1:
- logging.basicConfig(level=logging.DEBUG)
- unittest.TextTestRunner(verbosity=verbosity).run(tests)
+ parser = argparse.ArgumentParser()
+ known_args, remaining_args = parser.parse_known_args()
+ default_args = '-a !notdefault'
+ run(argv=[__file__, default_args] + remaining_args)
-def suite(testsuite="all"):
- tests = unittest.TestSuite()
- if testsuite == "all":
- tests.addTest(unittest.makeSuite(SQSConnectionTest))
- tests.addTest(unittest.makeSuite(S3ConnectionTest))
- tests.addTest(unittest.makeSuite(EC2ConnectionTest))
- tests.addTest(unittest.makeSuite(SDBConnectionTest))
- tests.addTest(unittest.makeSuite(AutoscaleConnectionTest))
- tests.addTest(unittest.makeSuite(CloudfrontSignedUrlsTest))
- tests.addTest(unittest.makeSuite(DynamoDBLayer1Test))
- tests.addTest(unittest.makeSuite(DynamoDBLayer2Test))
- elif testsuite == "s3":
- tests.addTest(unittest.makeSuite(S3ConnectionTest))
- tests.addTest(unittest.makeSuite(S3BucketTest))
- tests.addTest(unittest.makeSuite(S3KeyTest))
- tests.addTest(unittest.makeSuite(S3MultiPartUploadTest))
- tests.addTest(unittest.makeSuite(S3VersionTest))
- tests.addTest(unittest.makeSuite(S3EncryptionTest))
- tests.addTest(unittest.makeSuite(S3MultiDeleteTest))
- elif testsuite == "ssl":
- tests.addTest(unittest.makeSuite(CertValidationTest))
- elif testsuite == "s3mfa":
- tests.addTest(unittest.makeSuite(S3MFATest))
- elif testsuite == "gs":
- tests.addTest(unittest.makeSuite(GSConnectionTest))
- tests.addTest(unittest.makeSuite(ResumableDownloadTests))
- tests.addTest(unittest.makeSuite(ResumableUploadTests))
- elif testsuite == "sqs":
- tests.addTest(unittest.makeSuite(SQSConnectionTest))
- elif testsuite == "ec2":
- tests.addTest(unittest.makeSuite(EC2ConnectionTest))
- tests.addTest(unittest.makeSuite(AutoscaleConnectionTest))
- tests.addTest(unittest.makeSuite(ELBConnectionTest))
- tests.addTest(unittest.makeSuite(CloudWatchConnectionTest))
- elif testsuite == "autoscale":
- tests.addTest(unittest.makeSuite(AutoscaleConnectionTest))
- elif testsuite == "sdb":
- tests.addTest(unittest.makeSuite(SDBConnectionTest))
- elif testsuite == "cloudfront":
- tests.addTest(unittest.makeSuite(CloudfrontSignedUrlsTest))
- elif testsuite == "dynamodb":
- tests.addTest(unittest.makeSuite(DynamoDBLayer1Test))
- tests.addTest(unittest.makeSuite(DynamoDBLayer2Test))
- elif testsuite == "dynamodbL1":
- tests.addTest(unittest.makeSuite(DynamoDBLayer1Test))
- elif testsuite == "dynamodbL2":
- tests.addTest(unittest.makeSuite(DynamoDBLayer2Test))
- elif testsuite == "sts":
- tests.addTest(unittest.makeSuite(SessionTokenTest))
- elif testsuite == "swf":
- tests.addTest(unittest.makeSuite(SimpleWorkflowLayer1Test))
- tests.addTest(unittest.makeSuite(SwfL1WorkflowExecutionTest))
- elif testsuite == "swfL1":
- tests.addTest(unittest.makeSuite(SimpleWorkflowLayer1Test))
- tests.addTest(unittest.makeSuite(SwfL1WorkflowExecutionTest))
- else:
- raise ValueError("Invalid choice.")
- return tests
if __name__ == "__main__":
main()
View
62 tests/unit/__init__.py
@@ -0,0 +1,62 @@
+import unittest
+import httplib
+
+from mock import Mock
+
+
+class AWSMockServiceTestCase(unittest.TestCase):
+ """Base class for mocking aws services."""
+ # This param is used by the unittest module to display a full
+ # diff when assert*Equal methods produce an error message.
+ maxDiff = None
+ connection_class = None
+
+ def setUp(self):
+ self.https_connection = Mock(spec=httplib.HTTPSConnection)
+ self.https_connection_factory = (
+ Mock(return_value=self.https_connection), ())
+ self.service_connection = self.create_service_connection(
+ https_connection_factory=self.https_connection_factory,
+ aws_access_key_id='aws_access_key_id',
+ aws_secret_access_key='aws_secret_access_key')
+ self.actual_request = None
+ self.original_mexe = self.service_connection._mexe
+ self.service_connection._mexe = self._mexe_spy
+
+ def create_service_connection(self, **kwargs):
+ if self.connection_class is None:
+ raise ValueError("The connection_class class attribute must be "
+ "set to a non-None value.")
+ return self.connection_class(**kwargs)
+
+ def _mexe_spy(self, request, *args, **kwargs):
+ self.actual_request = request
+ return self.original_mexe(request, *args, **kwargs)
+
+ def create_response(self, status_code, reason='', body=None):
+ if body is None:
+ body = self.default_body()
+ response = Mock(spec=httplib.HTTPResponse)
+ response.status = status_code
+ response.read.return_value = body
+ response.reason = reason
+ return response
+
+ def assert_request_parameters(self, params, ignore_params_values=None):
+ """Verify the actual parameters sent to the service API."""
+ request_params = self.actual_request.params.copy()
+ if ignore_params_values is not None:
+ for param in ignore_params_values:
+ # We still want to check that the ignore_params_values params
+ # are in the request parameters, we just don't need to check
+ # their value.
+ self.assertIn(param, request_params)
+ del request_params[param]
+ self.assertDictEqual(request_params, params)
+
+ def set_http_response(self, status_code, reason='', body=None):
+ http_response = self.create_response(status_code, reason, body)
+ self.https_connection.getresponse.return_value = http_response
+
+ def default_body(self):
+ return ''
View
0 tests/cloudformation/__init__.py → tests/unit/cloudformation/__init__.py
File renamed without changes.
View
105 tests/cloudformation/test_connection.py → tests/unit/cloudformation/test_connection.py
@@ -6,6 +6,7 @@
from mock import Mock
+from tests.unit import AWSMockServiceTestCase
from boto.cloudformation.connection import CloudFormationConnection
@@ -37,64 +38,12 @@
}
"""
-
-class CloudFormationConnectionBase(unittest.TestCase):
- # This param is used by the unittest module to display a full
- # diff when assert*Equal methods produce an error message.
- maxDiff = None
+class CloudFormationConnectionBase(AWSMockServiceTestCase):
+ connection_class = CloudFormationConnection
def setUp(self):
- self.https_connection = Mock(spec=httplib.HTTPSConnection)
- self.https_connection_factory = (
- Mock(return_value=self.https_connection), ())
+ super(CloudFormationConnectionBase, self).setUp()
self.stack_id = u'arn:aws:cloudformation:us-east-1:18:stack/Name/id'
- self.cloud_formation = CloudFormationConnection(
- https_connection_factory=self.https_connection_factory,
- aws_access_key_id='aws_access_key_id',
- aws_secret_access_key='aws_secret_access_key')
- self.actual_request = None
- # We want to be able to verify the request params that
- # are sent to the CloudFormation service. By the time
- # we get to the boto.connection.AWSAuthConnection._mexe
- # method, all of the request params have been populated.
- # By patching out the _mexe method with self._mexe_spy,
- # we can record the actual http request that _mexe is passed
- # so that we can verify the http params (and anything
- # else about the request that we want).
- self.original_mexe = self.cloud_formation._mexe
- self.cloud_formation._mexe = self._mexe_spy
-
- def _mexe_spy(self, request, *args, **kwargs):
- self.actual_request = request
- return self.original_mexe(request, *args, **kwargs)
-
- def create_response(self, status_code, reason='', body=None):
- if body is None:
- body = self.default_body()
- response = Mock(spec=httplib.HTTPResponse)
- response.status = status_code
- response.read.return_value = body
- response.reason = reason
- return response
-
- def assert_request_parameters(self, params, ignore_params_values=None):
- """Verify the actual parameters sent to the service API."""
- request_params = self.actual_request.params.copy()
- if ignore_params_values is not None:
- for param in ignore_params_values:
- # We still want to check that the ignore_params_values params
- # are in the request parameters, we just don't need to check
- # their value.
- self.assertIn(param, request_params)
- del request_params[param]
- self.assertDictEqual(request_params, params)
-
- def set_http_response(self, status_code, reason='', body=None):
- http_response = self.create_response(status_code, reason, body)
- self.https_connection.getresponse.return_value = http_response
-
- def default_body(self):
- return json.dumps({})
class TestCloudFormationCreateStack(CloudFormationConnectionBase):
@@ -106,7 +55,7 @@ def default_body(self):
def test_create_stack_has_correct_request_params(self):
self.set_http_response(status_code=200)
- api_response = self.cloud_formation.create_stack(
+ api_response = self.service_connection.create_stack(
'stack_name', template_url='http://url',
template_body=SAMPLE_TEMPLATE,
parameters=[('KeyName', 'myKeyName')],
@@ -144,7 +93,7 @@ def test_create_stack_has_correct_request_params(self):
def test_create_stack_with_minimum_args(self):
# This will fail in practice, but the API docs only require stack_name.
self.set_http_response(status_code=200)
- api_response = self.cloud_formation.create_stack('stack_name')
+ api_response = self.service_connection.create_stack('stack_name')
self.assertEqual(api_response, self.stack_id)
self.assert_request_parameters({
'AWSAccessKeyId': 'aws_access_key_id',
@@ -160,8 +109,8 @@ def test_create_stack_with_minimum_args(self):
def test_create_stack_fails(self):
self.set_http_response(status_code=400, reason='Bad Request',
body='Invalid arg.')
- with self.assertRaises(self.cloud_formation.ResponseError):
- api_response = self.cloud_formation.create_stack(
+ with self.assertRaises(self.service_connection.ResponseError):
+ api_response = self.service_connection.create_stack(
'stack_name', template_body=SAMPLE_TEMPLATE,
parameters=[('KeyName', 'myKeyName')])
@@ -175,7 +124,7 @@ def default_body(self):
def test_update_stack_all_args(self):
self.set_http_response(status_code=200)
- api_response = self.cloud_formation.update_stack(
+ api_response = self.service_connection.update_stack(
'stack_name', template_url='http://url',
template_body=SAMPLE_TEMPLATE,
parameters=[('KeyName', 'myKeyName')],
@@ -203,7 +152,7 @@ def test_update_stack_all_args(self):
def test_update_stack_with_minimum_args(self):
self.set_http_response(status_code=200)
- api_response = self.cloud_formation.update_stack('stack_name')
+ api_response = self.service_connection.update_stack('stack_name')
self.assertEqual(api_response, self.stack_id)
self.assert_request_parameters({
'AWSAccessKeyId': 'aws_access_key_id',
@@ -219,8 +168,8 @@ def test_update_stack_with_minimum_args(self):
def test_update_stack_fails(self):
self.set_http_response(status_code=400, reason='Bad Request',
body='Invalid arg.')
- with self.assertRaises(self.cloud_formation.ResponseError):
- api_response = self.cloud_formation.update_stack(
+ with self.assertRaises(self.service_connection.ResponseError):
+ api_response = self.service_connection.update_stack(
'stack_name', template_body=SAMPLE_TEMPLATE,
parameters=[('KeyName', 'myKeyName')])
@@ -233,7 +182,7 @@ def default_body(self):
def test_delete_stack(self):
self.set_http_response(status_code=200)
- api_response = self.cloud_formation.delete_stack('stack_name')
+ api_response = self.service_connection.delete_stack('stack_name')
self.assertEqual(api_response, json.loads(self.default_body()))
self.assert_request_parameters({
'AWSAccessKeyId': 'aws_access_key_id',
@@ -247,8 +196,8 @@ def test_delete_stack(self):
def test_delete_stack_fails(self):
self.set_http_response(status_code=400)
- with self.assertRaises(self.cloud_formation.ResponseError):
- api_response = self.cloud_formation.delete_stack('stack_name')
+ with self.assertRaises(self.service_connection.ResponseError):
+ api_response = self.service_connection.delete_stack('stack_name')
class TestCloudFormationDescribeStackResource(CloudFormationConnectionBase):
@@ -257,7 +206,7 @@ def default_body(self):
def test_describe_stack_resource(self):
self.set_http_response(status_code=200)
- api_response = self.cloud_formation.describe_stack_resource(
+ api_response = self.service_connection.describe_stack_resource(
'stack_name', 'resource_id')
self.assertEqual(api_response, 'fake server response')
self.assert_request_parameters({
@@ -273,8 +222,8 @@ def test_describe_stack_resource(self):
def test_describe_stack_resource_fails(self):
self.set_http_response(status_code=400)
- with self.assertRaises(self.cloud_formation.ResponseError):
- api_response = self.cloud_formation.describe_stack_resource(
+ with self.assertRaises(self.service_connection.ResponseError):
+ api_response = self.service_connection.describe_stack_resource(
'stack_name', 'resource_id')
@@ -284,7 +233,7 @@ def default_body(self):
def test_get_template(self):
self.set_http_response(status_code=200)
- api_response = self.cloud_formation.get_template('stack_name')
+ api_response = self.service_connection.get_template('stack_name')
self.assertEqual(api_response, 'fake server response')
self.assert_request_parameters({
'AWSAccessKeyId': 'aws_access_key_id',
@@ -299,8 +248,8 @@ def test_get_template(self):
def test_get_template_fails(self):
self.set_http_response(status_code=400)
- with self.assertRaises(self.cloud_formation.ResponseError):
- api_response = self.cloud_formation.get_template('stack_name')
+ with self.assertRaises(self.service_connection.ResponseError):
+ api_response = self.service_connection.get_template('stack_name')
class TestCloudFormationGetStackevents(CloudFormationConnectionBase):
@@ -335,7 +284,7 @@ def default_body(self):
def test_describe_stack_events(self):
self.set_http_response(status_code=200)
- first, second = self.cloud_formation.describe_stack_events('stack_name', next_token='next_token')
+ first, second = self.service_connection.describe_stack_events('stack_name', next_token='next_token')
self.assertEqual(first.event_id, 'Event-1-Id')
self.assertEqual(first.logical_resource_id, 'MyStack')
self.assertEqual(first.physical_resource_id, 'MyStack_One')
@@ -398,7 +347,7 @@ def default_body(self):
def test_describe_stack_resources(self):
self.set_http_response(status_code=200)
- first, second = self.cloud_formation.describe_stack_resources(
+ first, second = self.service_connection.describe_stack_resources(
'stack_name', 'logical_resource_id', 'physical_resource_id')
self.assertEqual(first.description, None)
self.assertEqual(first.logical_resource_id, 'MyDBInstance')
@@ -473,7 +422,7 @@ def default_body(self):
def test_describe_stacks(self):
self.set_http_response(status_code=200)
- stack = self.cloud_formation.describe_stacks('MyStack')[0]
+ stack = self.service_connection.describe_stacks('MyStack')[0]
self.assertEqual(stack.creation_time,
datetime(2012, 5, 16, 22, 55, 31))
self.assertEqual(stack.description, 'My Description')
@@ -537,7 +486,7 @@ def default_body(self):
def test_list_stack_resources(self):
self.set_http_response(status_code=200)
- resources = self.cloud_formation.list_stack_resources('MyStack',
+ resources = self.service_connection.list_stack_resources('MyStack',
next_token='next_token')
self.assertEqual(len(resources), 2)
self.assertEqual(resources[0].last_updated_timestamp,
@@ -587,7 +536,7 @@ def default_body(self):
def test_list_stacks(self):
self.set_http_response(status_code=200)
- stacks = self.cloud_formation.list_stacks(['CREATE_IN_PROGRESS'],
+ stacks = self.service_connection.list_stacks(['CREATE_IN_PROGRESS'],
next_token='next_token')
self.assertEqual(len(stacks), 1)
self.assertEqual(stacks[0].stack_id,
@@ -638,7 +587,7 @@ def default_body(self):
def test_validate_template(self):
self.set_http_response(status_code=200)
- template = self.cloud_formation.validate_template(template_body=SAMPLE_TEMPLATE,
+ template = self.service_connection.validate_template(template_body=SAMPLE_TEMPLATE,
template_url='http://url')
self.assertEqual(template.description, 'My Description.')
self.assertEqual(len(template.template_parameters), 2)
View
0 tests/unit/cloudfront/__init__.py
No changes.
View
0 tests/cloudfront/test_signed_urls.py → tests/unit/cloudfront/test_signed_urls.py
File renamed without changes.
View
0 tests/emr/test_emr_responses.py → tests/unit/emr/test_emr_responses.py
File renamed without changes.
View
88 tests/utils/test_password.py → tests/unit/utils/test_utils.py
@@ -14,26 +14,25 @@
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
-# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
#
import unittest
+import hashlib
+import hmac
+from boto.utils import Password
+from boto.utils import pythonize_name
-import logging
-log = logging.getLogger(__file__)
class TestPassword(unittest.TestCase):
"""Test basic password functionality"""
-
- def clstest(self, cls):
-
- """Insure that password.__eq__ hashes test value before compare"""
- password=cls('foo')
- log.debug( "Password %s" % password )
+ def clstest(self, cls):
+ """Insure that password.__eq__ hashes test value before compare."""
+ password = cls('foo')
self.assertNotEquals(password, 'foo')
password.set('foo')
@@ -46,56 +45,65 @@ def clstest(self, cls):
self.assertEquals(password, 'foo')
self.assertEquals(password.str, hashed)
-
def test_aaa_version_1_9_default_behavior(self):
- from boto.utils import Password
self.clstest(Password)
def test_custom_hashclass(self):
-
- from boto.utils import Password
- import hashlib
-
class SHA224Password(Password):
- hashfunc=hashlib.sha224
+ hashfunc = hashlib.sha224
- password=SHA224Password()
+ password = SHA224Password()
password.set('foo')
- self.assertEquals( hashlib.sha224('foo').hexdigest(), str(password))
-
- def test_hmac(self):
- from boto.utils import Password
- import hmac
+ self.assertEquals(hashlib.sha224('foo').hexdigest(), str(password))
+ def test_hmac(self):
def hmac_hashfunc(cls, msg):
- log.debug("\n%s %s" % (cls.__class__, cls) )
return hmac.new('mysecretkey', msg)
class HMACPassword(Password):
- hashfunc=hmac_hashfunc
+ hashfunc = hmac_hashfunc
self.clstest(HMACPassword)
- password=HMACPassword()
+ password = HMACPassword()
password.set('foo')
-
- self.assertEquals(str(password), hmac.new('mysecretkey', 'foo').hexdigest())
- def test_constructor(self):
- from boto.utils import Password
- import hmac
+ self.assertEquals(str(password),
+ hmac.new('mysecretkey', 'foo').hexdigest())
- hmac_hashfunc = lambda msg: hmac.new('mysecretkey', msg )
+ def test_constructor(self):
+ hmac_hashfunc = lambda msg: hmac.new('mysecretkey', msg)
password = Password(hashfunc=hmac_hashfunc)
password.set('foo')
- self.assertEquals(password.str, hmac.new('mysecretkey', 'foo').hexdigest())
+ self.assertEquals(password.str,
+ hmac.new('mysecretkey', 'foo').hexdigest())
+
+
+class TestPythonizeName(unittest.TestCase):
+ def test_empty_string(self):
+ self.assertEqual(pythonize_name(''), '')
+
+ def test_all_lower_case(self):
+ self.assertEqual(pythonize_name('lowercase'), 'lowercase')
+
+ def test_all_upper_case(self):
+ self.assertEqual(pythonize_name('UPPERCASE'), 'uppercase')
+
+ def test_camel_case(self):
+ self.assertEqual(pythonize_name('OriginallyCamelCased'),
+ 'originally_camel_cased')
+
+ def test_already_pythonized(self):
+ self.assertEqual(pythonize_name('already_pythonized'),
+ 'already_pythonized')
+
+ def test_multiple_upper_cased_letters(self):
+ self.assertEqual(pythonize_name('HTTPRequest'), 'http_request')
+ self.assertEqual(pythonize_name('RequestForHTTP'), 'request_for_http')
+
+ def test_string_with_numbers(self):
+ self.assertEqual(pythonize_name('HTTPStatus200Ok'), 'http_status_200_ok')
+
-
-
if __name__ == '__main__':
- import sys
- sys.path = [ '../../' ] + sys.path
- #logging.basicConfig()
- #log.setLevel(logging.DEBUG)
- suite = unittest.TestLoader().loadTestsFromTestCase(TestPassword)
- unittest.TextTestRunner(verbosity=2).run(suite)
+ unittest.main()

0 comments on commit e62525b

Please sign in to comment.