Add manual .py test verifying isolation levels (#407)
This commit is contained in:
Tyler Neely
GitHub
parent
259cba5d43
commit
53fcd8ac4d
607
tests/manual/test_isolation_level.py
Executable file
607
tests/manual/test_isolation_level.py
Executable file
@ -0,0 +1,607 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""
|
||||
This script verifies whether the server is able to correctly
|
||||
detect and abort certain classes of transactional anomaly,
|
||||
inspired by the hermitage tests found at:
|
||||
https://github.com/ept/hermitage
|
||||
and based on the correctness definitions laid out in:
|
||||
https://pmg.csail.mit.edu/papers/adya-phd.pdf
|
||||
|
||||
As of the time of committing this script, the database
|
||||
is able to meet the conditions required for Snapshot Isolation
|
||||
and Consistent View.
|
||||
|
||||
All of these tests begin with a dataset of:
|
||||
Kv { key: 1, value: 10 }
|
||||
Kv { key: 2, value: 20 }
|
||||
|
||||
and then assert various transactional correctness properties.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import mgclient
|
||||
|
||||
|
||||
def setup(conn):
|
||||
conn.autocommit = True
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("CREATE CONSTRAINT ON (kv: Kv) ASSERT kv.key IS UNIQUE;")
|
||||
conn.commit()
|
||||
conn.autocommit = False
|
||||
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("MATCH (kv:Kv) DETACH DELETE kv;")
|
||||
conn.commit()
|
||||
|
||||
update(cursor, 1, 10, "instantiate key 1")
|
||||
conn.commit()
|
||||
update(cursor, 2, 20, "instantiate key 2")
|
||||
conn.commit()
|
||||
|
||||
assert_eq(get(cursor, 1), [10])
|
||||
assert_eq(get(cursor, 2), [20])
|
||||
|
||||
|
||||
def update(cursor, key, value, comment=""):
|
||||
cursor.execute(
|
||||
"""
|
||||
// %s
|
||||
MERGE (kv:Kv{ key: $key })
|
||||
SET kv.value = $value
|
||||
RETURN kv;
|
||||
"""
|
||||
% comment,
|
||||
{"key": key, "value": value},
|
||||
)
|
||||
ret = cursor.fetchall()
|
||||
if len(ret) != 1:
|
||||
print(f"expected ret to be len 1, but it's {ret}")
|
||||
assert len(ret) == 1
|
||||
return ret[0][0].properties["value"]
|
||||
|
||||
|
||||
def get(cursor, key):
|
||||
cursor.execute(
|
||||
"""
|
||||
MATCH (kv: Kv { key: $key })
|
||||
RETURN kv;
|
||||
""",
|
||||
{"key": key},
|
||||
)
|
||||
ret = cursor.fetchall()
|
||||
return [r[0].properties["value"] for r in ret]
|
||||
|
||||
|
||||
def assert_eq(a, b):
|
||||
if a != b:
|
||||
print("expected", a, "to be", b)
|
||||
assert a == b
|
||||
|
||||
|
||||
def assert_commit_fails(conn, failure="conflicting transactions"):
|
||||
try:
|
||||
conn.commit()
|
||||
# should always abort
|
||||
print("expected transaction to fail with", failure, "but it incorrectly committed successfully")
|
||||
assert False
|
||||
except mgclient.DatabaseError as e:
|
||||
assert failure in str(e), "expected exception containing text {failure} but instead it was {e}"
|
||||
|
||||
|
||||
def conflicting_update(cursor, key, value):
|
||||
try:
|
||||
update(cursor, key, value)
|
||||
assert False
|
||||
except mgclient.DatabaseError as e:
|
||||
assert str(e).startswith("Cannot resolve conflicting transactions")
|
||||
|
||||
|
||||
def select(cursor, key, expected):
|
||||
actual = get(cursor, key)
|
||||
if actual != expected:
|
||||
print("expected key", key, "to have value", expected, "but instead it had value", actual)
|
||||
assert False
|
||||
|
||||
|
||||
def select_all(cursor, expected):
|
||||
cursor.execute("MATCH (kv: Kv {}) RETURN kv")
|
||||
actual = [r[0].properties["value"] for r in cursor.fetchall()]
|
||||
assert actual == expected, "expected values {expected}, but instead it had values {actual}"
|
||||
|
||||
|
||||
"""
|
||||
G0: Write Cycles (dirty writes)
|
||||
"""
|
||||
|
||||
|
||||
def g0(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
update(cursor_1, 1, 11)
|
||||
conflicting_update(cursor_2, 1, 12)
|
||||
update(cursor_1, 2, 21)
|
||||
c1.commit()
|
||||
|
||||
get(cursor_1, 1)
|
||||
get(cursor_1, 2)
|
||||
|
||||
# if the above write to 1:12 was
|
||||
# able to be staged without throwing
|
||||
# an exception, these lines would be
|
||||
# necessary. but the interactive txn
|
||||
# from c2 is already aborted before
|
||||
# this point, so there's nothing we
|
||||
# need to do here.
|
||||
# update(cursor_2, 2, 22)
|
||||
# assert_commit_fails(c2)
|
||||
|
||||
select(cursor_1, 1, [11])
|
||||
select(cursor_1, 2, [21])
|
||||
|
||||
select(cursor_2, 1, [11])
|
||||
select(cursor_2, 2, [21])
|
||||
|
||||
c1.commit()
|
||||
c2.commit()
|
||||
|
||||
print("✓ g0 test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
G1a: Aborted Reads (dirty reads, cascaded aborts)
|
||||
"""
|
||||
|
||||
|
||||
def g1a(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
update(cursor_1, 1, 101)
|
||||
select(cursor_2, 1, [10])
|
||||
c1.rollback()
|
||||
|
||||
select(cursor_2, 1, [10])
|
||||
c2.commit()
|
||||
|
||||
print("✓ g1a test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
G1b: Intermediate Reads (dirty reads)
|
||||
"""
|
||||
|
||||
|
||||
def g1b(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
update(cursor_1, 1, 101)
|
||||
select(cursor_2, 1, [10])
|
||||
update(cursor_1, 1, 11)
|
||||
c1.commit()
|
||||
|
||||
select(cursor_2, 1, [10])
|
||||
c2.commit()
|
||||
|
||||
print("✓ g1b test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
G1c: Circular Information Flow (dirty reads)
|
||||
"""
|
||||
|
||||
|
||||
def g1c(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
update(cursor_1, 1, 11)
|
||||
update(cursor_2, 2, 22)
|
||||
|
||||
select(cursor_1, 2, [20])
|
||||
select(cursor_2, 1, [10])
|
||||
|
||||
c1.commit()
|
||||
c2.commit()
|
||||
|
||||
print("✓ g1c test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
OTV: Observed Transaction Vanishes
|
||||
"""
|
||||
|
||||
|
||||
def otv(c1, c2, c3):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
cursor_3 = c3.cursor()
|
||||
|
||||
update(cursor_1, 1, 11)
|
||||
update(cursor_1, 2, 19)
|
||||
conflicting_update(cursor_2, 1, 12)
|
||||
|
||||
c1.commit()
|
||||
|
||||
select(cursor_3, 1, [11])
|
||||
select(cursor_3, 2, [19])
|
||||
|
||||
# cursor_2 update not required due to its early-abort above
|
||||
|
||||
select(cursor_3, 1, [11])
|
||||
select(cursor_3, 2, [19])
|
||||
|
||||
c3.commit()
|
||||
|
||||
print("✓ otv test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
Predicate-Many-Preceders
|
||||
"""
|
||||
|
||||
|
||||
def pmp(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
select(cursor_1, 3, [])
|
||||
update(cursor_2, 3, 30)
|
||||
c2.commit()
|
||||
|
||||
select(cursor_1, 3, [])
|
||||
c1.commit()
|
||||
|
||||
print("✓ pmp test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
Predicate-Many-Preceders (write)
|
||||
"""
|
||||
|
||||
|
||||
def pmp_write(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
cursor_1.execute("MATCH (kv: Kv {}) SET kv.value = kv.value + 10 RETURN kv")
|
||||
|
||||
select(cursor_2, 1, [10])
|
||||
select(cursor_2, 2, [20])
|
||||
|
||||
try:
|
||||
cursor_2.execute("MATCH (kv:Kv { value: 20 }) DETACH DELETE kv;")
|
||||
assert False
|
||||
except mgclient.DatabaseError as e:
|
||||
assert "conflicting transactions" in str(e)
|
||||
|
||||
c1.commit()
|
||||
|
||||
select(cursor_1, 1, [20])
|
||||
select(cursor_1, 2, [30])
|
||||
|
||||
c1.commit()
|
||||
|
||||
print("✓ pmp_write test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
P4: Lost Update
|
||||
"""
|
||||
|
||||
|
||||
def p4(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
select(cursor_1, 1, [10])
|
||||
select(cursor_2, 1, [10])
|
||||
|
||||
update(cursor_1, 1, 11)
|
||||
conflicting_update(cursor_2, 1, 11)
|
||||
|
||||
c1.commit()
|
||||
|
||||
print("✓ p4 test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
G-single: Single Anti-dependency Cycles (read skew)
|
||||
"""
|
||||
|
||||
|
||||
def g_single(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
select(cursor_1, 1, [10])
|
||||
select(cursor_2, 1, [10])
|
||||
select(cursor_2, 2, [20])
|
||||
|
||||
update(cursor_2, 1, 12)
|
||||
update(cursor_2, 2, 18)
|
||||
c2.commit()
|
||||
|
||||
select(cursor_1, 2, [20])
|
||||
c1.commit()
|
||||
|
||||
print("✓ g_single test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
G-single: Single Anti-dependency Cycles (read skew) (dependencies)
|
||||
"""
|
||||
|
||||
|
||||
def g_single_dependencies(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
cursor_1.execute("MATCH (kv: Kv {}) WHERE kv.value % 5 = 0 RETURN kv")
|
||||
found = [r[0].properties["value"] for r in cursor_1.fetchall()]
|
||||
assert found == [10, 20]
|
||||
|
||||
cursor_2.execute("MATCH (kv: Kv { value: 10 }) SET kv.value = 12 RETURN kv")
|
||||
c2.commit()
|
||||
|
||||
cursor_1.execute("MATCH (kv: Kv {}) WHERE kv.value % 3 = 0 RETURN kv")
|
||||
found = [r[0].properties["value"] for r in cursor_1.fetchall()]
|
||||
assert found == []
|
||||
|
||||
c1.commit()
|
||||
|
||||
print("✓ g_single_dependencies test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
G-single: Single Anti-dependency Cycles (read skew) (write 1)
|
||||
"""
|
||||
|
||||
|
||||
def g_single_write_1(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
select(cursor_1, 1, [10])
|
||||
|
||||
select_all(cursor_2, [10, 20])
|
||||
|
||||
update(cursor_2, 1, 12)
|
||||
update(cursor_2, 2, 18)
|
||||
c2.commit()
|
||||
|
||||
try:
|
||||
cursor_1.execute("MATCH (kv:Kv { value: 20 }) DETACH DELETE kv;")
|
||||
assert False
|
||||
except mgclient.DatabaseError as e:
|
||||
assert "conflicting transactions" in str(e)
|
||||
|
||||
select_all(cursor_2, [12, 18])
|
||||
|
||||
c2.commit()
|
||||
|
||||
print("✓ g_single_write_1 test passed")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
G-single: Single Anti-dependency Cycles (read skew) (write 2)
|
||||
"""
|
||||
|
||||
|
||||
def g_single_write_2(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
select(cursor_1, 1, [10])
|
||||
select_all(cursor_2, [10, 20])
|
||||
|
||||
update(cursor_2, 1, 12)
|
||||
|
||||
cursor_1.execute("MATCH (kv:Kv { value: 20 }) DETACH DELETE kv;")
|
||||
|
||||
conflicting_update(cursor_2, 2, 18)
|
||||
|
||||
c1.rollback()
|
||||
# c2.commit()
|
||||
|
||||
print("✓ g_single_write_2 test passed (although the abort rate is pessimistically high)")
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
G2-item: Item Anti-dependency Cycles (write skew on disjoint read)
|
||||
"""
|
||||
|
||||
|
||||
def g2_item(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
cursor_1.execute("MATCH (kv: Kv {}) WHERE kv.key = 1 OR kv.key = 2 RETURN kv")
|
||||
found = [r[0].properties["value"] for r in cursor_1.fetchall()]
|
||||
assert found == [10, 20]
|
||||
|
||||
cursor_2.execute("MATCH (kv: Kv {}) WHERE kv.key = 1 OR kv.key = 2 RETURN kv")
|
||||
found = [r[0].properties["value"] for r in cursor_2.fetchall()]
|
||||
assert found == [10, 20]
|
||||
|
||||
update(cursor_1, 1, 11)
|
||||
update(cursor_2, 2, 21)
|
||||
|
||||
c1.commit()
|
||||
|
||||
try:
|
||||
assert_commit_fails(c2)
|
||||
print("✓ g2_item test passed")
|
||||
return True
|
||||
except:
|
||||
print(
|
||||
"X g2_item test failed - database exhibits write skew -",
|
||||
"writes based on invalidated reads should have failed,",
|
||||
"causing repeatable read (PL-2.99) and serializability (PL-3)",
|
||||
"to fail to be achieved",
|
||||
)
|
||||
return False
|
||||
|
||||
|
||||
"""
|
||||
G2: Anti-Dependency Cycles (write skew on predicate read)
|
||||
"""
|
||||
|
||||
|
||||
def g2(c1, c2):
|
||||
cursor_1 = c1.cursor()
|
||||
cursor_2 = c2.cursor()
|
||||
|
||||
cursor_1.execute("MATCH (kv: Kv {}) WHERE kv.value % 3 = 0 RETURN kv")
|
||||
found = [r[0].properties["value"] for r in cursor_1.fetchall()]
|
||||
assert found == []
|
||||
|
||||
cursor_2.execute("MATCH (kv: Kv {}) WHERE kv.value % 3 = 0 RETURN kv")
|
||||
found = [r[0].properties["value"] for r in cursor_2.fetchall()]
|
||||
assert found == []
|
||||
|
||||
update(cursor_1, 3, 30)
|
||||
update(cursor_2, 4, 42)
|
||||
|
||||
c1.commit()
|
||||
|
||||
try:
|
||||
assert_commit_fails(c2)
|
||||
print("✓ g2 test passed")
|
||||
return True
|
||||
except:
|
||||
print(
|
||||
"X g2 test failed - database exhibits write skew on predicate read -",
|
||||
"concurrent transactions that should have caused a predicate read to",
|
||||
"return data in one transaction actually returned nothing in both.",
|
||||
"Both transactions committed, but one of them should have failed due",
|
||||
"to having a predicate invalidated",
|
||||
)
|
||||
return False
|
||||
|
||||
|
||||
"""
|
||||
G2: Anti-Dependency Cycles (write skew on predicate read) (two edges)
|
||||
"""
|
||||
|
||||
|
||||
def g2_two_edges(c1, c2, c3):
|
||||
cursor_1 = c1.cursor()
|
||||
select_all(cursor_1, [10, 20])
|
||||
|
||||
cursor_2 = c2.cursor()
|
||||
cursor_2.execute("MATCH (kv: Kv { key: 2 }) SET kv.value = kv.value + 5 RETURN kv;")
|
||||
found = [r[0].properties["value"] for r in cursor_2.fetchall()]
|
||||
assert found == [25]
|
||||
c2.commit()
|
||||
|
||||
cursor_3 = c3.cursor()
|
||||
select_all(cursor_3, [10, 25])
|
||||
c3.commit()
|
||||
|
||||
try:
|
||||
conflicting_update(cursor_1, 1, 0)
|
||||
print("✓ g2_two_edges test passed")
|
||||
return True
|
||||
except:
|
||||
c3.rollback()
|
||||
print(
|
||||
"X g2_two_edges test failed: database exhibits write skew on predicate read -",
|
||||
"a transaction's read set was invalidated in two concurrent transactions,"
|
||||
"and it should have failed to commit if we want to be serializable",
|
||||
)
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("-s", "--host", default="127.0.0.1")
|
||||
parser.add_argument("-p", "--port", default=7687)
|
||||
args = parser.parse_args()
|
||||
|
||||
c1 = mgclient.connect(host=args.host, port=args.port)
|
||||
c1.autocommit = False
|
||||
c2 = mgclient.connect(host=args.host, port=args.port)
|
||||
c2.autocommit = False
|
||||
c3 = mgclient.connect(host=args.host, port=args.port)
|
||||
c3.autocommit = False
|
||||
|
||||
setup(c1)
|
||||
g0 = g0(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
g1a = g1a(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
g1b = g1b(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
g1c = g1c(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
otv = otv(c1, c2, c3)
|
||||
|
||||
setup(c1)
|
||||
pmp = pmp(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
pmp_write = pmp_write(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
p4 = p4(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
g_single = g_single(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
g_single_dependencies = g_single_dependencies(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
g_single_write_1 = g_single_write_1(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
g_single_write_2 = g_single_write_2(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
g2_item = g2_item(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
g2 = g2(c1, c2)
|
||||
|
||||
setup(c1)
|
||||
g2_two_edges = g2_two_edges(c1, c2, c3)
|
||||
|
||||
g1 = all([g1a, g1b, g1c])
|
||||
repeatable_read = all([g1, g2_item])
|
||||
snapshot_isolation = all(
|
||||
[g0, g1, otv, pmp, pmp_write, p4, g_single, g_single_dependencies, g_single_write_1, g_single_write_2]
|
||||
)
|
||||
full_serializability = all([snapshot_isolation, g2_item, g2, g2_two_edges])
|
||||
|
||||
print("")
|
||||
|
||||
print("results:")
|
||||
print("consistent view (PL-2+):", g1 and g_single)
|
||||
print("snapshot isolation (PL-SI):", snapshot_isolation)
|
||||
print("repeatable read (PL-2.99):", repeatable_read)
|
||||
print("full serializability (PL-3):", full_serializability)
|
||||
|
||||
print("")
|
||||
|
||||
print("more information about these anomalies can be found at:")
|
||||
print("\thttps://github.com/ept/hermitage")
|
||||
print("\thttps://pmg.csail.mit.edu/papers/adya-phd.pdf")
|
||||
Loading…
Reference in New Issue
Block a user