Source

dynamodb-mapper / real_dynamodb_tests.py

Full commit
#!/usr/bin/python

import unittest
from boto.exception import DynamoDBResponseError
from boto.dynamodb.exceptions import DynamoDBKeyNotFoundError
from dynamodb_mapper.model import DynamoDBModel, ConnectionBorg, OverwriteError, ConflictError, autoincrement_int

PREFIX=u"jttest-"

"""
The goal of theses tests is to check data is saved as expected and errors really
triggered when expected.
Most tests and scenarios are done in unitests where boto is mocked.
All the DDBM internals are suppose to work/pass

Test uses ``consistent_read=True`` to avoid side effects
"""

class DoomMap(DynamoDBModel):
    __table__ = PREFIX+u"doom_map"
    __hash_key__ = u"episode"
    __range_key__ = u"map"
    __schema__ = {
        u"episode": int,
        u"map": int,
        u"name": unicode,
        u"cheats": set,
    }
    __defaults__ = {
        u"cheats": set([u"Konami"]),
    }

class Trac(DynamoDBModel):
    __table__ = PREFIX+u"trac"
    __hash_key__ = u"id"
    __schema__ = {
        u"id": autoincrement_int,
        u"title": unicode,
        u"description": unicode,
    }

def init_tables():
    conn = ConnectionBorg()
    try:
        conn.create_table(DoomMap, 10, 10, wait_for_active=True)
    except DynamoDBResponseError, e:
        if e.body['message'].find(u"Duplicate table name") < 0:
            raise
    try:
        conn.create_table(Trac, 10, 10, wait_for_active=True)
    except DynamoDBResponseError, e:
        if e.body['message'].find(u"Duplicate table name") < 0:
            raise

# tests
class TestSequenceFunctions(unittest.TestCase):

    def test_regular_life_cycle(self):
        # create an object
        doommap = DoomMap(episode=1, map=1, name=u"entry level")
        doommap.save()
        # make sure it is there
        doommap2 = DoomMap.get(1,1, consistent_read=True)
        self.assertEqual(doommap._raw_data, doommap2._raw_data)
        # delete it
        doommap2.delete()
        # make sure it's gone
        self.assertRaises(
            DynamoDBKeyNotFoundError,
            DoomMap.get,
            1, 1, consistent_read=True
        )

    def test_create_overwriting_new_item(self):
        # create original object
        doommap = DoomMap(episode=1, map=1, name=u"entry level")
        doommap.save()
        # Attempts to overwrite it
        doommap2 = DoomMap(episode=1, map=1, name=u"---OVERWRITING--- entry level")
        self.assertRaises(
            OverwriteError,
            doommap2.save,
            raise_on_conflict=True
        )
        # Attempts to delete it, should not even trigger a DB request :)
        self.assertRaises(
            ConflictError,
            doommap2.delete,
            raise_on_conflict=True
        )
        # delete it
        doommap2.delete()

    def test_concurent_updates(self):
        # create original object
        doommap = DoomMap(episode=1, map=1, name=u"entry level")
        doommap.save()
        # Read a new copy of the object, edit and save it
        doommap2 = DoomMap.get(1, 1, consistent_read=True)
        doommap2.name = u"enhanced entry level"
        doommap2.save()
        # Attempts a "lost update"
        doommap.name = u"---LOST UPDATE--- enhanced entry level"
        self.assertRaises(
            ConflictError,
            doommap.save,
            raise_on_conflict=True
        )
        # Attempts a "lost delete"
        self.assertRaises(
            ConflictError,
            doommap.delete,
            raise_on_conflict=True
        )
        # make sure the doommap2 survived
        doommap3 = DoomMap.get(1, 1, consistent_read=True)
        self.assertEqual(doommap3._raw_data, doommap2._raw_data)
        # delete it
        doommap2.delete()

    def test_key_change(self):
        # create original object
        doommap = DoomMap(episode=1, map=1, name=u"entry level")
        doommap.save()
        # read it from the DB
        doommap2 = DoomMap.get(1, 1, consistent_read=True)
        # attempts to change the IDs
        doommap2.episode = 4
        doommap2.map = 2
        self.assertRaises(
            ConflictError,
            doommap2.save,
            raise_on_conflict=True
        )
        # re-insert with new IDs
        doommap2.save()
        # make sure it is there
        doommap3 = DoomMap.get(4, 2, consistent_read=True)
        self.assertEqual(doommap3._raw_data, doommap2._raw_data)
        # delete them
        doommap.delete()
        doommap3.delete()

    def test_get_bunch_batch(self):
        bunch = 120
        #insert these bunch of objects
        for i in range(bunch):
            DoomMap(episode=i, map=i*i, name=u"secret level"+str(i)).save()
        #get all these
        keys = [(i, i*i) for i in range(bunch)]
        data = list(DoomMap.get_batch(keys))

        self.assertEqual(bunch, len(data))

    def test_auto_inc(self):
        #insert a first item
        ticket1 = Trac(title="auto_inc is not working", description="better not")
        ticket1.save()
        self.assertNotEqual(0, ticket1.id)
        #insert another ticket
        ticket2 = Trac(title="no idea what the bug is", description="random")
        ticket2.save()
        self.assertNotEqual(0, ticket2.id)
        self.assertEqual(ticket1.id+1, ticket2.id)
        #delete
        ticket1.delete()
        ticket2.delete()


if __name__ == '__main__':
    init_tables()
    unittest.main()