Commits

Jean-Tiare Le Bigot  committed ba37a00

add test for schema reloading. Tests are back to 100% :)

  • Participants
  • Parent commits ea15791

Comments (0)

Files changed (4)

File ddbmock/database/db.py

         if name in self.data:
             raise ResourceInUseException("Table {} already exists".format(name))
         if len(self.data) >= config.MAX_TABLES:
-            raise LimitExceededException("Table limit reached. You can have more than {} tables simultaneously".format(config.MAX_TABLES))
+            raise LimitExceededException("Table limit reached. You can not have more than {} tables simultaneously".format(config.MAX_TABLES))
 
         self.data[name] = Table.from_dict(data)
-        self.store[name, None] = self.data[name]
+        self.store[name, False] = self.data[name]
         return self.data[name]
 
     def _internal_delete_table(self, name):
         if name in self.data:
             self.data[name].store.truncate()  # FIXME: should be moved in table
             del self.data[name]
-            del self.store[name, None]
+            del self.store[name, False]
 
     def delete_table(self, name):
         if name not in self.data:

File tests/functional/internal/__init__.py

+# -*- coding: utf-8 -*-

File tests/functional/internal/test_db_schema_persistence.py

+# -*- coding: utf-8 -*-
+
+import unittest, mock
+
+# This module both validates schema persitence/reloading and that sqlite backend is
+# actually working
+
+TABLE_NAME = "tabloid"
+
+TABLE_RT = 45
+TABLE_WT = 123
+
+HASH_KEY = {"AttributeName":"hash_key","AttributeType":"N"}
+RANGE_KEY = {"AttributeName":"range_key","AttributeType":"S"}
+
+class TestDBSchemaPersist(unittest.TestCase):
+    @mock.patch('ddbmock.database.db.Store')
+    def test_db_schema_persistence(self, m_store):
+        from ddbmock.database.db import DynamoDB
+        from ddbmock.database import storage
+        from ddbmock.database.storage.sqlite import Store
+
+        data = {
+            "TableName": TABLE_NAME,
+            "KeySchema": {
+                "HashKeyElement": HASH_KEY,
+                "RangeKeyElement": RANGE_KEY,
+            },
+            "ProvisionedThroughput": {
+                "ReadCapacityUnits": TABLE_RT,
+                "WriteCapacityUnits": TABLE_WT,
+            }
+        }
+
+        m_store.return_value = Store("whatever")
+
+        # reset internal state + force SQLite
+        old_internal_state = DynamoDB._shared_data
+        DynamoDB._shared_data = {'data': {}, 'store': None}
+
+        # data creation round
+        dynamodb = DynamoDB()
+        dynamodb.create_table(TABLE_NAME, data)
+
+        # reset internal state (again) to simulate reload
+        DynamoDB._shared_data = {'data': {}, 'store': None}
+        dynamodb = DynamoDB()
+
+        table = dynamodb.data[TABLE_NAME]
+        self.assertEqual(TABLE_NAME, table.name)
+        self.assertEqual(TABLE_RT, table.rt)
+        self.assertEqual(TABLE_WT, table.wt)
+
+        DynamoDB._shared_data = old_internal_state

File tests/unit/test_db.py

         from ddbmock.database import dynamodb
 
         dynamodb.data[TABLE_NAME] = mock.Mock()
-        dynamodb.store[TABLE_NAME, None] = dynamodb.data[TABLE_NAME]
+        dynamodb.store[TABLE_NAME, False] = dynamodb.data[TABLE_NAME]
 
     def test_internal_delete_table(self):
         from ddbmock.database import dynamodb
         dynamodb.create_table(TABLE_NAME2, data)
 
         m_store.__setitem__.assert_called_with(
-                                               (TABLE_NAME2,None),
+                                               (TABLE_NAME2, False),
                                                dynamodb.data[TABLE_NAME2],
                                               )