Commits

Jean-Tiare Le Bigot committed fd0eac0

full transactions test coverage

Comments (0)

Files changed (2)

dynamodb_mapper/model.py

             default = default()
         # Check default value consitency
         if not isinstance(default, schema_type):
-            raise TypeError("Default value is of type {}. Expected: {}".format(schema_type.__name__, default.__class__.__name__))
+            raise TypeError("Expected default value of type {}, got: {}".format(schema_type.__name__, default.__class__.__name__))
         else:
             return default
 

dynamodb_mapper/tests/test_transactions.py

         "energy": int
     }
 
+class Reward(DynamoDBModel):
+    __table__ = "rewards"
+    __hash_key__ = u'user_id'
+    __range_key__ = u'name'
+    __schema__ = {
+        'user_id': int,
+        'name': unicode,
+        'collected': bool,
+    }
 
 class InsufficientEnergyError(Exception):
     """Raised when a transaction would make a User's energy negative."""
     """Raised when the universe ceases to exist."""
     pass
 
+class collectRewards(Transaction):
+    """A sample transaction using the new system to work on multiple targets
+    It also relies on the basic schema provided in the base class.
+
+    In this testsm the getter returns a brand new object each time. this is probably
+    not what you want to do in real code.
+
+    Note that any function is supported, not only lambas.
+    """
+    __table__ = "rewards"
+
+    def _get_transactors(self):
+        return [
+            (
+                lambda: Reward(user_id=1, name=u"level up", collected=False),
+                lambda target: setattr(target, "collected", True)
+            ),
+            (
+                lambda: Reward(user_id=1, name=u"5 days", collected=False),
+                lambda target: setattr(target, "collected", True)
+            )
+        ]
 
 class UserEnergyTransaction(Transaction):
     """A sample transaction that adds/removes energy to a User."""
     transient = True
 
 
+
 class TestTransaction(unittest.TestCase):
     def _get_default_user(self):
         return User.from_dict({"id": USER_ID, "energy": ENERGY})
         self.assertEquals(m_transaction_save.call_count, 0)
 
         self.assertEquals(m_user_instance.energy, 0)
+        self.assertEquals(t.status, "done")
 
     @mock.patch("dynamodb_mapper.transactions.Transaction.save")
     @mock.patch("dynamodb_mapper.model.DynamoDBModel.save")
         self.assertEquals(m_user_save.call_count, Transaction.MAX_RETRIES)
         m_transaction_save.assert_called()
         self.assertEquals(t.status, "running")
+
+    def test_get_2_transactors(self):
+        t = collectRewards()
+        transactors = t._get_transactors()
+
+        self.assertEquals(len(transactors), 2)
+
+    def test_legacy_get_transactors(self):
+        t = UserEnergyTransaction()
+        transactors = t._get_transactors()
+
+        self.assertEquals(len(transactors), 1)
+        self.assertEquals(transactors[0][0], t._get_target)
+        self.assertEquals(transactors[0][1], t._alter_target)
+
+    @mock.patch("dynamodb_mapper.transactions.Transaction.save")
+    @mock.patch.object(Reward, "save")
+    def test_commit_2_targets(self, m_reward_save, m_transaction_save):
+        t = collectRewards()
+        t.commit()
+
+        self.assertEquals(m_reward_save.call_count, 2)
+        m_transaction_save.assert_called()
+        self.assertEquals(t.status, "done")