HiveBrain v1.2.0
Get Started
← Back to all entries
patternpythonMinor

Testing boto connection

Submitted by: @import:stackexchange-codereview··
0
Viewed 0 times
bototestingconnection

Problem

I am writing a unit test to be sure the code can connect to aws via boto, since I will be putting the credentials in place via some other process than code deployment. So far, I've only figured out how to check if I have a valid connection by looking at an S3 bucket, but am hoping someone has a more direct way to check.

import unittest
import boto

class TestTest(unittest.TestCase):
    def setUp(self):
        self.s3 = boto.connect_s3()

    def tearDown(self):
        pass

    def test_connection(self):
        assert self.s3 is not None

        # must be a better way to ensure connection works
        bucket = boto.s3.bucket.Bucket(connection=self.s3, name='test_this_bucket')
        all_keys = bucket.get_all_keys()
        assert len(all_keys) == 0

Solution

If you want just to test connection, checking boto.connect_s3() is good enough. According to the docs it raises an exception if something goes wrong.

In case if you want to do more advanced scenario, you can try to make another test with bucket creation and few keys inside.

I will structure the tests the following way:

import unittest
from time import time, sleep

import boto

class BasicConnectionTest(unittest.TestCase):
    def test_connection_works(self):
        # Please add some credentials if necessary
        connection = boto.connect_s3()
        # when using unittest, I'm trying to stick to helper methods
        # because they are more verbose
        self.assertIsNotNone(connection)

class AdvancedConnectionTest(unittest.TestCase):
    def setUp(self):
        self.connection = boto.connect_s3()

    def test_bucket_can_be_created(self):
        bucket = self.create_bucket()
        self.assertIsNotNone(bucket)
        self.assertEqual(len(bucket.get_all_keys()), 0)

    def test_keys_can_be_created(self):
        bucket = self.create_bucket()
        key = self.create_key(bucket, "mykey")
        key.set_contents_from_string("Hello World!")
        sleep(2)
        contents = key.get_contents_as_string()
        self.assertEqual(contents, "Hello World!")

    def create_bucket(self):
        bucket = self.connection.create_bucket('boto-demo-%s' % int(time()))
        self.addCleanup(bucket.delete)
        return bucket

    def create_key(self, bucket, name):
        key = bucket.new_key('mykey')
        self.addCleanup(key.delete)
        return key


In this case you will be safe that if something will go wrong in your tests, addCleanup will take care of removing bucket and key.

It is also possible to test some configs, e.g. http://boto.readthedocs.org/en/latest/ref/s3.html#boto.s3.bucket.Bucket.get_lifecycle_config

Just experiment :)

Code Snippets

import unittest
from time import time, sleep

import boto


class BasicConnectionTest(unittest.TestCase):
    def test_connection_works(self):
        # Please add some credentials if necessary
        connection = boto.connect_s3()
        # when using unittest, I'm trying to stick to helper methods
        # because they are more verbose
        self.assertIsNotNone(connection)


class AdvancedConnectionTest(unittest.TestCase):
    def setUp(self):
        self.connection = boto.connect_s3()

    def test_bucket_can_be_created(self):
        bucket = self.create_bucket()
        self.assertIsNotNone(bucket)
        self.assertEqual(len(bucket.get_all_keys()), 0)

    def test_keys_can_be_created(self):
        bucket = self.create_bucket()
        key = self.create_key(bucket, "mykey")
        key.set_contents_from_string("Hello World!")
        sleep(2)
        contents = key.get_contents_as_string()
        self.assertEqual(contents, "Hello World!")

    def create_bucket(self):
        bucket = self.connection.create_bucket('boto-demo-%s' % int(time()))
        self.addCleanup(bucket.delete)
        return bucket

    def create_key(self, bucket, name):
        key = bucket.new_key('mykey')
        self.addCleanup(key.delete)
        return key

Context

StackExchange Code Review Q#96950, answer score: 6

Revisions (0)

No revisions yet.