mirror of
https://github.com/GSA/notifications-api.git
synced 2026-02-02 17:31:14 -05:00
add tests for new test-key handling
This commit is contained in:
@@ -150,7 +150,7 @@ def dao_get_all_letter_jobs():
|
||||
).filter(
|
||||
Template.template_type == LETTER_TYPE,
|
||||
# test letter jobs (or from research mode services) are created with a different filename,
|
||||
# exclude them so we don't see them on the send to CSV
|
||||
# exclude them so we don't see them on the send to CSV
|
||||
Job.original_file_name != LETTER_TEST_API_FILENAME
|
||||
).order_by(
|
||||
desc(Job.created_at)
|
||||
|
||||
@@ -836,13 +836,16 @@ def test_get_notification_by_id(notify_db, notify_db_session, sample_template):
|
||||
assert notification_from_db.scheduled_notification.scheduled_for == datetime(2017, 5, 5, 14, 15)
|
||||
|
||||
|
||||
def test_get_notifications_by_reference(notify_db, notify_db_session, sample_service):
|
||||
def test_get_notifications_by_reference(sample_template):
|
||||
client_reference = 'some-client-ref'
|
||||
assert len(Notification.query.all()) == 0
|
||||
sample_notification(notify_db, notify_db_session, client_reference=client_reference)
|
||||
sample_notification(notify_db, notify_db_session, client_reference=client_reference)
|
||||
sample_notification(notify_db, notify_db_session, client_reference='other-ref')
|
||||
all_notifications = get_notifications_for_service(sample_service.id, client_reference=client_reference).items
|
||||
create_notification(sample_template, client_reference=client_reference)
|
||||
create_notification(sample_template, client_reference=client_reference)
|
||||
create_notification(sample_template, client_reference='other-ref')
|
||||
all_notifications = get_notifications_for_service(
|
||||
sample_template.service_id,
|
||||
client_reference=client_reference
|
||||
).items
|
||||
assert len(all_notifications) == 2
|
||||
|
||||
|
||||
@@ -1066,22 +1069,22 @@ def test_should_not_delete_notification_history(notify_db, notify_db_session, sa
|
||||
|
||||
|
||||
@freeze_time("2016-01-10")
|
||||
def test_should_limit_notifications_return_by_day_limit_plus_one(notify_db, notify_db_session, sample_service):
|
||||
def test_should_limit_notifications_return_by_day_limit_plus_one(sample_template):
|
||||
assert len(Notification.query.all()) == 0
|
||||
|
||||
# create one notification a day between 1st and 9th
|
||||
for i in range(1, 11):
|
||||
past_date = '2016-01-{0:02d}'.format(i)
|
||||
with freeze_time(past_date):
|
||||
sample_notification(notify_db, notify_db_session, created_at=datetime.utcnow(), status="failed")
|
||||
create_notification(sample_template, created_at=datetime.utcnow(), status="failed")
|
||||
|
||||
all_notifications = Notification.query.all()
|
||||
assert len(all_notifications) == 10
|
||||
|
||||
all_notifications = get_notifications_for_service(sample_service.id, limit_days=10).items
|
||||
all_notifications = get_notifications_for_service(sample_template.service_id, limit_days=10).items
|
||||
assert len(all_notifications) == 10
|
||||
|
||||
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1).items
|
||||
all_notifications = get_notifications_for_service(sample_template.service_id, limit_days=1).items
|
||||
assert len(all_notifications) == 2
|
||||
|
||||
|
||||
@@ -1302,42 +1305,20 @@ def test_dao_timeout_notifications_doesnt_affect_letters(sample_letter_template)
|
||||
assert updated == 0
|
||||
|
||||
|
||||
def test_should_return_notifications_excluding_jobs_by_default(notify_db, notify_db_session, sample_service):
|
||||
assert len(Notification.query.all()) == 0
|
||||
def test_should_return_notifications_excluding_jobs_by_default(sample_template, sample_job, sample_api_key):
|
||||
with_job = create_notification(sample_template, job=sample_job)
|
||||
without_job = create_notification(sample_template, api_key_id=sample_api_key.id)
|
||||
|
||||
job = sample_job(notify_db, notify_db_session)
|
||||
with_job = sample_notification(
|
||||
notify_db, notify_db_session, created_at=datetime.utcnow(), status="delivered", job=job
|
||||
)
|
||||
without_job = sample_notification(
|
||||
notify_db, notify_db_session, created_at=datetime.utcnow(), status="delivered"
|
||||
)
|
||||
include_jobs = get_notifications_for_service(sample_template.service_id, include_jobs=True).items
|
||||
assert len(include_jobs) == 2
|
||||
|
||||
all_notifications = Notification.query.all()
|
||||
assert len(all_notifications) == 2
|
||||
exclude_jobs_by_default = get_notifications_for_service(sample_template.service_id).items
|
||||
assert len(exclude_jobs_by_default) == 1
|
||||
assert exclude_jobs_by_default[0].id == without_job.id
|
||||
|
||||
all_notifications = get_notifications_for_service(sample_service.id).items
|
||||
assert len(all_notifications) == 1
|
||||
assert all_notifications[0].id == without_job.id
|
||||
|
||||
|
||||
def test_should_return_notifications_including_jobs(notify_db, notify_db_session, sample_service):
|
||||
assert len(Notification.query.all()) == 0
|
||||
|
||||
job = sample_job(notify_db, notify_db_session)
|
||||
with_job = sample_notification(
|
||||
notify_db, notify_db_session, created_at=datetime.utcnow(), status="delivered", job=job
|
||||
)
|
||||
|
||||
all_notifications = Notification.query.all()
|
||||
assert len(all_notifications) == 1
|
||||
|
||||
all_notifications = get_notifications_for_service(sample_service.id).items
|
||||
assert len(all_notifications) == 0
|
||||
|
||||
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True).items
|
||||
assert len(all_notifications) == 1
|
||||
assert all_notifications[0].id == with_job.id
|
||||
exclude_jobs_manually = get_notifications_for_service(sample_template.service_id, include_jobs=False).items
|
||||
assert len(exclude_jobs_manually) == 1
|
||||
assert exclude_jobs_manually[0].id == without_job.id
|
||||
|
||||
|
||||
def test_get_notifications_created_by_api_or_csv_are_returned_correctly_excluding_test_key_notifications(
|
||||
|
||||
@@ -5,6 +5,7 @@ from app import db
|
||||
from app.dao.jobs_dao import dao_create_job
|
||||
from app.dao.service_inbound_api_dao import save_service_inbound_api
|
||||
from app.models import (
|
||||
ApiKey,
|
||||
Service,
|
||||
User,
|
||||
Template,
|
||||
@@ -120,6 +121,14 @@ def create_notification(
|
||||
sent_at = sent_at or datetime.utcnow()
|
||||
updated_at = updated_at or datetime.utcnow()
|
||||
|
||||
if not job and not api_key_id:
|
||||
# we didn't specify in test - lets create it
|
||||
existing_api_key = ApiKey.query.filter(ApiKey.service == template.service, ApiKey.key_type == key_type).first()
|
||||
if existing_api_key:
|
||||
api_key_id = existing_api_key.id
|
||||
else:
|
||||
api_key_id = create_api_key(template.service, key_type=key_type).id
|
||||
|
||||
data = {
|
||||
'id': uuid.uuid4(),
|
||||
'to': to_field,
|
||||
@@ -253,3 +262,17 @@ def create_rate(start_date, value, notification_type):
|
||||
db.session.add(rate)
|
||||
db.session.commit()
|
||||
return rate
|
||||
|
||||
|
||||
def create_api_key(service, key_type=KEY_TYPE_NORMAL):
|
||||
api_key = ApiKey(
|
||||
service=service,
|
||||
name='live api key',
|
||||
created_by=service.created_by,
|
||||
key_type=key_type,
|
||||
id=uuid.uuid4(),
|
||||
secret=uuid.uuid4()
|
||||
)
|
||||
db.session.add(api_key)
|
||||
db.session.commit()
|
||||
return api_key
|
||||
|
||||
@@ -2,6 +2,8 @@ import uuid
|
||||
|
||||
from flask import json
|
||||
|
||||
from app.variables import LETTER_TEST_API_FILENAME
|
||||
|
||||
from tests import create_authorization_header
|
||||
|
||||
|
||||
@@ -41,7 +43,7 @@ def test_send_letter_jobs_throws_validation_error(client, mocker):
|
||||
assert not mock_celery.called
|
||||
|
||||
|
||||
def test_send_letter_jobs_throws_validation_error(client, sample_letter_job):
|
||||
def test_get_letter_jobs_excludes_non_letter_jobs(client, sample_letter_job, sample_job):
|
||||
auth_header = create_authorization_header()
|
||||
response = client.get(
|
||||
path='/letter-jobs',
|
||||
@@ -53,3 +55,11 @@ def test_send_letter_jobs_throws_validation_error(client, sample_letter_job):
|
||||
assert json_resp['data'][0]['id'] == str(sample_letter_job.id)
|
||||
assert json_resp['data'][0]['service_name']['name'] == sample_letter_job.service.name
|
||||
assert json_resp['data'][0]['job_status'] == 'pending'
|
||||
|
||||
|
||||
def test_get_letter_jobs_excludes_test_jobs(admin_request, sample_letter_job):
|
||||
sample_letter_job.original_file_name = LETTER_TEST_API_FILENAME
|
||||
|
||||
json_resp = admin_request.get('letter-job.get_letter_jobs')
|
||||
|
||||
assert len(json_resp['data']) == 0
|
||||
|
||||
@@ -8,6 +8,7 @@ from app.models import Notification
|
||||
from app.notifications.process_letter_notifications import create_letter_api_job
|
||||
from app.notifications.process_letter_notifications import create_letter_notification
|
||||
from app.v2.errors import InvalidRequest
|
||||
from app.variables import LETTER_API_FILENAME
|
||||
|
||||
from tests.app.db import create_service
|
||||
from tests.app.db import create_template
|
||||
@@ -37,7 +38,7 @@ def test_create_job_creates_job(sample_letter_template):
|
||||
job = create_letter_api_job(sample_letter_template)
|
||||
|
||||
assert job == Job.query.one()
|
||||
assert job.original_file_name == 'letter submitted via api'
|
||||
assert job.original_file_name == LETTER_API_FILENAME
|
||||
assert job.service == sample_letter_template.service
|
||||
assert job.template_id == sample_letter_template.id
|
||||
assert job.template_version == sample_letter_template.version
|
||||
|
||||
@@ -6,6 +6,7 @@ from app import encryption
|
||||
from app.models import (
|
||||
ServiceWhitelist,
|
||||
Notification,
|
||||
SMS_TYPE,
|
||||
MOBILE_TYPE,
|
||||
EMAIL_TYPE,
|
||||
NOTIFICATION_CREATED,
|
||||
@@ -18,6 +19,7 @@ from tests.app.conftest import (
|
||||
sample_template as create_sample_template,
|
||||
sample_notification_with_job as create_sample_notification_with_job
|
||||
)
|
||||
from tests.app.db import create_notification
|
||||
|
||||
|
||||
@pytest.mark.parametrize('mobile_number', [
|
||||
@@ -148,21 +150,56 @@ def test_notification_for_csv_returns_bst_correctly(notify_db, notify_db_session
|
||||
assert serialized['created_at'] == 'Monday 27 March 2017 at 00:01'
|
||||
|
||||
|
||||
def test_notification_personalisation_getter_returns_empty_dict_from_None(sample_notification):
|
||||
sample_notification._personalisation = None
|
||||
assert sample_notification.personalisation == {}
|
||||
def test_notification_personalisation_getter_returns_empty_dict_from_None():
|
||||
noti = Notification()
|
||||
noti._personalisation = None
|
||||
assert noti.personalisation == {}
|
||||
|
||||
|
||||
def test_notification_personalisation_getter_always_returns_empty_dict(sample_notification):
|
||||
sample_notification._personalisation = encryption.encrypt({})
|
||||
assert sample_notification.personalisation == {}
|
||||
def test_notification_personalisation_getter_always_returns_empty_dict():
|
||||
noti = Notification()
|
||||
noti._personalisation = encryption.encrypt({})
|
||||
assert noti.personalisation == {}
|
||||
|
||||
|
||||
@pytest.mark.parametrize('input_value', [
|
||||
None,
|
||||
{}
|
||||
])
|
||||
def test_notification_personalisation_setter_always_sets_empty_dict(sample_notification, input_value):
|
||||
sample_notification.personalisation = input_value
|
||||
def test_notification_personalisation_setter_always_sets_empty_dict(input_value):
|
||||
noti = Notification()
|
||||
noti.personalisation = input_value
|
||||
|
||||
assert sample_notification._personalisation == encryption.encrypt({})
|
||||
assert noti._personalisation == encryption.encrypt({})
|
||||
|
||||
|
||||
def test_notification_subject_is_none_for_sms():
|
||||
assert Notification(notification_type=SMS_TYPE).subject is None
|
||||
|
||||
|
||||
def test_notification_subject_fills_in_placeholders_for_email(sample_email_template_with_placeholders):
|
||||
noti = create_notification(sample_email_template_with_placeholders, personalisation={'name': 'hello'})
|
||||
assert noti.subject == 'hello'
|
||||
|
||||
|
||||
def test_notification_subject_fills_in_placeholders_for_letter(sample_letter_template):
|
||||
sample_letter_template.subject = '((name))'
|
||||
noti = create_notification(sample_letter_template, personalisation={'name': 'hello'})
|
||||
assert noti.subject == 'hello'
|
||||
|
||||
|
||||
def test_letter_notification_serializes_with_address(client, sample_letter_notification):
|
||||
sample_letter_notification.personalisation = {
|
||||
'address_line_1': 'foo',
|
||||
'address_line_3': 'bar',
|
||||
'address_line_5': None,
|
||||
'postcode': 'SW1 1AA'
|
||||
}
|
||||
res = sample_letter_notification.serialize()
|
||||
assert res['line_1'] == 'foo'
|
||||
assert res['line_2'] is None
|
||||
assert res['line_3'] == 'bar'
|
||||
assert res['line_4'] is None
|
||||
assert res['line_5'] is None
|
||||
assert res['line_6'] is None
|
||||
assert res['postcode'] == 'SW1 1AA'
|
||||
|
||||
@@ -13,6 +13,8 @@ from app.models import LETTER_TYPE
|
||||
from app.models import Notification
|
||||
from app.models import SMS_TYPE
|
||||
from app.v2.errors import RateLimitError
|
||||
from app.variables import LETTER_TEST_API_FILENAME
|
||||
from app.variables import LETTER_API_FILENAME
|
||||
|
||||
from tests import create_authorization_header
|
||||
from tests.app.db import create_service
|
||||
@@ -53,6 +55,7 @@ def test_post_letter_notification_returns_201(client, sample_letter_template, mo
|
||||
resp_json = letter_request(client, data, service_id=sample_letter_template.service_id)
|
||||
|
||||
job = Job.query.one()
|
||||
assert job.original_file_name == LETTER_API_FILENAME
|
||||
notification = Notification.query.one()
|
||||
notification_id = notification.id
|
||||
assert resp_json['id'] == str(notification_id)
|
||||
@@ -202,6 +205,7 @@ def test_post_letter_notification_doesnt_queue_task(
|
||||
letter_request(client, data, service_id=service.id, key_type=key_type)
|
||||
|
||||
job = Job.query.one()
|
||||
assert job.original_file_name == LETTER_TEST_API_FILENAME
|
||||
assert not real_task.called
|
||||
fake_task.assert_called_once_with([str(job.id)], queue='research-mode-tasks')
|
||||
|
||||
|
||||
Reference in New Issue
Block a user