Merge pull request #727 from alphagov/exc-handl

remove 500/unplanned exception handlers in test
This commit is contained in:
Leo Hemsted
2016-11-10 13:47:13 +00:00
committed by GitHub
2 changed files with 40 additions and 31 deletions
@@ -698,14 +698,15 @@ def test_should_delete_notification_and_return_error_if_sqs_fails(
save_model_api_key(api_key) save_model_api_key(api_key)
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id)) auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
response = client.post( with pytest.raises(Exception) as exc:
path='/notifications/{}'.format(template_type), response = client.post(
data=json.dumps(data), path='/notifications/{}'.format(template_type),
headers=[('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(auth_header))]) data=json.dumps(data),
headers=[('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(auth_header))])
mocked.assert_called_once_with([fake_uuid], queue='send-{}'.format(template_type)) mocked.assert_called_once_with([fake_uuid], queue='send-{}'.format(template_type))
assert str(exc.value) == 'failed to talk to SQS'
assert response.status_code == 500
assert not notifications_dao.get_notification_by_id(fake_uuid) assert not notifications_dao.get_notification_by_id(fake_uuid)
assert not NotificationHistory.query.get(fake_uuid) assert not NotificationHistory.query.get(fake_uuid)
+34 -26
View File
@@ -2,7 +2,6 @@ from contextlib import contextmanager
import os import os
import boto3 import boto3
from unittest import mock
import pytest import pytest
from alembic.command import upgrade from alembic.command import upgrade
from alembic.config import Config from alembic.config import Config
@@ -13,16 +12,31 @@ from app import create_app, db
@pytest.fixture(scope='session') @pytest.fixture(scope='session')
def notify_api(request): def notify_api():
app = create_app() 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 = app.app_context()
ctx.push() ctx.push()
def teardown(): yield app
ctx.pop()
request.addfinalizer(teardown) ctx.pop()
return app
@pytest.fixture(scope='function') @pytest.fixture(scope='function')
@@ -32,7 +46,7 @@ def client(notify_api):
@pytest.fixture(scope='session') @pytest.fixture(scope='session')
def notify_db(notify_api, request): def notify_db(notify_api):
Migrate(notify_api, db) Migrate(notify_api, db)
Manager(db, MigrateCommand) Manager(db, MigrateCommand)
BASE_DIR = os.path.dirname(os.path.dirname(__file__)) BASE_DIR = os.path.dirname(os.path.dirname(__file__))
@@ -43,36 +57,30 @@ def notify_db(notify_api, request):
with notify_api.app_context(): with notify_api.app_context():
upgrade(config, 'head') upgrade(config, 'head')
def teardown(): yield db
db.session.remove()
db.get_engine(notify_api).dispose()
request.addfinalizer(teardown) db.session.remove()
return db db.get_engine(notify_api).dispose()
@pytest.fixture(scope='function') @pytest.fixture(scope='function')
def notify_db_session(request, notify_db): def notify_db_session(notify_db):
def teardown(): 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"]:
notify_db.engine.execute(tbl.delete())
notify_db.session.commit()
request.addfinalizer(teardown) 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"]:
notify_db.engine.execute(tbl.delete())
notify_db.session.commit()
@pytest.fixture(scope='function') @pytest.fixture(scope='function')
def os_environ(request): def os_environ(mocker):
env_patch = mock.patch('os.environ', {}) mocker.patch('os.environ', {})
request.addfinalizer(env_patch.stop)
return env_patch.start()
@pytest.fixture(scope='function') @pytest.fixture(scope='function')
def sqs_client_conn(request): def sqs_client_conn():
boto3.setup_default_session(region_name='eu-west-1') boto3.setup_default_session(region_name='eu-west-1')
return boto3.resource('sqs') return boto3.resource('sqs')