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

Cleaning up tests.

Browse Source 
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
This commit is contained in:
Cliff Hill
2024-02-28 12:43:33 -05:00
parent 9523cc1d97
commit 8c6046b03b
33 changed files with 268 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/dao/fact_billing_dao.py
+4 -2
View File
@@ -408,7 +408,7 @@ def _query_for_billing_data(notification_type, start_date, end_date, service):
func.count().label("notifications_sent"),
)
.filter(
NotificationAllTimeView.status.in_(NotificationStatus.sent_emails),
NotificationAllTimeView.status.in_(NotificationStatus.sent_email_types),
NotificationAllTimeView.key_type.in_((KeyType.NORMAL, KeyType.TEAM)),
NotificationAllTimeView.created_at >= start_date,
NotificationAllTimeView.created_at < end_date,
@@ -440,7 +440,9 @@ def _query_for_billing_data(notification_type, start_date, end_date, service):
func.count().label("notifications_sent"),
)
.filter(
NotificationAllTimeView.status.in_(NotificationStatus.billable_sms),
NotificationAllTimeView.status.in_(
NotificationStatus.billable_sms_types
),
NotificationAllTimeView.key_type.in_((KeyType.NORMAL, KeyType.TEAM)),
NotificationAllTimeView.created_at >= start_date,
NotificationAllTimeView.created_at < end_date,
app/delivery/send_to_providers.py
+1 -1
View File
@@ -155,7 +155,7 @@ def send_email_to_provider(notification):
def update_notification_to_sending(notification, provider):
notification.sent_at = datetime.utcnow()
notification.sent_by = provider.name
if notification.status not in NotificationStatus.completed:
if notification.status not in NotificationStatus.completed_types:
notification.status = NotificationStatus.SENDING
dao_update_notification(notification)
app/enums.py
+7 -7
View File
@@ -52,7 +52,7 @@ class NotificationStatus(Enum):
VIRUS_SCAN_FAILED = "virus-scan-failed"
@property
def failed(self) -> tuple["NotificationStatus", ...]:
def failed_types(self) -> tuple["NotificationStatus", ...]:
cls = type(self)
return (
cls.TECHNICAL_FAILURE,
@@ -63,7 +63,7 @@ class NotificationStatus(Enum):
)
@property
def completed(self) -> tuple["NotificationStatus", ...]:
def completed_types(self) -> tuple["NotificationStatus", ...]:
cls = type(self)
return (
cls.SENT,
@@ -76,12 +76,12 @@ class NotificationStatus(Enum):
)
@property
def success(self) -> tuple["NotificationStatus", ...]:
def success_types(self) -> tuple["NotificationStatus", ...]:
cls = type(self)
return (cls.SENT, cls.DELIVERED)
@property
def billable(self) -> tuple["NotificationStatus", ...]:
def billable_types(self) -> tuple["NotificationStatus", ...]:
cls = type(self)
return (
cls.SENDING,
@@ -94,7 +94,7 @@ class NotificationStatus(Enum):
)
@property
def billable_sms(self) -> tuple["NotificationStatus", ...]:
def billable_sms_types(self) -> tuple["NotificationStatus", ...]:
cls = type(self)
return (
cls.SENDING,
@@ -106,7 +106,7 @@ class NotificationStatus(Enum):
)
@property
def sent_emails(self) -> tuple["NotificationStatus", ...]:
def sent_email_types(self) -> tuple["NotificationStatus", ...]:
cls = type(self)
return (
cls.SENDING,
@@ -116,7 +116,7 @@ class NotificationStatus(Enum):
)
@property
def non_billable(self) -> tuple["NotificationStatus", ...]:
def non_billable_types(self) -> tuple["NotificationStatus", ...]:
self._non_billable: tuple["NotificationStatus", ...]
try:
return self._non_billable
app/models.py
+2 -2
View File
@@ -1581,7 +1581,7 @@ class Notification(db.Model):
self._personalisation = encryption.encrypt(personalisation or {})
def completed_at(self):
if self.status in NotificationStatus.completed:
if self.status in NotificationStatus.completed_types:
return self.updated_at.strftime(DATETIME_FORMAT)
return None
@@ -1621,7 +1621,7 @@ class Notification(db.Model):
def _substitute_status_str(_status):
return (
NotificationStatus.failed
NotificationStatus.failed_types
if _status == NotificationStatus.FAILED
else [_status]
)
tests/__init__.py
+3 -2
View File
@@ -5,10 +5,11 @@ from notifications_python_client.authentication import create_jwt_token
from app.dao.api_key_dao import save_model_api_key
from app.dao.services_dao import dao_fetch_service_by_id
from app.models import KEY_TYPE_NORMAL, ApiKey
from app.enums import KeyType
from app.models import ApiKey
def create_service_authorization_header(service_id, key_type=KEY_TYPE_NORMAL):
def create_service_authorization_header(service_id, key_type=KeyType.NORMAL):
client_id = str(service_id)
secrets = ApiKey.query.filter_by(service_id=service_id, key_type=key_type).all()
tests/app/celery/test_reporting_tasks.py
+7 -10
View File
@@ -13,14 +13,11 @@ from app.celery.reporting_tasks import (
)
from app.config import QueueNames
from app.dao.fact_billing_dao import get_rate
from app.enums import NotificationType, KeyType
from app.models import (
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
FactBilling,
FactNotificationStatus,
Notification,
NotificationType,
)
from tests.app.db import (
create_notification,
@@ -431,12 +428,12 @@ def test_create_nightly_notification_status_for_service_and_day(notify_db_sessio
# team API key notifications are included
create_notification(
template=second_template, status="sending", key_type=KEY_TYPE_TEAM
template=second_template, status="sending", key_type=KeyType.TEAM
)
# test notifications are ignored
create_notification(
template=second_template, status="sending", key_type=KEY_TYPE_TEST
template=second_template, status="sending", key_type=KeyType.TEST
)
# historical notifications are included
@@ -471,7 +468,7 @@ def test_create_nightly_notification_status_for_service_and_day(notify_db_sessio
assert email_delivered_row.notification_type == "email"
assert email_delivered_row.notification_status == "delivered"
assert email_delivered_row.notification_count == 1
assert email_delivered_row.key_type == KEY_TYPE_NORMAL
assert email_delivered_row.key_type == KeyType.NORMAL
email_sending_row = new_fact_data[1]
assert email_sending_row.template_id == second_template.id
@@ -479,7 +476,7 @@ def test_create_nightly_notification_status_for_service_and_day(notify_db_sessio
assert email_sending_row.notification_type == "email"
assert email_sending_row.notification_status == "failed"
assert email_sending_row.notification_count == 1
assert email_sending_row.key_type == KEY_TYPE_NORMAL
assert email_sending_row.key_type == KeyType.NORMAL
email_failure_row = new_fact_data[2]
assert email_failure_row.local_date == process_day
@@ -489,7 +486,7 @@ def test_create_nightly_notification_status_for_service_and_day(notify_db_sessio
assert email_failure_row.notification_type == "email"
assert email_failure_row.notification_status == "sending"
assert email_failure_row.notification_count == 1
assert email_failure_row.key_type == KEY_TYPE_TEAM
assert email_failure_row.key_type == KeyType.TEAM
sms_delivered_row = new_fact_data[3]
assert sms_delivered_row.template_id == first_template.id
@@ -497,7 +494,7 @@ def test_create_nightly_notification_status_for_service_and_day(notify_db_sessio
assert sms_delivered_row.notification_type == "sms"
assert sms_delivered_row.notification_status == "delivered"
assert sms_delivered_row.notification_count == 1
assert sms_delivered_row.key_type == KEY_TYPE_NORMAL
assert sms_delivered_row.key_type == KeyType.NORMAL
def test_create_nightly_notification_status_for_service_and_day_overwrites_old_data(
tests/app/celery/test_scheduled_tasks.py
+20 -25
View File
@@ -19,12 +19,7 @@ from app.celery.scheduled_tasks import (
)
from app.config import QueueNames, Test
from app.dao.jobs_dao import dao_get_job_by_id
from app.models import (
JOB_STATUS_ERROR,
JOB_STATUS_FINISHED,
JOB_STATUS_IN_PROGRESS,
JOB_STATUS_PENDING,
)
from app.enums import JobStatus
from tests.app import load_example_csv
from tests.app.db import create_job, create_notification, create_template
@@ -156,7 +151,7 @@ def test_check_job_status_task_calls_process_incomplete_jobs(mocker, sample_temp
notification_count=3,
created_at=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
create_notification(template=sample_template, job=job)
check_job_status()
@@ -174,7 +169,7 @@ def test_check_job_status_task_calls_process_incomplete_jobs_when_scheduled_job_
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
check_job_status()
@@ -190,7 +185,7 @@ def test_check_job_status_task_calls_process_incomplete_jobs_for_pending_schedul
notification_count=3,
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_PENDING,
job_status=JobStatus.PENDING,
)
check_job_status()
@@ -207,7 +202,7 @@ def test_check_job_status_task_does_not_call_process_incomplete_jobs_for_non_sch
template=sample_template,
notification_count=3,
created_at=datetime.utcnow() - timedelta(hours=2),
job_status=JOB_STATUS_PENDING,
job_status=JobStatus.PENDING,
)
check_job_status()
@@ -224,7 +219,7 @@ def test_check_job_status_task_calls_process_incomplete_jobs_for_multiple_jobs(
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
job_2 = create_job(
template=sample_template,
@@ -232,7 +227,7 @@ def test_check_job_status_task_calls_process_incomplete_jobs_for_multiple_jobs(
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
check_job_status()
@@ -249,21 +244,21 @@ def test_check_job_status_task_only_sends_old_tasks(mocker, sample_template):
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
create_job(
template=sample_template,
notification_count=3,
created_at=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=29),
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
create_job(
template=sample_template,
notification_count=3,
created_at=datetime.utcnow() - timedelta(minutes=50),
scheduled_for=datetime.utcnow() - timedelta(minutes=29),
job_status=JOB_STATUS_PENDING,
job_status=JobStatus.PENDING,
)
check_job_status()
@@ -279,21 +274,21 @@ def test_check_job_status_task_sets_jobs_to_error(mocker, sample_template):
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
job_2 = create_job(
template=sample_template,
notification_count=3,
created_at=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=29),
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
check_job_status()
# job 2 not in celery task
mock_celery.assert_called_once_with([[str(job.id)]], queue=QueueNames.JOBS)
assert job.job_status == JOB_STATUS_ERROR
assert job_2.job_status == JOB_STATUS_IN_PROGRESS
assert job.job_status == JobStatus.ERROR
assert job_2.job_status == JobStatus.IN_PROGRESS
def test_replay_created_notifications(notify_db_session, sample_service, mocker):
@@ -352,14 +347,14 @@ def test_check_job_status_task_does_not_raise_error(sample_template):
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
)
create_job(
template=sample_template,
notification_count=3,
created_at=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
)
check_job_status()
@@ -389,7 +384,7 @@ def test_check_for_missing_rows_in_completed_jobs_ignores_old_and_new_jobs(
job = create_job(
template=sample_email_template,
notification_count=5,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - offset,
)
for i in range(0, 4):
@@ -411,7 +406,7 @@ def test_check_for_missing_rows_in_completed_jobs(mocker, sample_email_template)
job = create_job(
template=sample_email_template,
notification_count=5,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - timedelta(minutes=20),
)
for i in range(0, 4):
@@ -438,7 +433,7 @@ def test_check_for_missing_rows_in_completed_jobs_calls_save_email(
job = create_job(
template=sample_email_template,
notification_count=5,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - timedelta(minutes=20),
)
for i in range(0, 4):
@@ -468,7 +463,7 @@ def test_check_for_missing_rows_in_completed_jobs_uses_sender_id(
job = create_job(
template=sample_email_template,
notification_count=5,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - timedelta(minutes=20),
)
for i in range(0, 4):
tests/app/celery/test_test_key_tasks.py
+5 -4
View File
@@ -12,7 +12,8 @@ from app.celery.test_key_tasks import (
sns_callback,
)
from app.config import QueueNames
from app.models import NOTIFICATION_DELIVERED, NOTIFICATION_FAILED, Notification
from app.enums import NotificationStatus
from app.models import Notification
from tests.conftest import Matcher
dvla_response_file_matcher = Matcher(
@@ -28,7 +29,7 @@ def test_make_sns_callback(notify_api, rmock, mocker):
)
n = Notification()
n.id = 1234
n.status = NOTIFICATION_DELIVERED
n.status = NotificationStatus.DELIVERED
get_notification_by_id.return_value = n
rmock.request("POST", endpoint, json={"status": "success"}, status_code=200)
send_sms_response("sns", "1234")
@@ -45,7 +46,7 @@ def test_callback_logs_on_api_call_failure(notify_api, rmock, mocker):
)
n = Notification()
n.id = 1234
n.status = NOTIFICATION_FAILED
n.status = NotificationStatus.FAILED
get_notification_by_id.return_value = n
rmock.request(
@@ -81,7 +82,7 @@ def test_delivered_sns_callback(mocker):
)
n = Notification()
n.id = 1234
n.status = NOTIFICATION_DELIVERED
n.status = NotificationStatus.DELIVERED
get_notification_by_id.return_value = n
data = json.loads(sns_callback("1234"))
tests/app/conftest.py
+13 -19
View File
@@ -17,14 +17,10 @@ from app.dao.organization_dao import dao_create_organization
from app.dao.services_dao import dao_add_user_to_service, dao_create_service
from app.dao.templates_dao import dao_create_template
from app.dao.users_dao import create_secret_code, create_user_code
from app.enums import KeyType, NotificationStatus, ServicePermissionType, TemplateType, RecipientType
from app.history_meta import create_history
from app.models import (
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
NOTIFICATION_STATUS_TYPES_COMPLETED,
ApiKey,
GuestListRecipientType,
InvitedUser,
Job,
Notification,
@@ -36,10 +32,8 @@ from app.models import (
Service,
ServiceEmailReplyTo,
ServiceGuestList,
ServicePermissionType,
Template,
TemplateHistory,
TemplateType,
)
from tests import create_admin_authorization_header
from tests.app.db import (
@@ -77,7 +71,7 @@ def create_sample_notification(
billable_units=1,
personalisation=None,
api_key=None,
key_type=KEY_TYPE_NORMAL,
key_type=KeyType.NORMAL,
sent_by=None,
international=False,
client_reference=None,
@@ -129,7 +123,7 @@ def create_sample_notification(
"key_type": api_key.key_type if api_key else key_type,
"sent_by": sent_by,
"updated_at": created_at
if status in NOTIFICATION_STATUS_TYPES_COMPLETED
if status in NotificationStatus.completed_types
else None,
"client_reference": client_reference,
"rate_multiplier": rate_multiplier,
@@ -345,7 +339,7 @@ def sample_api_key(notify_db_session):
"service": service,
"name": uuid.uuid4(),
"created_by": service.created_by,
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
}
api_key = ApiKey(**data)
save_model_api_key(api_key)
@@ -356,14 +350,14 @@ def sample_api_key(notify_db_session):
def sample_test_api_key(sample_api_key):
service = create_service(check_if_service_exists=True)
return create_api_key(service, key_type=KEY_TYPE_TEST)
return create_api_key(service, key_type=KeyType.TEST)
@pytest.fixture(scope="function")
def sample_team_api_key(sample_api_key):
service = create_service(check_if_service_exists=True)
return create_api_key(service, key_type=KEY_TYPE_TEAM)
return create_api_key(service, key_type=KeyType.TEAM)
@pytest.fixture(scope="function")
@@ -426,7 +420,7 @@ def sample_notification_with_job(notify_db_session):
billable_units=1,
personalisation=None,
api_key=None,
key_type=KEY_TYPE_NORMAL,
key_type=KeyType.NORMAL,
)
@@ -437,10 +431,10 @@ def sample_notification(notify_db_session):
template = create_template(service=service)
api_key = ApiKey.query.filter(
ApiKey.service == template.service, ApiKey.key_type == KEY_TYPE_NORMAL
ApiKey.service == template.service, ApiKey.key_type == KeyType.NORMAL
).first()
if not api_key:
api_key = create_api_key(template.service, key_type=KEY_TYPE_NORMAL)
api_key = create_api_key(template.service, key_type=KeyType.NORMAL)
notification_id = uuid.uuid4()
to = "+447700900855"
@@ -504,7 +498,7 @@ def sample_email_notification(notify_db_session):
"personalisation": None,
"notification_type": template.template_type,
"api_key_id": None,
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
"job_row_number": 1,
}
notification = Notification(**data)
@@ -517,7 +511,7 @@ def sample_notification_history(notify_db_session, sample_template):
created_at = datetime.utcnow()
sent_at = datetime.utcnow()
notification_type = sample_template.template_type
api_key = create_api_key(sample_template.service, key_type=KEY_TYPE_NORMAL)
api_key = create_api_key(sample_template.service, key_type=KeyType.NORMAL)
notification_history = NotificationHistory(
id=uuid.uuid4(),
@@ -527,7 +521,7 @@ def sample_notification_history(notify_db_session, sample_template):
status="created",
created_at=created_at,
notification_type=notification_type,
key_type=KEY_TYPE_NORMAL,
key_type=KeyType.NORMAL,
api_key=api_key,
api_key_id=api_key and api_key.id,
sent_at=sent_at,
@@ -844,7 +838,7 @@ def notify_service(notify_db_session, sample_user):
def sample_service_guest_list(notify_db_session):
service = create_service(check_if_service_exists=True)
guest_list_user = ServiceGuestList.from_string(
service.id, GuestListRecipientType.EMAIL, "guest_list_user@digital.fake.gov"
service.id, RecipientType.EMAIL, "guest_list_user@digital.fake.gov"
)
notify_db_session.add(guest_list_user)
tests/app/dao/notification_dao/test_notification_dao.py
+24 -35
View File
@@ -28,19 +28,8 @@ from app.dao.notifications_dao import (
update_notification_status_by_id,
update_notification_status_by_reference,
)
from app.models import (
JOB_STATUS_IN_PROGRESS,
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
NOTIFICATION_DELIVERED,
NOTIFICATION_SENT,
NOTIFICATION_STATUS_TYPES,
Job,
Notification,
NotificationHistory,
NotificationType,
)
from app.enums import JobStatus, KeyType, NotificationStatus, NotificationType
from app.models import Job, Notification, NotificationHistory
from tests.app.db import (
create_ft_notification_status,
create_job,
@@ -139,13 +128,13 @@ def test_should_not_update_status_by_reference_if_from_country_with_no_delivery_
sample_template,
):
notification = create_notification(
sample_template, status=NOTIFICATION_SENT, reference="foo"
sample_template, status=NotificationStatus.SENT, reference="foo"
)
res = update_notification_status_by_reference("foo", "failed")
assert res is None
assert notification.status == NOTIFICATION_SENT
assert notification.status == NotificationStatus.SENT
def test_should_not_update_status_by_id_if_sent_to_country_with_unknown_delivery_receipts(
@@ -153,7 +142,7 @@ def test_should_not_update_status_by_id_if_sent_to_country_with_unknown_delivery
):
notification = create_notification(
sample_template,
status=NOTIFICATION_SENT,
status=NotificationStatus.SENT,
international=True,
phone_prefix="249", # sudan has no delivery receipts (or at least, that we know about)
)
@@ -161,7 +150,7 @@ def test_should_not_update_status_by_id_if_sent_to_country_with_unknown_delivery
res = update_notification_status_by_id(notification.id, "delivered")
assert res is None
assert notification.status == NOTIFICATION_SENT
assert notification.status == NotificationStatus.SENT
def test_should_not_update_status_by_id_if_sent_to_country_with_carrier_delivery_receipts(
@@ -169,7 +158,7 @@ def test_should_not_update_status_by_id_if_sent_to_country_with_carrier_delivery
):
notification = create_notification(
sample_template,
status=NOTIFICATION_SENT,
status=NotificationStatus.SENT,
international=True,
phone_prefix="1", # americans only have carrier delivery receipts
)
@@ -177,7 +166,7 @@ def test_should_not_update_status_by_id_if_sent_to_country_with_carrier_delivery
res = update_notification_status_by_id(notification.id, "delivered")
assert res is None
assert notification.status == NOTIFICATION_SENT
assert notification.status == NotificationStatus.SENT
def test_should_not_update_status_by_id_if_sent_to_country_with_delivery_receipts(
@@ -185,7 +174,7 @@ def test_should_not_update_status_by_id_if_sent_to_country_with_delivery_receipt
):
notification = create_notification(
sample_template,
status=NOTIFICATION_SENT,
status=NotificationStatus.SENT,
international=True,
phone_prefix="7", # russians have full delivery receipts
)
@@ -193,7 +182,7 @@ def test_should_not_update_status_by_id_if_sent_to_country_with_delivery_receipt
res = update_notification_status_by_id(notification.id, "delivered")
assert res == notification
assert notification.status == NOTIFICATION_DELIVERED
assert notification.status == NotificationStatus.DELIVERED
def test_should_not_update_status_by_reference_if_not_sending(sample_template):
@@ -540,15 +529,15 @@ def test_get_all_notifications_for_job_by_status(sample_job):
get_notifications_for_job, sample_job.service.id, sample_job.id
)
for status in NOTIFICATION_STATUS_TYPES:
for status in NotificationStatus:
create_notification(template=sample_job.template, job=sample_job, status=status)
# assert len(notifications().items) == len(NOTIFICATION_STATUS_TYPES)
# assert len(notifications().items) == len(NotificationStatus)
assert len(notifications(filter_dict={"status": status}).items) == 1
assert (
len(notifications(filter_dict={"status": NOTIFICATION_STATUS_TYPES[:3]}).items)
len(notifications(filter_dict={"status": NotificationStatus[:3]}).items)
== 3
)
@@ -681,7 +670,7 @@ def _notification_json(sample_template, job_id=None, id=None, status=None):
"created_at": datetime.utcnow(),
"billable_units": 1,
"notification_type": sample_template.template_type,
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
}
if job_id:
data.update({"job_id": job_id})
@@ -861,13 +850,13 @@ def test_get_notifications_with_a_live_api_key_type(
# only those created with normal API key, no jobs
all_notifications = get_notifications_for_service(
sample_job.service.id, limit_days=1, key_type=KEY_TYPE_NORMAL
sample_job.service.id, limit_days=1, key_type=KeyType.NORMAL
).items
assert len(all_notifications) == 1
# only those created with normal API key, with jobs
all_notifications = get_notifications_for_service(
sample_job.service.id, limit_days=1, include_jobs=True, key_type=KEY_TYPE_NORMAL
sample_job.service.id, limit_days=1, include_jobs=True, key_type=KeyType.NORMAL
).items
assert len(all_notifications) == 2
@@ -899,13 +888,13 @@ def test_get_notifications_with_a_test_api_key_type(
# only those created with test API key, no jobs
all_notifications = get_notifications_for_service(
sample_job.service_id, limit_days=1, key_type=KEY_TYPE_TEST
sample_job.service_id, limit_days=1, key_type=KeyType.TEST
).items
assert len(all_notifications) == 1
# only those created with test API key, no jobs, even when requested
all_notifications = get_notifications_for_service(
sample_job.service_id, limit_days=1, include_jobs=True, key_type=KEY_TYPE_TEST
sample_job.service_id, limit_days=1, include_jobs=True, key_type=KeyType.TEST
).items
assert len(all_notifications) == 1
@@ -937,13 +926,13 @@ def test_get_notifications_with_a_team_api_key_type(
# only those created with team API key, no jobs
all_notifications = get_notifications_for_service(
sample_job.service_id, limit_days=1, key_type=KEY_TYPE_TEAM
sample_job.service_id, limit_days=1, key_type=KeyType.TEAM
).items
assert len(all_notifications) == 1
# only those created with team API key, no jobs, even when requested
all_notifications = get_notifications_for_service(
sample_job.service_id, limit_days=1, include_jobs=True, key_type=KEY_TYPE_TEAM
sample_job.service_id, limit_days=1, include_jobs=True, key_type=KeyType.TEAM
).items
assert len(all_notifications) == 1
@@ -988,7 +977,7 @@ def test_should_exclude_test_key_notifications_by_default(
assert len(all_notifications) == 3
all_notifications = get_notifications_for_service(
sample_job.service_id, limit_days=1, key_type=KEY_TYPE_TEST
sample_job.service_id, limit_days=1, key_type=KeyType.TEST
).items
assert len(all_notifications) == 1
@@ -1006,7 +995,7 @@ def test_dao_get_notifications_by_recipient(sample_template):
template=sample_template, **recipient_to_search_for
)
create_notification(
template=sample_template, key_type=KEY_TYPE_TEST, **recipient_to_search_for
template=sample_template, key_type=KeyType.TEST, **recipient_to_search_for
)
create_notification(
template=sample_template,
@@ -1538,7 +1527,7 @@ def test_dao_get_last_notification_added_for_job_id_valid_job_id(sample_template
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
create_notification(sample_template, job, 0)
create_notification(sample_template, job, 1)
@@ -1554,7 +1543,7 @@ def test_dao_get_last_notification_added_for_job_id_no_notifications(sample_temp
created_at=datetime.utcnow() - timedelta(hours=2),
scheduled_for=datetime.utcnow() - timedelta(minutes=31),
processing_started=datetime.utcnow() - timedelta(minutes=31),
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
assert dao_get_last_notification_added_for_job_id(job.id) is None
tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py
+5 -7
View File
@@ -7,10 +7,8 @@ from app.dao.notifications_dao import (
insert_notification_history_delete_notifications,
move_notifications_to_notification_history,
)
from app.enums import KeyType
from app.models import (
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
Notification,
NotificationHistory,
)
@@ -137,13 +135,13 @@ def test_move_notifications_just_deletes_test_key_notifications(sample_template)
delete_time = datetime(2020, 6, 1, 12)
one_second_before = delete_time - timedelta(seconds=1)
create_notification(
template=sample_template, created_at=one_second_before, key_type=KEY_TYPE_NORMAL
template=sample_template, created_at=one_second_before, key_type=KeyType.NORMAL
)
create_notification(
template=sample_template, created_at=one_second_before, key_type=KEY_TYPE_TEAM
template=sample_template, created_at=one_second_before, key_type=KeyType.TEAM
)
create_notification(
template=sample_template, created_at=one_second_before, key_type=KEY_TYPE_TEST
template=sample_template, created_at=one_second_before, key_type=KeyType.TEST
)
result = move_notifications_to_notification_history(
@@ -156,7 +154,7 @@ def test_move_notifications_just_deletes_test_key_notifications(sample_template)
assert NotificationHistory.query.count() == 2
assert (
NotificationHistory.query.filter(
NotificationHistory.key_type == KEY_TYPE_TEST
NotificationHistory.key_type == KeyType.TEST
).count()
== 0
)
tests/app/dao/test_api_key_dao.py
+8 -7
View File
@@ -11,7 +11,8 @@ from app.dao.api_key_dao import (
get_unsigned_secrets,
save_model_api_key,
)
from app.models import KEY_TYPE_NORMAL, ApiKey
from app.enums import KeyType
from app.models import ApiKey
def test_save_api_key_should_create_new_api_key_and_history(sample_service):
@@ -20,7 +21,7 @@ def test_save_api_key_should_create_new_api_key_and_history(sample_service):
"service": sample_service,
"name": sample_service.name,
"created_by": sample_service.created_by,
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
}
)
save_model_api_key(api_key)
@@ -92,7 +93,7 @@ def test_should_not_allow_duplicate_key_names_per_service(sample_api_key, fake_u
"service": sample_api_key.service,
"name": sample_api_key.name,
"created_by": sample_api_key.created_by,
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
}
)
with pytest.raises(IntegrityError):
@@ -105,7 +106,7 @@ def test_save_api_key_can_create_key_with_same_name_if_other_is_expired(sample_s
"service": sample_service,
"name": "normal api key",
"created_by": sample_service.created_by,
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
"expiry_date": datetime.utcnow(),
}
)
@@ -115,7 +116,7 @@ def test_save_api_key_can_create_key_with_same_name_if_other_is_expired(sample_s
"service": sample_service,
"name": "normal api key",
"created_by": sample_service.created_by,
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
}
)
save_model_api_key(api_key)
@@ -134,7 +135,7 @@ def test_save_api_key_should_not_create_new_service_history(sample_service):
"service": sample_service,
"name": sample_service.name,
"created_by": sample_service.created_by,
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
}
)
save_model_api_key(api_key)
@@ -151,7 +152,7 @@ def test_should_not_return_revoked_api_keys_older_than_7_days(
"service": sample_service,
"name": sample_service.name,
"created_by": sample_service.created_by,
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
"expiry_date": datetime.utcnow() - timedelta(days=days_old),
}
)
tests/app/dao/test_fact_billing_dao.py
+3 -2
View File
@@ -21,7 +21,8 @@ from app.dao.fact_billing_dao import (
query_organization_sms_usage_for_year,
)
from app.dao.organization_dao import dao_add_service_to_organization
from app.models import NOTIFICATION_STATUS_TYPES, FactBilling
from app.enums import NotificationStatus
from app.models import FactBilling
from tests.app.db import (
create_annual_billing,
create_ft_billing,
@@ -301,7 +302,7 @@ def test_fetch_billing_data_for_day_bills_correctly_for_status(notify_db_session
service = create_service()
sms_template = create_template(service=service, template_type="sms")
email_template = create_template(service=service, template_type="email")
for status in NOTIFICATION_STATUS_TYPES:
for status in NotificationStatus:
create_notification(template=sms_template, status=status)
create_notification(template=email_template, status=status)
today = datetime.utcnow()
tests/app/dao/test_fact_notification_status_dao.py
+16 -26
View File
@@ -17,18 +17,8 @@ from app.dao.fact_notification_status_dao import (
get_total_notifications_for_date_range,
update_fact_notification_status,
)
from app.enums import KeyType, NotificationStatus
from app.models import (
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
NOTIFICATION_CREATED,
NOTIFICATION_DELIVERED,
NOTIFICATION_FAILED,
NOTIFICATION_PENDING,
NOTIFICATION_PERMANENT_FAILURE,
NOTIFICATION_SENDING,
NOTIFICATION_SENT,
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_TEMPORARY_FAILURE,
FactNotificationStatus,
NotificationType,
TemplateType,
@@ -63,7 +53,7 @@ def test_fetch_notification_status_for_service_by_month(notify_db_session):
create_ft_notification_status(date(2018, 1, 3), "sms", service_2)
# not included - test keys
create_ft_notification_status(
date(2018, 1, 3), "sms", service_1, key_type=KEY_TYPE_TEST
date(2018, 1, 3), "sms", service_1, key_type=KeyType.TEST
)
results = sorted(
@@ -118,7 +108,7 @@ def test_fetch_notification_status_for_service_for_day(notify_db_session):
create_notification(
service_1.templates[0],
created_at=datetime(2018, 6, 1, 12, 0, 0),
key_type=KEY_TYPE_TEAM,
key_type=KeyType.TEAM,
)
create_notification(
service_1.templates[0],
@@ -130,7 +120,7 @@ def test_fetch_notification_status_for_service_for_day(notify_db_session):
create_notification(
service_1.templates[0],
created_at=datetime(2018, 6, 1, 12, 0, 0),
key_type=KEY_TYPE_TEST,
key_type=KeyType.TEST,
)
# wrong service
@@ -638,69 +628,69 @@ def test_fetch_monthly_notification_statuses_per_service(notify_db_session):
date(2019, 4, 30),
notification_type="sms",
service=service_one,
notification_status=NOTIFICATION_DELIVERED,
notification_status=NotificationStatus.DELIVERED,
)
create_ft_notification_status(
date(2019, 3, 1),
notification_type="email",
service=service_one,
notification_status=NOTIFICATION_SENDING,
notification_status=NotificationStatus.SENDING,
count=4,
)
create_ft_notification_status(
date(2019, 3, 1),
notification_type="email",
service=service_one,
notification_status=NOTIFICATION_PENDING,
notification_status=NotificationStatus.PENDING,
count=1,
)
create_ft_notification_status(
date(2019, 3, 2),
notification_type="email",
service=service_one,
notification_status=NOTIFICATION_TECHNICAL_FAILURE,
notification_status=NotificationStatus.TECHNICAL_FAILURE,
count=2,
)
create_ft_notification_status(
date(2019, 3, 7),
notification_type="email",
service=service_one,
notification_status=NOTIFICATION_FAILED,
notification_status=NotificationStatus.FAILED,
count=1,
)
create_ft_notification_status(
date(2019, 3, 10),
notification_type="sms",
service=service_two,
notification_status=NOTIFICATION_PERMANENT_FAILURE,
notification_status=NotificationStatus.PERMANENT_FAILURE,
count=1,
)
create_ft_notification_status(
date(2019, 3, 10),
notification_type="sms",
service=service_two,
notification_status=NOTIFICATION_PERMANENT_FAILURE,
notification_status=NotificationStatus.PERMANENT_FAILURE,
count=1,
)
create_ft_notification_status(
date(2019, 3, 13),
notification_type="sms",
service=service_one,
notification_status=NOTIFICATION_SENT,
notification_status=NotificationStatus.SENT,
count=1,
)
create_ft_notification_status(
date(2019, 4, 1),
notification_type="sms",
service=service_two,
notification_status=NOTIFICATION_TEMPORARY_FAILURE,
notification_status=NotificationStatus.TEMPORARY_FAILURE,
count=10,
)
create_ft_notification_status(
date(2019, 3, 31),
notification_type="sms",
service=service_one,
notification_status=NOTIFICATION_DELIVERED,
notification_status=NotificationStatus.DELIVERED,
)
results = fetch_monthly_notification_statuses_per_service(
@@ -784,13 +774,13 @@ def test_fetch_monthly_notification_statuses_per_service_for_rows_that_should_be
create_ft_notification_status(
date(2019, 3, 15),
service=valid_service,
notification_status=NOTIFICATION_CREATED,
notification_status=NotificationStatus.CREATED,
)
# notification created by inactive service
create_ft_notification_status(date(2019, 3, 15), service=inactive_service)
# notification created with test key
create_ft_notification_status(
date(2019, 3, 12), service=valid_service, key_type=KEY_TYPE_TEST
date(2019, 3, 12), service=valid_service, key_type=KeyType.TEST
)
# notification created by trial mode service
create_ft_notification_status(date(2019, 3, 19), service=restricted_service)
tests/app/dao/test_invited_user_dao.py
+8 -7
View File
@@ -12,7 +12,8 @@ from app.dao.invited_user_dao import (
get_invited_users_for_service,
save_invited_user,
)
from app.models import INVITE_EXPIRED, InvitedUser
from app.enums import InvitedUserStatus
from app.models import InvitedUser
from tests.app.db import create_invited_user
@@ -122,11 +123,11 @@ def test_should_delete_all_invitations_more_than_one_day_old(
make_invitation(sample_user, sample_service, age=timedelta(hours=48))
make_invitation(sample_user, sample_service, age=timedelta(hours=48))
assert (
len(InvitedUser.query.filter(InvitedUser.status != INVITE_EXPIRED).all()) == 2
len(InvitedUser.query.filter(InvitedUser.status != InvitedUserStatus.EXPIRED).all()) == 2
)
expire_invitations_created_more_than_two_days_ago()
assert (
len(InvitedUser.query.filter(InvitedUser.status != INVITE_EXPIRED).all()) == 0
len(InvitedUser.query.filter(InvitedUser.status != InvitedUserStatus.EXPIRED).all()) == 0
)
@@ -149,20 +150,20 @@ def test_should_not_delete_invitations_less_than_two_days_old(
)
assert (
len(InvitedUser.query.filter(InvitedUser.status != INVITE_EXPIRED).all()) == 2
len(InvitedUser.query.filter(InvitedUser.status != InvitedUserStatus.EXPIRED).all()) == 2
)
expire_invitations_created_more_than_two_days_ago()
assert (
len(InvitedUser.query.filter(InvitedUser.status != INVITE_EXPIRED).all()) == 1
len(InvitedUser.query.filter(InvitedUser.status != InvitedUserStatus.EXPIRED).all()) == 1
)
assert (
InvitedUser.query.filter(InvitedUser.status != INVITE_EXPIRED)
InvitedUser.query.filter(InvitedUser.status != InvitedUserStatus.EXPIRED)
.first()
.email_address
== "valid@2.com"
)
assert (
InvitedUser.query.filter(InvitedUser.status == INVITE_EXPIRED)
InvitedUser.query.filter(InvitedUser.status == InvitedUserStatus.EXPIRED)
.first()
.email_address
== "expired@1.com"
tests/app/dao/test_jobs_dao.py
+9 -8
View File
@@ -18,7 +18,8 @@ from app.dao.jobs_dao import (
find_jobs_with_missing_rows,
find_missing_row_for_job,
)
from app.models import JOB_STATUS_FINISHED, Job, NotificationType, TemplateType
from app.enums import JobStatus
from app.models import Job, NotificationType, TemplateType
from tests.app.db import (
create_job,
create_notification,
@@ -379,7 +380,7 @@ def test_find_jobs_with_missing_rows(sample_email_template):
healthy_job = create_job(
template=sample_email_template,
notification_count=3,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - timedelta(minutes=20),
)
for i in range(0, 3):
@@ -387,7 +388,7 @@ def test_find_jobs_with_missing_rows(sample_email_template):
job_with_missing_rows = create_job(
template=sample_email_template,
notification_count=5,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - timedelta(minutes=20),
)
for i in range(0, 4):
@@ -405,7 +406,7 @@ def test_find_jobs_with_missing_rows_returns_nothing_for_a_job_completed_less_th
job = create_job(
template=sample_email_template,
notification_count=5,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - timedelta(minutes=9),
)
for i in range(0, 4):
@@ -422,7 +423,7 @@ def test_find_jobs_with_missing_rows_returns_nothing_for_a_job_completed_more_th
job = create_job(
template=sample_email_template,
notification_count=5,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - timedelta(days=1),
)
for i in range(0, 4):
@@ -455,7 +456,7 @@ def test_find_missing_row_for_job(sample_email_template):
job = create_job(
template=sample_email_template,
notification_count=5,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - timedelta(minutes=11),
)
create_notification(job=job, job_row_number=0)
@@ -472,7 +473,7 @@ def test_find_missing_row_for_job_more_than_one_missing_row(sample_email_templat
job = create_job(
template=sample_email_template,
notification_count=5,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - timedelta(minutes=11),
)
create_notification(job=job, job_row_number=0)
@@ -491,7 +492,7 @@ def test_find_missing_row_for_job_return_none_when_row_isnt_missing(
job = create_job(
template=sample_email_template,
notification_count=5,
job_status=JOB_STATUS_FINISHED,
job_status=JobStatus.FINISHED,
processing_finished=datetime.utcnow() - timedelta(minutes=11),
)
for i in range(0, 5):
tests/app/dao/test_services_dao.py
+10 -12
View File
@@ -40,10 +40,8 @@ from app.dao.services_dao import (
get_services_by_partial_name,
)
from app.dao.users_dao import create_user_code, save_model_user
from app.enums import KeyType
from app.models import (
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
ApiKey,
InvitedUser,
Job,
@@ -982,9 +980,9 @@ def test_dao_fetch_todays_stats_for_service(notify_db_session):
def test_dao_fetch_todays_stats_for_service_should_ignore_test_key(notify_db_session):
service = create_service()
template = create_template(service=service)
live_api_key = create_api_key(service=service, key_type=KEY_TYPE_NORMAL)
team_api_key = create_api_key(service=service, key_type=KEY_TYPE_TEAM)
test_api_key = create_api_key(service=service, key_type=KEY_TYPE_TEST)
live_api_key = create_api_key(service=service, key_type=KeyType.NORMAL)
team_api_key = create_api_key(service=service, key_type=KeyType.TEAM)
test_api_key = create_api_key(service=service, key_type=KeyType.TEST)
# two created email, one failed email, and one created sms
create_notification(
@@ -1224,9 +1222,9 @@ def test_dao_fetch_todays_stats_for_all_services_includes_all_keys_by_default(
notify_db_session,
):
template = create_template(service=create_service())
create_notification(template=template, key_type=KEY_TYPE_NORMAL)
create_notification(template=template, key_type=KEY_TYPE_TEAM)
create_notification(template=template, key_type=KEY_TYPE_TEST)
create_notification(template=template, key_type=KeyType.NORMAL)
create_notification(template=template, key_type=KeyType.TEAM)
create_notification(template=template, key_type=KeyType.TEST)
stats = dao_fetch_todays_stats_for_all_services()
@@ -1238,9 +1236,9 @@ def test_dao_fetch_todays_stats_for_all_services_can_exclude_from_test_key(
notify_db_session,
):
template = create_template(service=create_service())
create_notification(template=template, key_type=KEY_TYPE_NORMAL)
create_notification(template=template, key_type=KEY_TYPE_TEAM)
create_notification(template=template, key_type=KEY_TYPE_TEST)
create_notification(template=template, key_type=KeyType.NORMAL)
create_notification(template=template, key_type=KeyType.TEAM)
create_notification(template=template, key_type=KeyType.TEST)
stats = dao_fetch_todays_stats_for_all_services(include_from_test_key=False)
tests/app/dao/test_uploads_dao.py
+3 -3
View File
@@ -3,7 +3,7 @@ from datetime import datetime, timedelta
from freezegun import freeze_time
from app.dao.uploads_dao import dao_get_uploads_by_service_id
from app.models import JOB_STATUS_IN_PROGRESS, TemplateType
from app.enums import JobStatus, TemplateType
from tests.app.db import (
create_job,
create_notification,
@@ -89,13 +89,13 @@ def test_get_uploads_orders_by_processing_started_desc(sample_template):
sample_template,
processing_started=datetime.utcnow() - timedelta(days=1),
created_at=days_ago,
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
upload_2 = create_job(
sample_template,
processing_started=datetime.utcnow() - timedelta(days=2),
created_at=days_ago,
job_status=JOB_STATUS_IN_PROGRESS,
job_status=JobStatus.IN_PROGRESS,
)
results = dao_get_uploads_by_service_id(service_id=sample_template.service_id).items
tests/app/db.py
+7 -10
View File
@@ -26,8 +26,8 @@ from app.dao.service_sms_sender_dao import (
from app.dao.services_dao import dao_add_user_to_service, dao_create_service
from app.dao.templates_dao import dao_create_template, dao_update_template
from app.dao.users_dao import save_model_user
from app.enums import KeyType, RecipientType, TemplateType, ServicePermissionType
from app.models import (
KEY_TYPE_NORMAL,
AnnualBilling,
ApiKey,
Complaint,
@@ -36,7 +36,6 @@ from app.models import (
FactBilling,
FactNotificationStatus,
FactProcessingTime,
GuestListRecipientType,
InboundNumber,
InboundSms,
InvitedOrganizationUser,
@@ -53,11 +52,9 @@ from app.models import (
ServiceGuestList,
ServiceInboundApi,
ServicePermission,
ServicePermissionType,
ServiceSmsSender,
Template,
TemplateFolder,
TemplateType,
User,
WebauthnCredential,
)
@@ -240,7 +237,7 @@ def create_notification(
billable_units=1,
personalisation=None,
api_key=None,
key_type=KEY_TYPE_NORMAL,
key_type=KeyType.NORMAL,
sent_by=None,
client_reference=None,
rate_multiplier=None,
@@ -330,7 +327,7 @@ def create_notification_history(
updated_at=None,
billable_units=1,
api_key=None,
key_type=KEY_TYPE_NORMAL,
key_type=KeyType.NORMAL,
sent_by=None,
client_reference=None,
rate_multiplier=None,
@@ -519,7 +516,7 @@ def create_rate(start_date, value, notification_type):
return rate
def create_api_key(service, key_type=KEY_TYPE_NORMAL, key_name=None):
def create_api_key(service, key_type=KeyType.NORMAL, key_name=None):
id_ = uuid.uuid4()
name = key_name if key_name else "{} api key {}".format(key_type, id_)
@@ -725,15 +722,15 @@ def create_process_time(
def create_service_guest_list(service, email_address=None, mobile_number=None):
if email_address:
guest_list_user = ServiceGuestList.from_string(
service.id, GuestListRecipientType.EMAIL, email_address
service.id, RecipientType.EMAIL, email_address
)
elif mobile_number:
guest_list_user = ServiceGuestList.from_string(
service.id, GuestListRecipientType.MOBILE, mobile_number
service.id, RecipientType.MOBILE, mobile_number
)
else:
guest_list_user = ServiceGuestList.from_string(
service.id, GuestListRecipientType.EMAIL, "guest_list_user@digital.fake.gov"
service.id, RecipientType.EMAIL, "guest_list_user@digital.fake.gov"
)
db.session.add(guest_list_user)
tests/app/delivery/test_send_to_providers.py
+18 -26
View File
@@ -14,17 +14,9 @@ from app.dao import notifications_dao
from app.dao.provider_details_dao import get_provider_details_by_identifier
from app.delivery import send_to_providers
from app.delivery.send_to_providers import get_html_email_options, get_logo_url
from app.enums import BrandType, KeyType
from app.exceptions import NotificationTechnicalFailureException
from app.models import (
BRANDING_BOTH,
BRANDING_ORG,
BRANDING_ORG_BANNER,
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
EmailBranding,
Notification,
)
from app.models import EmailBranding, Notification
from app.serialised_models import SerialisedService
from tests.app.db import (
create_email_branding,
@@ -236,7 +228,7 @@ def test_should_have_sending_status_if_fake_callback_function_fails(
"app.delivery.send_to_providers.send_sms_response", side_effect=HTTPError
)
sample_notification.key_type = KEY_TYPE_TEST
sample_notification.key_type = KeyType.TEST
with pytest.raises(HTTPError):
send_to_providers.send_sms_to_provider(sample_notification)
assert sample_notification.status == "sending"
@@ -357,7 +349,7 @@ def test_get_html_email_renderer_should_return_for_normal_service(sample_service
@pytest.mark.parametrize(
"branding_type, govuk_banner",
[(BRANDING_ORG, False), (BRANDING_BOTH, True), (BRANDING_ORG_BANNER, False)],
[(BrandType.ORG, False), (BrandType.BOTH, True), (BrandType.ORG_BANNER, False)],
)
def test_get_html_email_renderer_with_branding_details(
branding_type, govuk_banner, notify_db_session, sample_service
@@ -380,7 +372,7 @@ def test_get_html_email_renderer_with_branding_details(
assert options["brand_text"] == "League of Justice"
assert options["brand_name"] == "Justice League"
if branding_type == BRANDING_ORG_BANNER:
if branding_type == BrandType.ORG_BANNER:
assert options["brand_banner"] is True
else:
assert options["brand_banner"] is False
@@ -405,7 +397,7 @@ def test_get_html_email_renderer_prepends_logo_path(notify_api):
)
email_branding = EmailBranding(
brand_type=BRANDING_ORG,
brand_type=BrandType.ORG,
colour="#000000",
logo="justice-league.png",
name="Justice League",
@@ -429,7 +421,7 @@ def test_get_html_email_renderer_handles_email_branding_without_logo(notify_api)
)
email_branding = EmailBranding(
brand_type=BRANDING_ORG_BANNER,
brand_type=BrandType.ORG_BANNER,
colour="#000000",
logo=None,
name="Justice League",
@@ -499,19 +491,19 @@ def test_update_notification_to_sending_does_not_update_status_from_a_final_stat
def __update_notification(notification_to_update, research_mode, expected_status):
if research_mode or notification_to_update.key_type == KEY_TYPE_TEST:
if research_mode or notification_to_update.key_type == KeyType.TEST:
notification_to_update.status = expected_status
@pytest.mark.parametrize(
"research_mode,key_type, billable_units, expected_status",
[
(True, KEY_TYPE_NORMAL, 0, "delivered"),
(False, KEY_TYPE_NORMAL, 1, "sending"),
(False, KEY_TYPE_TEST, 0, "sending"),
(True, KEY_TYPE_TEST, 0, "sending"),
(True, KEY_TYPE_TEAM, 0, "delivered"),
(False, KEY_TYPE_TEAM, 1, "sending"),
(True, KeyType.NORMAL, 0, "delivered"),
(False, KeyType.NORMAL, 1, "sending"),
(False, KeyType.TEST, 0, "sending"),
(True, KeyType.TEST, 0, "sending"),
(True, KeyType.TEAM, 0, "delivered"),
(False, KeyType.TEAM, 1, "sending"),
],
)
def test_should_update_billable_units_and_status_according_to_research_mode_and_key_type(
@@ -773,8 +765,8 @@ def test_get_html_email_options_return_email_branding_from_serialised_service(
email_options = get_html_email_options(service)
assert email_options is not None
assert email_options == {
"govuk_banner": branding.brand_type == BRANDING_BOTH,
"brand_banner": branding.brand_type == BRANDING_ORG_BANNER,
"govuk_banner": branding.brand_type == BrandType.BOTH,
"brand_banner": branding.brand_type == BrandType.ORG_BANNER,
"brand_colour": branding.colour,
"brand_logo": get_logo_url(current_app.config["ADMIN_BASE_URL"], branding.logo),
"brand_text": branding.text,
@@ -788,8 +780,8 @@ def test_get_html_email_options_add_email_branding_from_service(sample_service):
email_options = get_html_email_options(sample_service)
assert email_options is not None
assert email_options == {
"govuk_banner": branding.brand_type == BRANDING_BOTH,
"brand_banner": branding.brand_type == BRANDING_ORG_BANNER,
"govuk_banner": branding.brand_type == BrandType.BOTH,
"brand_banner": branding.brand_type == BrandType.ORG_BANNER,
"brand_colour": branding.colour,
"brand_logo": get_logo_url(current_app.config["ADMIN_BASE_URL"], branding.logo),
"brand_text": branding.text,
tests/app/email_branding/test_rest.py
+4 -3
View File
@@ -1,6 +1,7 @@
import pytest
from app.models import BRANDING_ORG, EmailBranding
from app.enums import BrandType
from app.models import EmailBranding
from tests.app.db import create_email_branding
@@ -59,7 +60,7 @@ def test_post_create_email_branding(admin_request, notify_db_session):
"name": "test email_branding",
"colour": "#0000ff",
"logo": "/images/test_x2.png",
"brand_type": BRANDING_ORG,
"brand_type": BrandType.ORG,
}
response = admin_request.post(
"email_branding.create_email_branding", _data=data, _expected_status=201
@@ -82,7 +83,7 @@ def test_post_create_email_branding_without_brand_type_defaults(
response = admin_request.post(
"email_branding.create_email_branding", _data=data, _expected_status=201
)
assert BRANDING_ORG == response["data"]["brand_type"]
assert BrandType.ORG == response["data"]["brand_type"]
def test_post_create_email_branding_without_logo_is_ok(
tests/app/job/test_rest.py
+4 -4
View File
@@ -9,7 +9,7 @@ from freezegun import freeze_time
import app.celery.tasks
from app.dao.templates_dao import dao_update_template
from app.models import JOB_STATUS_PENDING, JOB_STATUS_TYPES
from app.enums import JobStatus
from tests import create_admin_authorization_header
from tests.app.db import (
create_ft_notification_status,
@@ -782,11 +782,11 @@ def test_get_jobs_accepts_page_parameter(admin_request, sample_template):
@pytest.mark.parametrize(
"statuses_filter, expected_statuses",
[
("", JOB_STATUS_TYPES),
("pending", [JOB_STATUS_PENDING]),
("", JobStatus.TYPES),
("pending", [JobStatus.PENDING]),
(
"pending, in progress, finished, sending limits exceeded, scheduled, cancelled, ready to send, sent to dvla, error", # noqa
JOB_STATUS_TYPES,
JobStatus.TYPES,
),
# bad statuses are accepted, just return no data
("foo", []),
tests/app/notifications/test_process_notification.py
+2 -6
View File
@@ -11,12 +11,8 @@ from notifications_utils.recipients import (
)
from sqlalchemy.exc import SQLAlchemyError
from app.models import (
Notification,
NotificationHistory,
ServicePermissionType,
TemplateType,
)
from app.enums import ServicePermissionType, TemplateType
from app.models import Notification, NotificationHistory
from app.notifications.process_notifications import (
create_content_for_notification,
persist_notification,
tests/app/notifications/test_receive_notification.py
+2 -1
View File
@@ -5,7 +5,8 @@ from unittest import mock
import pytest
from flask import json
from app.models import InboundSms, ServicePermissionType
from app.models import InboundSms
from app.enums import ServicePermissionType
from app.notifications.receive_notifications import (
create_inbound_sms_object,
fetch_potential_service,
tests/app/notifications/test_rest.py
+30 -29
View File
@@ -8,7 +8,8 @@ from notifications_python_client.authentication import create_jwt_token
from app.dao.api_key_dao import save_model_api_key
from app.dao.notifications_dao import dao_update_notification
from app.dao.templates_dao import dao_update_template
from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST, ApiKey
from app.enums import KeyType
from app.models import ApiKey
from tests import create_service_authorization_header
from tests.app.db import create_api_key, create_notification
@@ -80,12 +81,12 @@ def test_get_notifications_empty_result(client, sample_api_key):
@pytest.mark.parametrize(
"api_key_type,notification_key_type",
[
(KEY_TYPE_NORMAL, KEY_TYPE_TEAM),
(KEY_TYPE_NORMAL, KEY_TYPE_TEST),
(KEY_TYPE_TEST, KEY_TYPE_NORMAL),
(KEY_TYPE_TEST, KEY_TYPE_TEAM),
(KEY_TYPE_TEAM, KEY_TYPE_NORMAL),
(KEY_TYPE_TEAM, KEY_TYPE_TEST),
(KeyType.NORMAL, KeyType.TEAM),
(KeyType.NORMAL, KeyType.TEST),
(KeyType.TEST, KeyType.NORMAL),
(KeyType.TEST, KeyType.TEAM),
(KeyType.TEAM, KeyType.NORMAL),
(KeyType.TEAM, KeyType.TEST),
],
)
def test_get_notification_from_different_api_key_works(
@@ -107,7 +108,7 @@ def test_get_notification_from_different_api_key_works(
assert response.status_code == 200
@pytest.mark.parametrize("key_type", [KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST])
@pytest.mark.parametrize("key_type", [KeyType.NORMAL, KeyType.TEAM, KeyType.TEST])
def test_get_notification_from_different_api_key_of_same_type_succeeds(
client, sample_notification, key_type
):
@@ -190,7 +191,7 @@ def test_normal_api_key_returns_notifications_created_from_jobs_and_from_api(
}
@pytest.mark.parametrize("key_type", [KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST])
@pytest.mark.parametrize("key_type", [KeyType.NORMAL, KeyType.TEAM, KeyType.TEST])
def test_get_all_notifications_only_returns_notifications_of_matching_type(
client,
sample_template,
@@ -200,19 +201,19 @@ def test_get_all_notifications_only_returns_notifications_of_matching_type(
key_type,
):
normal_notification = create_notification(
sample_template, api_key=sample_api_key, key_type=KEY_TYPE_NORMAL
sample_template, api_key=sample_api_key, key_type=KeyType.NORMAL
)
team_notification = create_notification(
sample_template, api_key=sample_team_api_key, key_type=KEY_TYPE_TEAM
sample_template, api_key=sample_team_api_key, key_type=KeyType.TEAM
)
test_notification = create_notification(
sample_template, api_key=sample_test_api_key, key_type=KEY_TYPE_TEST
sample_template, api_key=sample_test_api_key, key_type=KeyType.TEST
)
notification_objs = {
KEY_TYPE_NORMAL: normal_notification,
KEY_TYPE_TEAM: team_notification,
KEY_TYPE_TEST: test_notification,
KeyType.NORMAL: normal_notification,
KeyType.TEAM: team_notification,
KeyType.TEST: test_notification,
}
response = client.get(
@@ -227,13 +228,13 @@ def test_get_all_notifications_only_returns_notifications_of_matching_type(
assert notifications[0]["id"] == str(notification_objs[key_type].id)
@pytest.mark.parametrize("key_type", [KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST])
@pytest.mark.parametrize("key_type", [KeyType.NORMAL, KeyType.TEAM, KeyType.TEST])
def test_do_not_return_job_notifications_by_default(
client, sample_template, sample_job, key_type
):
team_api_key = create_api_key(sample_template.service, KEY_TYPE_TEAM)
normal_api_key = create_api_key(sample_template.service, KEY_TYPE_NORMAL)
test_api_key = create_api_key(sample_template.service, KEY_TYPE_TEST)
team_api_key = create_api_key(sample_template.service, KeyType.TEAM)
normal_api_key = create_api_key(sample_template.service, KeyType.NORMAL)
test_api_key = create_api_key(sample_template.service, KeyType.TEST)
create_notification(sample_template, job=sample_job)
normal_notification = create_notification(sample_template, api_key=normal_api_key)
@@ -241,9 +242,9 @@ def test_do_not_return_job_notifications_by_default(
test_notification = create_notification(sample_template, api_key=test_api_key)
notification_objs = {
KEY_TYPE_NORMAL: normal_notification,
KEY_TYPE_TEAM: team_notification,
KEY_TYPE_TEST: test_notification,
KeyType.NORMAL: normal_notification,
KeyType.TEAM: team_notification,
KeyType.TEST: test_notification,
}
response = client.get(
@@ -259,7 +260,7 @@ def test_do_not_return_job_notifications_by_default(
@pytest.mark.parametrize(
"key_type", [(KEY_TYPE_NORMAL, 2), (KEY_TYPE_TEAM, 1), (KEY_TYPE_TEST, 1)]
"key_type", [(KeyType.NORMAL, 2), (KeyType.TEAM, 1), (KeyType.TEST, 1)]
)
def test_only_normal_api_keys_can_return_job_notifications(
client,
@@ -271,19 +272,19 @@ def test_only_normal_api_keys_can_return_job_notifications(
key_type,
):
normal_notification = create_notification(
template=sample_template, api_key=sample_api_key, key_type=KEY_TYPE_NORMAL
template=sample_template, api_key=sample_api_key, key_type=KeyType.NORMAL
)
team_notification = create_notification(
template=sample_template, api_key=sample_team_api_key, key_type=KEY_TYPE_TEAM
template=sample_template, api_key=sample_team_api_key, key_type=KeyType.TEAM
)
test_notification = create_notification(
template=sample_template, api_key=sample_test_api_key, key_type=KEY_TYPE_TEST
template=sample_template, api_key=sample_test_api_key, key_type=KeyType.TEST
)
notification_objs = {
KEY_TYPE_NORMAL: normal_notification,
KEY_TYPE_TEAM: team_notification,
KEY_TYPE_TEST: test_notification,
KeyType.NORMAL: normal_notification,
KeyType.TEAM: team_notification,
KeyType.TEST: test_notification,
}
response = client.get(
tests/app/notifications/test_validators.py
+2 -7
View File
@@ -5,12 +5,7 @@ from notifications_utils import SMS_CHAR_COUNT_LIMIT
import app
from app.dao import templates_dao
from app.models import (
KEY_TYPE_NORMAL,
NotificationType,
ServicePermissionType,
TemplateType,
)
from app.enums import KeyType, NotificationType, ServicePermissionType, TemplateType
from app.notifications.process_notifications import create_content_for_notification
from app.notifications.sns_cert_validator import (
VALID_SNS_TOPICS,
@@ -783,6 +778,6 @@ def test_check_service_over_total_message_limit(mocker, sample_service):
get_redis_mock = mocker.patch("app.notifications.validators.redis_store.get")
get_redis_mock.return_value = None
service_stats = check_service_over_total_message_limit(
KEY_TYPE_NORMAL, sample_service
KeyType.NORMAL, sample_service
)
assert service_stats == 0
tests/app/platform_stats/test_rest.py
+1 -1
View File
@@ -4,7 +4,7 @@ import pytest
from freezegun import freeze_time
from app.errors import InvalidRequest
from app.models import TemplateType
from app.enums import TemplateType
from app.platform_stats.rest import validate_date_range_is_within_a_financial_year
from tests.app.db import (
create_ft_billing,
tests/app/public_contracts/test_GET_notification.py
+3 -2
View File
@@ -1,7 +1,8 @@
import pytest
from app.dao.api_key_dao import save_model_api_key
from app.models import KEY_TYPE_NORMAL, ApiKey
from app.enums import KeyType
from app.models import ApiKey
from app.v2.notifications.notification_schemas import (
get_notification_response,
get_notifications_response,
@@ -17,7 +18,7 @@ def _get_notification(client, notification, url):
service=notification.service,
name="api_key",
created_by=notification.service.created_by,
key_type=KEY_TYPE_NORMAL,
key_type=KeyType.NORMAL,
)
)
auth_header = create_service_authorization_header(
tests/app/service/test_api_key_endpoints.py
+4 -3
View File
@@ -3,7 +3,8 @@ import json
from flask import url_for
from app.dao.api_key_dao import expire_api_key
from app.models import KEY_TYPE_NORMAL, ApiKey
from app.enums import KeyType
from app.models import ApiKey
from tests import create_admin_authorization_header
from tests.app.db import create_api_key, create_service, create_user
@@ -14,7 +15,7 @@ def test_api_key_should_create_new_api_key_for_service(notify_api, sample_servic
data = {
"name": "some secret name",
"created_by": str(sample_service.created_by.id),
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
}
auth_header = create_admin_authorization_header()
response = client.post(
@@ -86,7 +87,7 @@ def test_api_key_should_create_multiple_new_api_key_for_service(
data = {
"name": "some secret name",
"created_by": str(sample_service.created_by.id),
"key_type": KEY_TYPE_NORMAL,
"key_type": KeyType.NORMAL,
}
auth_header = create_admin_authorization_header()
response = client.post(
tests/app/service/test_rest.py
+8 -13
View File
@@ -14,22 +14,17 @@ from app.dao.service_user_dao import dao_get_service_user
from app.dao.services_dao import dao_add_user_to_service, dao_remove_user_from_service
from app.dao.templates_dao import dao_redact_template
from app.dao.users_dao import save_model_user
from app.enums import KeyType, ServicePermissionType, TemplateType, NotificationType
from app.models import (
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
AnnualBilling,
EmailBranding,
InboundNumber,
Notification,
NotificationType,
Permission,
Service,
ServiceEmailReplyTo,
ServicePermission,
ServicePermissionType,
ServiceSmsSender,
TemplateType,
User,
)
from tests import create_admin_authorization_header
@@ -1882,7 +1877,7 @@ def test_get_all_notifications_for_service_including_ones_made_by_jobs(
mock_s3.return_value = "1"
# notification from_test_api_key
create_notification(sample_template, key_type=KEY_TYPE_TEST)
create_notification(sample_template, key_type=KeyType.TEST)
auth_header = create_admin_authorization_header()
@@ -2081,7 +2076,7 @@ def test_get_services_with_detailed_flag(client, sample_template):
notifications = [
create_notification(sample_template),
create_notification(sample_template),
create_notification(sample_template, key_type=KEY_TYPE_TEST),
create_notification(sample_template, key_type=KeyType.TEST),
]
resp = client.get(
"/service?detailed=True", headers=[create_admin_authorization_header()]
@@ -2101,11 +2096,11 @@ def test_get_services_with_detailed_flag(client, sample_template):
def test_get_services_with_detailed_flag_excluding_from_test_key(
client, sample_template
):
create_notification(sample_template, key_type=KEY_TYPE_NORMAL)
create_notification(sample_template, key_type=KEY_TYPE_TEAM)
create_notification(sample_template, key_type=KEY_TYPE_TEST)
create_notification(sample_template, key_type=KEY_TYPE_TEST)
create_notification(sample_template, key_type=KEY_TYPE_TEST)
create_notification(sample_template, key_type=KeyType.NORMAL)
create_notification(sample_template, key_type=KeyType.TEAM)
create_notification(sample_template, key_type=KeyType.TEST)
create_notification(sample_template, key_type=KeyType.TEST)
create_notification(sample_template, key_type=KeyType.TEST)
resp = client.get(
"/service?detailed=True&include_from_test_key=False",
tests/app/test_commands.py
+3 -5
View File
@@ -20,13 +20,11 @@ from app.commands import (
)
from app.dao.inbound_numbers_dao import dao_get_available_inbound_numbers
from app.dao.users_dao import get_user_by_email
from app.enums import KeyType, NotificationStatus, NotificationType
from app.models import (
KEY_TYPE_NORMAL,
NOTIFICATION_DELIVERED,
AnnualBilling,
Job,
Notification,
NotificationType,
Organization,
Service,
Template,
@@ -315,9 +313,9 @@ def test_fix_billable_units(notify_db_session, notify_api, sample_template):
notification = Notification.query.one()
notification.billable_units = 0
notification.notification_type = NotificationType.SMS
notification.status = NOTIFICATION_DELIVERED
notification.status = NotificationStatus.DELIVERED
notification.sent_at = None
notification.key_type = KEY_TYPE_NORMAL
notification.key_type = KeyType.NORMAL
notify_db_session.commit()
tests/app/test_model.py
+31 -25
View File
@@ -4,24 +4,23 @@ import pytest
from freezegun import freeze_time
from sqlalchemy.exc import IntegrityError
from app import encryption
from app.models import (
NOTIFICATION_CREATED,
NOTIFICATION_FAILED,
NOTIFICATION_PENDING,
NOTIFICATION_STATUS_TYPES_FAILED,
NOTIFICATION_TECHNICAL_FAILURE,
Agreement,
from app.enums import (
NotificationStatus,
RecipientType,
TemplateType,
AgreementStatus,
AgreementType,
)
from app import encryption
from app.models import (
Agreement,
AnnualBilling,
GuestListRecipientType,
Notification,
NotificationHistory,
Service,
ServiceGuestList,
ServicePermission,
TemplateType,
User,
VerifyCode,
filter_null_value_fields,
@@ -42,7 +41,7 @@ from tests.app.db import (
@pytest.mark.parametrize("mobile_number", ["+447700900855", "+12348675309"])
def test_should_build_service_guest_list_from_mobile_number(mobile_number):
service_guest_list = ServiceGuestList.from_string(
"service_id", GuestListRecipientType.MOBILE, mobile_number
"service_id", RecipientType.MOBILE, mobile_number
)
assert service_guest_list.recipient == mobile_number
@@ -76,30 +75,37 @@ def test_should_not_build_service_guest_list_from_invalid_contact(
"initial_statuses, expected_statuses",
[
# passing in single statuses as strings
(NOTIFICATION_FAILED, NOTIFICATION_STATUS_TYPES_FAILED),
(NOTIFICATION_CREATED, [NOTIFICATION_CREATED]),
(NOTIFICATION_TECHNICAL_FAILURE, [NOTIFICATION_TECHNICAL_FAILURE]),
(NotificationStatus.FAILED, NotificationStatus.failed_types),
(NotificationStatus.CREATED, [NotificationStatus.CREATED]),
(NotificationStatus.TECHNICAL_FAILURE, [NotificationStatus.TECHNICAL_FAILURE]),
# passing in lists containing single statuses
([NOTIFICATION_FAILED], NOTIFICATION_STATUS_TYPES_FAILED),
([NOTIFICATION_CREATED], [NOTIFICATION_CREATED]),
([NOTIFICATION_TECHNICAL_FAILURE], [NOTIFICATION_TECHNICAL_FAILURE]),
([NotificationStatus.FAILED], NotificationStatus.failed_types),
([NotificationStatus.CREATED], [NotificationStatus.CREATED]),
(
[NotificationStatus.TECHNICAL_FAILURE],
[NotificationStatus.TECHNICAL_FAILURE],
),
# passing in lists containing multiple statuses
(
[NOTIFICATION_FAILED, NOTIFICATION_CREATED],
NOTIFICATION_STATUS_TYPES_FAILED + [NOTIFICATION_CREATED],
[NotificationStatus.FAILED, NotificationStatus.CREATED],
list(NotificationStatus.failed_types) + [NotificationStatus.CREATED],
),
(
[NOTIFICATION_CREATED, NOTIFICATION_PENDING],
[NOTIFICATION_CREATED, NOTIFICATION_PENDING],
[NotificationStatus.CREATED, NotificationStatus.PENDING],
[NotificationStatus.CREATED, NotificationStatus.PENDING],
),
(
[NOTIFICATION_CREATED, NOTIFICATION_TECHNICAL_FAILURE],
[NOTIFICATION_CREATED, NOTIFICATION_TECHNICAL_FAILURE],
[NotificationStatus.CREATED, NotificationStatus.TECHNICAL_FAILURE],
[NotificationStatus.CREATED, NotificationStatus.TECHNICAL_FAILURE],
),
# checking we don't end up with duplicates
(
[NOTIFICATION_FAILED, NOTIFICATION_CREATED, NOTIFICATION_TECHNICAL_FAILURE],
NOTIFICATION_STATUS_TYPES_FAILED + [NOTIFICATION_CREATED],
[
NotificationStatus.FAILED,
NotificationStatus.CREATED,
NotificationStatus.TECHNICAL_FAILURE,
],
list(NotificationStatus.failed_types) + [NotificationStatus.CREATED],
),
],
)
tests/app/test_utils.py
+1 -1
View File
@@ -4,7 +4,7 @@ from datetime import date, datetime
import pytest
from freezegun import freeze_time
from app.models import ServicePermissionType
from app.enums import ServicePermissionType
from app.utils import (
format_sequential_number,
get_midnight_for_day_before,