test_dynamo_query.py

from boto.dynamodb2.fields import HashKey, RangeKey, KeysOnlyIndex, AllIndex, GlobalAllIndex
from boto.dynamodb2.table import Table
from boto.dynamodb2.types import NUMBER
from boto.dynamodb2.exceptions import ValidationException
import time
import uuid
import unittest

TABLE_NAME = "test_table"

class TestDynamoQuerying(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.table = Table.create(TABLE_NAME, schema = [
                HashKey("id"),
                RangeKey("index", data_type=NUMBER)
            ], throughput = {
                "read" : 2,
                "write" : 2
            }, global_indexes = [
                GlobalAllIndex("ordered-collection", parts = [
                    HashKey("collection", data_type=NUMBER),
                    RangeKey("index", data_type=NUMBER)
                ], throughput = {
                    "read" : 2,
                    "write" : 2
                })
            ]
        )
        
        while not cls.table.describe().get("Table", {}).get("TableStatus", None) == u'ACTIVE':
            time.sleep(5)
        
        with cls.table.batch_write() as batch:
            for coll in xrange(100,110):
                for i in xrange(10):
                    batch.put_item(data = {
                        "index" : i,
                        "id" : unicode(uuid.uuid4()),
                        "collection" : coll
                    })

    @classmethod
    def tearDownClass(cls):
        cls.table.delete()

    def test_collection_query(self):
        self.assertEqual(10, len(list(self.table.query(collection__eq=105, index="ordered-collection"))))
        self.assertEqual(5, len(list(self.table.query(collection__eq=104, index__gt=4, index="ordered-collection"))))
    
    def test_table_schema(self):
        self.assertRaises(ValidationException, list, self.table.query(index__eq=5))
        self.assertEqual(10, len(list(self.table.scan(index__eq=5))))
    
    def test_table_global_index_failure(self):
        table = Table(TABLE_NAME)
        self.assertEqual(10, len(list(table.query(collection__eq=105, index="ordered-collection"))))
    
    def test_table_schema_failure(self):
        table = Table(TABLE_NAME)
        self.assertEqual(10, len(list(table.scan(index__eq=5))))
        
if __name__ == '__main__':
    unittest.main()