在 SQLAlchemy 中模拟断开连接的最佳方法是什么答案

【问题标题】：What is the best way to mock disconnect within SQLAlchemy在 SQLAlchemy 中模拟断开连接的最佳方法是什么
【发布时间】：2022-01-26 14:20:35
【问题描述】：

我为 CRUD 应用程序编写测试，我想确保在永久或暂时丢失与数据库的连接时，它的行为在不同情况下是正确的。

现在我在会话中间使用 OperationalError side_effect 修补 session.execute。有没有更好或更真实的连接中断模拟？

我想要达到的效果是这样的：

    def test_interrupted_connection(self, client):
        query = text("SELECT 1")
        with sessionmaker.begin() as s:
            # Successful execute
            s.execute(query)
            # Failed execute
            with pytest.raises(OperationalError):
                # I guess here should be some monkeypatching or mocking
                # which will make SQLAlchemy think that connection
                # is lost during .execute() call below
                # But I don't know how it should look like.
                s.execute(query).

【问题讨论】：

什么不起作用？
我需要什么或如何模拟“with pytest.raises(...)”和 s.execute(query) 之间的断开连接
展示你的尝试。
也许可以看看 SQLAlchmey 的test suite 中的测试场景

标签： python unit-testing sqlalchemy pytest

【解决方案1】：

您无需更改任何内容，您的解决方案已经符合最佳实践。

【讨论】：

这是作为答案发布的，但它不会尝试回答问题。应该是评论。
问题是（按原样复制）：是否有更好或更真实的连接中断模拟？我对这个确切问题的回答是：不，您的解决方案已经符合最佳实践。所以它确实回答了他的问题。

【解决方案2】：

您可以使用使 SQL 连接无效

#db = create_engine('auth_url')
#conn = db.connect()
conn.invalidate()
db.dispose(

但是，如果您想知道您的应用程序在 DB 发生故障时将如何表现，那么最好的方法就是终止您的所有 DB 连接。这通常在您扩展 rds 时发生（停机时间很小），您会收到 No Active database found 错误，然后您可以检查应用程序的行为。

【讨论】：

在范围内我有 sessionmaker 工厂和 session 本身，我如何将 session 连接作为属性访问？
您必须在 main.py 中启动连接或在 main.py 中将其包裹在应用程序周围。您可以直接从 main.py 导入它
不，我不直接使用连接对象。我使用 sessionmaker 来生成在事务开始时打开连接的会话。