From 365c462e930eedf2d94b985608bc5d9a930634a1 Mon Sep 17 00:00:00 2001 From: Katie Smith Date: Thu, 15 Nov 2018 10:55:29 +0000 Subject: [PATCH] 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. --- app/dao/notifications_dao.py | 14 +++++++---- .../notification_dao/test_notification_dao.py | 23 ++++++++++++++++++- 2 files changed, 31 insertions(+), 6 deletions(-) diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py index a80cdbefe..0ff1ffa67 100644 --- a/app/dao/notifications_dao.py +++ b/app/dao/notifications_dao.py @@ -227,11 +227,15 @@ def get_notification_with_personalisation(service_id, notification_id, key_type) @statsd(namespace="dao") -def get_notification_by_id(notification_id, _raise=False): - if _raise: - return Notification.query.filter_by(id=notification_id).one() - else: - return Notification.query.filter_by(id=notification_id).first() +def get_notification_by_id(notification_id, service_id=None, _raise=False): + filters = [Notification.id == notification_id] + + if service_id: + 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): diff --git a/tests/app/dao/notification_dao/test_notification_dao.py b/tests/app/dao/notification_dao/test_notification_dao.py index 50c100192..c40b315d3 100644 --- a/tests/app/dao/notification_dao/test_notification_dao.py +++ b/tests/app/dao/notification_dao/test_notification_dao.py @@ -5,6 +5,8 @@ from functools import partial import pytest from freezegun import freeze_time from sqlalchemy.exc import SQLAlchemyError, IntegrityError +from sqlalchemy.orm.exc import NoResultFound + from app.dao.notifications_dao import ( dao_create_notification, @@ -520,7 +522,7 @@ def test_save_notification_with_no_job(sample_template, mmg_provider): 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, template=sample_template, 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) +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): client_reference = 'some-client-ref' assert len(Notification.query.all()) == 0