darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-02 17:31:14 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix test code

Browse Source
This commit is contained in:
Kenneth Kehl
2024-10-21 08:01:57 -07:00
parent 3d7ab71555
commit 38a2a87a15
3 changed files with 17 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
app/dao/complaint_dao.py
View File

@@ -16,13 +16,15 @@ def save_complaint(complaint):
def fetch_paginated_complaints(page=1): def fetch_paginated_complaints(page=1):
# return Complaint.query.order_by(desc(Complaint.created_at)).paginate(
# page=page, per_page=current_app.config["PAGE_SIZE"]
# )
page_size = current_app.config["PAGE_SIZE"] page_size = current_app.config["PAGE_SIZE"]
total_count = db.session.scalar(select(func.count()).select_from(Complaint)) total_count = db.session.scalar(select(func.count()).select_from(Complaint))
offset = (page - 1) * page_size offset = (page - 1) * page_size
stmt = select(Complaint).order_by(desc(Complaint.created_at)).offset(offset).limit(page_size) stmt = (
select(Complaint)
.order_by(desc(Complaint.created_at))
.offset(offset)
.limit(page_size)
)
result = db.session.execute(stmt).scalars().all() result = db.session.execute(stmt).scalars().all()
pagination = Pagination(result, page=page, per_page=page_size, total=total_count) pagination = Pagination(result, page=page, per_page=page_size, total=total_count)
return pagination return pagination

5
tests/app/dao/test_inbound_numbers_dao.py
View File

@@ -1,6 +1,8 @@
import pytest import pytest
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError from sqlalchemy.exc import IntegrityError
from app import db
from app.dao.inbound_numbers_dao import ( from app.dao.inbound_numbers_dao import (
dao_allocate_number_for_service, dao_allocate_number_for_service,
dao_get_available_inbound_numbers, dao_get_available_inbound_numbers,
@@ -35,7 +37,8 @@ def test_set_service_id_on_inbound_number(notify_db_session, sample_inbound_numb
dao_set_inbound_number_to_service(service.id, numbers[0]) dao_set_inbound_number_to_service(service.id, numbers[0])
res = InboundNumber.query.filter(InboundNumber.service_id == service.id).all() stmt = select(InboundNumber).filter(InboundNumber.service_id == service.id)
res = db.session.execute(stmt).all()
assert len(res) == 1 assert len(res) == 1
assert res[0].service_id == service.id assert res[0].service_id == service.id

10
tests/app/dao/test_inbound_sms_dao.py
View File

@@ -2,6 +2,7 @@ from datetime import datetime
from itertools import product from itertools import product
from freezegun import freeze_time from freezegun import freeze_time
from sqlalchemy import select
from app import db from app import db
from app.dao.inbound_sms_dao import ( from app.dao.inbound_sms_dao import (
@@ -141,7 +142,8 @@ def test_should_delete_inbound_sms_according_to_data_retention(notify_db_session
deleted_count = delete_inbound_sms_older_than_retention() deleted_count = delete_inbound_sms_older_than_retention()
history = InboundSmsHistory.query.all() stmt = select(InboundSmsHistory)
history = db.session.execute(stmt).all()
assert len(history) == 7 assert len(history) == 7
# four deleted for the 3-day service, two for the default seven days one, one for the 30 day # four deleted for the 3-day service, two for the default seven days one, one for the 30 day
@@ -171,7 +173,8 @@ def test_insert_into_inbound_sms_history_when_deleting_inbound_sms(sample_servic
create_inbound_sms(sample_service, created_at=datetime(2019, 12, 19, 20, 19)) create_inbound_sms(sample_service, created_at=datetime(2019, 12, 19, 20, 19))
delete_inbound_sms_older_than_retention() delete_inbound_sms_older_than_retention()
history = InboundSmsHistory.query.all() stmt = select(InboundSmsHistory)
history = db.session.execute(stmt).all()
assert len(history) == 1 assert len(history) == 1
for key_name in [ for key_name in [
@@ -226,7 +229,8 @@ def test_delete_inbound_sms_older_than_retention_does_nothing_when_database_conf
delete_inbound_sms_older_than_retention() delete_inbound_sms_older_than_retention()
history = InboundSmsHistory.query.all() stmt = select(InboundSmsHistory)
history = db.session.execute(stmt).all()
assert len(history) == 1 assert len(history) == 1
assert history[0].id == inbound_sms_id assert history[0].id == inbound_sms_id