import boto

from time import sleep
from unittest import TestCase

class TestKinesis(TestCase):
    def setUp(self):
        self.kinesis = boto.connect_kinesis()

    def tearDown(self):
        # Delete the stream even if there is a failure
        self.kinesis.delete_stream('test')

    def test_kinesis(self):
        kinesis = self.kinesis

        # Create a new stream
        kinesis.create_stream('test', 1)

        # Wait for the stream to be ready
        tries = 0
        while tries < 10:
            tries += 1
            sleep(15)
            response = kinesis.describe_stream('test')

            if response['StreamDescription']['StreamStatus'] == 'ACTIVE':
                break

        # Write some data to the stream
        data = 'Some data ...'
        response = kinesis.put_record('test', data, data)
        shard_id = response['ShardId']

        # Process some data from the stream
        response = kinesis.get_shard_iterator('test', shard_id, 'TRIM_HORIZON')
        shard_iterator = response['ShardIterator']

        # Wait for the data to show up
        tries = 0
        while tries < 20:
            tries += 1
            sleep(5)
            
            response = kinesis.get_next_records(shard_iterator, limit=5)
            shard_iterator = response['NextShardIterator']

            if len(response['Records']):
                break

        # Read the data, which should be the same as what we wrote
        self.assertEqual(1, len(response['Records']))
        self.assertEqual(data, response['Records'][0]['Data'])

@@ -1,12 +1,1136 @@
+# Copyright (c) 2013 Amazon.com, Inc. or its affiliates.%0A# All rights reserved.%0A#%0A# Permission is hereby granted, free of charge, to any person obtaining a%0A# copy of this software and associated documentation files (the%0A# %22Software%22), to deal in the Software without restriction, including%0A# without limitation the rights to use, copy, modify, merge, publish, dis-%0A# tribute, sublicense, and/or sell copies of the Software, and to permit%0A# persons to whom the Software is furnished to do so, subject to the fol-%0A# lowing conditions:%0A#%0A# The above copyright notice and this permission notice shall be included%0A# in all copies or substantial portions of the Software.%0A#%0A# THE SOFTWARE IS PROVIDED %22AS IS%22, WITHOUT WARRANTY OF ANY KIND, EXPRESS%0A# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-%0A# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT%0A# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,%0A# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,%0A# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS%0A# IN THE SOFTWARE.%0A%0A
 import boto%0A
@@ -1184,16 +1184,59 @@
 stCase%0A%0A
+%0Aclass TimeoutError(Exception):%0A    pass%0A%0A%0A
 class Te
@@ -1853,32 +1853,120 @@
            break
+%0A        else:%0A            raise TimeoutError('Stream is still not active, aborting...')
 %0A%0A        # Writ
@@ -2610,16 +2610,16 @@
 rds'%5D):%0A
-
         
@@ -2631,16 +2631,94 @@
    break
+%0A        else:%0A            raise TimeoutError('No records found, aborting...')
 %0A%0A      

Add a license and handle timeouts if the stream never becomes active or no records are ever read within the given time limits.