darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-23 15:56:45 -04:00
Code Issues Packages Projects Releases Wiki Activity

fix test

Browse Source
This commit is contained in:
Kenneth Kehl
2025-01-31 12:30:47 -08:00
parent c266cdaefe
commit b5b7885ce4
2 changed files with 8 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
app/dao/notifications_dao.py
View File

@@ -268,35 +268,25 @@ def get_notifications_for_job(
def get_recent_notifications_for_job( def get_recent_notifications_for_job(
service_id, job_id, filter_dict=None, page=1, page_size=None, status=None service_id, job_id, filter_dict=None, page=1, page_size=None
): ):
if page_size is None: if page_size is None:
page_size = current_app.config["PAGE_SIZE"] page_size = current_app.config["PAGE_SIZE"]
if status is None: stmt = select(Notification).where(
stmt = select(Notification).where( Notification.service_id == service_id,
Notification.service_id == service_id, Notification.job_id == job_id,
Notification.job_id == job_id, )
)
else:
stmt = select(Notification).where(
Notification.service_id == service_id,
Notification.job_id == job_id,
Notification.status == status,
)
stmt = _filter_query(stmt, filter_dict) stmt = _filter_query(stmt, filter_dict)
stmt = stmt.order_by(desc(Notification.job_row_number)) stmt = stmt.order_by(desc(Notification.job_row_number))
print(f"STMT {stmt}") print(f"STMT {stmt}")
results = db.session.execute(stmt).scalars().all() results = db.session.execute(stmt).scalars().all()
print(f"RESULTS {results}")
page_size = current_app.config["PAGE_SIZE"] page_size = current_app.config["PAGE_SIZE"]
offset = (page - 1) * page_size offset = (page - 1) * page_size
paginated_results = results[offset : offset + page_size] paginated_results = results[offset : offset + page_size]
print(
f"PAGINATED RESULTS {paginated_results} page_size {page_size} offset {offset}"
)
pagination = Pagination(paginated_results, page, page_size, len(results)) pagination = Pagination(paginated_results, page, page_size, len(results))
return pagination return pagination

3
tests/app/job/test_rest.py
View File

@@ -518,11 +518,12 @@ def test_get_recent_notifications_for_job_in_reverse_order_of_job_number(
assert resp["notifications"][0]["status"] == "virus-scan-failed" assert resp["notifications"][0]["status"] == "virus-scan-failed"
assert resp["notifications"][0]["job_row_number"] == 13 assert resp["notifications"][0]["job_row_number"] == 13
filter_dict = {"status": NotificationStatus.DELIVERED}
resp = admin_request.get( resp = admin_request.get(
"job.get_recent_notifications_for_service_job", "job.get_recent_notifications_for_service_job",
service_id=main_job.service_id, service_id=main_job.service_id,
job_id=main_job.id, job_id=main_job.id,
status=NotificationStatus.DELIVERED, filter_dict=filter_dict,
) )
assert len(resp["notifications"]) == 1 assert len(resp["notifications"]) == 1