darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-09-11 10:28:55 -04:00
Code Issues Packages Projects Releases Wiki Activity

Update update_notification_status_by_id DAO function

Browse Source 
Replaced `.with_for_lockmode()`, which is now deprecated, with
`.with_for_update() - https://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.with_lockmode

The function should update any statuses that are not 'final', so added
`pending-virus-check` to the list of statuses that the function can
update.
This commit is contained in:
Katie Smith
2018-12-03 12:05:23 +00:00
parent 7ec3dbd667
commit 902e1b403a
2 changed files with 12 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/dao/notifications_dao.py
+4 -2
View File
@@ -39,6 +39,7 @@ from app.models import (
NOTIFICATION_DELIVERED,
NOTIFICATION_SENDING,
NOTIFICATION_PENDING,
NOTIFICATION_PENDING_VIRUS_CHECK,
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_TEMPORARY_FAILURE,
NOTIFICATION_PERMANENT_FAILURE,
@@ -145,13 +146,14 @@ def _update_notification_status(notification, status):
@statsd(namespace="dao")
@transactional
def update_notification_status_by_id(notification_id, status, sent_by=None):
notification = Notification.query.with_lockmode("update").filter(
notification = Notification.query.with_for_update().filter(
Notification.id == notification_id,
or_(
Notification.status == NOTIFICATION_CREATED,
Notification.status == NOTIFICATION_SENDING,
Notification.status == NOTIFICATION_PENDING,
Notification.status == NOTIFICATION_SENT
Notification.status == NOTIFICATION_SENT,
Notification.status == NOTIFICATION_PENDING_VIRUS_CHECK
)).first()
if not notification:
tests/app/dao/notification_dao/test_notification_dao.py
+8
View File
@@ -141,6 +141,14 @@ def test_should_update_status_by_id_if_created(notify_db, notify_db_session):
assert updated.status == 'failed'
def test_should_update_status_by_id_if_pending_virus_check(notify_db, notify_db_session):
notification = sample_notification(notify_db, notify_db_session, status='pending-virus-check')
assert Notification.query.get(notification.id).status == 'pending-virus-check'
updated = update_notification_status_by_id(notification.id, 'cancelled')
assert Notification.query.get(notification.id).status == 'cancelled'
assert updated.status == 'cancelled'
def test_should_update_status_by_id_and_set_sent_by(notify_db, notify_db_session):
notification = sample_notification(notify_db, notify_db_session, status='sending')