使用SQLAlchemy模拟连接错误

Question

在生产中，我的Flask/SQLAlchemy应用程序在RDS Aurora上随机抛出psycopg2.OperationalError: server closed the connection unexpectedly。在我找到问题之前，我希望我的单元测试能确保我正确处理这个问题，例如我的回滚机制是有效的，等等。

现在，我模拟Session.commit方法抛出假异常，但这不会(afaik？(使会话处于需要实际回滚的失败状态。

在我的本地基于Docker Compose的开发环境中，有哪些可靠的方法可以模拟实际的连接故障？

Answer 1

使用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)

为了得到操作错误，我做了：

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

psycopg2.OperationalError:fesendauth:未提供密码