isolation_level_test.py

# /// script
# requires-python = ">=3.13"
# dependencies = [
#     "mysql-connector-python",
# ]
# ///

import mysql.connector

# Database configuration
dbconfig = {
    "database": "test",
    "user": "root",
    "password": "",
    "host": "127.0.0.1",
    "port": 4000,
}

# Table and data setup
TABLE_NAME = "isolation_test"
TEST_ROW_ID = 1
TEST_INITIAL_VALUE = "Initial Value"
TEST_UPDATED_VALUE = "Updated Value"


def setup_database():
    """Ensure the table exists and insert a test row."""
    conn = None
    try:
        conn = mysql.connector.connect(**dbconfig)
        cursor = conn.cursor()

        # Create table
        cursor.execute(f"DROP TABLE IF EXISTS {TABLE_NAME};")
        cursor.execute(f"""
            CREATE TABLE {TABLE_NAME} (
                id INT PRIMARY KEY,
                data VARCHAR(255)
            );
        """)

        # Insert a test row
        cursor.execute(f"INSERT INTO {TABLE_NAME} (id, data) VALUES (%s, %s);", (TEST_ROW_ID, TEST_INITIAL_VALUE))
        conn.commit()
        print("Table setup completed.")

    except Exception as ex:
        print(f"Error in setup: {ex}")
    finally:
        if conn is not None and conn.is_connected():
            conn.close()


def run_test(isolation_level):
    """Run the isolation level test."""
    conn1 = conn2 = conn3 = None
    try:
        print(f"\nTesting with isolation level: {isolation_level}")
        
        # Start connection 1
        conn1 = mysql.connector.connect(**dbconfig)
        conn1.start_transaction(isolation_level=isolation_level)
        cursor1 = conn1.cursor()

        # Start connection 2
        conn2 = mysql.connector.connect(**dbconfig)
        conn2.start_transaction(isolation_level=isolation_level)
        cursor2 = conn2.cursor()

        # Update row in transaction 1
        cursor1.execute(f"UPDATE {TABLE_NAME} SET data = %s WHERE id = %s;", (TEST_UPDATED_VALUE, TEST_ROW_ID))
        conn1.commit()  # Commit the update
        print("Transaction 1: Updated and committed the row.")

        # Transaction 2 selects the row
        # conn2.commit()
        cursor2.execute(f"SELECT data FROM {TABLE_NAME} WHERE id = %s;", (TEST_ROW_ID,))
        result2 = cursor2.fetchone()
        print(f"Transaction 2: Row data = {result2[0]}")

        # Report visibility in Transaction 2
        if result2[0] == TEST_UPDATED_VALUE:
            print("Transaction 2: Can see the committed update.")
        else:
            print("Transaction 2: Cannot see the committed update.")

        # Start connection 3
        conn3 = mysql.connector.connect(**dbconfig)
        cursor3 = conn3.cursor()

        # Transaction 3 selects the row after commit
        cursor3.execute(f"SELECT data FROM {TABLE_NAME} WHERE id = %s;", (TEST_ROW_ID,))
        result3 = cursor3.fetchone()
        print(f"Transaction 3: Row data = {result3[0]}")

    except Exception as ex:
        print(f"Error during test: {ex}")
    finally:
        # Cleanup connections
        for conn in (conn1, conn2, conn3):
            if conn is not None and conn.is_connected():
                conn.close()


if __name__ == "__main__":
    setup_database()
    for isolation in ["REPEATABLE READ", "READ COMMITTED"]:
        run_test(isolation)