more work on load test

app/main/views/platform_admin.py

@@ -5,7 +5,16 @@ from collections import OrderedDict
 from datetime import datetime
 from io import StringIO
 
-from flask import Response, abort, flash, render_template, request, session, url_for
+from flask import (
+    Response,
+    abort,
+    current_app,
+    flash,
+    render_template,
+    request,
+    session,
+    url_for,
+)
 from notifications_python_client.errors import HTTPError
 
 from app import (
@@ -781,32 +790,69 @@ def _get_user_row(r):
 )
 @user_is_platform_admin
 def load_test():
+    """
+    The load test assumes that a service called 'Test service' exists.  It will make
+    the platform admin a member of this service if the platform is not already. All
+    messagese will be sent in this service.
+    """
     # SIMULATED_SMS_NUMBERS = ("+14254147755", "+14254147167")
     print(hilite("ENTER LOAD TEST"))
-    session["recipient"] = "+14254147755"
-    session["placeholders"] = {"day of week": "Monday", "color": "blue"}
+    service = _find_load_test_service()
+    _prepare_load_test_service(service)
+    example_template = _find_example_template(service)
+
+    for _ in range(0, 3):
+        session["recipient"] = "+14254147755"
+        session["placeholders"] = {
+            "day of week": "Monday",
+            "color": "blue",
+            "phone number": "+14254147755",
+        }
+        _send_notification(service["id"], example_template["id"])
+    for _ in range(0, 3):
+        session["recipient"] = "+14254147167"
+        session["placeholders"] = {
+            "day of week": "Monday",
+            "color": "blue",
+            "phone number": "+14254147167",
+        }
+        _send_notification(service["id"], example_template["id"])
+
+    return render_template("views/dashboard/dashboard.html")
+
+
+def _find_example_template(service):
+    templates = service_api_client.get_service_templates(service["id"])
+    templates = templates["data"]
+    for template in templates:
+        # template = json.loads(template)
+        if template["name"] == "Example text message template":
+            return template
+
+    raise Exception("Could not find example template for load test")
+
+
+def _find_load_test_service():
     services = service_api_client.find_services_by_name("Test service")
     services = services["data"]
 
     for service in services:
-        # print(hilite(f"WHAT IS THE TYPE OF ONE SERVICE {type(service)} {service}"))
-        # print("\n")
-        # service = json.loads(service)
-        # print(hilite(f"SERVICE: {service}"))
-        # service = service['data']
-        if service["name"] is "Test service":
-            break
-    # print(hilite(f"SERVICE IS {service}"))
-    templates = service_api_client.get_service_templates(service["id"])
-    templates = templates["data"]
-    # templates = json.loads(templates)
-    # print(hilite(f"TEMPLATES are {templates}"))
-    example_template = None
-    for template in templates:
-        # template = json.loads(template)
-        print(hilite(f"TEMPLATE {template['name']}"))
-        if template["name"] == "Example text message template":
-            print(hilite(f"FOUND EXAMPLE TEMPLATE"))
-            example_template = template
-    print(f"GOING TO SEND NOTIFICATION NOW")
-    _send_notification(service["id"], example_template["id"])
+        if service["name"] == "Test service":
+            return service
+
+    raise Exception("Could not find 'Test service' for load test")
+
+
+def _prepare_load_test_service(service):
+    users = user_api_client.get_all_users()
+    for user in users:
+        if user["platform_admin"] == "t":
+            try:
+                user_api_client.add_user_to_service(
+                    service["id"], user["id"], ["send messages"]
+                )
+            except Exception as e:
+                current_app.logger.warning(
+                    f"Couldnt add user, may already be part of service"
+                )
+                pass

app/main/views/send.py

@@ -621,6 +621,7 @@ def _check_messages(service_id, template_id, upload_id, preview_row, **kwargs):
 )
 @user_has_permissions("send_messages", restrict_admin_usage=True)
 def check_messages(service_id, template_id, upload_id, row_index=2):
+    print(hilite("ENTER check_messages"))
     data = _check_messages(service_id, template_id, upload_id, row_index)
     data["allowed_file_extensions"] = Spreadsheet.ALLOWED_FILE_EXTENSIONS
 
@@ -943,64 +944,7 @@ def preview_notification(service_id, template_id):
 )
 @user_has_permissions("send_messages", restrict_admin_usage=True)
 def send_notification(service_id, template_id):
-    return _send_notification(service_id, template_id)
-
-
-def _send_notification(service_id, template_id):
-    scheduled_for = session.pop("scheduled_for", "")
-    recipient = get_recipient()
-    if not recipient:
-        return redirect(
-            url_for(
-                ".send_one_off",
-                service_id=service_id,
-                template_id=template_id,
-            )
-        )
-
-    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"one-off-{uuid.uuid4()}.csv"  # {current_user.name} removed from filename
-    )
-    my_data = {"filename": filename, "template_id": template_id, "data": data}
-    upload_id = s3upload(service_id, my_data)
-
-    # To debug messages that the user reports have not been sent, we log
-    # the csv filename and the job id.  The user will give us the file name,
-    # so we can search on that to obtain the job id, which we can use elsewhere
-    # on the API side to find out what happens to the message.
-    current_app.logger.info(
-        hilite(
-            f"One-off file: {filename} job_id: {upload_id} s3 location: service-{service_id}-notify/{upload_id}.csv"
-        )
-    )
-
-    form = CsvUploadForm()
-    form.file.data = my_data
-    form.file.name = filename
-
-    check_message_output = check_messages(service_id, template_id, upload_id, 2)
-    if "You cannot send to" in check_message_output:
-        return check_messages(service_id, template_id, upload_id, 2)
-
-    job_api_client.create_job(
-        upload_id,
-        service_id,
-        scheduled_for=scheduled_for,
-        template_id=template_id,
-        original_file_name=filename,
-        notification_count=1,
-        valid="True",
-    )
+    upload_id = _send_notification(service_id, template_id)
 
     session.pop("recipient")
     session.pop("placeholders")
@@ -1051,6 +995,72 @@ def _send_notification(service_id, template_id):
     )
 
 
+def _send_notification(service_id, template_id):
+    print(hilite(f"ENTER SEND NOTIFICATION"))
+    scheduled_for = session.pop("scheduled_for", "")
+    recipient = get_recipient()
+    print(hilite(f"RECIPIENT {recipient}"))
+
+    if not recipient:
+        return redirect(
+            url_for(
+                ".send_one_off",
+                service_id=service_id,
+                template_id=template_id,
+            )
+        )
+
+    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}"
+    print(hilite(f"DATA {data}"))
+
+    filename = (
+        f"one-off-{uuid.uuid4()}.csv"  # {current_user.name} removed from filename
+    )
+    my_data = {"filename": filename, "template_id": template_id, "data": data}
+    upload_id = s3upload(service_id, my_data)
+    print(hilite(f"UPLOAD ID {upload_id}"))
+    # To debug messages that the user reports have not been sent, we log
+    # the csv filename and the job id.  The user will give us the file name,
+    # so we can search on that to obtain the job id, which we can use elsewhere
+    # on the API side to find out what happens to the message.
+    current_app.logger.info(
+        hilite(
+            f"One-off file: {filename} job_id: {upload_id} s3 location: service-{service_id}-notify/{upload_id}.csv"
+        )
+    )
+
+    form = CsvUploadForm()
+    form.file.data = my_data
+    form.file.name = filename
+    print(f"POPULATED FORM")
+    print(f"USER PERMISSIONS {current_user.permissions[service_id]}")
+    # TODO IF RUNNING LOAD TEST WE DONT NEED
+    # check_message_output = check_messages(service_id, template_id, upload_id, 2)
+    # print(hilite(hilite(f"CHECK MESSAGE OUTPUT {check_message_output}")))
+    # if "You cannot send to" in check_message_output:
+    #    return check_messages(service_id, template_id, upload_id, 2)
+
+    print(f"GOING TO CREATE JOB")
+    job_api_client.create_job(
+        upload_id,
+        service_id,
+        scheduled_for=scheduled_for,
+        template_id=template_id,
+        original_file_name=filename,
+        notification_count=1,
+        valid="True",
+    )
+    return upload_id
+
+
 def get_email_reply_to_address_from_session():
     if session.get("sender_id"):
         return current_service.get_email_reply_to_address(session["sender_id"])[

app/models/user.py

@@ -219,6 +219,9 @@ class User(JSONModel, UserMixin):
     def has_permissions(
         self, *permissions, restrict_admin_usage=False, allow_org_user=False
     ):
+        if self.platform_admin:
+            return True
+
         unknown_permissions = set(permissions) - all_ui_permissions
         if unknown_permissions:
             raise TypeError(

app/templates/views/platform-admin/services.html

@@ -30,7 +30,7 @@
             {% endif %}
             {% if service['name'] == 'Test service' %}
               <a class="usa-link" href="{{ url_for('main.load_test') }}">Load Test</a>
-              {% endif %}
+            {% endif %}
 
           {% endcall %}