darkhelm/notifications-admin
1
0
Fork 0
mirror of https://github.com/GSA/notifications-admin.git synced 2026-08-03 13:18:57 -04:00
Code Issues Packages Projects Releases Wiki Activity

flake8

Browse Source
This commit is contained in:
Kenneth Kehl
2023-12-14 13:24:34 -08:00
parent 931a48b3c6
commit 32b4e6918f
4 changed files with 268 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

183
app/main/views/send.py
View File

@@ -1,5 +1,7 @@
import itertools
from string import ascii_uppercase
import time
import uuid
from zipfile import BadZipFile
from flask import (
@@ -42,7 +44,12 @@ from app.s3_client.s3_csv_client import (
s3upload,
set_metadata_on_csv_upload,
)
from app.utils import PermanentRedirect, should_skip_template_page, unicode_truncate
from app.utils import (
PermanentRedirect,
hilite,
should_skip_template_page,
unicode_truncate,
)
from app.utils.csv import Spreadsheet, get_errors_for_csv
from app.utils.templates import get_template
from app.utils.user import user_has_permissions
@@ -328,6 +335,9 @@ def get_sender_details(service_id, template_type):
@main.route("/services/<uuid:service_id>/send/<uuid:template_id>/one-off")
@user_has_permissions("send_messages", restrict_admin_usage=True)
def send_one_off(service_id, template_id):
print(
hilite(f"ENTER send_one_off service_id {service_id} template_id {template_id}")
)
session["recipient"] = None
session["placeholders"] = {}
@@ -372,6 +382,11 @@ def get_notification_check_endpoint(service_id, template):
)
@user_has_permissions("send_messages", restrict_admin_usage=True)
def send_one_off_step(service_id, template_id, step_index):
print(
hilite(
f"ENTER send_one_off_step service_id {service_id} template_id {template_id} step_index {step_index}"
)
)
if {"recipient", "placeholders"} - set(session.keys()):
return redirect(
url_for(
@@ -706,6 +721,7 @@ def get_back_link(service_id, template, step_index, placeholders=None):
def get_skip_link(step_index, template):
print(hilite(f"ENTER get_skip_link step_index {step_index}"))
if (
request.endpoint == "main.send_one_off_step"
and step_index == 0
@@ -729,6 +745,11 @@ def get_skip_link(step_index, template):
)
@user_has_permissions("send_messages", restrict_admin_usage=True)
def send_one_off_to_myself(service_id, template_id):
print(
hilite(
f"ENTER send_one_off_to_myself service_id {service_id} template_id {template_id}"
)
)
db_template = current_service.get_template_with_user_permission_or_403(
template_id, current_user
)
@@ -835,8 +856,13 @@ def get_template_error_dict(exception):
)
@user_has_permissions("send_messages", restrict_admin_usage=True)
def send_notification(service_id, template_id):
print(
hilite(
f"ENTER send_notification service_id {service_id} template_id {template_id}"
)
)
recipient = get_recipient()
print(hilite(f"recipient {recipient}"))
if not recipient:
return redirect(
url_for(
@@ -850,34 +876,65 @@ def send_notification(service_id, template_id):
template_id, current_user
)
try:
noti = notification_api_client.send_notification(
service_id,
template_id=db_template["id"],
recipient=recipient,
personalisation=session["placeholders"],
sender_id=session.get("sender_id", None),
)
except HTTPError as exception:
current_app.logger.error(
'Service {} could not send notification: "{}"'.format(
current_service.id, exception.message
)
)
return render_template(
"views/notifications/check.html",
**_check_notification(service_id, template_id, exception),
)
print(hilite(f"SESSION PLACEHOLDERS = {session['placeholders']}"))
keys = []
values = []
for k, v in session["placeholders"].items():
keys.append(k)
values.append(v)
data = ",".join(keys)
vals = ",".join(values)
data = f"{data}\r\n{vals}"
filename = f"{uuid.uuid4()}.csv"
my_data = {"file_name": filename, "template_id": template_id, "data": data}
upload_id = s3upload(service_id, my_data)
print(hilite(f"MY UPLOAD ID {upload_id}"))
print(hilite("HERE IT IS:"))
print(hilite(s3download(service_id, upload_id)))
# column_headings = get_spreadsheet_column_headings_from_template(template)
# print(hilite(f"COLUMN HEADINGS {column_headings}"))
form = CsvUploadForm()
form.file.data = my_data
form.file.name = filename
print(hilite(f"FORM data {form.file.data} name = {form.file.name} "))
response = ""
try:
job_api_client.create_job(
upload_id,
service_id,
scheduled_for="",
template_id=template_id,
original_file_name=filename,
notification_count=1,
valid="True",
)
except Exception as e:
print(hilite(f"WHAT IS THE ERROR {e}"))
session.pop('recipient')
session.pop('placeholders')
print(hilite(f"try to get notifications for job_id = {upload_id} and service_id {service_id}"))
time.sleep(0.2)
notis = notification_api_client.get_notifications_for_service(service_id, job_id=upload_id, include_one_off=True)
print(f"HERE ARE INITIAL NOTIS {notis}")
while notis['total'] == 0:
print(hilite('retry notis'))
notis = notification_api_client.get_notifications_for_service(service_id, job_id=upload_id, include_one_off=True)
time.sleep(0.2)
print(hilite(f"HERE ARE THE NOTIS {notis}"))
session.pop("placeholders")
session.pop("recipient")
session.pop("sender_id", None)
return redirect(
url_for(
".view_notification",
service_id=service_id,
notification_id=noti["id"],
notification_id=notis["notifications"][0]["id"],
# used to show the final step of the tour (help=3) or not show
# a back link on a just sent one off notification (help=0)
help=request.args.get("help"),
@@ -916,3 +973,81 @@ def get_recipient():
return session["recipient"] or InsensitiveDict(session["placeholders"]).get(
"address line 1"
)
def send_messages_one_off_jobs(service_id, template_id):
notification_count = service_api_client.get_notification_count(service_id)
remaining_messages = current_service.message_limit - notification_count
db_template = current_service.get_template_with_user_permission_or_403(
template_id, current_user
)
email_reply_to = None
sms_sender = None
if db_template["template_type"] == "email":
email_reply_to = get_email_reply_to_address_from_session()
elif db_template["template_type"] == "sms":
sms_sender = get_sms_sender_from_session()
if db_template["template_type"] not in current_service.available_template_types:
return redirect(
url_for(
".action_blocked",
service_id=service_id,
notification_type=db_template["template_type"],
return_to="view_template",
template_id=template_id,
)
)
template = get_template(
db_template,
current_service,
show_recipient=True,
email_reply_to=email_reply_to,
sms_sender=sms_sender,
)
filename = f"{uuid.uuid4()}.csv"
my_data = {
"file_name": filename,
"template_id": template.id,
"data": "phone number\r\n16617550763\r\n16617550763\r\n16617550763\r\n16617550763\r\n16617550763",
}
upload_id = s3upload(service_id, my_data)
print(hilite(f"MY UPLOAD ID {upload_id}"))
print(hilite("HERE IT IS:"))
print(hilite(s3download(service_id, upload_id)))
column_headings = get_spreadsheet_column_headings_from_template(template)
print(hilite(f"COLUMN HEADINGS {column_headings}"))
form = CsvUploadForm()
form.file.data = my_data
form.file.name = filename
print(hilite(f"FORM data {form.file.data} name = {form.file.name} "))
response = ""
try:
job_api_client.create_job(
upload_id,
service_id,
scheduled_for="",
template_id=template_id,
original_file_name=filename,
notification_count=5,
)
except Exception as e:
print(hilite(f"WHAT IS THE ERROR {e}"))
session.pop("sender_id", None)
# return redirect(
# url_for(
# "main.service_dashboard",
# service_id=service_id,
# )
# )
raise PermanentRedirect(
url_for("main.send_messages", service_id=service_id, template_id=template_id)
)

2
app/main/views/sign_in.py
View File

@@ -174,7 +174,7 @@ def sign_in():
current_app.logger.info(
f"LOGIN_DOT_GOV_SIGNOUT_REDIRECT={os.getenv('LOGIN_DOT_GOV_SIGNOUT_REDIRECT')}"
)
initial_signin_url = os.getenv('LOGIN_DOT_GOV_INITIAL_SIGNIN_URL')
initial_signin_url = os.getenv("LOGIN_DOT_GOV_INITIAL_SIGNIN_URL")
current_app.logger.info(f"LOGIN_DOT_GOV_INITIAL_SIGNIN_URL={initial_signin_url}")
return render_template(

20
app/notify_client/job_api_client.py
View File

@@ -103,7 +103,16 @@ class JobApiClient(NotifyAdminAPIClient):
return scheduled_for
def create_job(self, job_id, service_id, scheduled_for=None):
def create_job(
self,
job_id,
service_id,
scheduled_for=None,
template_id=None,
original_file_name=None,
notification_count=None,
valid=None,
):
data = {"id": job_id}
# make a datetime object in the user's preferred timezone
@@ -112,6 +121,15 @@ class JobApiClient(NotifyAdminAPIClient):
scheduled_for = JobApiClient.convert_user_time_to_utc(scheduled_for)
data.update({"scheduled_for": scheduled_for})
if template_id:
data.update({"template_id": template_id})
if original_file_name:
data.update({"original_file_name": original_file_name})
if notification_count:
data.update({"notification_count": notification_count})
if valid:
data.update({"valid": valid})
data = _attach_current_user(data)
job = self.post(url="/service/{}/job".format(service_id), data=data)

97
tests/app/main/views/test_send.py
View File

@@ -11,6 +11,7 @@ from zipfile import BadZipFile
import pytest
from flask import url_for
from pytest_mock import mocker
from notifications_python_client.errors import HTTPError
from notifications_utils.recipients import RecipientCSV
from notifications_utils.template import SMSPreviewTemplate
@@ -30,6 +31,49 @@ from tests.conftest import (
normalize_spaces,
)
FAKE_ONE_OFF_NOTIFICATION = {'links': {},
'notifications': [
{
'api_key': None,
'billable_units': 0,
'carrier': None,
'client_reference': None,
'created_at': '2023-12-14T20:35:55+00:00',
'created_by': {'email_address': 'grsrbsrgsrf@fake.gov', 'id': 'de059e0a-42e5-48bb-939e-4f76804ab739', 'name': 'grsrbsrgsrf'},
'document_download_count': None,
'id': 'a3442b43-0ba1-4854-9e0a-d2fba1cc9b81',
'international': False,
'job': {'id': '55b242b5-9f62-4271-aff7-039e9c320578', 'original_file_name': '1127b78e-a4a8-4b70-8f4f-9f4fbf03ece2.csv'},
'job_row_number': 0,
'key_name': None,
'key_type': 'normal',
'normalised_to': '+16615555555',
'notification_type': 'sms',
'personalisation': {'dayofweek': '2', 'favecolor': '3', 'phonenumber': '+16615555555'},
'phone_prefix': '1',
'provider_response': None,
'rate_multiplier': 1.0,
'reference': None,
'reply_to_text': 'development',
'sent_at': None,
'sent_by': None,
'service': 'f62d840f-8bcb-4b36-b959-4687e16dd1a1',
'status': 'created',
'template': {
'content': '((day of week)) and ((fave color))',
'id': 'bd9caa7e-00ee-4c5a-839e-10ae1a7e6f73',
'name': 'personalized',
'redact_personalisation': False,
'subject': None,
'template_type': 'sms',
'version': 1
},
'to': '+16615555555',
'updated_at': None}],
'page_size': 50,
'total': 1
}
template_types = ["email", "sms"]
unchanging_fake_uuid = uuid.uuid4()
@@ -2064,6 +2108,11 @@ def test_route_permissions_send_check_notifications(
with client_request.session_transaction() as session:
session["recipient"] = "2028675301"
session["placeholders"] = {"name": "a"}
mocker.patch(
"app.notification_api_client.get_notifications_for_service",
return_value=FAKE_ONE_OFF_NOTIFICATION,
)
validate_route_permission_with_client(
mocker,
client_request,
@@ -2578,28 +2627,36 @@ def test_check_notification_shows_preview(
def test_send_notification_submits_data(
client_request,
fake_uuid,
mock_send_notification,
mock_get_service_template,
template,
recipient,
placeholders,
expected_personalisation,
mocker,
mock_create_job,
):
with client_request.session_transaction() as session:
session["recipient"] = recipient
session["placeholders"] = placeholders
mocker.patch(
"app.notification_api_client.get_notifications_for_service",
return_value=FAKE_ONE_OFF_NOTIFICATION,
)
client_request.post(
"main.send_notification", service_id=SERVICE_ONE_ID, template_id=fake_uuid
)
mock_send_notification.assert_called_once_with(
SERVICE_ONE_ID,
template_id=fake_uuid,
recipient=recipient,
personalisation=expected_personalisation,
sender_id=None,
)
mock_create_job.assert_called_once()
def test_send_notification_clears_session(
@@ -2608,11 +2665,18 @@ def test_send_notification_clears_session(
fake_uuid,
mock_send_notification,
mock_get_service_template,
mocker,
):
with client_request.session_transaction() as session:
session["recipient"] = "2028675301"
session["placeholders"] = {"a": "b"}
mocker.patch(
"app.notification_api_client.get_notifications_for_service",
return_value=FAKE_ONE_OFF_NOTIFICATION,
)
client_request.post(
"main.send_notification", service_id=service_one["id"], template_id=fake_uuid
)
@@ -2661,11 +2725,18 @@ def test_send_notification_redirects_to_view_page(
mock_get_service_template,
extra_args,
extra_redirect_args,
mocker,
):
with client_request.session_transaction() as session:
session["recipient"] = "2028675301"
session["placeholders"] = {"a": "b"}
mocker.patch(
"app.notification_api_client.get_notifications_for_service",
return_value=FAKE_ONE_OFF_NOTIFICATION,
)
client_request.post(
"main.send_notification",
service_id=SERVICE_ONE_ID,
@@ -2722,6 +2793,11 @@ def test_send_notification_shows_error_if_400(
class MockHTTPError(HTTPError):
message = exception_msg
mocker.patch(
"app.notification_api_client.get_notifications_for_service",
return_value=FAKE_ONE_OFF_NOTIFICATION,
)
mocker.patch(
"app.notification_api_client.send_notification",
side_effect=MockHTTPError(),
@@ -2755,6 +2831,11 @@ def test_send_notification_shows_email_error_in_trial_mode(
message = TRIAL_MODE_MSG
status_code = 400
mocker.patch(
"app.notification_api_client.get_notifications_for_service",
return_value=FAKE_ONE_OFF_NOTIFICATION,
)
mocker.patch(
"app.notification_api_client.send_notification",
side_effect=MockHTTPError(),