Add base test class for model db CRUD tests.
This way we don't need to copy paste and repeat this code.
@@ -18,6 +18,47 @@ import mongoengine
from st2common.models import db
from st2common.models.db import stormbase
from st2common.persistence.base import Access
+from st2common.exceptions.db import StackStormDBObjectNotFoundError
+
+__all__ = [
+ 'BaseDBModelCRUDTestCase',
+
+ 'FakeModelDB',
+ 'FakeModelDB'
+]
+
+
+class BaseDBModelCRUDTestCase(object):
+ model_class = None
+ persistance_class = None
+ model_class_kwargs = {}
+ update_attribute_name = None
+
+ def test_crud_operations(self):
+ # 1. Test create
+ model_db = self.model_class(**self.model_class_kwargs)
+ saved_db = self.persistance_class.add_or_update(model_db)
+
+ retrieved_db = self.persistance_class.get_by_id(saved_db.id)
+ self.assertEqual(saved_db.id, retrieved_db.id)
+
+ for attribute_name, attribute_value in self.model_class_kwargs.items():
+ self.assertEqual(getattr(saved_db, attribute_name), attribute_value)
+ self.assertEqual(getattr(retrieved_db, attribute_name), attribute_value)
+
+ # 2. Test update
+ setattr(model_db, self.update_attribute_name, 'updated')
+ saved_db = self.persistance_class.add_or_update(model_db)
+ self.assertEqual(getattr(saved_db, self.update_attribute_name), 'updated')
+
+ retrieved_db = self.persistance_class.get_by_id(saved_db.id)
+ self.assertEqual(saved_db.id, retrieved_db.id)
+ self.assertEqual(getattr(retrieved_db, self.update_attribute_name), 'updated')
+
+ # 3. Test delete
+ self.persistance_class.delete(model_db)
+ self.assertRaises(StackStormDBObjectNotFoundError, self.persistance_class.get_by_id,
+ model_db.id)
class FakeModelDB(stormbase.StormBaseDB):
