mirror of
https://github.com/GSA/notifications-api.git
synced 2026-02-03 01:41:05 -05:00
@@ -15,7 +15,7 @@ from app.dao.notifications_dao import (
|
||||
update_notification_status_by_id,
|
||||
)
|
||||
from app.delivery import send_to_providers
|
||||
from app.enum import NotificationStatus
|
||||
from app.enums import NotificationStatus
|
||||
from app.exceptions import NotificationTechnicalFailureException
|
||||
|
||||
# This is the amount of time to wait after sending an sms message before we check the aws logs and look for delivery
|
||||
|
||||
@@ -408,7 +408,9 @@ def _query_for_billing_data(notification_type, start_date, end_date, service):
|
||||
func.count().label("notifications_sent"),
|
||||
)
|
||||
.filter(
|
||||
NotificationAllTimeView.status.in_(NotificationStatus.sent_email_types),
|
||||
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,
|
||||
@@ -441,7 +443,7 @@ def _query_for_billing_data(notification_type, start_date, end_date, service):
|
||||
)
|
||||
.filter(
|
||||
NotificationAllTimeView.status.in_(
|
||||
NotificationStatus.billable_sms_types
|
||||
NotificationStatus.billable_sms_types()
|
||||
),
|
||||
NotificationAllTimeView.key_type.in_((KeyType.NORMAL, KeyType.TEAM)),
|
||||
NotificationAllTimeView.created_at >= start_date,
|
||||
|
||||
@@ -9,7 +9,7 @@ class PermissionDAO(DAOClass):
|
||||
model = Permission
|
||||
|
||||
def add_default_service_permissions_for_user(self, user, service):
|
||||
for name in PermissionType.defaults:
|
||||
for name in PermissionType.defaults():
|
||||
permission = Permission(permission=name, user=user, service=service)
|
||||
self.create_instance(permission, _commit=False)
|
||||
|
||||
|
||||
@@ -275,7 +275,7 @@ def dao_create_service(
|
||||
raise ValueError("Can't create a service without a user")
|
||||
|
||||
if service_permissions is None:
|
||||
service_permissions = ServicePermissionType.defaults
|
||||
service_permissions = ServicePermissionType.defaults()
|
||||
|
||||
organization = dao_get_organization_by_email_address(user.email_address)
|
||||
|
||||
|
||||
@@ -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_types:
|
||||
if notification.status not in NotificationStatus.completed_types():
|
||||
notification.status = NotificationStatus.SENDING
|
||||
|
||||
dao_update_notification(notification)
|
||||
|
||||
46
app/enums.py
46
app/enums.py
@@ -51,9 +51,8 @@ class NotificationStatus(Enum):
|
||||
VALIDATION_FAILED = "validation-failed"
|
||||
VIRUS_SCAN_FAILED = "virus-scan-failed"
|
||||
|
||||
@property
|
||||
def failed_types(self) -> tuple["NotificationStatus", ...]:
|
||||
cls = type(self)
|
||||
@classmethod
|
||||
def failed_types(cls) -> tuple["NotificationStatus", ...]:
|
||||
return (
|
||||
cls.TECHNICAL_FAILURE,
|
||||
cls.TEMPORARY_FAILURE,
|
||||
@@ -62,9 +61,8 @@ class NotificationStatus(Enum):
|
||||
cls.VIRUS_SCAN_FAILED,
|
||||
)
|
||||
|
||||
@property
|
||||
def completed_types(self) -> tuple["NotificationStatus", ...]:
|
||||
cls = type(self)
|
||||
@classmethod
|
||||
def completed_types(cls) -> tuple["NotificationStatus", ...]:
|
||||
return (
|
||||
cls.SENT,
|
||||
cls.DELIVERED,
|
||||
@@ -75,14 +73,12 @@ class NotificationStatus(Enum):
|
||||
cls.CANCELLED,
|
||||
)
|
||||
|
||||
@property
|
||||
def success_types(self) -> tuple["NotificationStatus", ...]:
|
||||
cls = type(self)
|
||||
@classmethod
|
||||
def success_types(cls) -> tuple["NotificationStatus", ...]:
|
||||
return (cls.SENT, cls.DELIVERED)
|
||||
|
||||
@property
|
||||
def billable_types(self) -> tuple["NotificationStatus", ...]:
|
||||
cls = type(self)
|
||||
@classmethod
|
||||
def billable_types(cls) -> tuple["NotificationStatus", ...]:
|
||||
return (
|
||||
cls.SENDING,
|
||||
cls.SENT,
|
||||
@@ -93,9 +89,8 @@ class NotificationStatus(Enum):
|
||||
cls.PERMANENT_FAILURE,
|
||||
)
|
||||
|
||||
@property
|
||||
def billable_sms_types(self) -> tuple["NotificationStatus", ...]:
|
||||
cls = type(self)
|
||||
@classmethod
|
||||
def billable_sms_types(cls) -> tuple["NotificationStatus", ...]:
|
||||
return (
|
||||
cls.SENDING,
|
||||
cls.SENT, # internationally
|
||||
@@ -105,9 +100,8 @@ class NotificationStatus(Enum):
|
||||
cls.PERMANENT_FAILURE,
|
||||
)
|
||||
|
||||
@property
|
||||
def sent_email_types(self) -> tuple["NotificationStatus", ...]:
|
||||
cls = type(self)
|
||||
@classmethod
|
||||
def sent_email_types(cls) -> tuple["NotificationStatus", ...]:
|
||||
return (
|
||||
cls.SENDING,
|
||||
cls.DELIVERED,
|
||||
@@ -115,14 +109,9 @@ class NotificationStatus(Enum):
|
||||
cls.PERMANENT_FAILURE,
|
||||
)
|
||||
|
||||
@property
|
||||
def non_billable_types(self) -> tuple["NotificationStatus", ...]:
|
||||
self._non_billable: tuple["NotificationStatus", ...]
|
||||
try:
|
||||
return self._non_billable
|
||||
except AttributeError:
|
||||
self._non_billable = tuple(set(type(self)) - set(self.billable))
|
||||
return self._non_billable
|
||||
@classmethod
|
||||
def non_billable_types(cls) -> tuple["NotificationStatus", ...]:
|
||||
return tuple({i for i in cls} - set(cls.billable_types()))
|
||||
|
||||
|
||||
class PermissionType(Enum):
|
||||
@@ -135,9 +124,8 @@ class PermissionType(Enum):
|
||||
PLATFORM_ADMIN = "platform_admin"
|
||||
VIEW_ACTIVITY = "view_activity"
|
||||
|
||||
@property
|
||||
def defaults(self) -> tuple["PermissionType", ...]:
|
||||
cls = type(self)
|
||||
@classmethod
|
||||
def defaults(cls) -> tuple["PermissionType", ...]:
|
||||
return (
|
||||
cls.MANAGE_USERS,
|
||||
cls.MANAGE_TEMPLATES,
|
||||
|
||||
@@ -232,10 +232,12 @@ class ServiceUser(db.Model):
|
||||
primary_key=True,
|
||||
)
|
||||
|
||||
__table_args__ = UniqueConstraint(
|
||||
"user_id",
|
||||
"service_id",
|
||||
name="uix_user_to_service",
|
||||
__table_args__ = (
|
||||
UniqueConstraint(
|
||||
"user_id",
|
||||
"service_id",
|
||||
name="uix_user_to_service",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@@ -1581,7 +1583,7 @@ class Notification(db.Model):
|
||||
self._personalisation = encryption.encrypt(personalisation or {})
|
||||
|
||||
def completed_at(self):
|
||||
if self.status in NotificationStatus.completed_types:
|
||||
if self.status in NotificationStatus.completed_types():
|
||||
return self.updated_at.strftime(DATETIME_FORMAT)
|
||||
|
||||
return None
|
||||
@@ -1621,7 +1623,7 @@ class Notification(db.Model):
|
||||
|
||||
def _substitute_status_str(_status):
|
||||
return (
|
||||
NotificationStatus.failed_types
|
||||
NotificationStatus.failed_types()
|
||||
if _status == NotificationStatus.FAILED
|
||||
else [_status]
|
||||
)
|
||||
|
||||
@@ -129,7 +129,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 NotificationStatus.completed_types
|
||||
if status in NotificationStatus.completed_types()
|
||||
else None,
|
||||
"client_reference": client_reference,
|
||||
"rate_multiplier": rate_multiplier,
|
||||
|
||||
@@ -74,11 +74,11 @@ def test_should_not_build_service_guest_list_from_invalid_contact(
|
||||
"initial_statuses, expected_statuses",
|
||||
[
|
||||
# passing in single statuses as strings
|
||||
(NotificationStatus.FAILED, NotificationStatus.failed_types),
|
||||
(NotificationStatus.FAILED, NotificationStatus.failed_types()),
|
||||
(NotificationStatus.CREATED, [NotificationStatus.CREATED]),
|
||||
(NotificationStatus.TECHNICAL_FAILURE, [NotificationStatus.TECHNICAL_FAILURE]),
|
||||
# passing in lists containing single statuses
|
||||
([NotificationStatus.FAILED], NotificationStatus.failed_types),
|
||||
([NotificationStatus.FAILED], NotificationStatus.failed_types()),
|
||||
([NotificationStatus.CREATED], [NotificationStatus.CREATED]),
|
||||
(
|
||||
[NotificationStatus.TECHNICAL_FAILURE],
|
||||
@@ -87,7 +87,7 @@ def test_should_not_build_service_guest_list_from_invalid_contact(
|
||||
# passing in lists containing multiple statuses
|
||||
(
|
||||
[NotificationStatus.FAILED, NotificationStatus.CREATED],
|
||||
list(NotificationStatus.failed_types) + [NotificationStatus.CREATED],
|
||||
list(NotificationStatus.failed_types()) + [NotificationStatus.CREATED],
|
||||
),
|
||||
(
|
||||
[NotificationStatus.CREATED, NotificationStatus.PENDING],
|
||||
@@ -104,7 +104,7 @@ def test_should_not_build_service_guest_list_from_invalid_contact(
|
||||
NotificationStatus.CREATED,
|
||||
NotificationStatus.TECHNICAL_FAILURE,
|
||||
],
|
||||
list(NotificationStatus.failed_types) + [NotificationStatus.CREATED],
|
||||
list(NotificationStatus.failed_types()) + [NotificationStatus.CREATED],
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user