Simplify test config for now

This is in prep for switching to running a single test worker for the time being.  We will spend time figuring out how to get multiple concurrent workers going again separately as we seem to have several issues taking place right now with that type of test setup.

Signed-off-by: Carlo Costino <carlo.costino@gsa.gov>
This commit is contained in:
Carlo Costino
2023-08-15 18:03:49 -04:00
parent 17e13c3b3b
commit 206d5fccb7

View File

@@ -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):