Getting testcontainers working with marshmallow

Question

I am trying to get the python testcontainer working with marshmallow to run a test but I don't seem to be able to get it working. I always get the connection refused when I run the test. I have created a sqlalchemy engine and tested it, and this works fine but when passing the testcontainer connection string as a marshmallow config, it just doesn't work. Below is my base test class.

class BaseTestCase(TestCase):
    base_url = '/api/v1'

    def create_app(self):
        config = MySqlContainer('mysql:8.0.19')
        with config as mysql:
            print(mysql.get_connection_url())
            e = sqlalchemy.create_engine(mysql.get_connection_url())
            result = e.execute("select version()")
            for row in result:
                print("Printing::::::::::::::::::::::" + str(row))
            result.close()
            logging.getLogger('connexion.operation').setLevel('ERROR')
            connex_app = connexion.App(__name__, specification_dir='../../api/')
            connex_app.app.json_encoder = JSONEncoder
            connex_app.add_api('static/openapi.yaml')
            app = connex_app.app
            app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
            app.config['SQLALCHEMY_DATABASE_URI'] = mysql.get_connection_url()
            app.config['LOG_LEVEL'] = 'DEBUG'

            bcrypt.init_app(app)
            db.init_app(app)
            ma.init_app(app)

            print("Finished setting up")

            return app

I can get this to work when I use sqlite as the connection string.

class BaseTestCase(TestCase):
    base_url = '/api/v1'

    def create_app(self):
        class Config:
            PORT = 5000
            SQLALCHEMY_TRACK_MODIFICATIONS = False
            FLASK_ENV = 'local'
            SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
            LOG_LEVEL = 'DEBUG'

        logging.getLogger('connexion.operation').setLevel('ERROR')
        connex_app = connexion.App(__name__, specification_dir='../../api/')
        connex_app.app.json_encoder = JSONEncoder
        connex_app.add_api('static/openapi.yaml')
        app = connex_app.app
        app.config.from_object(Config)
        bcrypt.init_app(app)
        db.init_app(app)
        ma.init_app(app)

        print("Finished setting up test")

        return app

Any help will be much appreciated.

Paul

Answer 1

I decided to use testcontainers.compose and the connection string seems to work.

class BaseTestCase(TestCase):
    base_url = '/api/v1'

    def create_app(self):
        self.compose = testcontainers.compose.DockerCompose(".")
        self.compose.start()
        time.sleep(10)

        class Config:
            PORT = 3306
            SQLALCHEMY_TRACK_MODIFICATIONS = False
            FLASK_ENV = 'local'
            SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://root:password@0.0.0.0:3306/test-db'
            LOG_LEVEL = 'DEBUG'

        logging.getLogger('connexion.operation').setLevel('ERROR')
        connex_app = connexion.App(__name__, specification_dir='../../api/')
        connex_app.app.json_encoder = JSONEncoder
        connex_app.add_api('static/openapi.yaml')
        app = connex_app.app
        app.config.from_object(Config)
        bcrypt.init_app(app)
        db.init_app(app)
        ma.init_app(app)

        print("Finished setting up test")

        return app