mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-23 17:01:35 -05:00
6c79ddbe41
If the template is marked as priority the notification will be sent using the `notify` queue. The `notify` queue is a low volume queue, messages here will not be queue behind a large job and should be delivered with in a more consistent time frame. - Added templates.process_type and templates_history.process_type column. - Added a template_process_type table to handle the enum for templates.process_type, initial values are normal and priority https://www.pivotaltracker.com/story/show/135429147
109 lines
3.0 KiB
Python
109 lines
3.0 KiB
Python
from contextlib import contextmanager
|
|
import os
|
|
|
|
import boto3
|
|
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 app import create_app, db
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def notify_api():
|
|
app = create_app()
|
|
|
|
# deattach server-error error handlers - error_handler_spec looks like:
|
|
# {'blueprint_name': {
|
|
# status_code: [error_handlers],
|
|
# None: [ tuples of (exception, )]
|
|
# }}
|
|
for error_handlers in app.error_handler_spec.values():
|
|
error_handlers.pop(500, None)
|
|
if None in error_handlers:
|
|
error_handlers[None] = [
|
|
exception_handler
|
|
for exception_handler in error_handlers[None]
|
|
if exception_handler[0] != Exception
|
|
]
|
|
if error_handlers[None] == []:
|
|
error_handlers.pop(None)
|
|
|
|
ctx = app.app_context()
|
|
ctx.push()
|
|
|
|
yield app
|
|
|
|
ctx.pop()
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def client(notify_api):
|
|
with notify_api.test_request_context(), notify_api.test_client() as client:
|
|
yield client
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def notify_db(notify_api):
|
|
assert db.engine.url.database != 'notification_api', 'dont run tests against main db'
|
|
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')
|
|
|
|
yield db
|
|
|
|
db.session.remove()
|
|
db.get_engine(notify_api).dispose()
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def notify_db_session(notify_db):
|
|
yield notify_db
|
|
|
|
notify_db.session.remove()
|
|
for tbl in reversed(notify_db.metadata.sorted_tables):
|
|
if tbl.name not in ["provider_details",
|
|
"key_types",
|
|
"branding_type",
|
|
"job_status",
|
|
"provider_details_history",
|
|
"template_process_type"]:
|
|
notify_db.engine.execute(tbl.delete())
|
|
notify_db.session.commit()
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def os_environ(mocker):
|
|
mocker.patch('os.environ', {})
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def sqs_client_conn():
|
|
boto3.setup_default_session(region_name='eu-west-1')
|
|
return boto3.resource('sqs')
|
|
|
|
|
|
def pytest_generate_tests(metafunc):
|
|
# Copied from https://gist.github.com/pfctdayelise/5719730
|
|
idparametrize = getattr(metafunc.function, 'idparametrize', None)
|
|
if idparametrize:
|
|
argnames, testdata = idparametrize.args
|
|
ids, argvalues = zip(*sorted(testdata.items()))
|
|
metafunc.parametrize(argnames, argvalues, ids=ids)
|
|
|
|
|
|
@contextmanager
|
|
def set_config(app, name, value):
|
|
old_val = app.config.get(name)
|
|
app.config[name] = value
|
|
yield
|
|
app.config[name] = old_val
|