adding Transact testing utility with transact pytest fixture (#119)

django_sorcery/pytest_plugin.py

@@ -4,9 +4,16 @@
 import pytest
 
 from .db.profiler import SQLAlchemyProfiler
+from .testing import Transact
 
 
 @pytest.fixture(scope="function")
 def sqlalchemy_profiler():
     with SQLAlchemyProfiler() as profiler:
         yield profiler
+
+
+@pytest.fixture(scope="function")
+def transact():
+    with Transact() as t:
+        yield t

django_sorcery/testing.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+from __future__ import absolute_import, print_function, unicode_literals
+
+from sqlalchemy import event
+from sqlalchemy.engine import Engine
+
+from .db import databases
+
+
+class CommitException(BaseException):
+    pass
+
+
+class Transact(object):
+    @staticmethod
+    def _commit(connection, *args, **kwargs):
+        raise CommitException("Commits are not allowed")
+
+    def start(self):
+        self.hook()
+
+    def stop(self):
+        self.end()
+        self.unhook()
+
+    def hook(self):
+        if not event.contains(Engine, "commit", self._commit):
+            event.listen(Engine, "commit", self._commit)
+        if not event.contains(Engine, "commit_twophase", self._commit):
+            event.listen(Engine, "commit_twophase", self._commit)
+
+    def unhook(self):
+        if event.contains(Engine, "commit", self._commit):
+            event.remove(Engine, "commit", self._commit)
+        if event.contains(Engine, "commit_twophase", self._commit):
+            event.remove(Engine, "commit_twophase", self._commit)
+
+    def end(self):
+        databases.rollback()
+        databases.remove()
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.stop()

tests/test_pytest_plugin.py

@@ -1,7 +1,10 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import, print_function, unicode_literals
 
-from django_sorcery.pytest_plugin import sqlalchemy_profiler  # noqa
+import pytest
+
+from django_sorcery.pytest_plugin import sqlalchemy_profiler, transact  # noqa
+from django_sorcery.testing import CommitException
 
 from .testapp.models import Business, Owner, db
 
@@ -35,3 +38,17 @@ def test_profiler(sqlalchemy_profiler):  # noqa
 
     db.rollback()
     db.remove()
+
+
+def test_transact(transact):  # noqa
+    db.add(Owner(first_name="foo", last_name="bar"))
+    db.flush()
+
+    assert Owner.objects.count() == 1
+
+    with pytest.raises(CommitException):
+        db.commit()
+
+    transact.stop()
+
+    assert Owner.objects.count() == 0