What is the best way to mock disconnect within SQLAlchemy

Question

I write tests for CRUD application and i want to be sure that its behaviour is correct during different scenarios when connection to DB permanently or temporarily lost.

For now I patch session.execute in the middle of session with OperationalError side_effect. Is there better or\and more realistic mocking of connection interruption?

What I want to achive looks like that:

    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). 

Answer 1

You don't need to change anything, your solution is already in accordance with the best practice.

Answer 2

You can Invalidate the SQL connection by using

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

But if you want to know how your application will behave in case of DB failure, The best way to Kill all your DB connections. This generally happens when you scale your rds(there is a small downtime) and you will get the error of No Active database found and then you can check the behavior of your application.