start adding tests for consistency on version_number [data_integrity_test]

hjwp · hjwp · commit b4ffd279b0c8 · 2020-01-08T20:41:48.000Z
diff --git a/src/allocation/adapters/orm.py b/src/allocation/adapters/orm.py
@@ -20,7 +20,7 @@
 products = Table(
     'products', metadata,
     Column('sku', String(255), primary_key=True),
-    # Column('version_number', Integer, nullable=False, default=0),
+    Column('version_number', Integer, nullable=False, server_default='0'),
 )
 
 batches = Table(
diff --git a/src/allocation/adapters/repository.py b/src/allocation/adapters/repository.py
@@ -22,4 +22,5 @@ def add(self, product):
         self.session.add(product)
 
     def get(self, sku):
+        print(sku, type(sku))
         return self.session.query(model.Product).filter_by(sku=sku).first()
diff --git a/tests/__init__.py b/tests/__init__.py
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -59,13 +59,16 @@ def postgres_db():
     metadata.create_all(engine)
     return engine
 
-
 @pytest.fixture
-def postgres_session(postgres_db):
+def postgres_session_factory(postgres_db):
     start_mappers()
-    yield sessionmaker(bind=postgres_db)()
+    yield sessionmaker(bind=postgres_db)
     clear_mappers()
 
+@pytest.fixture
+def postgres_session(postgres_session_factory):
+    return postgres_session_factory()
+
 
 @pytest.fixture
 def restart_api():
diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py
diff --git a/tests/e2e/test_api.py b/tests/e2e/test_api.py
@@ -3,18 +3,7 @@
 import requests
 
 from allocation import config
-
-def random_suffix():
-    return uuid.uuid4().hex[:6]
-
-def random_sku(name=''):
-    return f'sku-{name}-{random_suffix()}'
-
-def random_batchref(name=''):
-    return f'batch-{name}-{random_suffix()}'
-
-def random_orderid(name=''):
-    return f'order-{name}-{random_suffix()}'
+from ..random_refs import random_sku, random_batchref, random_orderid
 
 
 def post_to_add_batch(ref, sku, qty, eta):
diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py
diff --git a/tests/integration/test_uow.py b/tests/integration/test_uow.py
@@ -1,12 +1,18 @@
+# pylint: disable=broad-except
+import threading
+import time
+import traceback
+from typing import List
 import pytest
 from allocation.domain import model
 from allocation.service_layer import unit_of_work
+from ..random_refs import random_sku, random_batchref, random_orderid
 
 
-def insert_batch(session, ref, sku, qty, eta):
+def insert_batch(session, ref, sku, qty, eta, product_version=1):
     session.execute(
-        'INSERT INTO products (sku) VALUES (:sku)',
-        dict(sku=sku),
+        'INSERT INTO products (sku, version_number) VALUES (:sku, :version)',
+        dict(sku=sku, version=product_version),
     )
     session.execute(
         'INSERT INTO batches (reference, sku, _purchased_quantity, eta)'
@@ -67,3 +73,53 @@ class MyException(Exception):
     new_session = session_factory()
     rows = list(new_session.execute('SELECT * FROM "batches"'))
     assert rows == []
+
+
+def try_to_allocate(orderid, sku, exceptions):
+    line = model.OrderLine(orderid, sku, 10)
+    try:
+        with unit_of_work.SqlAlchemyUnitOfWork() as uow:
+            product = uow.products.get(sku=sku)
+            product.allocate(line)
+            time.sleep(0.2)
+            uow.commit()
+    except Exception as e:
+        print(traceback.format_exc())
+        exceptions.append(e)
+
+
+def test_concurrent_updates_to_version_are_not_allowed(postgres_session_factory):
+    sku, batch = random_sku(), random_batchref()
+    session = postgres_session_factory()
+    insert_batch(session, batch, sku, 100, eta=None, product_version=1)
+    session.commit()
+
+    order1, order2 = random_orderid(1), random_orderid(2)
+    exceptions = []  # type: List[Exception]
+    try_to_allocate_order1 = lambda: try_to_allocate(order1, sku, exceptions)
+    try_to_allocate_order2 = lambda: try_to_allocate(order2, sku, exceptions)
+    thread1 = threading.Thread(target=try_to_allocate_order1)
+    thread2 = threading.Thread(target=try_to_allocate_order2)
+    thread1.start()
+    thread2.start()
+    thread1.join()
+    thread2.join()
+
+    [[version]] = session.execute(
+        "SELECT version_number FROM products WHERE sku=:sku",
+        dict(sku=sku),
+    )
+    assert version == 2
+    [exception] = exceptions
+    assert 'could not serialize access due to concurrent update' in str(exception)
+
+    orders = list(session.execute(
+        "SELECT orderid FROM allocations"
+        " JOIN batches ON allocations.batch_id = batches.id"
+        " JOIN order_lines ON allocations.orderline_id = order_lines.id"
+        " WHERE order_lines.sku=:sku",
+        dict(sku=sku),
+    ))
+    assert len(orders) == 1
+    with unit_of_work.SqlAlchemyUnitOfWork() as uow:
+        uow.session.execute('select 1')
diff --git a/tests/random_refs.py b/tests/random_refs.py
@@ -0,0 +1,13 @@
+import uuid
+
+def random_suffix():
+    return uuid.uuid4().hex[:6]
+
+def random_sku(name=''):
+    return f'sku-{name}-{random_suffix()}'
+
+def random_batchref(name=''):
+    return f'batch-{name}-{random_suffix()}'
+
+def random_orderid(name=''):
+    return f'order-{name}-{random_suffix()}'

Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,7 @@`
`20`	`20`	`products = Table(`
`21`	`21`	`'products', metadata,`
`22`	`22`	`Column('sku', String(255), primary_key=True),`
`23`		`- # Column('version_number', Integer, nullable=False, default=0),`
	`23`	`+ Column('version_number', Integer, nullable=False, server_default='0'),`
`24`	`24`	`)`
`25`	`25`
`26`	`26`	`batches = Table(`