mirror of
https://github.com/GSA/notifications-api.git
synced 2026-08-25 00:33:41 -04:00
Cleaning up "email" from code.
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
This commit is contained in:
@@ -155,7 +155,7 @@ def test_get_yearly_usage_by_monthly_from_ft_billing(admin_request, notify_db_se
|
||||
)
|
||||
|
||||
sms_template = create_template(service=service, template_type=TemplateType.SMS)
|
||||
email_template = create_template(service=service, template_type="email")
|
||||
email_template = create_template(service=service, template_type=TemplateType.EMAIL)
|
||||
|
||||
for dt in (date(2016, 1, 28), date(2016, 8, 10), date(2016, 12, 26)):
|
||||
create_ft_billing(local_date=dt, template=sms_template, rate=0.0162)
|
||||
@@ -171,7 +171,11 @@ def test_get_yearly_usage_by_monthly_from_ft_billing(admin_request, notify_db_se
|
||||
|
||||
assert len(json_response) == 3 # 3 billed months for SMS
|
||||
|
||||
email_rows = [row for row in json_response if row["notification_type"] == "email"]
|
||||
email_rows = [
|
||||
row
|
||||
for row in json_response
|
||||
if row["notification_type"] == NotificationType.EMAIL
|
||||
]
|
||||
assert len(email_rows) == 0
|
||||
|
||||
sms_row = next(
|
||||
|
||||
@@ -104,10 +104,10 @@ def test_will_remove_csv_files_for_jobs_older_than_retention_period(
|
||||
days_of_retention=30,
|
||||
)
|
||||
sms_template_service_1 = create_template(service=service_1)
|
||||
email_template_service_1 = create_template(service=service_1, template_type="email")
|
||||
email_template_service_1 = create_template(service=service_1, template_type=TemplateType.EMAIL,)
|
||||
|
||||
sms_template_service_2 = create_template(service=service_2)
|
||||
email_template_service_2 = create_template(service=service_2, template_type="email")
|
||||
email_template_service_2 = create_template(service=service_2, template_type=TemplateType.EMAIL,)
|
||||
|
||||
four_days_ago = datetime.utcnow() - timedelta(days=4)
|
||||
eight_days_ago = datetime.utcnow() - timedelta(days=8)
|
||||
@@ -153,7 +153,7 @@ def test_delete_email_notifications_older_than_retentions_calls_child_task(
|
||||
"app.celery.nightly_tasks._delete_notifications_older_than_retention_by_type"
|
||||
)
|
||||
delete_email_notifications_older_than_retention()
|
||||
mocked_notifications.assert_called_once_with("email")
|
||||
mocked_notifications.assert_called_once_with(NotificationType.EMAIL)
|
||||
|
||||
|
||||
@freeze_time("2021-12-13T10:00")
|
||||
@@ -352,10 +352,10 @@ def test_delete_notifications_task_calls_task_for_services_that_have_sent_notifi
|
||||
create_template(service_will_delete_1)
|
||||
create_template(service_will_delete_2)
|
||||
nothing_to_delete_sms_template = create_template(
|
||||
service_nothing_to_delete, template_type=TemplateType.SMS
|
||||
service_nothing_to_delete, template_type=TemplateType.SMS,
|
||||
)
|
||||
nothing_to_delete_email_template = create_template(
|
||||
service_nothing_to_delete, template_type="email"
|
||||
service_nothing_to_delete, template_type=TemplateType.EMAIL,
|
||||
)
|
||||
|
||||
# will be deleted as service has no custom retention, but past our default 7 days
|
||||
|
||||
@@ -64,14 +64,14 @@ def test_send_delivery_status_to_service_post_https_request_to_service_with_encr
|
||||
assert request_mock.request_history[0].headers["Content-type"] == "application/json"
|
||||
assert request_mock.request_history[0].headers[
|
||||
"Authorization"
|
||||
] == "Bearer {}".format(callback_api.bearer_token)
|
||||
] == f"Bearer {callback_api.bearer_token}"
|
||||
|
||||
|
||||
def test_send_complaint_to_service_posts_https_request_to_service_with_encrypted_data(
|
||||
notify_db_session,
|
||||
):
|
||||
with freeze_time("2001-01-01T12:00:00"):
|
||||
callback_api, template = _set_up_test_data("email", "complaint")
|
||||
callback_api, template = _set_up_test_data(NotificationType.EMAIL, CallbackType.COMPLAINT,)
|
||||
|
||||
notification = create_notification(template=template)
|
||||
complaint = create_complaint(
|
||||
@@ -102,7 +102,7 @@ def test_send_complaint_to_service_posts_https_request_to_service_with_encrypted
|
||||
)
|
||||
assert request_mock.request_history[0].headers[
|
||||
"Authorization"
|
||||
] == "Bearer {}".format(callback_api.bearer_token)
|
||||
] == f"Bearer {callback_api.bearer_token}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@@ -113,7 +113,7 @@ def test_send_complaint_to_service_posts_https_request_to_service_with_encrypted
|
||||
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
|
||||
):
|
||||
callback_api, template = _set_up_test_data(notification_type, "delivery_status")
|
||||
callback_api, template = _set_up_test_data(notification_type, CallbackType.DELIVERY_STATUS,)
|
||||
datestr = datetime(2017, 6, 20)
|
||||
notification = create_notification(
|
||||
template=template,
|
||||
@@ -143,7 +143,7 @@ def test__send_data_to_service_callback_api_retries_if_request_returns_error_cod
|
||||
def test__send_data_to_service_callback_api_does_not_retry_if_request_returns_404_with_encrypted_data(
|
||||
notify_db_session, mocker, notification_type
|
||||
):
|
||||
callback_api, template = _set_up_test_data(notification_type, "delivery_status")
|
||||
callback_api, template = _set_up_test_data(notification_type, CallbackType.DELIVERY_STATUS,)
|
||||
datestr = datetime(2017, 6, 20)
|
||||
notification = create_notification(
|
||||
template=template,
|
||||
@@ -178,7 +178,7 @@ def test_send_delivery_status_to_service_succeeds_if_sent_at_is_none(
|
||||
created_at=datestr,
|
||||
updated_at=datestr,
|
||||
sent_at=None,
|
||||
status="technical-failure",
|
||||
status=NotificationStatus.TECHNICAL_FAILURE,
|
||||
)
|
||||
encrypted_data = _set_up_data_for_status_update(callback_api, notification)
|
||||
mocked = mocker.patch(
|
||||
|
||||
@@ -116,7 +116,7 @@ def test_should_process_sms_job(sample_job, mocker):
|
||||
queue="database-tasks",
|
||||
)
|
||||
job = jobs_dao.dao_get_job_by_id(sample_job.id)
|
||||
assert job.job_status == "finished"
|
||||
assert job.job_status == JobStatus.FINISHED
|
||||
|
||||
|
||||
def test_should_process_sms_job_with_sender_id(sample_job, mocker, fake_uuid):
|
||||
@@ -138,7 +138,7 @@ def test_should_process_sms_job_with_sender_id(sample_job, mocker, fake_uuid):
|
||||
|
||||
|
||||
def test_should_not_process_job_if_already_pending(sample_template, mocker):
|
||||
job = create_job(template=sample_template, job_status="scheduled")
|
||||
job = create_job(template=sample_template, job_status=JobStatus.SCHEDULED)
|
||||
|
||||
mocker.patch("app.celery.tasks.s3.get_job_and_metadata_from_s3")
|
||||
mocker.patch("app.celery.tasks.process_row")
|
||||
@@ -153,7 +153,7 @@ def test_should_process_job_if_send_limits_are_not_exceeded(
|
||||
notify_api, notify_db_session, mocker
|
||||
):
|
||||
service = create_service(message_limit=10)
|
||||
template = create_template(service=service, template_type="email")
|
||||
template = create_template(service=service, template_type=TemplateType.EMAIL)
|
||||
job = create_job(template=template, notification_count=10)
|
||||
|
||||
mocker.patch(
|
||||
@@ -169,7 +169,7 @@ def test_should_process_job_if_send_limits_are_not_exceeded(
|
||||
service_id=str(job.service.id), job_id=str(job.id)
|
||||
)
|
||||
job = jobs_dao.dao_get_job_by_id(job.id)
|
||||
assert job.job_status == "finished"
|
||||
assert job.job_status == JobStatus.FINISHED
|
||||
tasks.save_email.apply_async.assert_called_with(
|
||||
(
|
||||
str(job.service_id),
|
||||
@@ -238,7 +238,7 @@ def test_should_process_email_job(email_job_with_placeholders, mocker):
|
||||
queue="database-tasks",
|
||||
)
|
||||
job = jobs_dao.dao_get_job_by_id(email_job_with_placeholders.id)
|
||||
assert job.job_status == "finished"
|
||||
assert job.job_status == JobStatus.FINISHED
|
||||
|
||||
|
||||
def test_should_process_email_job_with_sender_id(
|
||||
@@ -293,7 +293,7 @@ def test_should_process_all_sms_job(sample_job_with_placeholdered_template, mock
|
||||
}
|
||||
assert tasks.save_sms.apply_async.call_count == 10
|
||||
job = jobs_dao.dao_get_job_by_id(sample_job_with_placeholdered_template.id)
|
||||
assert job.job_status == "finished"
|
||||
assert job.job_status == JobStatus.FINISHED
|
||||
|
||||
|
||||
# -------------- process_row tests -------------- #
|
||||
@@ -311,7 +311,7 @@ def test_process_row_sends_letter_task(
|
||||
):
|
||||
mocker.patch("app.celery.tasks.create_uuid", return_value="noti_uuid")
|
||||
task_mock = mocker.patch(
|
||||
"app.celery.tasks.{}.apply_async".format(expected_function)
|
||||
f"app.celery.tasks.{expected_function}.apply_async"
|
||||
)
|
||||
encrypt_mock = mocker.patch("app.celery.tasks.encryption.encrypt")
|
||||
template = Mock(id="template_id", template_type=template_type)
|
||||
@@ -1041,7 +1041,7 @@ def test_send_inbound_sms_to_service_post_https_request_to_service(
|
||||
assert request_mock.request_history[0].headers["Content-type"] == "application/json"
|
||||
assert request_mock.request_history[0].headers[
|
||||
"Authorization"
|
||||
] == "Bearer {}".format(inbound_api.bearer_token)
|
||||
] == f"Bearer {inbound_api.bearer_token}"
|
||||
|
||||
|
||||
def test_send_inbound_sms_to_service_does_not_send_request_when_inbound_sms_does_not_exist(
|
||||
|
||||
@@ -19,6 +19,7 @@ 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 (
|
||||
JobStatus,
|
||||
KeyType,
|
||||
NotificationStatus,
|
||||
RecipientType,
|
||||
@@ -160,7 +161,7 @@ def service_factory(sample_user):
|
||||
user=user,
|
||||
check_if_service_exists=True,
|
||||
)
|
||||
if template_type == "email":
|
||||
if template_type == TemplateType.EMAIL:
|
||||
create_template(
|
||||
service,
|
||||
template_name="Template Name",
|
||||
@@ -381,7 +382,7 @@ def sample_job(notify_db_session):
|
||||
"notification_count": 1,
|
||||
"created_at": datetime.utcnow(),
|
||||
"created_by": service.created_by,
|
||||
"job_status": "pending",
|
||||
"job_status": JobStatus.PENDING,
|
||||
"scheduled_for": None,
|
||||
"processing_started": None,
|
||||
"archived": False,
|
||||
@@ -405,7 +406,7 @@ def sample_job_with_placeholdered_template(
|
||||
def sample_scheduled_job(sample_template_with_placeholders):
|
||||
return create_job(
|
||||
sample_template_with_placeholders,
|
||||
job_status="scheduled",
|
||||
job_status=JobStatus.SCHEDULED,
|
||||
scheduled_for=(datetime.utcnow() + timedelta(minutes=60)).isoformat(),
|
||||
)
|
||||
|
||||
@@ -420,7 +421,7 @@ def sample_notification_with_job(notify_db_session):
|
||||
job=job,
|
||||
job_row_number=None,
|
||||
to_field=None,
|
||||
status="created",
|
||||
status=NotificationStatus.CREATED,
|
||||
reference=None,
|
||||
created_at=None,
|
||||
sent_at=None,
|
||||
@@ -525,7 +526,7 @@ def sample_notification_history(notify_db_session, sample_template):
|
||||
service=sample_template.service,
|
||||
template_id=sample_template.id,
|
||||
template_version=sample_template.version,
|
||||
status="created",
|
||||
status=NotificationStatus.CREATED,
|
||||
created_at=created_at,
|
||||
notification_type=notification_type,
|
||||
key_type=KeyType.NORMAL,
|
||||
@@ -572,7 +573,7 @@ def sample_expired_user(notify_db_session):
|
||||
"permissions": "send_messages,manage_service,manage_api_keys",
|
||||
"folder_permissions": ["folder_1_id", "folder_2_id"],
|
||||
"created_at": datetime.utcnow() - timedelta(days=3),
|
||||
"status": "expired",
|
||||
"status": NotificationStatus.EXPIRED,
|
||||
}
|
||||
expired_user = InvitedUser(**data)
|
||||
save_invited_user(expired_user)
|
||||
@@ -639,7 +640,7 @@ def email_2fa_code_template(notify_service):
|
||||
"((url))"
|
||||
),
|
||||
subject="Sign in to GOV.UK Notify",
|
||||
template_type="email",
|
||||
template_type=TemplateType.EMAIL,
|
||||
)
|
||||
|
||||
|
||||
@@ -650,7 +651,7 @@ def email_verification_template(notify_service):
|
||||
user=notify_service.users[0],
|
||||
template_config_name="NEW_USER_EMAIL_VERIFICATION_TEMPLATE_ID",
|
||||
content="((user_name)) use ((url)) to complete registration",
|
||||
template_type="email",
|
||||
template_type=TemplateType.EMAIL,
|
||||
)
|
||||
|
||||
|
||||
@@ -665,7 +666,7 @@ def invitation_email_template(notify_service):
|
||||
template_config_name="INVITATION_EMAIL_TEMPLATE_ID",
|
||||
content=content,
|
||||
subject="Invitation to ((service_name))",
|
||||
template_type="email",
|
||||
template_type=TemplateType.EMAIL,
|
||||
)
|
||||
|
||||
|
||||
@@ -677,7 +678,7 @@ def org_invite_email_template(notify_service):
|
||||
template_config_name="ORGANIZATION_INVITATION_EMAIL_TEMPLATE_ID",
|
||||
content="((user_name)) ((organization_name)) ((url))",
|
||||
subject="Invitation to ((organization_name))",
|
||||
template_type="email",
|
||||
template_type=TemplateType.EMAIL,
|
||||
)
|
||||
|
||||
|
||||
@@ -689,7 +690,7 @@ def password_reset_email_template(notify_service):
|
||||
template_config_name="PASSWORD_RESET_TEMPLATE_ID",
|
||||
content="((user_name)) you can reset password by clicking ((url))",
|
||||
subject="Reset your password",
|
||||
template_type="email",
|
||||
template_type=TemplateType.EMAIL,
|
||||
)
|
||||
|
||||
|
||||
@@ -701,7 +702,7 @@ def verify_reply_to_address_email_template(notify_service):
|
||||
template_config_name="REPLY_TO_EMAIL_ADDRESS_VERIFICATION_TEMPLATE_ID",
|
||||
content="Hi,This address has been provided as the reply-to email address so we are verifying if it's working",
|
||||
subject="Your GOV.UK Notify reply-to email address",
|
||||
template_type="email",
|
||||
template_type=TemplateType.EMAIL,
|
||||
)
|
||||
|
||||
|
||||
@@ -713,7 +714,7 @@ def team_member_email_edit_template(notify_service):
|
||||
template_config_name="TEAM_MEMBER_EDIT_EMAIL_TEMPLATE_ID",
|
||||
content="Hi ((name)) ((servicemanagername)) changed your email to ((email address))",
|
||||
subject="Your GOV.UK Notify email address has changed",
|
||||
template_type="email",
|
||||
template_type=TemplateType.EMAIL,
|
||||
)
|
||||
|
||||
|
||||
@@ -737,7 +738,7 @@ def already_registered_template(notify_service):
|
||||
user=notify_service.users[0],
|
||||
template_config_name="ALREADY_REGISTERED_EMAIL_TEMPLATE_ID",
|
||||
content=content,
|
||||
template_type="email",
|
||||
template_type=TemplateType.EMAIL,
|
||||
)
|
||||
|
||||
|
||||
@@ -753,7 +754,7 @@ def change_email_confirmation_template(notify_service):
|
||||
user=notify_service.users[0],
|
||||
template_config_name="CHANGE_EMAIL_CONFIRMATION_TEMPLATE_ID",
|
||||
content=content,
|
||||
template_type="email",
|
||||
template_type=TemplateType.EMAIL,
|
||||
)
|
||||
return template
|
||||
|
||||
@@ -771,7 +772,7 @@ def mou_signed_templates(notify_service):
|
||||
notify_service,
|
||||
notify_service.users[0],
|
||||
config_name,
|
||||
"email",
|
||||
TemplateType.EMAIL,
|
||||
content="\n".join(
|
||||
next(
|
||||
x
|
||||
@@ -862,7 +863,7 @@ def sample_inbound_numbers(sample_service):
|
||||
inbound_numbers.append(create_inbound_number(number="1", provider="sns"))
|
||||
inbound_numbers.append(
|
||||
create_inbound_number(
|
||||
number="2", provider="sns", active=False, service_id=service.id
|
||||
number="2", provider="sns", active=False, service_id=service.id,
|
||||
)
|
||||
)
|
||||
return inbound_numbers
|
||||
|
||||
@@ -7,6 +7,7 @@ from app.dao.complaint_dao import (
|
||||
fetch_paginated_complaints,
|
||||
save_complaint,
|
||||
)
|
||||
from app.enums import TemplateType
|
||||
from app.models import Complaint
|
||||
from tests.app.db import (
|
||||
create_complaint,
|
||||
@@ -72,8 +73,8 @@ def test_fetch_complaint_by_service_returns_empty_list(sample_service):
|
||||
def test_fetch_complaint_by_service_return_many(notify_db_session):
|
||||
service_1 = create_service(service_name="first")
|
||||
service_2 = create_service(service_name="second")
|
||||
template_1 = create_template(service=service_1, template_type="email")
|
||||
template_2 = create_template(service=service_2, template_type="email")
|
||||
template_1 = create_template(service=service_1, template_type=TemplateType.EMAIL)
|
||||
template_2 = create_template(service=service_2, template_type=TemplateType.EMAIL)
|
||||
notification_1 = create_notification(template=template_1)
|
||||
notification_2 = create_notification(template=template_2)
|
||||
notification_3 = create_notification(template=template_2)
|
||||
@@ -138,13 +139,13 @@ def test_fetch_count_of_complaints(sample_email_notification):
|
||||
)
|
||||
|
||||
count_of_complaints = fetch_count_of_complaints(
|
||||
start_date=datetime(2018, 6, 7), end_date=datetime(2018, 6, 7)
|
||||
start_date=datetime(2018, 6, 7), end_date=datetime(2018, 6, 7),
|
||||
)
|
||||
assert count_of_complaints == 5
|
||||
|
||||
|
||||
def test_fetch_count_of_complaints_returns_zero(notify_db_session):
|
||||
count_of_complaints = fetch_count_of_complaints(
|
||||
start_date=datetime(2018, 6, 7), end_date=datetime(2018, 6, 7)
|
||||
start_date=datetime(2018, 6, 7), end_date=datetime(2018, 6, 7),
|
||||
)
|
||||
assert count_of_complaints == 0
|
||||
|
||||
@@ -57,7 +57,7 @@ def test_fetch_service_data_retention_returns_empty_list_when_no_rows_for_servic
|
||||
|
||||
|
||||
def test_fetch_service_data_retention_by_id(sample_service):
|
||||
email_data_retention = insert_service_data_retention(sample_service.id, "email", 3)
|
||||
email_data_retention = insert_service_data_retention(sample_service.id, NotificationType.EMAIL, 3,)
|
||||
insert_service_data_retention(sample_service.id, NotificationType.SMS, 13)
|
||||
result = fetch_service_data_retention_by_id(
|
||||
sample_service.id, email_data_retention.id
|
||||
|
||||
@@ -511,7 +511,7 @@ def test_dao_fetch_live_services_data(sample_user):
|
||||
create_service(service_name="restricted", restricted=True)
|
||||
create_service(service_name="not_active", active=False)
|
||||
create_service(service_name="not_live", count_as_live=False)
|
||||
email_template = create_template(service=service, template_type="email")
|
||||
email_template = create_template(service=service, template_type=TemplateType.EMAIL)
|
||||
dao_add_service_to_organization(service=service, organization_id=org.id)
|
||||
# two sms billing records for 1st service within current financial year:
|
||||
create_ft_billing(local_date="2019-04-20", template=sms_template)
|
||||
@@ -1217,8 +1217,8 @@ def test_dao_fetch_todays_stats_for_all_services_only_includes_today(notify_db_s
|
||||
stats = dao_fetch_todays_stats_for_all_services()
|
||||
|
||||
stats = {row.status: row.count for row in stats}
|
||||
assert "delivered" not in stats
|
||||
assert stats["failed"] == 1
|
||||
assert NotificationStatus.DELIVERED not in stats
|
||||
assert stats[NotificationStatus.FAILED] == 1
|
||||
|
||||
|
||||
def test_dao_fetch_todays_stats_for_all_services_groups_correctly(notify_db_session):
|
||||
|
||||
@@ -141,7 +141,7 @@ def create_service(
|
||||
created_by=user
|
||||
if user
|
||||
else create_user(
|
||||
email="{}@digital.cabinet-office.gov.uk".format(uuid.uuid4())
|
||||
email=f"{uuid.uuid4()}@digital.cabinet-office.gov.uk"
|
||||
),
|
||||
prefix_sms=prefix_sms,
|
||||
organization_type=organization_type,
|
||||
@@ -212,7 +212,7 @@ def create_template(
|
||||
contact_block_id=None,
|
||||
):
|
||||
data = {
|
||||
"name": template_name or "{} Template Name".format(template_type),
|
||||
"name": template_name or f"{template_type} Template Name",
|
||||
"template_type": template_type,
|
||||
"content": content,
|
||||
"service": service,
|
||||
@@ -353,7 +353,7 @@ def create_notification_history(
|
||||
if created_at is None:
|
||||
created_at = datetime.utcnow()
|
||||
|
||||
if status != "created":
|
||||
if status != NotificationStatus.CREATED:
|
||||
sent_at = sent_at or datetime.utcnow()
|
||||
updated_at = updated_at or datetime.utcnow()
|
||||
|
||||
@@ -498,7 +498,7 @@ def create_service_callback_api(
|
||||
|
||||
|
||||
def create_email_branding(
|
||||
id=None, colour="blue", logo="test_x2.png", name="test_org_1", text="DisplayName"
|
||||
id=None, colour="blue", logo="test_x2.png", name="test_org_1", text="DisplayName",
|
||||
):
|
||||
data = {
|
||||
"colour": colour,
|
||||
@@ -529,7 +529,7 @@ def create_rate(start_date, value, notification_type):
|
||||
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_)
|
||||
name = key_name if key_name else f"{key_type} api key {id_}"
|
||||
|
||||
api_key = ApiKey(
|
||||
service=service,
|
||||
@@ -644,7 +644,7 @@ def create_organization(
|
||||
|
||||
|
||||
def create_invited_org_user(
|
||||
organization, invited_by, email_address="invite@example.com"
|
||||
organization, invited_by, email_address="invite@example.com",
|
||||
):
|
||||
invited_org_user = InvitedOrganizationUser(
|
||||
email_address=email_address,
|
||||
@@ -752,7 +752,7 @@ def create_complaint(service=None, notification=None, created_at=None):
|
||||
if not service:
|
||||
service = create_service()
|
||||
if not notification:
|
||||
template = create_template(service=service, template_type="email")
|
||||
template = create_template(service=service, template_type=TemplateType.EMAIL)
|
||||
notification = create_notification(template=template)
|
||||
|
||||
complaint = Complaint(
|
||||
@@ -907,7 +907,7 @@ def set_up_usage_data(start_date):
|
||||
financial_year_start=year,
|
||||
)
|
||||
org_1 = create_organization(
|
||||
name="Org for {}".format(service_1_sms_and_letter.name),
|
||||
name=f"Org for {service_1_sms_and_letter.name}",
|
||||
purchase_order_number="org1 purchase order number",
|
||||
billing_contact_names="org1 billing contact names",
|
||||
billing_contact_email_addresses="org1@billing.contact email@addresses.gov.uk",
|
||||
@@ -918,13 +918,13 @@ def set_up_usage_data(start_date):
|
||||
)
|
||||
|
||||
create_ft_billing(
|
||||
local_date=one_week_earlier, template=sms_template_1, billable_unit=2, rate=0.11
|
||||
local_date=one_week_earlier, template=sms_template_1, billable_unit=2, rate=0.11,
|
||||
)
|
||||
create_ft_billing(
|
||||
local_date=start_date, template=sms_template_1, billable_unit=2, rate=0.11
|
||||
local_date=start_date, template=sms_template_1, billable_unit=2, rate=0.11,
|
||||
)
|
||||
create_ft_billing(
|
||||
local_date=two_days_later, template=sms_template_1, billable_unit=1, rate=0.11
|
||||
local_date=two_days_later, template=sms_template_1, billable_unit=1, rate=0.11,
|
||||
)
|
||||
|
||||
# service with emails only:
|
||||
@@ -933,7 +933,7 @@ def set_up_usage_data(start_date):
|
||||
service=service_with_emails, template_type=TemplateType.EMAIL
|
||||
)
|
||||
org_2 = create_organization(
|
||||
name="Org for {}".format(service_with_emails.name),
|
||||
name=f"Org for {service_with_emails.name}",
|
||||
)
|
||||
dao_add_service_to_organization(
|
||||
service=service_with_emails, organization_id=org_2.id
|
||||
|
||||
@@ -119,8 +119,8 @@ def test_check_template_is_for_notification_type_fails_when_template_type_does_n
|
||||
notification_type=notification_type, template_type=template_type
|
||||
)
|
||||
assert e.value.status_code == 400
|
||||
error_message = "{0} template is not suitable for {1} notification".format(
|
||||
template_type, notification_type
|
||||
error_message = (
|
||||
f"{template_type} template is not suitable for {notification_type} notification"
|
||||
)
|
||||
assert e.value.message == error_message
|
||||
assert e.value.fields == [{"template": error_message}]
|
||||
@@ -474,7 +474,7 @@ def test_validate_template_calls_all_validators_exception_message_too_long(
|
||||
check_char_count=False,
|
||||
)
|
||||
|
||||
mock_check_type.assert_called_once_with("email", "email")
|
||||
mock_check_type.assert_called_once_with(NotificationType.EMAIL, TemplateType.EMAIL)
|
||||
mock_check_if_active.assert_called_once_with(template)
|
||||
mock_create_conent.assert_called_once_with(template, {})
|
||||
mock_check_not_empty.assert_called_once_with("content")
|
||||
@@ -499,7 +499,7 @@ def test_check_service_over_api_rate_limit_when_exceed_rate_limit_request_fails_
|
||||
check_service_over_api_rate_limit(serialised_service, serialised_api_key)
|
||||
|
||||
assert app.redis_store.exceeded_rate_limit.called_with(
|
||||
"{}-{}".format(str(sample_service.id), api_key.key_type),
|
||||
f"{sample_service.id}-{api_key.key_type}",
|
||||
sample_service.rate_limit,
|
||||
60,
|
||||
)
|
||||
@@ -527,7 +527,7 @@ def test_check_service_over_api_rate_limit_when_rate_limit_has_not_exceeded_limi
|
||||
|
||||
check_service_over_api_rate_limit(serialised_service, serialised_api_key)
|
||||
assert app.redis_store.exceeded_rate_limit.called_with(
|
||||
"{}-{}".format(str(sample_service.id), api_key.key_type), 3000, 60
|
||||
f"{sample_service.id}-{api_key.key_type}", 3000, 60,
|
||||
)
|
||||
|
||||
|
||||
@@ -583,7 +583,7 @@ def test_validate_and_format_recipient_fails_when_international_number_and_servi
|
||||
assert e.value.fields == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("key_type", ["test", "normal"])
|
||||
@pytest.mark.parametrize("key_type", [KeyType.TEST, KeyType.NORMAL])
|
||||
def test_validate_and_format_recipient_succeeds_with_international_numbers_if_service_does_allow_int_sms(
|
||||
key_type, sample_service_full_permissions
|
||||
):
|
||||
@@ -597,7 +597,7 @@ def test_validate_and_format_recipient_succeeds_with_international_numbers_if_se
|
||||
def test_validate_and_format_recipient_fails_when_no_recipient():
|
||||
with pytest.raises(BadRequestError) as e:
|
||||
validate_and_format_recipient(
|
||||
None, "key_type", "service", "NotificationType.SMS"
|
||||
None, KeyType.NORMAL, "service", NotificationType.SMS,
|
||||
)
|
||||
assert e.value.status_code == 400
|
||||
assert e.value.message == "Recipient can't be empty"
|
||||
@@ -632,8 +632,9 @@ def test_check_service_email_reply_to_id_where_service_id_is_not_found(
|
||||
assert e.value.status_code == 400
|
||||
assert (
|
||||
e.value.message
|
||||
== "email_reply_to_id {} does not exist in database for service id {}".format(
|
||||
reply_to_address.id, fake_uuid
|
||||
== (
|
||||
f"email_reply_to_id {reply_to_address.id} does not exist in database for i"
|
||||
f"service id {fake_uuid}"
|
||||
)
|
||||
)
|
||||
|
||||
@@ -648,8 +649,9 @@ def test_check_service_email_reply_to_id_where_reply_to_id_is_not_found(
|
||||
assert e.value.status_code == 400
|
||||
assert (
|
||||
e.value.message
|
||||
== "email_reply_to_id {} does not exist in database for service id {}".format(
|
||||
fake_uuid, sample_service.id
|
||||
== (
|
||||
f"email_reply_to_id {fake_uuid} does not exist in database for service "
|
||||
f"id {sample_service.id}"
|
||||
)
|
||||
)
|
||||
|
||||
@@ -683,8 +685,9 @@ def test_check_service_sms_sender_id_where_service_id_is_not_found(
|
||||
assert e.value.status_code == 400
|
||||
assert (
|
||||
e.value.message
|
||||
== "sms_sender_id {} does not exist in database for service id {}".format(
|
||||
sms_sender.id, fake_uuid
|
||||
== (
|
||||
f"sms_sender_id {sms_sender.id} does not exist in database for service "
|
||||
f"id {fake_uuid}"
|
||||
)
|
||||
)
|
||||
|
||||
@@ -697,8 +700,9 @@ def test_check_service_sms_sender_id_where_sms_sender_is_not_found(
|
||||
assert e.value.status_code == 400
|
||||
assert (
|
||||
e.value.message
|
||||
== "sms_sender_id {} does not exist in database for service id {}".format(
|
||||
fake_uuid, sample_service.id
|
||||
== (
|
||||
f"sms_sender_id {fake_uuid} does not exist in database for service "
|
||||
f"id {sample_service.id}"
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@@ -151,7 +151,7 @@ def test_find_services_by_name_handles_no_results(
|
||||
mock_get_services_by_partial_name = mocker.patch(
|
||||
"app.service.rest.get_services_by_partial_name", return_value=[]
|
||||
)
|
||||
response = admin_request.get("service.find_services_by_name", service_name="ABC")[
|
||||
response = admin_request.get("service.find_services_by_name", service_name="ABC",)[
|
||||
"data"
|
||||
]
|
||||
mock_get_services_by_partial_name.assert_called_once_with("ABC")
|
||||
@@ -330,7 +330,7 @@ def test_get_service_by_id_and_user(client, sample_service, sample_user):
|
||||
create_reply_to_email(service=sample_service, email_address="new@service.com")
|
||||
auth_header = create_admin_authorization_header()
|
||||
resp = client.get(
|
||||
"/service/{}?user_id={}".format(sample_service.id, sample_user.id),
|
||||
f"/service/{sample_service.id}?user_id={sample_user.id}",
|
||||
headers=[auth_header],
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
@@ -345,7 +345,7 @@ def test_get_service_by_id_should_404_if_no_service_for_user(notify_api, sample_
|
||||
service_id = str(uuid.uuid4())
|
||||
auth_header = create_admin_authorization_header()
|
||||
resp = client.get(
|
||||
"/service/{}?user_id={}".format(service_id, sample_user.id),
|
||||
f"/service/{service_id}?user_id={sample_user.id}",
|
||||
headers=[auth_header],
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
@@ -448,7 +448,7 @@ def test_create_service_with_domain_sets_organization(
|
||||
create_domain("fake.gov", another_org.id)
|
||||
create_domain("cabinetoffice.gov.uk", another_org.id)
|
||||
|
||||
sample_user.email_address = "test@{}".format(domain)
|
||||
sample_user.email_address = f"test@{domain}"
|
||||
|
||||
data = {
|
||||
"name": "created service",
|
||||
@@ -657,7 +657,7 @@ def test_should_not_create_service_with_duplicate_name(
|
||||
json_resp = resp.json
|
||||
assert json_resp["result"] == "error"
|
||||
assert (
|
||||
"Duplicate service name '{}'".format(sample_service.name)
|
||||
f"Duplicate service name '{sample_service.name}'"
|
||||
in json_resp["message"]["name"]
|
||||
)
|
||||
|
||||
@@ -685,7 +685,7 @@ def test_create_service_should_throw_duplicate_key_constraint_for_existing_email
|
||||
json_resp = resp.json
|
||||
assert json_resp["result"] == "error"
|
||||
assert (
|
||||
"Duplicate service name '{}'".format(service_name)
|
||||
f"Duplicate service name '{service_name}'"
|
||||
in json_resp["message"]["name"]
|
||||
)
|
||||
|
||||
@@ -712,7 +712,7 @@ def test_update_service(client, notify_db_session, sample_service):
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(sample_service.id),
|
||||
f"/service/{sample_service.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -735,7 +735,7 @@ def test_cant_update_service_org_type_to_random_value(client, sample_service):
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(sample_service.id),
|
||||
f"/service/{sample_service.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -784,7 +784,7 @@ def test_update_service_change_email_branding(
|
||||
|
||||
def test_update_service_flags(client, sample_service):
|
||||
auth_header = create_admin_authorization_header()
|
||||
resp = client.get("/service/{}".format(sample_service.id), headers=[auth_header])
|
||||
resp = client.get(f"/service/{sample_service.id}", headers=[auth_header])
|
||||
json_resp = resp.json
|
||||
assert resp.status_code == 200
|
||||
assert json_resp["data"]["name"] == sample_service.name
|
||||
@@ -793,7 +793,7 @@ def test_update_service_flags(client, sample_service):
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(sample_service.id),
|
||||
f"/service/{sample_service.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -879,7 +879,7 @@ def test_update_service_flags_with_service_without_default_service_permissions(
|
||||
}
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(service_with_no_permissions.id),
|
||||
f"/service/{service_with_no_permissions.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -911,7 +911,7 @@ def test_update_service_flags_will_remove_service_permissions(
|
||||
data = {"permissions": [ServicePermissionType.SMS, ServicePermissionType.EMAIL]}
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(service.id),
|
||||
f"/service/{service.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -935,7 +935,7 @@ def test_update_permissions_will_override_permission_flags(
|
||||
data = {"permissions": [ServicePermissionType.INTERNATIONAL_SMS]}
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(service_with_no_permissions.id),
|
||||
f"/service/{service_with_no_permissions.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -955,7 +955,7 @@ def test_update_service_permissions_will_add_service_permissions(
|
||||
data = {"permissions": [ServicePermissionType.EMAIL, ServicePermissionType.SMS]}
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(sample_service.id),
|
||||
f"/service/{sample_service.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -986,7 +986,7 @@ def test_add_service_permission_will_add_permission(
|
||||
data = {"permissions": [permission_to_add]}
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(service_with_no_permissions.id),
|
||||
f"/service/{service_with_no_permissions.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -1014,7 +1014,7 @@ def test_update_permissions_with_an_invalid_permission_will_raise_error(
|
||||
}
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(sample_service.id),
|
||||
f"/service/{sample_service.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -1023,7 +1023,7 @@ def test_update_permissions_with_an_invalid_permission_will_raise_error(
|
||||
assert resp.status_code == 400
|
||||
assert result["result"] == "error"
|
||||
assert (
|
||||
"Invalid Service Permission: '{}'".format(invalid_permission)
|
||||
f"Invalid Service Permission: '{invalid_permission}'"
|
||||
in result["message"]["permissions"]
|
||||
)
|
||||
|
||||
@@ -1042,7 +1042,7 @@ def test_update_permissions_with_duplicate_permissions_will_raise_error(
|
||||
}
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(sample_service.id),
|
||||
f"/service/{sample_service.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -1070,7 +1070,7 @@ def test_should_not_update_service_with_duplicate_name(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(sample_service.id),
|
||||
f"/service/{sample_service.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -1078,7 +1078,7 @@ def test_should_not_update_service_with_duplicate_name(
|
||||
json_resp = resp.json
|
||||
assert json_resp["result"] == "error"
|
||||
assert (
|
||||
"Duplicate service name '{}'".format(service_name)
|
||||
f"Duplicate service name '{service_name}'"
|
||||
in json_resp["message"]["name"]
|
||||
)
|
||||
|
||||
@@ -1102,7 +1102,7 @@ def test_should_not_update_service_with_duplicate_email_from(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(sample_service.id),
|
||||
f"/service/{sample_service.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -1110,9 +1110,9 @@ def test_should_not_update_service_with_duplicate_email_from(
|
||||
json_resp = resp.json
|
||||
assert json_resp["result"] == "error"
|
||||
assert (
|
||||
"Duplicate service name '{}'".format(service_name)
|
||||
f"Duplicate service name '{service_name}'"
|
||||
in json_resp["message"]["name"]
|
||||
or "Duplicate service name '{}'".format(email_from)
|
||||
or f"Duplicate service name '{email_from}'"
|
||||
in json_resp["message"]["name"]
|
||||
)
|
||||
|
||||
@@ -1127,7 +1127,7 @@ def test_update_service_should_404_if_id_is_invalid(notify_api):
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}".format(missing_service_id),
|
||||
f"/service/{missing_service_id}",
|
||||
data=json.dumps(data),
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
@@ -1141,7 +1141,7 @@ def test_get_users_by_service(notify_api, sample_service):
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.get(
|
||||
"/service/{}/users".format(sample_service.id),
|
||||
f"/service/{sample_service.id}/users",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
|
||||
@@ -1162,7 +1162,7 @@ def test_get_users_for_service_returns_empty_list_if_no_users_associated_with_se
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
response = client.get(
|
||||
"/service/{}/users".format(sample_service.id),
|
||||
f"/service/{sample_service.id}/users",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
result = json.loads(response.get_data(as_text=True))
|
||||
@@ -1179,7 +1179,7 @@ def test_get_users_for_service_returns_404_when_service_does_not_exist(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
response = client.get(
|
||||
"/service/{}/users".format(service_id),
|
||||
f"/service/{service_id}/users",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
assert response.status_code == 404
|
||||
@@ -1215,15 +1215,13 @@ def test_default_permissions_are_added_for_user_service(
|
||||
auth_header_fetch = create_admin_authorization_header()
|
||||
|
||||
resp = client.get(
|
||||
"/service/{}?user_id={}".format(
|
||||
json_resp["data"]["id"], sample_user.id
|
||||
),
|
||||
f"/service/{json_resp['data']['id']}?user_id={sample_user.id}",
|
||||
headers=[auth_header_fetch],
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
header = create_admin_authorization_header()
|
||||
response = client.get(
|
||||
url_for("user.get_user", user_id=sample_user.id), headers=[header]
|
||||
url_for("user.get_user", user_id=sample_user.id), headers=[header],
|
||||
)
|
||||
assert response.status_code == 200
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
@@ -1246,7 +1244,7 @@ def test_add_existing_user_to_another_service_with_all_permissions(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.get(
|
||||
"/service/{}/users".format(sample_service.id),
|
||||
f"/service/{sample_service.id}/users",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
|
||||
@@ -1284,7 +1282,7 @@ def test_add_existing_user_to_another_service_with_all_permissions(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}/users/{}".format(sample_service.id, user_to_add.id),
|
||||
f"/service/{sample_service.id}/users/{user_to_add.id}",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
data=json.dumps(data),
|
||||
)
|
||||
@@ -1295,7 +1293,7 @@ def test_add_existing_user_to_another_service_with_all_permissions(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.get(
|
||||
"/service/{}".format(sample_service.id),
|
||||
f"/service/{sample_service.id}",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
@@ -1312,13 +1310,13 @@ def test_add_existing_user_to_another_service_with_all_permissions(
|
||||
json_resp = resp.json
|
||||
permissions = json_resp["data"]["permissions"][str(sample_service.id)]
|
||||
expected_permissions = [
|
||||
"send_texts",
|
||||
"send_emails",
|
||||
"manage_users",
|
||||
"manage_settings",
|
||||
"manage_templates",
|
||||
"manage_api_keys",
|
||||
"view_activity",
|
||||
PermissionType.SEND_TEXTS,
|
||||
PermissionType.SEND_EMAILS,
|
||||
PermissionType.MANAGE_USERS,
|
||||
PermissionType.MANAGE_SETTINGS,
|
||||
PermissionType.MANAGE_TEMPLATES,
|
||||
PermissionType.MANAGE_API_KEYS,
|
||||
PermissionType.VIEW_ACTIVITY,
|
||||
]
|
||||
assert sorted(expected_permissions) == sorted(permissions)
|
||||
|
||||
@@ -1348,7 +1346,7 @@ def test_add_existing_user_to_another_service_with_send_permissions(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}/users/{}".format(sample_service.id, user_to_add.id),
|
||||
f"/service/{sample_service.id}/users/{user_to_add.id}",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
data=json.dumps(data),
|
||||
)
|
||||
@@ -1395,7 +1393,7 @@ def test_add_existing_user_to_another_service_with_manage_permissions(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}/users/{}".format(sample_service.id, user_to_add.id),
|
||||
f"/service/{sample_service.id}/users/{user_to_add.id}",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
data=json.dumps(data),
|
||||
)
|
||||
@@ -1446,7 +1444,7 @@ def test_add_existing_user_to_another_service_with_folder_permissions(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}/users/{}".format(sample_service.id, user_to_add.id),
|
||||
f"/service/{sample_service.id}/users/{user_to_add.id}",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
data=json.dumps(data),
|
||||
)
|
||||
@@ -1481,7 +1479,7 @@ def test_add_existing_user_to_another_service_with_manage_api_keys(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}/users/{}".format(sample_service.id, user_to_add.id),
|
||||
f"/service/{sample_service.id}/users/{user_to_add.id}",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
data=json.dumps(data),
|
||||
)
|
||||
@@ -1519,12 +1517,12 @@ def test_add_existing_user_to_non_existing_service_returns404(
|
||||
incorrect_id = uuid.uuid4()
|
||||
|
||||
data = {
|
||||
"permissions": ["send_messages", "manage_service", "manage_api_keys"]
|
||||
"permissions": [PermissionType.SEND_EMAILS, PermissionType.SEND_TEXTS, PermissionType.MANAGE_USERS, PermissionType.MANAGE_SETTINGS, PermissionType.MANAGE_TEMPLATES, PermissionType.MANAGE_API_KEYS,]
|
||||
}
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}/users/{}".format(incorrect_id, user_to_add.id),
|
||||
f"/service/{incorrect_id}/users/{user_to_add.id}",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
data=json.dumps(data),
|
||||
)
|
||||
@@ -1545,19 +1543,20 @@ def test_add_existing_user_of_service_to_service_returns400(
|
||||
existing_user_id = sample_service.users[0].id
|
||||
|
||||
data = {
|
||||
"permissions": ["send_messages", "manage_service", "manage_api_keys"]
|
||||
"permissions": [PermissionType.SEND_EMAILS, PermissionType.SEND_TEXTS, PermissionType.MANAGE_USERS, PermissionType.MANAGE_SETTINGS, PermissionType.MANAGE_TEMPLATES, PermissionType.MANAGE_API_KEYS,]
|
||||
}
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}/users/{}".format(sample_service.id, existing_user_id),
|
||||
f"/service/{sample_service.id}/users/{existing_user_id}",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
data=json.dumps(data),
|
||||
)
|
||||
|
||||
result = resp.json
|
||||
expected_message = "User id: {} already part of service id: {}".format(
|
||||
existing_user_id, sample_service.id
|
||||
expected_message = (
|
||||
f"User id: {existing_user_id} already part of service "
|
||||
f"id: {sample_service.id}"
|
||||
)
|
||||
|
||||
assert resp.status_code == 400
|
||||
@@ -1573,12 +1572,12 @@ def test_add_unknown_user_to_service_returns404(
|
||||
incorrect_id = 9876
|
||||
|
||||
data = {
|
||||
"permissions": ["send_messages", "manage_service", "manage_api_keys"]
|
||||
"permissions": [PermissionType.SEND_EMAILS, PermissionType.SEND_TEXTS, PermissionType.MANAGE_USERS, PermissionType.MANAGE_SETTINGS, PermissionType.MANAGE_TEMPLATES, PermissionType.MANAGE_API_KEYS,]
|
||||
}
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
resp = client.post(
|
||||
"/service/{}/users/{}".format(sample_service.id, incorrect_id),
|
||||
f"/service/{sample_service.id}/users/{incorrect_id}",
|
||||
headers=[("Content-Type", "application/json"), auth_header],
|
||||
data=json.dumps(data),
|
||||
)
|
||||
@@ -1660,7 +1659,7 @@ def test_get_service_and_api_key_history(notify_api, sample_service, sample_api_
|
||||
with notify_api.test_client() as client:
|
||||
auth_header = create_admin_authorization_header()
|
||||
response = client.get(
|
||||
path="/service/{}/history".format(sample_service.id),
|
||||
path=f"/service/{sample_service.id}/history",
|
||||
headers=[auth_header],
|
||||
)
|
||||
assert response.status_code == 200
|
||||
@@ -1691,7 +1690,7 @@ def test_get_all_notifications_for_service_in_order(client, notify_db_session):
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
response = client.get(
|
||||
path="/service/{}/notifications".format(service_1.id), headers=[auth_header]
|
||||
path=f"/service/{service_1.id}/notifications", headers=[auth_header]
|
||||
)
|
||||
|
||||
resp = json.loads(response.get_data(as_text=True))
|
||||
@@ -1790,9 +1789,7 @@ def test_get_all_notifications_for_service_formatted_for_csv(client, sample_temp
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
response = client.get(
|
||||
path="/service/{}/notifications?format_for_csv=True".format(
|
||||
sample_template.service_id
|
||||
),
|
||||
path=f"/service/{sample_template.service_id}/notifications?format_for_csv=True",
|
||||
headers=[auth_header],
|
||||
)
|
||||
|
||||
@@ -1809,7 +1806,7 @@ def test_get_all_notifications_for_service_formatted_for_csv(client, sample_temp
|
||||
def test_get_notification_for_service_without_uuid(client, notify_db_session):
|
||||
service_1 = create_service(service_name="1", email_from="1")
|
||||
response = client.get(
|
||||
path="/service/{}/notifications/{}".format(service_1.id, "foo"),
|
||||
path=f"/service/{service_1.id}/notifications/{'foo'}",
|
||||
headers=[create_admin_authorization_header()],
|
||||
)
|
||||
assert response.status_code == 404
|
||||
@@ -1832,7 +1829,7 @@ def test_get_notification_for_service(client, notify_db_session):
|
||||
|
||||
for notification in service_1_notifications:
|
||||
response = client.get(
|
||||
path="/service/{}/notifications/{}".format(service_1.id, notification.id),
|
||||
path=f"/service/{service_1.id}/notifications/{notification.id}",
|
||||
headers=[create_admin_authorization_header()],
|
||||
)
|
||||
resp = json.loads(response.get_data(as_text=True))
|
||||
@@ -1840,7 +1837,7 @@ def test_get_notification_for_service(client, notify_db_session):
|
||||
assert response.status_code == 200
|
||||
|
||||
service_2_response = client.get(
|
||||
path="/service/{}/notifications/{}".format(service_2.id, notification.id),
|
||||
path=f"/service/{service_2.id}/notifications/{notification.id}",
|
||||
headers=[create_admin_authorization_header()],
|
||||
)
|
||||
assert service_2_response.status_code == 404
|
||||
@@ -1909,8 +1906,9 @@ def test_get_all_notifications_for_service_including_ones_made_by_jobs(
|
||||
auth_header = create_admin_authorization_header()
|
||||
|
||||
response = client.get(
|
||||
path="/service/{}/notifications?include_from_test_key={}".format(
|
||||
sample_service.id, include_from_test_key
|
||||
path=(
|
||||
f"/service/{sample_service.id}/notifications?include_from_test_key="
|
||||
f"{include_from_test_key}"
|
||||
),
|
||||
headers=[auth_header],
|
||||
)
|
||||
@@ -2084,9 +2082,7 @@ def test_get_detailed_service(
|
||||
with freeze_time("2000-01-02T12:00:00"):
|
||||
create_notification(template=sample_template, status="created")
|
||||
resp = client.get(
|
||||
"/service/{}?detailed=True&today_only={}".format(
|
||||
sample_service.id, today_only
|
||||
),
|
||||
f"/service/{sample_service.id}?detailed=True&today_only={today_only}",
|
||||
headers=[create_admin_authorization_header()],
|
||||
)
|
||||
|
||||
@@ -2197,10 +2193,10 @@ def test_get_detailed_services_groups_by_service(notify_db_session):
|
||||
service_1_template = create_template(service_1)
|
||||
service_2_template = create_template(service_2)
|
||||
|
||||
create_notification(service_1_template, status="created")
|
||||
create_notification(service_2_template, status="created")
|
||||
create_notification(service_1_template, status="delivered")
|
||||
create_notification(service_1_template, status="created")
|
||||
create_notification(service_1_template, status=NotificationStatus.CREATED)
|
||||
create_notification(service_2_template, status=NotificationStatus.CREATED)
|
||||
create_notification(service_1_template, status=NotificationStatus.DELIVERED)
|
||||
create_notification(service_1_template, status=NotificationStatus.CREATED)
|
||||
|
||||
data = get_detailed_services(
|
||||
start_date=datetime.utcnow().date(), end_date=datetime.utcnow().date()
|
||||
@@ -2210,13 +2206,13 @@ def test_get_detailed_services_groups_by_service(notify_db_session):
|
||||
assert len(data) == 2
|
||||
assert data[0]["id"] == str(service_1.id)
|
||||
assert data[0]["statistics"] == {
|
||||
NotificationType.EMAIL.value: {"delivered": 0, "failed": 0, "requested": 0},
|
||||
NotificationType.SMS.value: {"delivered": 1, "failed": 0, "requested": 3},
|
||||
NotificationType.EMAIL.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
|
||||
NotificationType.SMS.value: {NotificationStatus.DELIVERED: 1, NotificationStatus.FAILED: 0, "requested": 3,},
|
||||
}
|
||||
assert data[1]["id"] == str(service_2.id)
|
||||
assert data[1]["statistics"] == {
|
||||
NotificationType.EMAIL.value: {"delivered": 0, "failed": 0, "requested": 0},
|
||||
NotificationType.SMS.value: {"delivered": 0, "failed": 0, "requested": 1},
|
||||
NotificationType.EMAIL.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
|
||||
NotificationType.SMS.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 1,},
|
||||
}
|
||||
|
||||
|
||||
@@ -2239,13 +2235,13 @@ def test_get_detailed_services_includes_services_with_no_notifications(
|
||||
assert len(data) == 2
|
||||
assert data[0]["id"] == str(service_1.id)
|
||||
assert data[0]["statistics"] == {
|
||||
NotificationType.EMAIL.value: {"delivered": 0, "failed": 0, "requested": 0},
|
||||
NotificationType.SMS.value: {"delivered": 0, "failed": 0, "requested": 1},
|
||||
NotificationType.EMAIL.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
|
||||
NotificationType.SMS.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 1,},
|
||||
}
|
||||
assert data[1]["id"] == str(service_2.id)
|
||||
assert data[1]["statistics"] == {
|
||||
NotificationType.EMAIL.value: {"delivered": 0, "failed": 0, "requested": 0},
|
||||
NotificationType.SMS.value: {"delivered": 0, "failed": 0, "requested": 0},
|
||||
NotificationType.EMAIL.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
|
||||
NotificationType.SMS.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
|
||||
}
|
||||
|
||||
|
||||
@@ -2265,8 +2261,8 @@ def test_get_detailed_services_only_includes_todays_notifications(sample_templat
|
||||
|
||||
assert len(data) == 1
|
||||
assert data[0]["statistics"] == {
|
||||
NotificationType.EMAIL.value: {"delivered": 0, "failed": 0, "requested": 0},
|
||||
NotificationType.SMS.value: {"delivered": 0, "failed": 0, "requested": 3},
|
||||
NotificationType.EMAIL.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
|
||||
NotificationType.SMS.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 3,},
|
||||
}
|
||||
|
||||
|
||||
@@ -2311,13 +2307,13 @@ def test_get_detailed_services_for_date_range(
|
||||
|
||||
assert len(data) == 1
|
||||
assert data[0]["statistics"][NotificationType.EMAIL.value] == {
|
||||
"delivered": 0,
|
||||
"failed": 0,
|
||||
NotificationStatus.DELIVERED: 0,
|
||||
NotificationStatus.FAILED: 0,
|
||||
"requested": 0,
|
||||
}
|
||||
assert data[0]["statistics"][NotificationType.SMS.value] == {
|
||||
"delivered": 2,
|
||||
"failed": 0,
|
||||
NotificationStatus.DELIVERED: 2,
|
||||
NotificationStatus.FAILED: 0,
|
||||
"requested": 2,
|
||||
}
|
||||
|
||||
@@ -2338,9 +2334,8 @@ def test_search_for_notification_by_to_field(
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
"/service/{}/notifications?to={}&template_type={}".format(
|
||||
notification1.service_id, "jack@gmail.com", "email"
|
||||
),
|
||||
f"/service/{notification1.service_id}/notifications?to={'jack@gmail.com'}"
|
||||
f"&template_type={TemplateType.EMAIL}",
|
||||
headers=[create_admin_authorization_header()],
|
||||
)
|
||||
notifications = json.loads(response.get_data(as_text=True))["notifications"]
|
||||
@@ -2377,16 +2372,16 @@ def test_search_for_notification_by_to_field_return_multiple_matches(
|
||||
client, sample_template, sample_email_template
|
||||
):
|
||||
notification1 = create_notification(
|
||||
sample_template, to_field="+447700900855", normalised_to="447700900855"
|
||||
sample_template, to_field="+447700900855", normalised_to="447700900855",
|
||||
)
|
||||
notification2 = create_notification(
|
||||
sample_template, to_field=" +44 77009 00855 ", normalised_to="447700900855"
|
||||
sample_template, to_field=" +44 77009 00855 ", normalised_to="447700900855",
|
||||
)
|
||||
notification3 = create_notification(
|
||||
sample_template, to_field="+44770 0900 855", normalised_to="447700900855"
|
||||
sample_template, to_field="+44770 0900 855", normalised_to="447700900855",
|
||||
)
|
||||
notification4 = create_notification(
|
||||
sample_email_template, to_field="jack@gmail.com", normalised_to="jack@gmail.com"
|
||||
sample_email_template, to_field="jack@gmail.com", normalised_to="jack@gmail.com",
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
@@ -2414,7 +2409,7 @@ def test_search_for_notification_by_to_field_returns_next_link_if_more_than_50(
|
||||
):
|
||||
for _ in range(51):
|
||||
create_notification(
|
||||
sample_template, to_field="+447700900855", normalised_to="447700900855"
|
||||
sample_template, to_field="+447700900855", normalised_to="447700900855",
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
@@ -2438,7 +2433,7 @@ def test_search_for_notification_by_to_field_returns_no_next_link_if_50_or_less(
|
||||
):
|
||||
for _ in range(50):
|
||||
create_notification(
|
||||
sample_template, to_field="+447700900855", normalised_to="447700900855"
|
||||
sample_template, to_field="+447700900855", normalised_to="447700900855",
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
@@ -2492,7 +2487,7 @@ def test_update_service_does_not_call_send_notification_for_live_service(
|
||||
|
||||
auth_header = create_admin_authorization_header()
|
||||
resp = client.post(
|
||||
"service/{}".format(sample_service.id),
|
||||
f"service/{sample_service.id}",
|
||||
data=json.dumps(data),
|
||||
headers=[auth_header],
|
||||
content_type="application/json",
|
||||
@@ -3035,7 +3030,7 @@ def test_get_email_reply_to_address(client, notify_db_session):
|
||||
reply_to = create_reply_to_email(service, "test_a@mail.com")
|
||||
|
||||
response = client.get(
|
||||
"/service/{}/email-reply-to/{}".format(service.id, reply_to.id),
|
||||
f"/service/{service.id}/email-reply-to/{reply_to.id}",
|
||||
headers=[
|
||||
("Content-Type", "application/json"),
|
||||
create_admin_authorization_header(),
|
||||
@@ -3053,7 +3048,7 @@ def test_add_service_sms_sender_can_add_multiple_senders(client, notify_db_sessi
|
||||
"is_default": False,
|
||||
}
|
||||
response = client.post(
|
||||
"/service/{}/sms-sender".format(service.id),
|
||||
f"/service/{service.id}/sms-sender",
|
||||
data=json.dumps(data),
|
||||
headers=[
|
||||
("Content-Type", "application/json"),
|
||||
|
||||
@@ -42,7 +42,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", RecipientType.MOBILE, mobile_number
|
||||
"service_id", RecipientType.MOBILE, mobile_number,
|
||||
)
|
||||
|
||||
assert service_guest_list.recipient == mobile_number
|
||||
@@ -51,7 +51,7 @@ def test_should_build_service_guest_list_from_mobile_number(mobile_number):
|
||||
@pytest.mark.parametrize("email_address", ["test@example.com"])
|
||||
def test_should_build_service_guest_list_from_email_address(email_address):
|
||||
service_guest_list = ServiceGuestList.from_string(
|
||||
"service_id", RecipientType.EMAIL, email_address
|
||||
"service_id", RecipientType.EMAIL, email_address,
|
||||
)
|
||||
|
||||
assert service_guest_list.recipient == email_address
|
||||
@@ -220,7 +220,7 @@ def test_notification_subject_is_none_for_sms(sample_service):
|
||||
assert notification.subject is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("template_type", ["email"])
|
||||
@pytest.mark.parametrize("template_type", [TemplateType.EMAIL])
|
||||
def test_notification_subject_fills_in_placeholders(sample_service, template_type):
|
||||
template = create_template(
|
||||
service=sample_service, template_type=template_type, subject="((name))"
|
||||
|
||||
Reference in New Issue
Block a user