How to test DB connection failures mid execution

[andre-qumulo]

I am upgrading from flask-sqlalchemy 2.5.1 to 3.1.1 and have a test case that used to work but now isn't.

The test case looks like:

self.app.config['SQLALCHEMY_DATABASE_URI'] = 'not a path'
with assertLogs(self, level=logging.DEBUG) as cm:
    resp = self.client.get('/status/')
self.assertEqual(resp.status_code, 500)

The route on the app looks like:

        with db.engine.connect() as conn:
            log.info('Start fetching migrations current revision')
            context = MigrationContext.configure(conn)
            current = context.get_current_revision()
            db_resp = {'status': 'Pass', 'error_msg': f'current migration {current}'}
            log.info('End fetching migrations current revision')

            log.info('Start DB query for SomeModel')
            SomeModel.query.first()
            log.info('End DB query for SomeModel')

What I'm wondering is if people think this is a change with flask-sqalchemy in how it handles updating the config (maybe SqlAlchemy used to reference the config everytime when doing a db.engine.connect() and now caches the values somehow or if there's something else I'm missing.

I recognize that changing the DB URL mid-execution of a flask app isn't something we'd normally do but we're not sure how else to simulate a DB failure outside of putting test paths in production code (which we'd like to avoid).

[andre-qumulo]

I ended up changing my testing strategy. Our code creates a postgresql server in python so we passed a reference up to the test and stopped the server.

class PostgresClient:
    port: int
    server: PostgresServer

    def __init__(self, port: int, server: PostgresServer) -> None:
        self.port = port
        self.server = server

    def get_url(self) -> str:
        return f'postgresql://localhost:{self.port}/{DEFAULT_DB}'

    def connect(self, **kwargs: Any) -> psycopg2.extensions.connection:
        return psycopg2.connect(host='localhost', port=self.port, **kwargs)

    def create_db(self, dbname: str) -> None:
        conn = self.connect(dbname=DEFAULT_DB)
        conn.autocommit = True
        conn.cursor().execute(f'CREATE DATABASE {dbname};')
        conn.close()

    def stop_server(self) -> None:
        self.server.stop()

    def start_server(self) -> None:
        self.server.start(redirect_to_dev_null=False, port=self.port)


@contextlib.contextmanager
def postgres_server(
    fs: posix_fs.FileSystem,
    port: Optional[int] = None,
    data_dir: Optional[str] = None,
    init_clean: bool = True,
) -> Iterator[PostgresClient]:
    """
    Context manager for a postgres server instance.
    @param: fs -- FileSystem where database files should be stored
    @param: port -- Optional explicit port to expose database access
    @param: data_dir -- Optional existing directory where files should be stored
    @param: init_clean -- When True, removes database files and runs pg_ctl init
    """

    actual_data_dir = data_dir or fs.mkdtemp()

    server = PostgresServer(fs.realpath(actual_data_dir))
    if init_clean:
        fs.clean_dir(actual_data_dir, ignore_missing=True)
        server.init_clean(redirect_to_dev_null=False)
    port = server.start(redirect_to_dev_null=False, port=port)

    try:
        yield PostgresClient(port=port, server=server)
    finally:
        server.stop()
        if not data_dir:
            fs.rmtree(actual_data_dir, ignore_missing=True)

and then our test becomes:

        self.service.postgres_client.stop_server()
        with assertLogs(self, level=logging.DEBUG):
            resp = self.client.get('/status/')
        self.assertEqual(resp.status_code, 500)
        self.assertEqual(resp.json['database']['status'], 'Fail')
        self.assertIn(
            f'connection to server at "localhost" (127.0.0.1), port {self.service.postgres_client.port} failed',
            resp.json['database']['error_msg'],
        )