Simulating connection errors with SQLAlchemy

user_209184 0
Viewed 112

On production, my Flask / SQLAlchemy app randomly throws psycopg2.OperationalError: server closed the connection unexpectedly on RDS Aurora. Until I track down the issue, I would like my unit tests to ensure that I handle this properly, e.g. that my rollback mechanism is effective, etc.

Right now, I mock the Session.commit method to throw a fake exception, but this does not (afaik?) leave the session in a failed state that needs actual rollback.

What are some reliable ways to simulate an actual connection failure in my local, Docker Compose-based development environment?

database-connection sqlalchemy unit-testing
1 Answers

Rollback with psycopg2

conn = psycopg2.connect(
    f"host='{credentials['host']}' port={credentials['port']} dbname='{credentials['dbname']}' user='{credentials['user']}' password='{credentials['password']}'"
)
print("Connected to database")
print("Setting autocommit false")
conn.autocommit = False

func_name= 'procedure_update_users'

print(f"Executing function {func_name}...")
try:
    cur = conn.cursor()
    query = f"""SELECT {func_name}();"""
    print(query)
    cur.execute(query)
    row = cur.fetchone()
    conn.commit()
    print(f"Function {func_name} executed.")
except Exception as e:
    print(e)
    conn.rollback()
    raise Exception(e)

To get OperationalError I did:

try:
    cur.execute("LOCK TABLE mytable IN ACCESS EXCLUSIVE MODE NOWAIT")
except psycopg2.OperationalError as e:
    cur.execute("rollback")
    raise Exception(e)

psycopg2.OperationalError: fe_sendauth: no password supplied

user_53984900
Related