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

Merge pull request #1362 from GSA/notify-api-1320

Browse Source 
upgrade core daos to sqlalchemy 2.0
This commit is contained in:
Kenneth Kehl
2024-10-28 13:01:02 -07:00
committed by GitHub
parent 616f6bebc2 bd7d072571
commit 2d2661f304
5 changed files with 273 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.ds.baseline
View File

@@ -249,7 +249,7 @@
"filename": "tests/app/dao/test_users_dao.py", "filename": "tests/app/dao/test_users_dao.py",
"hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8", "hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8",
"is_verified": false, "is_verified": false,
"line_number": 52, "line_number": 69,
"is_secret": false "is_secret": false
}, },
{ {
@@ -257,7 +257,7 @@
"filename": "tests/app/dao/test_users_dao.py", "filename": "tests/app/dao/test_users_dao.py",
"hashed_secret": "f2c57870308dc87f432e5912d4de6f8e322721ba", "hashed_secret": "f2c57870308dc87f432e5912d4de6f8e322721ba",
"is_verified": false, "is_verified": false,
"line_number": 176, "line_number": 194,
"is_secret": false "is_secret": false
} }
], ],
@@ -384,5 +384,5 @@
} }
] ]
}, },
"generated_at": "2024-09-27T16:42:53Z" "generated_at": "2024-10-11T19:26:50Z"
} }

106
app/dao/notifications_dao.py
View File

@@ -1,7 +1,7 @@
from datetime import timedelta from datetime import timedelta
from flask import current_app from flask import current_app
from sqlalchemy import asc, desc, or_, select, text, union from sqlalchemy import asc, delete, desc, func, or_, select, text, union, update
from sqlalchemy.orm import joinedload from sqlalchemy.orm import joinedload
from sqlalchemy.orm.exc import NoResultFound from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.sql import functions from sqlalchemy.sql import functions
@@ -109,11 +109,12 @@ def _update_notification_status(
def update_notification_status_by_id( def update_notification_status_by_id(
notification_id, status, sent_by=None, provider_response=None, carrier=None notification_id, status, sent_by=None, provider_response=None, carrier=None
): ):
notification = ( stmt = (
Notification.query.with_for_update() select(Notification)
.with_for_update()
.filter(Notification.id == notification_id) .filter(Notification.id == notification_id)
.first()
) )
notification = db.session.execute(stmt).scalars().first()
if not notification: if not notification:
current_app.logger.info( current_app.logger.info(
@@ -156,9 +157,8 @@ def update_notification_status_by_id(
@autocommit @autocommit
def update_notification_status_by_reference(reference, status): def update_notification_status_by_reference(reference, status):
# this is used to update emails # this is used to update emails
notification = Notification.query.filter( stmt = select(Notification).filter(Notification.reference == reference)
Notification.reference == reference notification = db.session.execute(stmt).scalars().first()
).first()
if not notification: if not notification:
current_app.logger.error( current_app.logger.error(
@@ -200,19 +200,20 @@ def get_notifications_for_job(
def dao_get_notification_count_for_job_id(*, job_id): def dao_get_notification_count_for_job_id(*, job_id):
return Notification.query.filter_by(job_id=job_id).count() stmt = select(func.count(Notification.id)).filter_by(job_id=job_id)
return db.session.execute(stmt).scalar()
def dao_get_notification_count_for_service(*, service_id): def dao_get_notification_count_for_service(*, service_id):
notification_count = Notification.query.filter_by(service_id=service_id).count() stmt = select(func.count(Notification.id)).filter_by(service_id=service_id)
return notification_count return db.session.execute(stmt).scalar()
def dao_get_failed_notification_count(): def dao_get_failed_notification_count():
failed_count = Notification.query.filter_by( stmt = select(func.count(Notification.id)).filter_by(
status=NotificationStatus.FAILED status=NotificationStatus.FAILED
).count() )
return failed_count return db.session.execute(stmt).scalar()
def get_notification_with_personalisation(service_id, notification_id, key_type): def get_notification_with_personalisation(service_id, notification_id, key_type):
@@ -220,11 +221,12 @@ def get_notification_with_personalisation(service_id, notification_id, key_type)
if key_type: if key_type:
filter_dict["key_type"] = key_type filter_dict["key_type"] = key_type
return ( stmt = (
Notification.query.filter_by(**filter_dict) select(Notification)
.filter_by(**filter_dict)
.options(joinedload(Notification.template)) .options(joinedload(Notification.template))
.one()
) )
return db.session.execute(stmt).scalars().one()
def get_notification_by_id(notification_id, service_id=None, _raise=False): def get_notification_by_id(notification_id, service_id=None, _raise=False):
@@ -233,9 +235,13 @@ def get_notification_by_id(notification_id, service_id=None, _raise=False):
if service_id: if service_id:
filters.append(Notification.service_id == service_id) filters.append(Notification.service_id == service_id)
query = Notification.query.filter(*filters) stmt = select(Notification).filter(*filters)
return query.one() if _raise else query.first() return (
db.session.execute(stmt).scalars().one()
if _raise
else db.session.execute(stmt).scalars().first()
)
def get_notifications_for_service( def get_notifications_for_service(
@@ -415,12 +421,13 @@ def move_notifications_to_notification_history(
deleted += delete_count_per_call deleted += delete_count_per_call
# Deleting test Notifications, test notifications are not persisted to NotificationHistory # Deleting test Notifications, test notifications are not persisted to NotificationHistory
Notification.query.filter( stmt = delete(Notification).filter(
Notification.notification_type == notification_type, Notification.notification_type == notification_type,
Notification.service_id == service_id, Notification.service_id == service_id,
Notification.created_at < timestamp_to_delete_backwards_from, Notification.created_at < timestamp_to_delete_backwards_from,
Notification.key_type == KeyType.TEST, Notification.key_type == KeyType.TEST,
).delete(synchronize_session=False) )
db.session.execute(stmt)
db.session.commit() db.session.commit()
return deleted return deleted
@@ -442,8 +449,9 @@ def dao_timeout_notifications(cutoff_time, limit=100000):
current_statuses = [NotificationStatus.SENDING, NotificationStatus.PENDING] current_statuses = [NotificationStatus.SENDING, NotificationStatus.PENDING]
new_status = NotificationStatus.TEMPORARY_FAILURE new_status = NotificationStatus.TEMPORARY_FAILURE
notifications = ( stmt = (
Notification.query.filter( select(Notification)
.filter(
Notification.created_at < cutoff_time, Notification.created_at < cutoff_time,
Notification.status.in_(current_statuses), Notification.status.in_(current_statuses),
Notification.notification_type.in_( Notification.notification_type.in_(
@@ -451,14 +459,15 @@ def dao_timeout_notifications(cutoff_time, limit=100000):
), ),
) )
.limit(limit) .limit(limit)
.all()
) )
notifications = db.session.execute(stmt).scalars().all()
Notification.query.filter( stmt = (
Notification.id.in_([n.id for n in notifications]), update(Notification)
).update( .filter(Notification.id.in_([n.id for n in notifications]))
{"status": new_status, "updated_at": updated_at}, synchronize_session=False .values({"status": new_status, "updated_at": updated_at})
) )
db.session.execute(stmt)
db.session.commit() db.session.commit()
return notifications return notifications
@@ -466,15 +475,23 @@ def dao_timeout_notifications(cutoff_time, limit=100000):
@autocommit @autocommit
def dao_update_notifications_by_reference(references, update_dict): def dao_update_notifications_by_reference(references, update_dict):
updated_count = Notification.query.filter( stmt = (
Notification.reference.in_(references) update(Notification)
).update(update_dict, synchronize_session=False) .filter(Notification.reference.in_(references))
.values(update_dict)
)
result = db.session.execute(stmt)
updated_count = result.rowcount
updated_history_count = 0 updated_history_count = 0
if updated_count != len(references): if updated_count != len(references):
updated_history_count = NotificationHistory.query.filter( stmt = (
NotificationHistory.reference.in_(references) update(NotificationHistory)
).update(update_dict, synchronize_session=False) .filter(NotificationHistory.reference.in_(references))
.values(update_dict)
)
result = db.session.execute(stmt)
updated_history_count = result.rowcount
return updated_count, updated_history_count return updated_count, updated_history_count
@@ -541,18 +558,21 @@ def dao_get_notifications_by_recipient_or_reference(
def dao_get_notification_by_reference(reference): def dao_get_notification_by_reference(reference):
return Notification.query.filter(Notification.reference == reference).one() stmt = select(Notification).filter(Notification.reference == reference)
return db.session.execute(stmt).scalars().one()
def dao_get_notification_history_by_reference(reference): def dao_get_notification_history_by_reference(reference):
try: try:
# This try except is necessary because in test keys and research mode does not create notification history. # This try except is necessary because in test keys and research mode does not create notification history.
# Otherwise we could just search for the NotificationHistory object # Otherwise we could just search for the NotificationHistory object
return Notification.query.filter(Notification.reference == reference).one() stmt = select(Notification).filter(Notification.reference == reference)
return db.session.execute(stmt).scalars().one()
except NoResultFound: except NoResultFound:
return NotificationHistory.query.filter( stmt = select(NotificationHistory).filter(
NotificationHistory.reference == reference NotificationHistory.reference == reference
).one() )
return db.session.execute(stmt).scalars().one()
def dao_get_notifications_processing_time_stats(start_date, end_date): def dao_get_notifications_processing_time_stats(start_date, end_date):
@@ -590,11 +610,12 @@ def dao_get_notifications_processing_time_stats(start_date, end_date):
def dao_get_last_notification_added_for_job_id(job_id): def dao_get_last_notification_added_for_job_id(job_id):
last_notification_added = ( stmt = (
Notification.query.filter(Notification.job_id == job_id) select(Notification)
.filter(Notification.job_id == job_id)
.order_by(Notification.job_row_number.desc()) .order_by(Notification.job_row_number.desc())
.first()
) )
last_notification_added = db.session.execute(stmt).scalars().first()
return last_notification_added return last_notification_added
@@ -602,11 +623,12 @@ def dao_get_last_notification_added_for_job_id(job_id):
def notifications_not_yet_sent(should_be_sending_after_seconds, notification_type): def notifications_not_yet_sent(should_be_sending_after_seconds, notification_type):
older_than_date = utc_now() - timedelta(seconds=should_be_sending_after_seconds) older_than_date = utc_now() - timedelta(seconds=should_be_sending_after_seconds)
notifications = Notification.query.filter( stmt = select(Notification).filter(
Notification.created_at <= older_than_date, Notification.created_at <= older_than_date,
Notification.notification_type == notification_type, Notification.notification_type == notification_type,
Notification.status == NotificationStatus.CREATED, Notification.status == NotificationStatus.CREATED,
).all() )
notifications = db.session.execute(stmt).scalars().all()
return notifications return notifications

55
app/dao/users_dao.py
View File

@@ -4,7 +4,7 @@ from secrets import randbelow
import sqlalchemy import sqlalchemy
from flask import current_app from flask import current_app
from sqlalchemy import func, text from sqlalchemy import delete, func, select, text
from sqlalchemy.orm import joinedload from sqlalchemy.orm import joinedload
from app import db from app import db
@@ -37,8 +37,8 @@ def get_login_gov_user(login_uuid, email_address):
login.gov uuids are. Eventually the code that checks by email address login.gov uuids are. Eventually the code that checks by email address
should be removed. should be removed.
""" """
stmt = select(User).filter_by(login_uuid=login_uuid)
user = User.query.filter_by(login_uuid=login_uuid).first() user = db.session.execute(stmt).scalars().first()
if user: if user:
if user.email_address != email_address: if user.email_address != email_address:
try: try:
@@ -54,7 +54,8 @@ def get_login_gov_user(login_uuid, email_address):
return user return user
# Remove this 1 July 2025, all users should have login.gov uuids by now # Remove this 1 July 2025, all users should have login.gov uuids by now
user = User.query.filter(User.email_address.ilike(email_address)).first() stmt = select(User).filter(User.email_address.ilike(email_address))
user = db.session.execute(stmt).scalars().first()
if user: if user:
save_user_attribute(user, {"login_uuid": login_uuid}) save_user_attribute(user, {"login_uuid": login_uuid})
@@ -102,24 +103,27 @@ def create_user_code(user, code, code_type):
def get_user_code(user, code, code_type): def get_user_code(user, code, code_type):
# Get the most recent codes to try and reduce the # Get the most recent codes to try and reduce the
# time searching for the correct code. # time searching for the correct code.
codes = VerifyCode.query.filter_by(user=user, code_type=code_type).order_by( stmt = (
VerifyCode.created_at.desc() select(VerifyCode)
.filter_by(user=user, code_type=code_type)
.order_by(VerifyCode.created_at.desc())
) )
codes = db.session.execute(stmt).scalars().all()
return next((x for x in codes if x.check_code(code)), None) return next((x for x in codes if x.check_code(code)), None)
def delete_codes_older_created_more_than_a_day_ago(): def delete_codes_older_created_more_than_a_day_ago():
deleted = ( stmt = delete(VerifyCode).filter(
db.session.query(VerifyCode) VerifyCode.created_at < utc_now() - timedelta(hours=24)
.filter(VerifyCode.created_at < utc_now() - timedelta(hours=24))
.delete()
) )
deleted = db.session.execute(stmt)
db.session.commit() db.session.commit()
return deleted return deleted
def use_user_code(id): def use_user_code(id):
verify_code = VerifyCode.query.get(id) verify_code = db.session.get(VerifyCode, id)
verify_code.code_used = True verify_code.code_used = True
db.session.add(verify_code) db.session.add(verify_code)
db.session.commit() db.session.commit()
@@ -131,36 +135,42 @@ def delete_model_user(user):
def delete_user_verify_codes(user): def delete_user_verify_codes(user):
VerifyCode.query.filter_by(user=user).delete() stmt = delete(VerifyCode).filter_by(user=user)
db.session.execute(stmt)
db.session.commit() db.session.commit()
def count_user_verify_codes(user): def count_user_verify_codes(user):
query = VerifyCode.query.filter( stmt = select(func.count(VerifyCode.id)).filter(
VerifyCode.user == user, VerifyCode.user == user,
VerifyCode.expiry_datetime > utc_now(), VerifyCode.expiry_datetime > utc_now(),
VerifyCode.code_used.is_(False), VerifyCode.code_used.is_(False),
) )
return query.count() result = db.session.execute(stmt).scalar()
return result or 0
def get_user_by_id(user_id=None): def get_user_by_id(user_id=None):
if user_id: if user_id:
return User.query.filter_by(id=user_id).one() stmt = select(User).filter_by(id=user_id)
return User.query.filter_by().all() return db.session.execute(stmt).scalars().one()
return get_users()
def get_users(): def get_users():
return User.query.all() stmt = select(User)
return db.session.execute(stmt).scalars().all()
def get_user_by_email(email): def get_user_by_email(email):
return User.query.filter(func.lower(User.email_address) == func.lower(email)).one() stmt = select(User).filter(func.lower(User.email_address) == func.lower(email))
return db.session.execute(stmt).scalars().one()
def get_users_by_partial_email(email): def get_users_by_partial_email(email):
email = escape_special_characters(email) email = escape_special_characters(email)
return User.query.filter(User.email_address.ilike("%{}%".format(email))).all() stmt = select(User).filter(User.email_address.ilike("%{}%".format(email)))
return db.session.execute(stmt).scalars().all()
def increment_failed_login_count(user): def increment_failed_login_count(user):
@@ -188,16 +198,17 @@ def get_user_and_accounts(user_id):
# TODO: With sqlalchemy 2.0 change as below because of the breaking change # TODO: With sqlalchemy 2.0 change as below because of the breaking change
# at User.organizations.services, we need to verify that the below subqueryload # at User.organizations.services, we need to verify that the below subqueryload
# that we have put is functionally doing the same thing as before # that we have put is functionally doing the same thing as before
return ( stmt = (
User.query.filter(User.id == user_id) select(User)
.filter(User.id == user_id)
.options( .options(
# eagerly load the user's services and organizations, and also the service's org and vice versa # eagerly load the user's services and organizations, and also the service's org and vice versa
# (so we can see if the user knows about it) # (so we can see if the user knows about it)
joinedload(User.services).joinedload(Service.organization), joinedload(User.services).joinedload(Service.organization),
joinedload(User.organizations).subqueryload(Organization.services), joinedload(User.organizations).subqueryload(Organization.services),
) )
.one()
) )
return db.session.execute(stmt).scalars().unique().one()
@autocommit @autocommit

234
tests/app/dao/notification_dao/test_notification_dao.py
View File

@@ -4,9 +4,11 @@ from functools import partial
import pytest import pytest
from freezegun import freeze_time from freezegun import freeze_time
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError, SQLAlchemyError from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from sqlalchemy.orm.exc import NoResultFound from sqlalchemy.orm.exc import NoResultFound
from app import db
from app.dao.notifications_dao import ( from app.dao.notifications_dao import (
dao_create_notification, dao_create_notification,
dao_delete_notifications_by_id, dao_delete_notifications_by_id,
@@ -55,7 +57,10 @@ def test_should_by_able_to_update_status_by_reference(
notification = Notification(**data) notification = Notification(**data)
dao_create_notification(notification) dao_create_notification(notification)
assert Notification.query.get(notification.id).status == NotificationStatus.SENDING assert (
db.session.get(Notification, notification.id).status
== NotificationStatus.SENDING
)
notification.reference = "reference" notification.reference = "reference"
dao_update_notification(notification) dao_update_notification(notification)
@@ -64,7 +69,8 @@ def test_should_by_able_to_update_status_by_reference(
) )
assert updated.status == NotificationStatus.DELIVERED assert updated.status == NotificationStatus.DELIVERED
assert ( assert (
Notification.query.get(notification.id).status == NotificationStatus.DELIVERED db.session.get(Notification, notification.id).status
== NotificationStatus.DELIVERED
) )
@@ -81,7 +87,10 @@ def test_should_by_able_to_update_status_by_id(
dao_create_notification(notification) dao_create_notification(notification)
assert notification.status == NotificationStatus.SENDING assert notification.status == NotificationStatus.SENDING
assert Notification.query.get(notification.id).status == NotificationStatus.SENDING assert (
db.session.get(Notification, notification.id).status
== NotificationStatus.SENDING
)
with freeze_time("2000-01-02 12:00:00"): with freeze_time("2000-01-02 12:00:00"):
updated = update_notification_status_by_id( updated = update_notification_status_by_id(
@@ -92,7 +101,8 @@ def test_should_by_able_to_update_status_by_id(
assert updated.status == NotificationStatus.DELIVERED assert updated.status == NotificationStatus.DELIVERED
assert updated.updated_at == datetime(2000, 1, 2, 12, 0, 0) assert updated.updated_at == datetime(2000, 1, 2, 12, 0, 0)
assert ( assert (
Notification.query.get(notification.id).status == NotificationStatus.DELIVERED db.session.get(Notification, notification.id).status
== NotificationStatus.DELIVERED
) )
assert notification.updated_at == datetime(2000, 1, 2, 12, 0, 0) assert notification.updated_at == datetime(2000, 1, 2, 12, 0, 0)
assert notification.status == NotificationStatus.DELIVERED assert notification.status == NotificationStatus.DELIVERED
@@ -107,15 +117,17 @@ def test_should_not_update_status_by_id_if_not_sending_and_does_not_update_job(
job=sample_job, job=sample_job,
) )
assert ( assert (
Notification.query.get(notification.id).status == NotificationStatus.DELIVERED db.session.get(Notification, notification.id).status
== NotificationStatus.DELIVERED
) )
assert not update_notification_status_by_id( assert not update_notification_status_by_id(
notification.id, NotificationStatus.FAILED notification.id, NotificationStatus.FAILED
) )
assert ( assert (
Notification.query.get(notification.id).status == NotificationStatus.DELIVERED db.session.get(Notification, notification.id).status
== NotificationStatus.DELIVERED
) )
assert sample_job == Job.query.get(notification.job_id) assert sample_job == db.session.get(Job, notification.job_id)
def test_should_not_update_status_by_reference_if_not_sending_and_does_not_update_job( def test_should_not_update_status_by_reference_if_not_sending_and_does_not_update_job(
@@ -128,20 +140,22 @@ def test_should_not_update_status_by_reference_if_not_sending_and_does_not_updat
job=sample_job, job=sample_job,
) )
assert ( assert (
Notification.query.get(notification.id).status == NotificationStatus.DELIVERED db.session.get(Notification, notification.id).status
== NotificationStatus.DELIVERED
) )
assert not update_notification_status_by_reference( assert not update_notification_status_by_reference(
"reference", NotificationStatus.FAILED "reference", NotificationStatus.FAILED
) )
assert ( assert (
Notification.query.get(notification.id).status == NotificationStatus.DELIVERED db.session.get(Notification, notification.id).status
== NotificationStatus.DELIVERED
) )
assert sample_job == Job.query.get(notification.job_id) assert sample_job == db.session.get(Job, notification.job_id)
def test_should_update_status_by_id_if_created(sample_template, sample_notification): def test_should_update_status_by_id_if_created(sample_template, sample_notification):
assert ( assert (
Notification.query.get(sample_notification.id).status db.session.get(Notification, sample_notification.id).status
== NotificationStatus.CREATED == NotificationStatus.CREATED
) )
updated = update_notification_status_by_id( updated = update_notification_status_by_id(
@@ -149,7 +163,7 @@ def test_should_update_status_by_id_if_created(sample_template, sample_notificat
NotificationStatus.FAILED, NotificationStatus.FAILED,
) )
assert ( assert (
Notification.query.get(sample_notification.id).status db.session.get(Notification, sample_notification.id).status
== NotificationStatus.FAILED == NotificationStatus.FAILED
) )
assert updated.status == NotificationStatus.FAILED assert updated.status == NotificationStatus.FAILED
@@ -244,11 +258,17 @@ def test_should_not_update_status_by_reference_if_not_sending(sample_template):
status=NotificationStatus.CREATED, status=NotificationStatus.CREATED,
reference="reference", reference="reference",
) )
assert Notification.query.get(notification.id).status == NotificationStatus.CREATED assert (
db.session.get(Notification, notification.id).status
== NotificationStatus.CREATED
)
updated = update_notification_status_by_reference( updated = update_notification_status_by_reference(
"reference", NotificationStatus.FAILED "reference", NotificationStatus.FAILED
) )
assert Notification.query.get(notification.id).status == NotificationStatus.CREATED assert (
db.session.get(Notification, notification.id).status
== NotificationStatus.CREATED
)
assert not updated assert not updated
@@ -264,14 +284,18 @@ def test_should_by_able_to_update_status_by_id_from_pending_to_delivered(
assert update_notification_status_by_id( assert update_notification_status_by_id(
notification_id=notification.id, status=NotificationStatus.PENDING notification_id=notification.id, status=NotificationStatus.PENDING
) )
assert Notification.query.get(notification.id).status == NotificationStatus.PENDING assert (
db.session.get(Notification, notification.id).status
== NotificationStatus.PENDING
)
assert update_notification_status_by_id( assert update_notification_status_by_id(
notification.id, notification.id,
NotificationStatus.DELIVERED, NotificationStatus.DELIVERED,
) )
assert ( assert (
Notification.query.get(notification.id).status == NotificationStatus.DELIVERED db.session.get(Notification, notification.id).status
== NotificationStatus.DELIVERED
) )
@@ -289,7 +313,10 @@ def test_should_by_able_to_update_status_by_id_from_pending_to_temporary_failure
notification_id=notification.id, notification_id=notification.id,
status=NotificationStatus.PENDING, status=NotificationStatus.PENDING,
) )
assert Notification.query.get(notification.id).status == NotificationStatus.PENDING assert (
db.session.get(Notification, notification.id).status
== NotificationStatus.PENDING
)
assert update_notification_status_by_id( assert update_notification_status_by_id(
notification.id, notification.id,
@@ -297,7 +324,7 @@ def test_should_by_able_to_update_status_by_id_from_pending_to_temporary_failure
) )
assert ( assert (
Notification.query.get(notification.id).status db.session.get(Notification, notification.id).status
== NotificationStatus.TEMPORARY_FAILURE == NotificationStatus.TEMPORARY_FAILURE
) )
@@ -312,14 +339,17 @@ def test_should_by_able_to_update_status_by_id_from_sending_to_permanent_failure
) )
notification = Notification(**data) notification = Notification(**data)
dao_create_notification(notification) dao_create_notification(notification)
assert Notification.query.get(notification.id).status == NotificationStatus.SENDING assert (
db.session.get(Notification, notification.id).status
== NotificationStatus.SENDING
)
assert update_notification_status_by_id( assert update_notification_status_by_id(
notification.id, notification.id,
status=NotificationStatus.PERMANENT_FAILURE, status=NotificationStatus.PERMANENT_FAILURE,
) )
assert ( assert (
Notification.query.get(notification.id).status db.session.get(Notification, notification.id).status
== NotificationStatus.PERMANENT_FAILURE == NotificationStatus.PERMANENT_FAILURE
) )
@@ -331,7 +361,10 @@ def test_should_not_update_status_once_notification_status_is_delivered(
template=sample_email_template, template=sample_email_template,
status=NotificationStatus.SENDING, status=NotificationStatus.SENDING,
) )
assert Notification.query.get(notification.id).status == NotificationStatus.SENDING assert (
db.session.get(Notification, notification.id).status
== NotificationStatus.SENDING
)
notification.reference = "reference" notification.reference = "reference"
dao_update_notification(notification) dao_update_notification(notification)
@@ -340,7 +373,8 @@ def test_should_not_update_status_once_notification_status_is_delivered(
NotificationStatus.DELIVERED, NotificationStatus.DELIVERED,
) )
assert ( assert (
Notification.query.get(notification.id).status == NotificationStatus.DELIVERED db.session.get(Notification, notification.id).status
== NotificationStatus.DELIVERED
) )
update_notification_status_by_reference( update_notification_status_by_reference(
@@ -348,7 +382,8 @@ def test_should_not_update_status_once_notification_status_is_delivered(
NotificationStatus.FAILED, NotificationStatus.FAILED,
) )
assert ( assert (
Notification.query.get(notification.id).status == NotificationStatus.DELIVERED db.session.get(Notification, notification.id).status
== NotificationStatus.DELIVERED
) )
@@ -370,7 +405,7 @@ def test_create_notification_creates_notification_with_personalisation(
sample_template_with_placeholders, sample_template_with_placeholders,
sample_job, sample_job,
): ):
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
data = create_notification( data = create_notification(
template=sample_template_with_placeholders, template=sample_template_with_placeholders,
@@ -379,8 +414,8 @@ def test_create_notification_creates_notification_with_personalisation(
status=NotificationStatus.CREATED, status=NotificationStatus.CREATED,
) )
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
notification_from_db = Notification.query.all()[0] notification_from_db = _get_notification_query_all()[0]
assert notification_from_db.id assert notification_from_db.id
assert data.to == notification_from_db.to assert data.to == notification_from_db.to
assert data.job_id == notification_from_db.job_id assert data.job_id == notification_from_db.job_id
@@ -393,15 +428,15 @@ def test_create_notification_creates_notification_with_personalisation(
def test_save_notification_creates_sms(sample_template, sample_job): def test_save_notification_creates_sms(sample_template, sample_job):
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
data = _notification_json(sample_template, job_id=sample_job.id) data = _notification_json(sample_template, job_id=sample_job.id)
notification = Notification(**data) notification = Notification(**data)
dao_create_notification(notification) dao_create_notification(notification)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
notification_from_db = Notification.query.all()[0] notification_from_db = _get_notification_query_all()[0]
assert notification_from_db.id assert notification_from_db.id
assert "1" == notification_from_db.to assert "1" == notification_from_db.to
assert data["job_id"] == notification_from_db.job_id assert data["job_id"] == notification_from_db.job_id
@@ -412,16 +447,36 @@ def test_save_notification_creates_sms(sample_template, sample_job):
assert notification_from_db.status == NotificationStatus.CREATED assert notification_from_db.status == NotificationStatus.CREATED
def _get_notification_query_all():
stmt = select(Notification)
return db.session.execute(stmt).scalars().all()
def _get_notification_query_one():
stmt = select(Notification)
return db.session.execute(stmt).scalars().one()
def _get_notification_query_count():
stmt = select(func.count(Notification.id))
return db.session.execute(stmt).scalar() or 0
def _get_notification_history_query_count():
stmt = select(func.count(NotificationHistory.id))
return db.session.execute(stmt).scalar() or 0
def test_save_notification_and_create_email(sample_email_template, sample_job): def test_save_notification_and_create_email(sample_email_template, sample_job):
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
data = _notification_json(sample_email_template, job_id=sample_job.id) data = _notification_json(sample_email_template, job_id=sample_job.id)
notification = Notification(**data) notification = Notification(**data)
dao_create_notification(notification) dao_create_notification(notification)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
notification_from_db = Notification.query.all()[0] notification_from_db = _get_notification_query_all()[0]
assert notification_from_db.id assert notification_from_db.id
assert "1" == notification_from_db.to assert "1" == notification_from_db.to
assert data["job_id"] == notification_from_db.job_id assert data["job_id"] == notification_from_db.job_id
@@ -433,29 +488,29 @@ def test_save_notification_and_create_email(sample_email_template, sample_job):
def test_save_notification(sample_email_template, sample_job): def test_save_notification(sample_email_template, sample_job):
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
data = _notification_json(sample_email_template, job_id=sample_job.id) data = _notification_json(sample_email_template, job_id=sample_job.id)
notification_1 = Notification(**data) notification_1 = Notification(**data)
notification_2 = Notification(**data) notification_2 = Notification(**data)
dao_create_notification(notification_1) dao_create_notification(notification_1)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
dao_create_notification(notification_2) dao_create_notification(notification_2)
assert Notification.query.count() == 2 assert _get_notification_query_count() == 2
def test_save_notification_does_not_creates_history(sample_email_template, sample_job): def test_save_notification_does_not_creates_history(sample_email_template, sample_job):
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
data = _notification_json(sample_email_template, job_id=sample_job.id) data = _notification_json(sample_email_template, job_id=sample_job.id)
notification_1 = Notification(**data) notification_1 = Notification(**data)
dao_create_notification(notification_1) dao_create_notification(notification_1)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
assert NotificationHistory.query.count() == 0 assert _get_notification_history_query_count() == 0
def test_update_notification_with_research_mode_service_does_not_create_or_update_history( def test_update_notification_with_research_mode_service_does_not_create_or_update_history(
@@ -464,14 +519,14 @@ def test_update_notification_with_research_mode_service_does_not_create_or_updat
sample_template.service.research_mode = True sample_template.service.research_mode = True
notification = create_notification(template=sample_template) notification = create_notification(template=sample_template)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
assert NotificationHistory.query.count() == 0 assert _get_notification_history_query_count() == 0
notification.status = NotificationStatus.DELIVERED notification.status = NotificationStatus.DELIVERED
dao_update_notification(notification) dao_update_notification(notification)
assert Notification.query.one().status == NotificationStatus.DELIVERED assert _get_notification_query_one().status == NotificationStatus.DELIVERED
assert NotificationHistory.query.count() == 0 assert _get_notification_history_query_count() == 0
def test_not_save_notification_and_not_create_stats_on_commit_error( def test_not_save_notification_and_not_create_stats_on_commit_error(
@@ -479,26 +534,26 @@ def test_not_save_notification_and_not_create_stats_on_commit_error(
): ):
random_id = str(uuid.uuid4()) random_id = str(uuid.uuid4())
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
data = _notification_json(sample_template, job_id=random_id) data = _notification_json(sample_template, job_id=random_id)
notification = Notification(**data) notification = Notification(**data)
with pytest.raises(SQLAlchemyError): with pytest.raises(SQLAlchemyError):
dao_create_notification(notification) dao_create_notification(notification)
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
assert Job.query.get(sample_job.id).notifications_sent == 0 assert db.session.get(Job, sample_job.id).notifications_sent == 0
def test_save_notification_and_increment_job(sample_template, sample_job, sns_provider): def test_save_notification_and_increment_job(sample_template, sample_job, sns_provider):
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
data = _notification_json(sample_template, job_id=sample_job.id) data = _notification_json(sample_template, job_id=sample_job.id)
notification = Notification(**data) notification = Notification(**data)
dao_create_notification(notification) dao_create_notification(notification)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
notification_from_db = Notification.query.all()[0] notification_from_db = _get_notification_query_all()[0]
assert notification_from_db.id assert notification_from_db.id
assert "1" == notification_from_db.to assert "1" == notification_from_db.to
assert data["job_id"] == notification_from_db.job_id assert data["job_id"] == notification_from_db.job_id
@@ -510,21 +565,21 @@ def test_save_notification_and_increment_job(sample_template, sample_job, sns_pr
notification_2 = Notification(**data) notification_2 = Notification(**data)
dao_create_notification(notification_2) dao_create_notification(notification_2)
assert Notification.query.count() == 2 assert _get_notification_query_count() == 2
def test_save_notification_and_increment_correct_job(sample_template, sns_provider): def test_save_notification_and_increment_correct_job(sample_template, sns_provider):
job_1 = create_job(sample_template) job_1 = create_job(sample_template)
job_2 = create_job(sample_template) job_2 = create_job(sample_template)
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
data = _notification_json(sample_template, job_id=job_1.id) data = _notification_json(sample_template, job_id=job_1.id)
notification = Notification(**data) notification = Notification(**data)
dao_create_notification(notification) dao_create_notification(notification)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
notification_from_db = Notification.query.all()[0] notification_from_db = _get_notification_query_all()[0]
assert notification_from_db.id assert notification_from_db.id
assert "1" == notification_from_db.to assert "1" == notification_from_db.to
assert data["job_id"] == notification_from_db.job_id assert data["job_id"] == notification_from_db.job_id
@@ -537,14 +592,14 @@ def test_save_notification_and_increment_correct_job(sample_template, sns_provid
def test_save_notification_with_no_job(sample_template, sns_provider): def test_save_notification_with_no_job(sample_template, sns_provider):
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
data = _notification_json(sample_template) data = _notification_json(sample_template)
notification = Notification(**data) notification = Notification(**data)
dao_create_notification(notification) dao_create_notification(notification)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
notification_from_db = Notification.query.all()[0] notification_from_db = _get_notification_query_all()[0]
assert notification_from_db.id assert notification_from_db.id
assert "1" == notification_from_db.to assert "1" == notification_from_db.to
assert data["service"] == notification_from_db.service assert data["service"] == notification_from_db.service
@@ -592,7 +647,7 @@ def test_get_notification_by_id_when_notification_exists_for_different_service(
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(_get_notification_query_all()) == 0
create_notification(sample_template, client_reference=client_reference) create_notification(sample_template, client_reference=client_reference)
create_notification(sample_template, client_reference=client_reference) create_notification(sample_template, client_reference=client_reference)
create_notification(sample_template, client_reference="other-ref") create_notification(sample_template, client_reference="other-ref")
@@ -603,14 +658,14 @@ def test_get_notifications_by_reference(sample_template):
def test_save_notification_no_job_id(sample_template): def test_save_notification_no_job_id(sample_template):
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
data = _notification_json(sample_template) data = _notification_json(sample_template)
notification = Notification(**data) notification = Notification(**data)
dao_create_notification(notification) dao_create_notification(notification)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
notification_from_db = Notification.query.all()[0] notification_from_db = _get_notification_query_all()[0]
assert notification_from_db.id assert notification_from_db.id
assert "1" == notification_from_db.to assert "1" == notification_from_db.to
assert data["service"] == notification_from_db.service assert data["service"] == notification_from_db.service
@@ -687,13 +742,13 @@ def test_update_notification_sets_status(sample_notification):
assert sample_notification.status == NotificationStatus.CREATED assert sample_notification.status == NotificationStatus.CREATED
sample_notification.status = NotificationStatus.FAILED sample_notification.status = NotificationStatus.FAILED
dao_update_notification(sample_notification) dao_update_notification(sample_notification)
notification_from_db = Notification.query.get(sample_notification.id) notification_from_db = db.session.get(Notification, sample_notification.id)
assert notification_from_db.status == NotificationStatus.FAILED assert notification_from_db.status == NotificationStatus.FAILED
@freeze_time("2016-01-10") @freeze_time("2016-01-10")
def test_should_limit_notifications_return_by_day_limit_plus_one(sample_template): def test_should_limit_notifications_return_by_day_limit_plus_one(sample_template):
assert len(Notification.query.all()) == 0 assert len(_get_notification_query_all()) == 0
# create one notification a day between 1st and 9th, # create one notification a day between 1st and 9th,
# with assumption that the local timezone is EST # with assumption that the local timezone is EST
@@ -706,7 +761,7 @@ def test_should_limit_notifications_return_by_day_limit_plus_one(sample_template
status=NotificationStatus.FAILED, status=NotificationStatus.FAILED,
) )
all_notifications = Notification.query.all() all_notifications = _get_notification_query_all()
assert len(all_notifications) == 10 assert len(all_notifications) == 10
all_notifications = get_notifications_for_service( all_notifications = get_notifications_for_service(
@@ -722,19 +777,19 @@ def test_should_limit_notifications_return_by_day_limit_plus_one(sample_template
def test_creating_notification_does_not_add_notification_history(sample_template): def test_creating_notification_does_not_add_notification_history(sample_template):
create_notification(template=sample_template) create_notification(template=sample_template)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
assert NotificationHistory.query.count() == 0 assert _get_notification_history_query_count() == 0
def test_should_delete_notification_for_id(sample_template): def test_should_delete_notification_for_id(sample_template):
notification = create_notification(template=sample_template) notification = create_notification(template=sample_template)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
assert NotificationHistory.query.count() == 0 assert _get_notification_history_query_count() == 0
dao_delete_notifications_by_id(notification.id) dao_delete_notifications_by_id(notification.id)
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
def test_should_delete_notification_and_ignore_history_for_research_mode( def test_should_delete_notification_and_ignore_history_for_research_mode(
@@ -744,31 +799,32 @@ def test_should_delete_notification_and_ignore_history_for_research_mode(
notification = create_notification(template=sample_template) notification = create_notification(template=sample_template)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
dao_delete_notifications_by_id(notification.id) dao_delete_notifications_by_id(notification.id)
assert Notification.query.count() == 0 assert _get_notification_query_count() == 0
def test_should_delete_only_notification_with_id(sample_template): def test_should_delete_only_notification_with_id(sample_template):
notification_1 = create_notification(template=sample_template) notification_1 = create_notification(template=sample_template)
notification_2 = create_notification(template=sample_template) notification_2 = create_notification(template=sample_template)
assert Notification.query.count() == 2 assert _get_notification_query_count() == 2
dao_delete_notifications_by_id(notification_1.id) dao_delete_notifications_by_id(notification_1.id)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
assert Notification.query.first().id == notification_2.id stmt = select(Notification)
assert db.session.execute(stmt).scalars().first().id == notification_2.id
def test_should_delete_no_notifications_if_no_matching_ids(sample_template): def test_should_delete_no_notifications_if_no_matching_ids(sample_template):
create_notification(template=sample_template) create_notification(template=sample_template)
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
dao_delete_notifications_by_id(uuid.uuid4()) dao_delete_notifications_by_id(uuid.uuid4())
assert Notification.query.count() == 1 assert _get_notification_query_count() == 1
def _notification_json(sample_template, job_id=None, id=None, status=None): def _notification_json(sample_template, job_id=None, id=None, status=None):
@@ -814,16 +870,19 @@ def test_dao_timeout_notifications(sample_template):
temporary_failure_notifications = dao_timeout_notifications(utc_now()) temporary_failure_notifications = dao_timeout_notifications(utc_now())
assert len(temporary_failure_notifications) == 2 assert len(temporary_failure_notifications) == 2
assert Notification.query.get(created.id).status == NotificationStatus.CREATED assert db.session.get(Notification, created.id).status == NotificationStatus.CREATED
assert ( assert (
Notification.query.get(sending.id).status db.session.get(Notification, sending.id).status
== NotificationStatus.TEMPORARY_FAILURE == NotificationStatus.TEMPORARY_FAILURE
) )
assert ( assert (
Notification.query.get(pending.id).status db.session.get(Notification, pending.id).status
== NotificationStatus.TEMPORARY_FAILURE == NotificationStatus.TEMPORARY_FAILURE
) )
assert Notification.query.get(delivered.id).status == NotificationStatus.DELIVERED assert (
db.session.get(Notification, delivered.id).status
== NotificationStatus.DELIVERED
)
def test_dao_timeout_notifications_only_updates_for_older_notifications( def test_dao_timeout_notifications_only_updates_for_older_notifications(
@@ -842,8 +901,8 @@ def test_dao_timeout_notifications_only_updates_for_older_notifications(
temporary_failure_notifications = dao_timeout_notifications(utc_now()) temporary_failure_notifications = dao_timeout_notifications(utc_now())
assert len(temporary_failure_notifications) == 0 assert len(temporary_failure_notifications) == 0
assert Notification.query.get(sending.id).status == NotificationStatus.SENDING assert db.session.get(Notification, sending.id).status == NotificationStatus.SENDING
assert Notification.query.get(pending.id).status == NotificationStatus.PENDING assert db.session.get(Notification, pending.id).status == NotificationStatus.PENDING
def test_should_return_notifications_excluding_jobs_by_default( def test_should_return_notifications_excluding_jobs_by_default(
@@ -935,7 +994,7 @@ def test_get_notifications_created_by_api_or_csv_are_returned_correctly_excludin
key_type=sample_test_api_key.key_type, key_type=sample_test_api_key.key_type,
) )
all_notifications = Notification.query.all() all_notifications = _get_notification_query_all()
assert len(all_notifications) == 4 assert len(all_notifications) == 4
# returns all real API derived notifications # returns all real API derived notifications
@@ -982,7 +1041,7 @@ def test_get_notifications_with_a_live_api_key_type(
key_type=sample_test_api_key.key_type, key_type=sample_test_api_key.key_type,
) )
all_notifications = Notification.query.all() all_notifications = _get_notification_query_all()
assert len(all_notifications) == 4 assert len(all_notifications) == 4
# only those created with normal API key, no jobs # only those created with normal API key, no jobs
@@ -1114,7 +1173,7 @@ def test_should_exclude_test_key_notifications_by_default(
key_type=sample_test_api_key.key_type, key_type=sample_test_api_key.key_type,
) )
all_notifications = Notification.query.all() all_notifications = _get_notification_query_all()
assert len(all_notifications) == 4 assert len(all_notifications) == 4
all_notifications = get_notifications_for_service( all_notifications = get_notifications_for_service(
@@ -1757,10 +1816,10 @@ def test_dao_update_notifications_by_reference_updated_notifications(sample_temp
update_dict={"status": NotificationStatus.DELIVERED, "billable_units": 2}, update_dict={"status": NotificationStatus.DELIVERED, "billable_units": 2},
) )
assert updated_count == 2 assert updated_count == 2
updated_1 = Notification.query.get(notification_1.id) updated_1 = db.session.get(Notification, notification_1.id)
assert updated_1.billable_units == 2 assert updated_1.billable_units == 2
assert updated_1.status == NotificationStatus.DELIVERED assert updated_1.status == NotificationStatus.DELIVERED
updated_2 = Notification.query.get(notification_2.id) updated_2 = db.session.get(Notification, notification_2.id)
assert updated_2.billable_units == 2 assert updated_2.billable_units == 2
assert updated_2.status == NotificationStatus.DELIVERED assert updated_2.status == NotificationStatus.DELIVERED
@@ -1823,10 +1882,11 @@ def test_dao_update_notifications_by_reference_updates_history_when_one_of_two_n
assert updated_count == 1 assert updated_count == 1
assert updated_history_count == 1 assert updated_history_count == 1
assert ( assert (
Notification.query.get(notification2.id).status == NotificationStatus.DELIVERED db.session.get(Notification, notification2.id).status
== NotificationStatus.DELIVERED
) )
assert ( assert (
NotificationHistory.query.get(notification1.id).status db.session.get(NotificationHistory, notification1.id).status
== NotificationStatus.DELIVERED == NotificationStatus.DELIVERED
) )

35
tests/app/dao/test_users_dao.py
View File

@@ -3,6 +3,7 @@ from datetime import timedelta
import pytest import pytest
from freezegun import freeze_time from freezegun import freeze_time
from sqlalchemy import func, select
from sqlalchemy.exc import DataError from sqlalchemy.exc import DataError
from sqlalchemy.orm.exc import NoResultFound from sqlalchemy.orm.exc import NoResultFound
@@ -37,6 +38,21 @@ from tests.app.db import (
) )
def _get_user_query_count():
stmt = select(func.count(User.id))
return db.session.execute(stmt).scalar() or 0
def _get_user_query_first():
stmt = select(User)
return db.session.execute(stmt).scalars().first()
def _get_verify_code_query_count():
stmt = select(func.count(VerifyCode.id))
return db.session.execute(stmt).scalar() or 0
@freeze_time("2020-01-28T12:00:00") @freeze_time("2020-01-28T12:00:00")
@pytest.mark.parametrize( @pytest.mark.parametrize(
"phone_number, expected_phone_number", "phone_number, expected_phone_number",
@@ -55,8 +71,8 @@ def test_create_user(notify_db_session, phone_number, expected_phone_number):
} }
user = User(**data) user = User(**data)
save_model_user(user, password="password", validated_email_access=True) save_model_user(user, password="password", validated_email_access=True)
assert User.query.count() == 1 assert _get_user_query_count() == 1
user_query = User.query.first() user_query = _get_user_query_first()
assert user_query.email_address == email assert user_query.email_address == email
assert user_query.id == user.id assert user_query.id == user.id
assert user_query.mobile_number == expected_phone_number assert user_query.mobile_number == expected_phone_number
@@ -68,7 +84,7 @@ def test_get_all_users(notify_db_session):
create_user(email="1@test.com") create_user(email="1@test.com")
create_user(email="2@test.com") create_user(email="2@test.com")
assert User.query.count() == 2 assert _get_user_query_count() == 2
assert len(get_user_by_id()) == 2 assert len(get_user_by_id()) == 2
@@ -89,9 +105,9 @@ def test_get_user_invalid_id(notify_db_session):
def test_delete_users(sample_user): def test_delete_users(sample_user):
assert User.query.count() == 1 assert _get_user_query_count() == 1
delete_model_user(sample_user) delete_model_user(sample_user)
assert User.query.count() == 0 assert _get_user_query_count() == 0
def test_increment_failed_login_should_increment_failed_logins(sample_user): def test_increment_failed_login_should_increment_failed_logins(sample_user):
@@ -127,9 +143,9 @@ def test_get_user_by_email_is_case_insensitive(sample_user):
def test_should_delete_all_verification_codes_more_than_one_day_old(sample_user): def test_should_delete_all_verification_codes_more_than_one_day_old(sample_user):
make_verify_code(sample_user, age=timedelta(hours=24), code="54321") make_verify_code(sample_user, age=timedelta(hours=24), code="54321")
make_verify_code(sample_user, age=timedelta(hours=24), code="54321") make_verify_code(sample_user, age=timedelta(hours=24), code="54321")
assert VerifyCode.query.count() == 2 assert _get_verify_code_query_count() == 2
delete_codes_older_created_more_than_a_day_ago() delete_codes_older_created_more_than_a_day_ago()
assert VerifyCode.query.count() == 0 assert _get_verify_code_query_count() == 0
def test_should_not_delete_verification_codes_less_than_one_day_old(sample_user): def test_should_not_delete_verification_codes_less_than_one_day_old(sample_user):
@@ -138,9 +154,10 @@ def test_should_not_delete_verification_codes_less_than_one_day_old(sample_user)
) )
make_verify_code(sample_user, age=timedelta(hours=24), code="54321") make_verify_code(sample_user, age=timedelta(hours=24), code="54321")
assert VerifyCode.query.count() == 2 assert _get_verify_code_query_count() == 2
delete_codes_older_created_more_than_a_day_ago() delete_codes_older_created_more_than_a_day_ago()
assert VerifyCode.query.one()._code == "12345" stmt = select(VerifyCode)
assert db.session.execute(stmt).scalars().one()._code == "12345"
def make_verify_code(user, age=None, expiry_age=None, code="12335", code_used=False): def make_verify_code(user, age=None, expiry_age=None, code="12335", code_used=False):