I write tests for CRUD application and I want to be sure that its behavior is correct in different scenarios when connection to database is lost permanently or temporarily.
For now, I patch session.execute in the middle of the session with OperationalError side_effect. Is there a better or more realistic simulation of connection interruption?
What I want to achieve looks like this:
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).You can invalidate the SQL connection 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 database failure, the best way to kill all your database connections. This usually happens when you scale your rds (there is a little downtime) and you will get No active database found error and then you can check your application behavior.