start adding tests for consistency on version_number [data_integrity_test]

hjwp · hjwp · commit f59e9f7b6bb7 · 2024-03-01T15:01:35.000Z
diff --git a/src/allocation/adapters/orm.py b/src/allocation/adapters/orm.py
@@ -19,7 +19,7 @@
     "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
@@ -20,4 +20,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
@@ -62,12 +62,17 @@ def postgres_db():
 
 
 @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():
     (Path(__file__).parent / "../src/allocation/entrypoints/flask_app.py").touch()
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
@@ -1,24 +1,8 @@
-import uuid
 import pytest
 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 = 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 orders.rowcount == 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,17 @@
+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
`@@ -19,7 +19,7 @@`
`19`	`19`	`"products",`
`20`	`20`	`metadata,`
`21`	`21`	`Column("sku", String(255), primary_key=True),`
`22`		`- # Column('version_number', Integer, nullable=False, default=0),`
	`22`	`+ Column("version_number", Integer, nullable=False, server_default="0"),`
`23`	`23`	`)`
`24`	`24`
`25`	`25`	`batches = Table(`