diff --git a/tests/conftest.py b/tests/conftest.py index 37b88a62f..ad3ceebec 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,7 +2,6 @@ import os from contextlib import contextmanager import pytest -import sqlalchemy from alembic.command import upgrade from alembic.config import Config from flask import Flask @@ -12,35 +11,14 @@ from app.dao.provider_details_dao import get_provider_details_by_identifier @pytest.fixture(scope='session') -def notify_app(worker_id): +def notify_app(): app = Flask('test') - - # Override the SQLALCHEMY_DATABASE_URI config before the app is - # initialized to account for Flask-SQLAlchemy 3.0.x changes. - # What is ultimately happening is the create_engine call made with - # SQLAlchemy itself is now only happening at the time of calling - # init_app with Flask instead of at the time it is first accessed, - # which the _notify_db fixture method was relying on. - - # See the following for more information: - # https://github.com/pallets-eco/flask-sqlalchemy/pull/1087 - # https://flask-sqlalchemy.palletsprojects.com/en/3.0.x/api/#module-flask_sqlalchemy - app.config['SQLALCHEMY_DATABASE_URI'] = '{}_{}'.format( - os.getenv('SQLALCHEMY_DATABASE_TEST_URI', '').replace('postgres://', 'postgresql://'), - worker_id - ) - create_app(app) return app @pytest.fixture(scope='session') def notify_api(notify_app): - # deattach server-error error handlers - error_handler_spec looks like: - # {'blueprint_name': { - # status_code: [error_handlers], - # None: { ExceptionClass: error_handler } - # }} for error_handlers in notify_app.error_handler_spec.values(): error_handlers.pop(500, None) if None in error_handlers: @@ -66,63 +44,33 @@ def client(notify_api): yield client -def create_test_db(database_uri): - # get the - db_uri_parts = database_uri.split('/') - postgres_db_uri = '/'.join(db_uri_parts[:-1] + ['postgres']) - - postgres_db = sqlalchemy.create_engine( - postgres_db_uri, - echo=False, - isolation_level='AUTOCOMMIT', - client_encoding='utf8' - ) - try: - result = postgres_db.execute(sqlalchemy.sql.text('CREATE DATABASE {}'.format(db_uri_parts[-1]))) - result.close() - except sqlalchemy.exc.ProgrammingError: - # database "test_notification_api_master" already exists - pass - finally: - postgres_db.dispose() - - @pytest.fixture(scope='session') -def _notify_db(notify_api, worker_id): +def _notify_db(notify_api): """ Manages the connection to the database. Generally this shouldn't be used, instead you should use the `notify_db_session` fixture which also cleans up any data you've got left over after your test run. """ - - # Create a database for this worker thread; note that we still have - # to reset it here to point to the correct database. - from flask import current_app - current_app.config['SQLALCHEMY_DATABASE_URI'] += '_{}'.format( - worker_id - ) - create_test_db(current_app.config['SQLALCHEMY_DATABASE_URI']) - - BASE_DIR = os.path.dirname(os.path.dirname(__file__)) - ALEMBIC_CONFIG = os.path.join(BASE_DIR, 'migrations') - config = Config(ALEMBIC_CONFIG + '/alembic.ini') - config.set_main_option("script_location", ALEMBIC_CONFIG) - config.set_main_option( - "sqlalchemy.url", - current_app.config['SQLALCHEMY_DATABASE_URI'] - ) - with notify_api.app_context() as app_context: - # Run DB migrations. + db = app_context.app.extensions['sqlalchemy'] + assert 'test_notification_api' in db.engine.url.database, 'dont run tests against main db' + + BASE_DIR = os.path.dirname(os.path.dirname(__file__)) + ALEMBIC_CONFIG = os.path.join(BASE_DIR, 'migrations') + config = Config(ALEMBIC_CONFIG + '/alembic.ini') + config.set_main_option('script_location', ALEMBIC_CONFIG) + config.set_main_option( + 'sqlalchemy.url', + app_context.app.config['SQLALCHEMY_DATABASE_URI'] + ) + + # Run migrations on the test database. upgrade(config, 'head') - # Modify the database connection URL to point to the correct - # test database. - db = app_context.app.extensions['sqlalchemy'] - db.engine.url = current_app.config['SQLALCHEMY_DATABASE_URI'] - - # Return the DB object to the test calling for it. yield db + db.session.remove() + db.engine.dispose() + @pytest.fixture(scope='function') def sms_providers(_notify_db):