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

More cleanup.

Browse Source 
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
This commit is contained in:
Cliff Hill
2024-02-09 13:46:44 -05:00
parent 9cc3f1c1ff
commit 262bd8943a
3 changed files with 129 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
tests/app/celery/test_service_callback_tasks.py
View File

@@ -10,6 +10,7 @@ from app.celery.service_callback_tasks import (
send_complaint_to_service,
send_delivery_status_to_service,
)
from app.enums import CallbackType, NotificationStatus, NotificationType
from app.utils import DATETIME_FORMAT
from tests.app.db import (
create_complaint,
@@ -20,7 +21,7 @@ from tests.app.db import (
)
@pytest.mark.parametrize("notification_type", ["email", "sms"])
@pytest.mark.parametrize("notification_type", [NotificationType.EMAIL, NotificationType.SMS])
def test_send_delivery_status_to_service_post_https_request_to_service_with_encrypted_data(
notify_db_session, notification_type
):
@@ -32,7 +33,7 @@ def test_send_delivery_status_to_service_post_https_request_to_service_with_encr
created_at=datestr,
updated_at=datestr,
sent_at=datestr,
status="sent",
status=NotificationStatus.SENT,
)
encrypted_status_update = _set_up_data_for_status_update(callback_api, notification)
with requests_mock.Mocker() as request_mock:
@@ -102,7 +103,7 @@ def test_send_complaint_to_service_posts_https_request_to_service_with_encrypted
] == "Bearer {}".format(callback_api.bearer_token)
@pytest.mark.parametrize("notification_type", ["email", "sms"])
@pytest.mark.parametrize("notification_type", [NotificationType.EMAIL, NotificationType.SMS],)
@pytest.mark.parametrize("status_code", [429, 500, 503])
def test__send_data_to_service_callback_api_retries_if_request_returns_error_code_with_encrypted_data(
notify_db_session, mocker, notification_type, status_code
@@ -114,7 +115,7 @@ def test__send_data_to_service_callback_api_retries_if_request_returns_error_cod
created_at=datestr,
updated_at=datestr,
sent_at=datestr,
status="sent",
status=NotificationStatus.SENT,
)
encrypted_data = _set_up_data_for_status_update(callback_api, notification)
mocked = mocker.patch(
@@ -130,7 +131,7 @@ def test__send_data_to_service_callback_api_retries_if_request_returns_error_cod
assert mocked.call_args[1]["queue"] == "service-callbacks-retry"
@pytest.mark.parametrize("notification_type", ["email", "sms"])
@pytest.mark.parametrize("notification_type", [NotificationType.EMAIL, NotificationType.SMS],)
def test__send_data_to_service_callback_api_does_not_retry_if_request_returns_404_with_encrypted_data(
notify_db_session, mocker, notification_type
):
@@ -159,7 +160,7 @@ def test__send_data_to_service_callback_api_does_not_retry_if_request_returns_40
def test_send_delivery_status_to_service_succeeds_if_sent_at_is_none(
notify_db_session, mocker
):
callback_api, template = _set_up_test_data("email", "delivery_status")
callback_api, template = _set_up_test_data(NotificationType.EMAIL, CallbackType.DELIVERY_STATUS,)
datestr = datetime(2017, 6, 20)
notification = create_notification(
template=template,

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

@@ -194,7 +194,7 @@ def test_should_not_create_save_task_for_empty_file(sample_job, mocker):
service_id=str(sample_job.service.id), job_id=str(sample_job.id)
)
job = jobs_dao.dao_get_job_by_id(sample_job.id)
assert job.job_status == "finished"
assert job.job_status == JobStatus.FINISHED
assert tasks.save_sms.apply_async.called is False
@@ -420,13 +420,13 @@ def test_should_send_template_to_correct_sms_task_and_persist(
persisted_notification.template_version
== sample_template_with_placeholders.version
)
assert persisted_notification.status == "created"
assert persisted_notification.status == NotificationStatus.CREATED
assert persisted_notification.created_at <= datetime.utcnow()
assert not persisted_notification.sent_at
assert not persisted_notification.sent_by
assert not persisted_notification.job_id
assert persisted_notification.personalisation == {"name": "Jo"}
assert persisted_notification.notification_type == "sms"
assert persisted_notification.notification_type == NotificationType.SMS
mocked_deliver_sms.assert_called_once_with(
[str(persisted_notification.id)], queue="send-sms-tasks"
)
@@ -456,13 +456,13 @@ def test_should_save_sms_if_restricted_service_and_valid_number(
assert persisted_notification.to == "1"
assert persisted_notification.template_id == template.id
assert persisted_notification.template_version == template.version
assert persisted_notification.status == "created"
assert persisted_notification.status == NotificationStatus.CREATED
assert persisted_notification.created_at <= datetime.utcnow()
assert not persisted_notification.sent_at
assert not persisted_notification.sent_by
assert not persisted_notification.job_id
assert not persisted_notification.personalisation
assert persisted_notification.notification_type == "sms"
assert persisted_notification.notification_type == NotificationType.SMS
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[str(persisted_notification.id)], queue="send-sms-tasks"
)
@@ -475,7 +475,7 @@ def test_save_email_should_save_default_email_reply_to_text_on_notification(
create_reply_to_email(
service=service, email_address="reply_to@digital.fake.gov", is_default=True
)
template = create_template(service=service, template_type="email", subject="Hello")
template = create_template(service=service, template_type=TemplateType.EMAIL, subject="Hello",)
notification = _notification_json(template, to="test@example.com")
mocker.patch("app.celery.provider_tasks.deliver_email.apply_async")
@@ -536,7 +536,7 @@ def test_should_not_save_email_if_restricted_service_and_invalid_email_address(
):
user = create_user()
service = create_service(user=user, restricted=True)
template = create_template(service=service, template_type="email", subject="Hello")
template = create_template(service=service, template_type=TemplateType.EMAIL, subject="Hello",)
notification = _notification_json(template, to="test@example.com")
notification_id = uuid.uuid4()
@@ -551,7 +551,7 @@ def test_should_not_save_email_if_restricted_service_and_invalid_email_address(
def test_should_save_sms_template_to_and_persist_with_job_id(sample_job, mocker):
notification = _notification_json(
sample_job.template, to="+447234123123", job_id=sample_job.id, row_number=2
sample_job.template, to="+447234123123", job_id=sample_job.id, row_number=2,
)
mocker.patch("app.celery.provider_tasks.deliver_sms.apply_async")
@@ -566,14 +566,14 @@ def test_should_save_sms_template_to_and_persist_with_job_id(sample_job, mocker)
assert persisted_notification.to == "1"
assert persisted_notification.job_id == sample_job.id
assert persisted_notification.template_id == sample_job.template.id
assert persisted_notification.status == "created"
assert persisted_notification.status == NotificationStatus.CREATED
assert not persisted_notification.sent_at
assert persisted_notification.created_at >= now
assert not persisted_notification.sent_by
assert persisted_notification.job_row_number == 2
assert persisted_notification.api_key_id is None
assert persisted_notification.key_type == KeyType.NORMAL
assert persisted_notification.notification_type == "sms"
assert persisted_notification.notification_type == NotificationType.SMS
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[str(persisted_notification.id)], queue="send-sms-tasks"
@@ -638,13 +638,13 @@ def test_should_use_email_template_and_persist(
)
assert persisted_notification.created_at >= now
assert not persisted_notification.sent_at
assert persisted_notification.status == "created"
assert persisted_notification.status == NotificationStatus.CREATED
assert not persisted_notification.sent_by
assert persisted_notification.job_row_number == 1
assert persisted_notification.personalisation == {"name": "Jo"}
assert persisted_notification.api_key_id is None
assert persisted_notification.key_type == KeyType.NORMAL
assert persisted_notification.notification_type == "email"
assert persisted_notification.notification_type == NotificationType.EMAIL
provider_tasks.deliver_email.apply_async.assert_called_once_with(
[str(persisted_notification.id)], queue="send-email-tasks"
@@ -680,9 +680,9 @@ def test_save_email_should_use_template_version_from_job_not_latest(
assert persisted_notification.template_version == version_on_notification
assert persisted_notification.created_at >= now
assert not persisted_notification.sent_at
assert persisted_notification.status == "created"
assert persisted_notification.status == NotificationStatus.CREATED
assert not persisted_notification.sent_by
assert persisted_notification.notification_type == "email"
assert persisted_notification.notification_type == NotificationType.EMAIL
provider_tasks.deliver_email.apply_async.assert_called_once_with(
[str(persisted_notification.id)], queue="send-email-tasks"
)
@@ -708,12 +708,12 @@ def test_should_use_email_template_subject_placeholders(
assert (
persisted_notification.template_id == sample_email_template_with_placeholders.id
)
assert persisted_notification.status == "created"
assert persisted_notification.status == NotificationStatus.CREATED
assert persisted_notification.created_at >= now
assert not persisted_notification.sent_by
assert persisted_notification.personalisation == {"name": "Jo"}
assert not persisted_notification.reference
assert persisted_notification.notification_type == "email"
assert persisted_notification.notification_type == NotificationType.EMAIL
provider_tasks.deliver_email.apply_async.assert_called_once_with(
[str(persisted_notification.id)], queue="send-email-tasks"
)
@@ -726,11 +726,11 @@ def test_save_email_uses_the_reply_to_text_when_provided(sample_email_template,
service = sample_email_template.service
notification_id = uuid.uuid4()
service_email_reply_to_dao.add_reply_to_email_address_for_service(
service.id, "default@example.com", True
service.id, "default@example.com", True,
)
other_email_reply_to = (
service_email_reply_to_dao.add_reply_to_email_address_for_service(
service.id, "other@example.com", False
service.id, "other@example.com", False,
)
)
@@ -741,7 +741,7 @@ def test_save_email_uses_the_reply_to_text_when_provided(sample_email_template,
sender_id=other_email_reply_to.id,
)
persisted_notification = Notification.query.one()
assert persisted_notification.notification_type == "email"
assert persisted_notification.notification_type == NotificationType.EMAIL
assert persisted_notification.reply_to_text == "other@example.com"
@@ -754,7 +754,7 @@ def test_save_email_uses_the_default_reply_to_text_if_sender_id_is_none(
service = sample_email_template.service
notification_id = uuid.uuid4()
service_email_reply_to_dao.add_reply_to_email_address_for_service(
service.id, "default@example.com", True
service.id, "default@example.com", True,
)
save_email(
@@ -764,7 +764,7 @@ def test_save_email_uses_the_default_reply_to_text_if_sender_id_is_none(
sender_id=None,
)
persisted_notification = Notification.query.one()
assert persisted_notification.notification_type == "email"
assert persisted_notification.notification_type == NotificationType.EMAIL
assert persisted_notification.reply_to_text == "default@example.com"
@@ -787,11 +787,11 @@ def test_should_use_email_template_and_persist_without_personalisation(
assert persisted_notification.template_id == sample_email_template.id
assert persisted_notification.created_at >= now
assert not persisted_notification.sent_at
assert persisted_notification.status == "created"
assert persisted_notification.status == NotificationStatus.CREATED
assert not persisted_notification.sent_by
assert not persisted_notification.personalisation
assert not persisted_notification.reference
assert persisted_notification.notification_type == "email"
assert persisted_notification.notification_type == NotificationType.EMAIL
provider_tasks.deliver_email.apply_async.assert_called_once_with(
[str(persisted_notification.id)], queue="send-email-tasks"
)
@@ -925,7 +925,7 @@ def test_save_sms_uses_non_default_sms_sender_reply_to_text_if_provided(
service = create_service_with_defined_sms_sender(sms_sender_value="2028675309")
template = create_template(service=service)
new_sender = service_sms_sender_dao.dao_add_sms_sender_for_service(
service.id, "new-sender", False
service.id, "new-sender", False,
)
notification = _notification_json(template, to="202-867-5301")
@@ -952,7 +952,7 @@ def test_should_cancel_job_if_service_is_inactive(sample_service, sample_job, mo
process_job(sample_job.id)
job = jobs_dao.dao_get_job_by_id(sample_job.id)
assert job.job_status == "cancelled"
assert job.job_status == JobStatus.CANCELLED
s3.get_job_from_s3.assert_not_called()
tasks.process_row.assert_not_called()
@@ -1377,7 +1377,7 @@ def test_process_incomplete_jobs_sets_status_to_in_progress_and_resets_processin
@freeze_time("2020-03-25 14:30")
@pytest.mark.parametrize("notification_type", ["sms", "email"])
@pytest.mark.parametrize("notification_type", [NotificationType.SMS, NotificationType.EMAIL],)
def test_save_api_email_or_sms(mocker, sample_service, notification_type):
template = (
create_template(sample_service)
@@ -1427,7 +1427,7 @@ def test_save_api_email_or_sms(mocker, sample_service, notification_type):
@freeze_time("2020-03-25 14:30")
@pytest.mark.parametrize("notification_type", ["sms", "email"])
@pytest.mark.parametrize("notification_type", [NotificationType.SMS, NotificationType.EMAIL])
def test_save_api_email_dont_retry_if_notification_already_exists(
sample_service, mocker, notification_type
):
@@ -1492,13 +1492,13 @@ def test_save_api_email_dont_retry_if_notification_already_exists(
save_email,
"app.celery.provider_tasks.deliver_email.apply_async",
"test@example.com",
{"template_type": "email", "subject": "Hello"},
{"template_type": TemplateType.EMAIL, "subject": "Hello"},
),
(
save_sms,
"app.celery.provider_tasks.deliver_sms.apply_async",
"202-867-5309",
{"template_type": "sms"},
{"template_type": TemplateType.SMS},
),
),
)
@@ -1549,8 +1549,8 @@ def test_save_tasks_use_cached_service_and_template(
@pytest.mark.parametrize(
"notification_type, task_function, expected_queue, recipient",
(
("sms", save_api_sms, QueueNames.SEND_SMS, "+447700900855"),
("email", save_api_email, QueueNames.SEND_EMAIL, "jane.citizen@example.com"),
(NotificationType.SMS, save_api_sms, QueueNames.SEND_SMS, "+447700900855",),
(NotificationType.EMAIL, save_api_email, QueueNames.SEND_EMAIL, "jane.citizen@example.com",),
),
)
def test_save_api_tasks_use_cache(

182
tests/app/dao/test_fact_billing_dao.py
View File

@@ -21,8 +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.enums import NotificationStatus
from app.models import FactBilling
from app.enums import NotificationStatus, NotificationType, TemplateType
from app.models import FactBilling, Notification
from tests.app.db import (
create_annual_billing,
create_ft_billing,
@@ -39,8 +39,8 @@ from tests.app.db import (
def set_up_yearly_data():
service = create_service()
sms_template = create_template(service=service, template_type="sms")
email_template = create_template(service=service, template_type="email")
sms_template = create_template(service=service, template_type=TemplateType.SMS)
email_template = create_template(service=service, template_type=TemplateType.EMAIL)
# use different rates for adjacent financial years to make sure the query
# doesn't accidentally bleed over into them
@@ -68,7 +68,7 @@ def set_up_yearly_data():
def set_up_yearly_data_variable_rates():
service = create_service()
sms_template = create_template(service=service, template_type="sms")
sms_template = create_template(service=service, template_type=TemplateType.SMS)
create_ft_billing(local_date="2018-05-16", template=sms_template, rate=0.162)
create_ft_billing(
@@ -93,9 +93,9 @@ def test_fetch_billing_data_for_today_includes_data_with_the_right_key_type(
notify_db_session,
):
service = create_service()
template = create_template(service=service, template_type="email")
template = create_template(service=service, template_type=TemplateType.EMAIL)
for key_type in ["normal", "test", "team"]:
create_notification(template=template, status="delivered", key_type=key_type)
create_notification(template=template, status=NotificationStatus.DELIVERED, key_type=key_type,)
today = datetime.utcnow()
results = fetch_billing_data_for_day(today.date())
@@ -103,13 +103,13 @@ def test_fetch_billing_data_for_today_includes_data_with_the_right_key_type(
assert results[0].notifications_sent == 2
@pytest.mark.parametrize("notification_type", ["email", "sms"])
@pytest.mark.parametrize("notification_type", [NotificationType.EMAIL, NotificationType.SMS])
def test_fetch_billing_data_for_day_only_calls_query_for_permission_type(
notify_db_session, notification_type
):
service = create_service(service_permissions=[notification_type])
email_template = create_template(service=service, template_type="email")
sms_template = create_template(service=service, template_type="sms")
email_template = create_template(service=service, template_type=TemplateType.EMAIL)
sms_template = create_template(service=service, template_type=TemplateType.SMS)
create_notification(template=email_template, status="delivered")
create_notification(template=sms_template, status="delivered")
today = datetime.utcnow()
@@ -119,18 +119,18 @@ def test_fetch_billing_data_for_day_only_calls_query_for_permission_type(
assert len(results) == 1
@pytest.mark.parametrize("notification_type", ["email", "sms"])
@pytest.mark.parametrize("notification_type", [NotificationType.EMAIL, NotificationType.SMS],)
def test_fetch_billing_data_for_day_only_calls_query_for_all_channels(
notify_db_session, notification_type
):
service = create_service(service_permissions=[notification_type])
email_template = create_template(service=service, template_type="email")
sms_template = create_template(service=service, template_type="sms")
create_notification(template=email_template, status="delivered")
create_notification(template=sms_template, status="delivered")
email_template = create_template(service=service, template_type=TemplateType.EMAIL)
sms_template = create_template(service=service, template_type=TemplateType.SMS)
create_notification(template=email_template, status=NotificationStatus.DELIVERED)
create_notification(template=sms_template, status=NotificationStatus.DELIVERED)
today = datetime.utcnow()
results = fetch_billing_data_for_day(
process_day=today.date(), check_permissions=False
process_day=today.date(), check_permissions=False,
)
assert len(results) == 2
@@ -141,21 +141,21 @@ def test_fetch_billing_data_for_today_includes_data_with_the_right_date(
):
process_day = datetime(2018, 4, 1, 13, 30, 0)
service = create_service()
template = create_template(service=service, template_type="email")
create_notification(template=template, status="delivered", created_at=process_day)
template = create_template(service=service, template_type=TemplateType.EMAIL)
create_notification(template=template, status=NotificationStatus.DELIVERED, created_at=process_day,)
create_notification(
template=template,
status="delivered",
status=NotificationStatus.DELIVERED,
created_at=datetime(2018, 4, 1, 4, 23, 23),
)
create_notification(
template=template,
status="delivered",
status=NotificationStatus.DELIVERED,
created_at=datetime(2018, 4, 1, 0, 23, 23),
)
create_notification(
template=template, status="sending", created_at=process_day + timedelta(days=1)
template=template, status=NotificationStatus.SENDING, created_at=process_day + timedelta(days=1,)
)
day_under_test = process_day
@@ -168,10 +168,10 @@ def test_fetch_billing_data_for_day_is_grouped_by_template_and_notification_type
notify_db_session,
):
service = create_service()
email_template = create_template(service=service, template_type="email")
sms_template = create_template(service=service, template_type="sms")
create_notification(template=email_template, status="delivered")
create_notification(template=sms_template, status="delivered")
email_template = create_template(service=service, template_type=TemplateType.EMAIL)
sms_template = create_template(service=service, template_type=TemplateType.SMS)
create_notification(template=email_template, status=NotificationStatus.DELIVERED)
create_notification(template=sms_template, status=NotificationStatus.DELIVERED)
today = datetime.utcnow()
results = fetch_billing_data_for_day(today.date())
@@ -185,8 +185,8 @@ def test_fetch_billing_data_for_day_is_grouped_by_service(notify_db_session):
service_2 = create_service(service_name="Service 2")
email_template = create_template(service=service_1)
sms_template = create_template(service=service_2)
create_notification(template=email_template, status="delivered")
create_notification(template=sms_template, status="delivered")
create_notification(template=email_template, status=NotificationStatus.DELIVERED)
create_notification(template=sms_template, status=NotificationStatus.DELIVERED)
today = datetime.utcnow()
results = fetch_billing_data_for_day(today.date())
@@ -198,8 +198,8 @@ def test_fetch_billing_data_for_day_is_grouped_by_service(notify_db_session):
def test_fetch_billing_data_for_day_is_grouped_by_provider(notify_db_session):
service = create_service()
template = create_template(service=service)
create_notification(template=template, status="delivered", sent_by="sns")
create_notification(template=template, status="delivered", sent_by="sns")
create_notification(template=template, status=NotificationStatus.DELIVERED, sent_by="sns",)
create_notification(template=template, status=NotificationStatus.DELIVERED, sent_by="sns",)
today = datetime.utcnow()
results = fetch_billing_data_for_day(today.date())
@@ -211,8 +211,8 @@ def test_fetch_billing_data_for_day_is_grouped_by_provider(notify_db_session):
def test_fetch_billing_data_for_day_is_grouped_by_rate_mulitplier(notify_db_session):
service = create_service()
template = create_template(service=service)
create_notification(template=template, status="delivered", rate_multiplier=1)
create_notification(template=template, status="delivered", rate_multiplier=2)
create_notification(template=template, status=NotificationStatus.DELIVERED, rate_multiplier=1,)
create_notification(template=template, status=NotificationStatus.DELIVERED, rate_multiplier=2,)
today = datetime.utcnow()
results = fetch_billing_data_for_day(today.date())
@@ -224,8 +224,8 @@ def test_fetch_billing_data_for_day_is_grouped_by_rate_mulitplier(notify_db_sess
def test_fetch_billing_data_for_day_is_grouped_by_international(notify_db_session):
service = create_service()
sms_template = create_template(service=service)
create_notification(template=sms_template, status="delivered", international=True)
create_notification(template=sms_template, status="delivered", international=False)
create_notification(template=sms_template, status=NotificationStatus.DELIVERED, international=True,)
create_notification(template=sms_template, status=NotificationStatus.DELIVERED, international=False,)
today = datetime.utcnow()
results = fetch_billing_data_for_day(today.date())
@@ -235,13 +235,13 @@ def test_fetch_billing_data_for_day_is_grouped_by_international(notify_db_sessio
def test_fetch_billing_data_for_day_is_grouped_by_notification_type(notify_db_session):
service = create_service()
sms_template = create_template(service=service, template_type="sms")
email_template = create_template(service=service, template_type="email")
create_notification(template=sms_template, status="delivered")
create_notification(template=sms_template, status="delivered")
create_notification(template=sms_template, status="delivered")
create_notification(template=email_template, status="delivered")
create_notification(template=email_template, status="delivered")
sms_template = create_template(service=service, template_type=TemplateType.SMS)
email_template = create_template(service=service, template_type=TemplateType.EMAIL)
create_notification(template=sms_template, status=NotificationStatus.DELIVERED)
create_notification(template=sms_template, status=NotificationStatus.DELIVERED)
create_notification(template=sms_template, status=NotificationStatus.DELIVERED)
create_notification(template=email_template, status=NotificationStatus.DELIVERED)
create_notification(template=email_template, status=NotificationStatus.DELIVERED)
today = datetime.utcnow()
results = fetch_billing_data_for_day(today.date())
@@ -259,26 +259,26 @@ def test_fetch_billing_data_for_day_returns_empty_list(notify_db_session):
def test_fetch_billing_data_for_day_uses_correct_table(notify_db_session):
service = create_service()
create_service_data_retention(
service, notification_type="email", days_of_retention=3
service, notification_type=NotificationType.EMAIL, days_of_retention=3
)
sms_template = create_template(service=service, template_type="sms")
email_template = create_template(service=service, template_type="email")
sms_template = create_template(service=service, template_type=TemplateType.SMS)
email_template = create_template(service=service, template_type=TemplateType.EMAIL)
five_days_ago = datetime.utcnow() - timedelta(days=5)
create_notification(
template=sms_template, status="delivered", created_at=five_days_ago
template=sms_template, status=NotificationStatus.DELIVERED, created_at=five_days_ago,
)
create_notification_history(
template=email_template, status="delivered", created_at=five_days_ago
template=email_template, status=NotificationStatus.DELIVERED, created_at=five_days_ago,
)
results = fetch_billing_data_for_day(
process_day=five_days_ago.date(), service_id=service.id
)
assert len(results) == 2
assert results[0].notification_type == "sms"
assert results[0].notification_type == NotificationType.SMS
assert results[0].notifications_sent == 1
assert results[1].notification_type == "email"
assert results[1].notification_type == NotificationType.EMAIL
assert results[1].notifications_sent == 1
@@ -287,8 +287,8 @@ def test_fetch_billing_data_for_day_returns_list_for_given_service(notify_db_ses
service_2 = create_service(service_name="Service 2")
template = create_template(service=service)
template_2 = create_template(service=service_2)
create_notification(template=template, status="delivered")
create_notification(template=template_2, status="delivered")
create_notification(template=template, status=NotificationStatus.DELIVERED)
create_notification(template=template_2, status=NotificationStatus.DELIVERED)
today = datetime.utcnow()
results = fetch_billing_data_for_day(
@@ -300,8 +300,8 @@ def test_fetch_billing_data_for_day_returns_list_for_given_service(notify_db_ses
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")
sms_template = create_template(service=service, template_type=TemplateType.SMS)
email_template = create_template(service=service, template_type=TemplateType.EMAIL)
for status in NotificationStatus:
create_notification(template=sms_template, status=status)
create_notification(template=email_template, status=status)
@@ -310,17 +310,17 @@ def test_fetch_billing_data_for_day_bills_correctly_for_status(notify_db_session
process_day=today.date(), service_id=service.id
)
sms_results = [x for x in results if x.notification_type == "sms"]
email_results = [x for x in results if x.notification_type == "email"]
sms_results = [x for x in results if x.notification_type == NotificationType.SMS]
email_results = [x for x in results if x.notification_type == NotificationType.EMAIL]
# we expect as many rows as we check for notification types
assert 6 == sms_results[0].notifications_sent
assert 4 == email_results[0].notifications_sent
def test_get_rates_for_billing(notify_db_session):
create_rate(start_date=datetime.utcnow(), value=12, notification_type="email")
create_rate(start_date=datetime.utcnow(), value=22, notification_type="sms")
create_rate(start_date=datetime.utcnow(), value=33, notification_type="email")
create_rate(start_date=datetime.utcnow(), value=12, notification_type=NotificationType.EMAIL)
create_rate(start_date=datetime.utcnow(), value=22, notification_type=NotificationType.SMS)
create_rate(start_date=datetime.utcnow(), value=33, notification_type=NotificationType.EMAIL)
rates = get_rates_for_billing()
assert len(rates) == 3
@@ -329,17 +329,17 @@ def test_get_rates_for_billing(notify_db_session):
@freeze_time("2017-06-01 12:00")
def test_get_rate(notify_db_session):
create_rate(
start_date=datetime(2017, 5, 30, 23, 0), value=1.2, notification_type="email"
start_date=datetime(2017, 5, 30, 23, 0), value=1.2, notification_type=NotificationType.EMAIL,
)
create_rate(
start_date=datetime(2017, 5, 30, 23, 0), value=2.2, notification_type="sms"
start_date=datetime(2017, 5, 30, 23, 0), value=2.2, notification_type=NotificationType.SMS,
)
create_rate(
start_date=datetime(2017, 5, 30, 23, 0), value=3.3, notification_type="email"
start_date=datetime(2017, 5, 30, 23, 0), value=3.3, notification_type=NotificationType.EMAIL,
)
rates = get_rates_for_billing()
rate = get_rate(rates, notification_type="sms", date=date(2017, 6, 1))
rate = get_rate(rates, notification_type=NotificationType.SMS, date=date(2017, 6, 1))
assert rate == 2.2
@@ -351,14 +351,14 @@ def test_get_rate_chooses_right_rate_depending_on_date(
notify_db_session, date, expected_rate
):
create_rate(
start_date=datetime(2016, 1, 1, 0, 0), value=1.2, notification_type="sms"
start_date=datetime(2016, 1, 1, 0, 0), value=1.2, notification_type=NotificationType.SMS,
)
create_rate(
start_date=datetime(2018, 9, 30, 23, 0), value=2.2, notification_type="sms"
start_date=datetime(2018, 9, 30, 23, 0), value=2.2, notification_type=NotificationType.SMS,
)
rates = get_rates_for_billing()
rate = get_rate(rates, "sms", date)
rate = get_rate(rates, NotificationType.SMS, date)
assert rate == expected_rate
@@ -373,7 +373,7 @@ def test_fetch_monthly_billing_for_year(notify_db_session):
print(f"RESULTS {results}")
assert str(results[0].month) == "2016-01-01"
assert results[0].notification_type == "email"
assert results[0].notification_type == NotificationType.EMAIL
assert results[0].notifications_sent == 2
assert results[0].chargeable_units == 0
assert results[0].rate == Decimal("0")
@@ -382,7 +382,7 @@ def test_fetch_monthly_billing_for_year(notify_db_session):
assert results[0].charged_units == 0
assert str(results[1].month) == "2016-01-01"
assert results[1].notification_type == "sms"
assert results[1].notification_type == NotificationType.SMS
assert results[1].notifications_sent == 2
assert results[1].chargeable_units == 2
assert results[1].rate == Decimal("0.162")
@@ -397,7 +397,7 @@ def test_fetch_monthly_billing_for_year(notify_db_session):
def test_fetch_monthly_billing_for_year_variable_rates(notify_db_session):
service = set_up_yearly_data_variable_rates()
create_annual_billing(
service_id=service.id, free_sms_fragment_limit=6, financial_year_start=2018
service_id=service.id, free_sms_fragment_limit=6, financial_year_start=2018,
)
results = fetch_monthly_billing_for_year(service.id, 2018)
@@ -405,7 +405,7 @@ def test_fetch_monthly_billing_for_year_variable_rates(notify_db_session):
assert len(results) == 2
assert str(results[0].month) == "2018-05-01"
assert results[0].notification_type == "sms"
assert results[0].notification_type == NotificationType.SMS
assert results[0].notifications_sent == 1
assert results[0].chargeable_units == 4
assert results[0].rate == Decimal("0.015")
@@ -415,7 +415,7 @@ def test_fetch_monthly_billing_for_year_variable_rates(notify_db_session):
assert results[0].charged_units == 3
assert str(results[1].month) == "2018-05-01"
assert results[1].notification_type == "sms"
assert results[1].notification_type == NotificationType.SMS
assert results[1].notifications_sent == 2
assert results[1].chargeable_units == 5
assert results[1].rate == Decimal("0.162")
@@ -428,12 +428,12 @@ def test_fetch_monthly_billing_for_year_variable_rates(notify_db_session):
@freeze_time("2018-08-01 13:30:00")
def test_fetch_monthly_billing_for_year_adds_data_for_today(notify_db_session):
service = create_service()
template = create_template(service=service, template_type="sms")
template = create_template(service=service, template_type=TemplateType.SMS)
create_rate(
start_date=datetime.utcnow() - timedelta(days=1),
value=0.158,
notification_type="sms",
notification_type=NotificationType.SMS,
)
create_annual_billing(
service_id=service.id, free_sms_fragment_limit=1000, financial_year_start=2018
@@ -442,7 +442,7 @@ def test_fetch_monthly_billing_for_year_adds_data_for_today(notify_db_session):
for i in range(1, 32):
create_ft_billing(local_date="2018-07-{}".format(i), template=template)
create_notification(template=template, status="delivered")
create_notification(template=template, status=NotificationStatus.DELIVERED)
assert db.session.query(FactBilling.local_date).count() == 31
results = fetch_monthly_billing_for_year(service_id=service.id, year=2018)
@@ -459,7 +459,7 @@ def test_fetch_billing_totals_for_year(notify_db_session):
results = fetch_billing_totals_for_year(service_id=service.id, year=2016)
assert len(results) == 2
assert results[0].notification_type == "email"
assert results[0].notification_type == NotificationType.EMAIL
assert results[0].notifications_sent == 4
assert results[0].chargeable_units == 0
assert results[0].rate == Decimal("0")
@@ -467,7 +467,7 @@ def test_fetch_billing_totals_for_year(notify_db_session):
assert results[0].free_allowance_used == 0
assert results[0].charged_units == 0
assert results[1].notification_type == "sms"
assert results[1].notification_type == NotificationType.SMS
assert results[1].notifications_sent == 4
assert results[1].chargeable_units == 4
assert results[1].rate == Decimal("0.162")
@@ -488,7 +488,7 @@ def test_fetch_billing_totals_for_year_uses_current_annual_billing(notify_db_ses
result = next(
result
for result in fetch_billing_totals_for_year(service_id=service.id, year=2016)
if result.notification_type == "sms"
if result.notification_type == NotificationType.SMS
)
assert result.chargeable_units == 4
@@ -501,13 +501,13 @@ def test_fetch_billing_totals_for_year_uses_current_annual_billing(notify_db_ses
def test_fetch_billing_totals_for_year_variable_rates(notify_db_session):
service = set_up_yearly_data_variable_rates()
create_annual_billing(
service_id=service.id, free_sms_fragment_limit=6, financial_year_start=2018
service_id=service.id, free_sms_fragment_limit=6, financial_year_start=2018,
)
results = fetch_billing_totals_for_year(service_id=service.id, year=2018)
assert len(results) == 2
assert results[0].notification_type == "sms"
assert results[0].notification_type == NotificationType.SMS
assert results[0].notifications_sent == 1
assert results[0].chargeable_units == 4
assert results[0].rate == Decimal("0.015")
@@ -516,7 +516,7 @@ def test_fetch_billing_totals_for_year_variable_rates(notify_db_session):
assert results[0].free_allowance_used == 1
assert results[0].charged_units == 3
assert results[1].notification_type == "sms"
assert results[1].notification_type == NotificationType.SMS
assert results[1].notifications_sent == 2
assert results[1].chargeable_units == 5
assert results[1].rate == Decimal("0.162")
@@ -529,9 +529,9 @@ def test_fetch_billing_totals_for_year_variable_rates(notify_db_session):
def test_delete_billing_data(notify_db_session):
service_1 = create_service(service_name="1")
service_2 = create_service(service_name="2")
sms_template = create_template(service_1, "sms")
email_template = create_template(service_1, "email")
other_service_template = create_template(service_2, "sms")
sms_template = create_template(service_1, TemplateType.SMS)
email_template = create_template(service_1, TemplateType.EMAIL)
other_service_template = create_template(service_2, TemplateType.SMS)
existing_rows_to_delete = [ # noqa
create_ft_billing("2018-01-01", sms_template, billable_unit=1),
@@ -558,13 +558,13 @@ def test_fetch_sms_free_allowance_remainder_until_date_with_two_services(
org = create_organization(name="Org for {}".format(service.name))
dao_add_service_to_organization(service=service, organization_id=org.id)
create_annual_billing(
service_id=service.id, free_sms_fragment_limit=10, financial_year_start=2016
service_id=service.id, free_sms_fragment_limit=10, financial_year_start=2016,
)
create_ft_billing(
template=template, local_date=datetime(2016, 4, 20), billable_unit=2, rate=0.11
template=template, local_date=datetime(2016, 4, 20), billable_unit=2, rate=0.11,
)
create_ft_billing(
template=template, local_date=datetime(2016, 5, 20), billable_unit=3, rate=0.11
template=template, local_date=datetime(2016, 5, 20), billable_unit=3, rate=0.11,
)
service_2 = create_service(service_name="used free allowance")
@@ -693,7 +693,7 @@ def test_fetch_sms_billing_for_all_services_with_remainder(notify_db_session):
)
service_4 = create_service(service_name="d - email only")
email_template = create_template(service=service_4, template_type="email")
email_template = create_template(service=service_4, template_type=TemplateType.EMAIL)
org_4 = create_organization(name="Org for {}".format(service_4.name))
dao_add_service_to_organization(service=service_4, organization_id=org_4.id)
create_annual_billing(
@@ -827,13 +827,13 @@ def test_fetch_usage_year_for_organization(notify_db_session):
financial_year_start=2019,
)
dao_add_service_to_organization(
service=service_with_emails_for_org, organization_id=fixtures["org_1"].id
service=service_with_emails_for_org, organization_id=fixtures["org_1"].id,
)
template = create_template(
service=service_with_emails_for_org, template_type="email"
service=service_with_emails_for_org, template_type=TemplateType.EMAIL,
)
create_ft_billing(
local_date=datetime(2019, 5, 1), template=template, notifications_sent=1100
local_date=datetime(2019, 5, 1), template=template, notifications_sent=1100,
)
results = fetch_usage_year_for_organization(fixtures["org_1"].id, 2019)
@@ -877,7 +877,7 @@ def test_fetch_usage_year_for_organization_populates_ft_billing_for_today(
create_rate(
start_date=datetime.utcnow() - timedelta(days=1),
value=0.65,
notification_type="sms",
notification_type=NotificationType.SMS,
)
new_org = create_organization(name="New organization")
service = create_service()
@@ -892,7 +892,7 @@ def test_fetch_usage_year_for_organization_populates_ft_billing_for_today(
assert FactBilling.query.count() == 0
create_notification(template=template, status="delivered")
create_notification(template=template, status=NotificationStatus.DELIVERED)
results = fetch_usage_year_for_organization(
organization_id=new_org.id, year=current_year
@@ -1039,8 +1039,8 @@ def test_fetch_usage_year_for_organization_only_returns_data_for_live_services(
live_service = create_service(restricted=False)
sms_template = create_template(service=live_service)
trial_service = create_service(restricted=True, service_name="trial_service")
email_template = create_template(service=trial_service, template_type="email")
trial_sms_template = create_template(service=trial_service, template_type="sms")
email_template = create_template(service=trial_service, template_type=TemplateType.EMAIL)
trial_sms_template = create_template(service=trial_service, template_type=TemplateType.SMS)
dao_add_service_to_organization(service=live_service, organization_id=org.id)
dao_add_service_to_organization(service=trial_service, organization_id=org.id)
create_ft_billing(