Feat/scaffold UI test user on demand

app/__init__.py

@@ -215,7 +215,7 @@ def register_blueprint(application):
         template_statistics as template_statistics_blueprint,
     )
     from app.user.rest import user_blueprint
-
+    from app.cypress.rest import cypress_blueprint
     register_notify_blueprint(application, service_blueprint, requires_admin_auth, "/service")
 
     register_notify_blueprint(application, user_blueprint, requires_admin_auth, "/user")
@@ -270,6 +270,8 @@ def register_blueprint(application):
 
     register_notify_blueprint(application, template_category_blueprint, requires_admin_auth)
 
+    register_notify_blueprint(application, cypress_blueprint, requires_admin_auth, "/cypress")
+
     register_notify_blueprint(application, cache_blueprint, requires_cache_clear_auth)
 
 

app/config.py

@@ -347,6 +347,10 @@ class Config(object):
     DEFAULT_TEMPLATE_CATEGORY_MEDIUM = "f75d6706-21b7-437e-b93a-2c0ab771e28e"
     DEFAULT_TEMPLATE_CATEGORY_HIGH = "c4f87d7c-a55b-4c0f-91fe-e56c65bb1871"
 
+    # UUIDs for Cypress tests
+    CYPRESS_SERVICE_ID = "5c8a0501-2aa8-433a-ba51-cefb8063ab93"
+    CYPRESS_TEST_USER_ID = "5e8fdc9b-4080-430d-962a-8065a1a17274"
+
     # Allowed service IDs able to send HTML through their templates.
     ALLOW_HTML_SERVICE_IDS: List[str] = [id.strip() for id in os.getenv("ALLOW_HTML_SERVICE_IDS", "").split(",")]
 

app/cypress/rest.py

@@ -0,0 +1,145 @@
+import hashlib
+import logging
+import uuid
+
+from datetime import datetime, timedelta
+from flask import Blueprint, current_app, jsonify, request
+
+from app import db
+from app.dao.services_dao import dao_add_user_to_service
+from app.dao.templates_dao import dao_update_template
+from app.dao.users_dao import save_model_user
+from app.errors import register_errors
+from app.models import LoginEvent, Permission, Service, ServiceUser, Template, TemplateHistory, TemplateRedacted, User, VerifyCode
+
+""" 
+This module will be used by the cypress tests to create users on the fly whenever a test suite is run. 
+
+Additionally, this module will also be used to clean up test users periodically to keep the data footprint small.
+"""
+
+cypress_blueprint = Blueprint("cypress", __name__)
+register_errors(cypress_blueprint)
+
+@cypress_blueprint.route("/create_user/<email_name>", methods=["POST"])
+def create_test_user(email_name):
+    data = request.get_json()
+    password = data.get('password')
+
+    if current_app.config["NOTIFY_ENVIRONMENT"] == "production":
+        return jsonify(message="Forbidden"), 403
+
+    # Create the user
+    data = {
+        "id": uuid.uuid4(),
+        "name": "Notify UI testing account",
+        "email_address": f"notify-ui-tests+{email_name}@cds-snc.ca",
+        "password": hashlib.sha256((password + current_app.config["DANGEROUS_SALT"]).encode("utf-8")).hexdigest(), #"e01221bcb56f1fb931c0ca310e2a13e23390b267b4cd80dc36366b6c9ef8eb5d", # TODO: move this to a secret!
+        "mobile_number": "9025555555",
+        "state": "active",
+        "blocked": False,
+    }
+
+    user = User(**data)
+    save_model_user(user)
+
+    # add user to cypress service w/ full permissions
+    service = Service.query.filter_by(id="5c8a0501-2aa8-433a-ba51-cefb8063ab93").first()
+    permissions = []
+    for p in ["manage_users", "manage_templates", "manage_settings", "send_texts", "send_emails", "send_letters", "manage_api_keys", "view_activity"]:
+        permissions.append(Permission(permission=p))
+
+    dao_add_user_to_service(service, user, permissions=permissions)
+
+    return jsonify(user.serialize()), 201
+
+
+def _destroy_test_user(email_name):
+    CYPRESS_TEST_USER_ID = current_app.config["CYPRESS_TEST_USER_ID"]
+
+    user = User.query.filter_by(email_address=f"notify-ui-tests+{email_name}@cds-snc.ca").first()
+
+    if not user:
+        return
+
+    try:
+        # update the created_by field for each template to use id CYPRESS_TEST_USER_ID
+        templates = Template.query.filter_by(created_by=user).all()
+        for template in templates:
+            template.created_by_id = CYPRESS_TEST_USER_ID
+            dao_update_template(template)
+
+        # update the created_by field for each template to use id CYPRESS_TEST_USER_ID
+        history_templates = TemplateHistory.query.filter_by(created_by=user).all()
+        for templateh in history_templates:
+            templateh.created_by_id = CYPRESS_TEST_USER_ID
+            db.session.add(templateh)
+
+        # update the created_by field for each template_redacted to use id CYPRESS_TEST_USER_ID
+        redacted_templates = TemplateRedacted.query.filter_by(updated_by=user).all()
+        for templater in redacted_templates:
+            templater.updated_by_id = CYPRESS_TEST_USER_ID
+            db.session.add(templater)
+
+        # Update services create by this user to use CYPRESS_TEST_USER_ID
+        services = Service.query.filter_by(created_by=user).all()
+        for service in services:
+            service.created_by_id = CYPRESS_TEST_USER_ID
+            db.session.add(service)
+
+        # remove all the login events for this user
+        LoginEvent.query.filter_by(user=user).delete()
+
+        # remove all permissions for this user
+        Permission.query.filter_by(user=user).delete()
+
+        # remove user_to_service entries
+        ServiceUser.query.filter_by(user_id=user.id).delete()
+
+        # remove verify codes
+        VerifyCode.query.filter_by(user=user).delete()
+
+        # remove the user
+        User.query.filter_by(email_address=f"notify-ui-tests+{email_name}@cds-snc.ca").delete()
+
+        print("removal success: " + email_name)
+    except Exception as e:
+        print(f"Error cleaning up test user: {e}")
+        db.session.rollback()
+
+
+"""
+Endpoint for cleaning up stale users.  This endpoint will only be used internally by the Cypress tests.
+
+This endpoint is responsible for removing stale testing users from the database. 
+Stale users are identified as users whose email addresses match the pattern "%notify-ui-tests+%@cds-snc.ca%" and whose creation time is older than three hours ago.
+
+Returns:
+    A JSON response with a success message if the cleanup is successful, or an error message if an exception occurs during the cleanup process.
+"""
+@cypress_blueprint.route("/cleanup", methods=["GET"])
+def cleanup_stale_users():
+    if current_app.config["NOTIFY_ENVIRONMENT"] == "production":
+        return jsonify(message="Forbidden"), 403
+
+    three_hours_ago = datetime.utcnow() - timedelta(hours=3)
+    users = User.query.filter(User.email_address.like(f"%notify-ui-tests+%@cds-snc.ca%"),
+                              User.created_at < three_hours_ago).all()
+
+    # get the list of email_address property from users
+    users_emails = [user for user in users]
+
+    print('users to clean: ' + str(users_emails))
+
+    # loop through users and call destroy_user on each one
+    for user in users:
+        user_email = user.email_address.split("+")[1].split("@")[0]
+        print("Trying to remove:" + user_email)
+
+        try:
+            _destroy_test_user(user_email)
+        except Exception as e:
+            return jsonify(message="Error cleaning up"), 500
+
+    db.session.commit()
+    return jsonify(message="Zeds dead, baby"), 201