mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-15 17:52:26 -05:00
e9482c7fe1
* single-column static data table that currently contains two types: 'normal' and 'team' * key_type foreign-keyed from api_keys - must be not null - existing rows set to 'normal' * key_type foreign-keyed from notifications - nullable - existing rows set to null * api_key foreign-keyed from notifications - nullable - existing rows set to null
72 lines
1.8 KiB
Python
72 lines
1.8 KiB
Python
import os
|
|
|
|
import boto3
|
|
import mock
|
|
import pytest
|
|
from alembic.command import upgrade
|
|
from alembic.config import Config
|
|
from flask.ext.migrate import Migrate, MigrateCommand
|
|
from flask.ext.script import Manager
|
|
from sqlalchemy.schema import MetaData
|
|
|
|
from app import create_app, db
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def notify_api(request):
|
|
app = create_app()
|
|
ctx = app.app_context()
|
|
ctx.push()
|
|
|
|
def teardown():
|
|
ctx.pop()
|
|
|
|
request.addfinalizer(teardown)
|
|
return app
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def notify_db(notify_api, request):
|
|
Migrate(notify_api, db)
|
|
Manager(db, MigrateCommand)
|
|
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)
|
|
|
|
with notify_api.app_context():
|
|
upgrade(config, 'head')
|
|
|
|
def teardown():
|
|
db.session.remove()
|
|
db.get_engine(notify_api).dispose()
|
|
|
|
request.addfinalizer(teardown)
|
|
return db
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def notify_db_session(request, notify_db):
|
|
def teardown():
|
|
notify_db.session.remove()
|
|
for tbl in reversed(notify_db.metadata.sorted_tables):
|
|
if tbl.name not in ["provider_details", "key_types"]:
|
|
notify_db.engine.execute(tbl.delete())
|
|
notify_db.session.commit()
|
|
|
|
request.addfinalizer(teardown)
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def os_environ(request):
|
|
env_patch = mock.patch('os.environ', {})
|
|
request.addfinalizer(env_patch.stop)
|
|
|
|
return env_patch.start()
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def sqs_client_conn(request):
|
|
boto3.setup_default_session(region_name='eu-west-1')
|
|
return boto3.resource('sqs')
|