two more tests for rollback behaviour [chapter_05_uow_ends]

hjwp · hjwp · commit 469958ea6979 · 2020-01-08T20:41:36.000Z
diff --git a/tests/integration/test_uow.py b/tests/integration/test_uow.py
@@ -1,13 +1,16 @@
+import pytest
 from allocation.domain import model
 from allocation.service_layer import unit_of_work
 
+
 def insert_batch(session, ref, sku, qty, eta):
     session.execute(
         'INSERT INTO batches (reference, sku, _purchased_quantity, eta)'
         ' VALUES (:ref, :sku, :qty, :eta)',
         dict(ref=ref, sku=sku, qty=qty, eta=eta)
     )
 
+
 def get_allocated_batch_ref(session, orderid, sku):
     [[orderlineid]] = session.execute(
         'SELECT id FROM order_lines WHERE orderid=:orderid AND sku=:sku',
@@ -35,3 +38,28 @@ def test_uow_can_retrieve_a_batch_and_allocate_to_it(session_factory):
 
     batchref = get_allocated_batch_ref(session, 'o1', 'HIPSTER-WORKBENCH')
     assert batchref == 'batch1'
+
+
+def test_rolls_back_uncommitted_work_by_default(session_factory):
+    uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory)
+    with uow:
+        insert_batch(uow.session, 'batch1', 'MEDIUM-PLINTH', 100, None)
+
+    new_session = session_factory()
+    rows = list(new_session.execute('SELECT * FROM "batches"'))
+    assert rows == []
+
+
+def test_rolls_back_on_error(session_factory):
+    class MyException(Exception):
+        pass
+
+    uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory)
+    with pytest.raises(MyException):
+        with uow:
+            insert_batch(uow.session, 'batch1', 'LARGE-FORK', 100, None)
+            raise MyException()
+
+    new_session = session_factory()
+    rows = list(new_session.execute('SELECT * FROM "batches"'))
+    assert rows == []
