Fix IntegrityError due to autoflush

- catch IntegrityError on patch
- disable autoflush (IntegrityError was raised before commit due to
autoflush)

app/api/main.py

@@ -170,7 +170,11 @@ def patch_item(id_):
     for child in item.children:
         child.location = item.location
         child.status = item.status
-    db.session.commit()
+    try:
+        db.session.commit()
+    except sa.exc.IntegrityError as e:
+        db.session.rollback()
+        raise utils.InventoryError('IntegrityError', status_code=409)
     return jsonify(item.to_dict())
 
 

app/extensions.py

@@ -29,8 +29,7 @@ convention = {
 }
 
 metadata = MetaData(naming_convention=convention)
-db = SQLAlchemy(metadata=metadata)
-# db = SQLAlchemy(metadata=metadata, session_options={'autoflush': False})
+db = SQLAlchemy(metadata=metadata, session_options={'autoflush': False})
 migrate = Migrate(db=db)
 login_manager = LoginManager()
 ldap_manager = LDAP3LoginManager()

tests/functional/conftest.py

@@ -67,7 +67,7 @@ def session(db, request):
     """
     connection = db.engine.connect()
     transaction = connection.begin()
-    options = dict(bind=connection, binds={})
+    options = dict(bind=connection, binds={}, autoflush=False)
     session = db.create_scoped_session(options=options)
     session.begin_nested()
 

tests/functional/test_api.py

@@ -231,6 +231,7 @@ def test_get_item_fail(client, session, readonly_token):
 def test_get_item(client, session, readonly_token):
     # Create some items
     session.add(models.Status(name='Stock'))
+    session.flush()
     item1 = models.Item(serial_number='123456')
     item2 = models.Item(serial_number='234567', ics_id='AAA001', status='Stock')
     for item in (item1, item2):
@@ -288,6 +289,7 @@ def test_patch_item(client, session, user_token):
     # Create some items
     session.add(models.Status(name='Stock'))
     session.add(models.Status(name='In service'))
+    session.flush()
     item1 = models.Item(serial_number='123456')
     item2 = models.Item(serial_number='234567', ics_id='AAA001', status='Stock')
     for item in (item1, item2):
@@ -311,6 +313,18 @@ def test_patch_item(client, session, user_token):
     assert response.json['status'] == data['status']
 
 
+def test_patch_item_integrity_error(client, session, user_token):
+    # Create some items
+    item1 = models.Item(serial_number='123456', ics_id='AAA001')
+    item2 = models.Item(serial_number='234567')
+    for item in (item1, item2):
+        session.add(item)
+    session.commit()
+    data = {'ics_id': 'AAA001'}
+    response = patch(client, f'/api/items/{item2.id}', data=data, token=user_token)
+    check_response_message(response, 'IntegrityError', 409)
+
+
 def test_patch_item_parent(client, session, user_token):
     # Create some items
     session.add(models.Manufacturer(name='HP'))
@@ -319,6 +333,7 @@ def test_patch_item_parent(client, session, user_token):
     session.add(models.Status(name='In service'))
     session.add(models.Location(name='ESS'))
     session.add(models.Location(name='ICS lab'))
+    session.flush()
     item1 = models.Item(serial_number='123456', ics_id='AAA001', location='ICS lab', status='In service',
                         manufacturer='Dell')
     item2 = models.Item(serial_number='234567', ics_id='AAA002', status='Stock')
@@ -383,6 +398,7 @@ def test_get_items(client, session, readonly_token):
     session.add(models.Status(name='In service'))
     session.add(models.Location(name='ESS'))
     session.add(models.Location(name='ICS lab'))
+    session.flush()
     item1 = models.Item(serial_number='123456', ics_id='AAA001', location='ESS', status='In service',
                         manufacturer='Dell')
     item2 = models.Item(serial_number='234567', ics_id='AAA002', status='Stock')