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

upgrade test queries to sqlalchemy 2.0

Browse Source
This commit is contained in:
Kenneth Kehl
2024-10-30 07:41:26 -07:00
parent 99c199b9d2
commit b2ed52ffd4
1 changed files with 9 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
tests/app/celery/test_tasks.py
View File

@@ -8,9 +8,10 @@ import requests_mock
from celery.exceptions import Retry from celery.exceptions import Retry
from freezegun import freeze_time from freezegun import freeze_time
from requests import RequestException from requests import RequestException
from sqlalchemy import func, select
from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.exc import SQLAlchemyError
from app import encryption from app import db, encryption
from app.celery import provider_tasks, tasks from app.celery import provider_tasks, tasks
from app.celery.tasks import ( from app.celery.tasks import (
get_recipient_csv_and_template_and_sender_id, get_recipient_csv_and_template_and_sender_id,
@@ -1166,7 +1167,13 @@ def test_process_incomplete_job_sms(mocker, sample_template):
create_notification(sample_template, job, 0) create_notification(sample_template, job, 0)
create_notification(sample_template, job, 1) create_notification(sample_template, job, 1)
assert Notification.query.filter(Notification.job_id == job.id).count() == 2 stmt = (
select(func.count())
.select_from(Notification)
.where(Notification.job_id == job.id)
)
result = db.session.execute(stmt)
assert result.rowcount == 2
process_incomplete_job(str(job.id)) process_incomplete_job(str(job.id))