Update get_notification_by_id to take an optional service_id

It can be useful to get a notification by id while checking that the
notification belongs to a given service. This changes the
get_notification_by_id DAO function to optionally also filter by
service_id so that we can check this.
This commit is contained in:
Katie Smith
2018-11-15 10:55:29 +00:00
parent 902e1b403a
commit 365c462e93
2 changed files with 31 additions and 6 deletions

View File

@@ -227,11 +227,15 @@ def get_notification_with_personalisation(service_id, notification_id, key_type)
@statsd(namespace="dao") @statsd(namespace="dao")
def get_notification_by_id(notification_id, _raise=False): def get_notification_by_id(notification_id, service_id=None, _raise=False):
if _raise: filters = [Notification.id == notification_id]
return Notification.query.filter_by(id=notification_id).one()
else: if service_id:
return Notification.query.filter_by(id=notification_id).first() filters.append(Notification.service_id == service_id)
query = Notification.query.filter(*filters)
return query.one() if _raise else query.first()
def get_notifications(filter_dict=None): def get_notifications(filter_dict=None):

View File

@@ -5,6 +5,8 @@ from functools import partial
import pytest import pytest
from freezegun import freeze_time from freezegun import freeze_time
from sqlalchemy.exc import SQLAlchemyError, IntegrityError from sqlalchemy.exc import SQLAlchemyError, IntegrityError
from sqlalchemy.orm.exc import NoResultFound
from app.dao.notifications_dao import ( from app.dao.notifications_dao import (
dao_create_notification, dao_create_notification,
@@ -520,7 +522,7 @@ def test_save_notification_with_no_job(sample_template, mmg_provider):
assert notification_from_db.status == 'created' assert notification_from_db.status == 'created'
def test_get_notification_by_id(notify_db, notify_db_session, sample_template): def test_get_notification_with_personalisation_by_id(notify_db, notify_db_session, sample_template):
notification = sample_notification(notify_db=notify_db, notify_db_session=notify_db_session, notification = sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
template=sample_template, template=sample_template,
scheduled_for='2017-05-05 14:15', scheduled_for='2017-05-05 14:15',
@@ -534,6 +536,25 @@ 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) assert notification_from_db.scheduled_notification.scheduled_for == datetime(2017, 5, 5, 14, 15)
def test_get_notification_by_id_when_notification_exists(sample_notification):
notification_from_db = get_notification_by_id(sample_notification.id)
assert sample_notification == notification_from_db
def test_get_notification_by_id_when_notification_does_not_exist(notify_db_session, fake_uuid):
notification_from_db = get_notification_by_id(fake_uuid)
assert notification_from_db is None
def test_get_notification_by_id_when_notification_exists_for_different_service(sample_notification):
another_service = create_service(service_name='Another service')
with pytest.raises(NoResultFound):
get_notification_by_id(sample_notification.id, another_service.id, _raise=True)
def test_get_notifications_by_reference(sample_template): def test_get_notifications_by_reference(sample_template):
client_reference = 'some-client-ref' client_reference = 'some-client-ref'
assert len(Notification.query.all()) == 0 assert len(Notification.query.all()) == 0