Add _get_instance() tests

tests/conftest.py

@@ -151,7 +151,7 @@ def AtomicModel(atomic_db):
     class AtomicModel(db.Model):
         __tablename__ = 'test_atomic_model'
         id = db.Column(db.Integer, primary_key=True)
-        name = db.Column(db.String(60), primary_key=True)
+        name = db.Column(db.String(60))
         value = db.Column(db.String(60))
 
     db.create_all()

tests/test_atomic.py

@@ -58,9 +58,9 @@ def t1():
     def t2():
         with db.retry_on_integrity_error():
             db.session.merge(o)
-    debtors = AtomicModel.query.all()
-    assert len(debtors) == 1
-    assert debtors[0].value == '2'
+    objects = AtomicModel.query.all()
+    assert len(objects) == 1
+    assert objects[0].value == '2'
 
 
 @pytest.mark.skip('too slow')
@@ -85,6 +85,43 @@ def t():
     assert len(call_list) > 1
 
 
+def test_get_instance(atomic_db, AtomicModel):
+    db = atomic_db
+    o = AtomicModel(id=1, name='test', value='1')
+    assert o not in db.session
+    assert AtomicModel._get_instance(o) is None
+    db.session.add(o)
+    assert AtomicModel._get_instance(o) is o
+    assert o in db.session
+    pk = o.id
+    db.session.commit()
+    assert AtomicModel._get_instance(pk) in db.session
+    assert AtomicModel._get_instance((pk,)) in db.session
+    assert AtomicModel._get_instance(o) in db.session
+
+
+def test_lock_instance(atomic_db, AtomicModel):
+    db = atomic_db
+    o = AtomicModel(id=1, name='test', value='1')
+    assert o not in db.session
+    assert AtomicModel._lock_instance(o) is None
+    db.session.add(o)
+    assert AtomicModel._lock_instance(o) is o
+    assert o in db.session
+    pk = o.id
+    db.session.commit()
+    assert AtomicModel._lock_instance(pk) in db.session
+    assert AtomicModel._lock_instance((pk,)) in db.session
+    assert AtomicModel._lock_instance(o) in db.session
+
+
+def test_get_pk_values(atomic_db, AtomicModel):
+    o = AtomicModel(id=1, name='test', value='1')
+    assert AtomicModel._get_pk_values(o) == (o.id,)
+    assert AtomicModel._get_pk_values(1) == (1,)
+    assert AtomicModel._get_pk_values((1,)) == (1,)
+
+
 def test_create_sharding_key(ShardingKey):
     assert ShardingKey().sharding_key_value
     assert hasattr(ShardingKey, 'generate')