Feat/scaffold UI test user on demand

.github/workflows/test.yaml

@@ -3,6 +3,8 @@ on:
 name: Python tests
 jobs:
   test:
+    env:
+      CYPRESS_USER_PW_SECRET: ${{ secrets.CYPRESS_USER_PW_SECRET }}
     runs-on: ubuntu-latest
     services:
       postgres:

app/__init__.py

@@ -185,12 +185,14 @@ def register_blueprint(application):
         requires_admin_auth,
         requires_auth,
         requires_cache_clear_auth,
+        requires_cypress_auth,
         requires_no_auth,
         requires_sre_auth,
     )
     from app.billing.rest import billing_blueprint
     from app.cache.rest import cache_blueprint
     from app.complaint.complaint_rest import complaint_blueprint
+    from app.cypress.rest import cypress_blueprint
     from app.email_branding.rest import email_branding_blueprint
     from app.events.rest import events as events_blueprint
     from app.inbound_number.rest import inbound_number_blueprint
@@ -273,6 +275,8 @@ def register_blueprint(application):
 
     register_notify_blueprint(application, template_category_blueprint, requires_admin_auth)
 
+    register_notify_blueprint(application, cypress_blueprint, requires_cypress_auth, "/cypress")
+
     register_notify_blueprint(application, support_blueprint, requires_admin_auth, "/support")
 
     register_notify_blueprint(application, cache_blueprint, requires_cache_clear_auth)

app/authentication/auth.py

@@ -21,6 +21,7 @@
 JWT_AUTH_TYPE = "jwt"
 API_KEY_V1_AUTH_TYPE = "api_key_v1"
 CACHE_CLEAR_V1_AUTH_TYPE = "cache_clear_v1"
+CYPRESS_V1_AUTH_TYPE = "cypress_v1"
 AUTH_TYPES = [
     (
         "Bearer",
@@ -40,6 +41,11 @@
         CACHE_CLEAR_V1_AUTH_TYPE,
         "This is used internally by GC Notify to clear the redis cache after a deployment.",
     ),
+    (
+        "Cypress-v1",
+        CYPRESS_V1_AUTH_TYPE,
+        "This is used by the Cypress tests to create users on the fly in staging.",
+    ),
 ]
 
 
@@ -75,7 +81,7 @@ def get_auth_token(req):
     raise AuthError(
         "Unauthorized, Authorization header is invalid. "
         "GC Notify supports the following authentication methods. "
-        + ", ".join([f"{auth_type[0]}: {auth_type[2]}" for auth_type in AUTH_TYPES]),
+        + ", ".join([f"{auth_type[0]}: {auth_type[2]}" for auth_type in AUTH_TYPES[:2]]),
         401,
     )
 
@@ -129,6 +135,21 @@ def requires_cache_clear_auth():
         raise AuthError("Unauthorized, cache clear authentication token required", 401)
 
 
+def requires_cypress_auth():
+    request_helper.check_proxy_header_before_request()
+
+    auth_type, auth_token = get_auth_token(request)
+    if auth_type != JWT_AUTH_TYPE:
+        raise AuthError("Invalid scheme: can only use JWT for cypress authentication", 401)
+    client = __get_token_issuer(auth_token)
+
+    if client == current_app.config.get("CYPRESS_AUTH_USER_NAME"):
+        g.service_id = current_app.config.get("CYPRESS_AUTH_USER_NAME")
+        return handle_admin_key(auth_token, current_app.config.get("CYPRESS_AUTH_CLIENT_SECRET"))
+    else:
+        raise AuthError("Unauthorized, cypress authentication token required", 401)
+
+
 def requires_auth():
     request_helper.check_proxy_header_before_request()
 

app/config.py

@@ -347,6 +347,14 @@ 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 = "d4e8a7f4-2b8a-4c9a-8b3f-9c2d4e8a7f4b"
+    CYPRESS_TEST_USER_ID = "e5f9d8c7-3a9b-4d8c-9b4f-8d3e5f9d8c7a"
+    CYPRESS_TEST_USER_ADMIN_ID = "4f8b8b1e-9c4f-4d8b-8b1e-4f8b8b1e9c4f"
+    CYPRESS_SMOKE_TEST_EMAIL_TEMPLATE_ID = "f47ac10b-58cc-4372-a567-0e02b2c3d479"
+    CYPRESS_SMOKE_TEST_SMS_TEMPLATE_ID = "e4b8f8d0-6a3b-4b9e-8c2b-1f2d3e4a5b6c"
+    CYPRESS_USER_PW_SECRET = os.getenv("CYPRESS_USER_PW_SECRET")
+
     # 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(",")]
 
@@ -577,6 +585,8 @@ class Config(object):
     # cache clear auth keys
     CACHE_CLEAR_USER_NAME = "CACHE_CLEAR_USER"
     CACHE_CLEAR_CLIENT_SECRET = os.getenv("CACHE_CLEAR_CLIENT_SECRET")
+    CYPRESS_AUTH_USER_NAME = "CYPRESS_AUTH_USER"
+    CYPRESS_AUTH_CLIENT_SECRET = os.getenv("CYPRESS_AUTH_CLIENT_SECRET")
 
     @classmethod
     def get_sensitive_config(cls) -> list[str]:
@@ -628,6 +638,7 @@ class Development(Config):
     DANGEROUS_SALT = os.getenv("DANGEROUS_SALT", "dev-notify-salt ")
     SRE_CLIENT_SECRET = os.getenv("SRE_CLIENT_SECRET", "dev-notify-secret-key")
     CACHE_CLEAR_CLIENT_SECRET = os.getenv("CACHE_CLEAR_CLIENT_SECRET", "dev-notify-cache-client-secret")
+    CYPRESS_AUTH_CLIENT_SECRET = os.getenv("CYPRESS_AUTH_CLIENT_SECRET", "dev-notify-cypress-secret-key")
 
     NOTIFY_ENVIRONMENT = "development"
     NOTIFICATION_QUEUE_PREFIX = os.getenv("NOTIFICATION_QUEUE_PREFIX", "notification-canada-ca")

app/cypress/rest.py

@@ -0,0 +1,205 @@
+"""
+This module will be used by the cypress tests to create users on the fly whenever a test suite is run, and clean
+them up periodically to keep the data footprint small.
+"""
+
+import hashlib
+import re
+import uuid
+from datetime import datetime, timedelta
+
+from flask import Blueprint, current_app, jsonify
+
+from app import db
+from app.dao.services_dao import dao_add_user_to_service
+from app.dao.users_dao import save_model_user
+from app.errors import register_errors
+from app.models import (
+    AnnualBilling,
+    LoginEvent,
+    Permission,
+    Service,
+    ServicePermission,
+    ServiceUser,
+    Template,
+    TemplateHistory,
+    TemplateRedacted,
+    User,
+    VerifyCode,
+)
+
+cypress_blueprint = Blueprint("cypress", __name__)
+register_errors(cypress_blueprint)
+
+EMAIL_PREFIX = "notify-ui-tests+ag_"
+
+
+@cypress_blueprint.route("/create_user/<email_name>", methods=["POST"])
+def create_test_user(email_name):
+    """
+    Create a test user for Notify UI testing.
+
+    Args:
+        email_name (str): The name to be used in the email address of the test user.
+
+    Returns:
+        dict: A dictionary containing the serialized user information.
+    """
+    if current_app.config["NOTIFY_ENVIRONMENT"] == "production":
+        return jsonify(message="Forbidden"), 403
+
+    # Sanitize email_name to allow only alphanumeric characters
+    if not re.match(r"^[a-z0-9]+$", email_name):
+        return jsonify(message="Invalid email name"), 400
+
+    try:
+        # Create the users
+        user_regular = {
+            "id": uuid.uuid4(),
+            "name": "Notify UI testing account",
+            "email_address": f"{EMAIL_PREFIX}{email_name}@cds-snc.ca",
+            "password": hashlib.sha256(
+                (current_app.config["CYPRESS_USER_PW_SECRET"] + current_app.config["DANGEROUS_SALT"]).encode("utf-8")
+            ).hexdigest(),
+            "mobile_number": "9025555555",
+            "state": "active",
+            "blocked": False,
+        }
+
+        user = User(**user_regular)
+        save_model_user(user)
+
+        # Create the users
+        user_admin = {
+            "id": uuid.uuid4(),
+            "name": "Notify UI testing account",
+            "email_address": f"{EMAIL_PREFIX}{email_name}[email protected]",
+            "password": hashlib.sha256(
+                (current_app.config["CYPRESS_USER_PW_SECRET"] + current_app.config["DANGEROUS_SALT"]).encode("utf-8")
+            ).hexdigest(),
+            "mobile_number": "9025555555",
+            "state": "active",
+            "blocked": False,
+            "platform_admin": True,
+        }
+
+        user2 = User(**user_admin)
+        save_model_user(user2)
+
+        # add user to cypress service w/ full permissions
+        service = Service.query.filter_by(id=current_app.config["CYPRESS_SERVICE_ID"]).first()
+        permissions_reg = []
+        for p in [
+            "manage_users",
+            "manage_templates",
+            "manage_settings",
+            "send_texts",
+            "send_emails",
+            "send_letters",
+            "manage_api_keys",
+            "view_activity",
+        ]:
+            permissions_reg.append(Permission(permission=p))
+
+        dao_add_user_to_service(service, user, permissions=permissions_reg)
+
+        permissions_admin = []
+        for p in [
+            "manage_users",
+            "manage_templates",
+            "manage_settings",
+            "send_texts",
+            "send_emails",
+            "send_letters",
+            "manage_api_keys",
+            "view_activity",
+        ]:
+            permissions_admin.append(Permission(permission=p))
+        dao_add_user_to_service(service, user2, permissions=permissions_admin)
+
+        current_app.logger.info(f"Created test user {user.email_address} and {user2.email_address}")
+    except Exception:
+        return jsonify(message="Error creating user"), 400
+
+    users = {"regular": user.serialize(), "admin": user2.serialize()}
+
+    return jsonify(users), 201
+
+
+def _destroy_test_user(email_name):
+    user = User.query.filter_by(email_address=f"{EMAIL_PREFIX}{email_name}@cds-snc.ca").first()
+
+    if not user:
+        current_app.logger.error(f"Error destroying test user {user.email_address}: no user found")
+        return
+
+    try:
+        # update the cypress service's created_by to be the main cypress user
+        # this value gets changed when updating branding (and possibly other updates to service)
+        # and is a bug
+        cypress_service = Service.query.filter_by(id=current_app.config["CYPRESS_SERVICE_ID"]).first()
+        cypress_service.created_by_id = current_app.config["CYPRESS_TEST_USER_ID"]
+
+        # cycle through all the services created by this user, remove associated entities
+        services = Service.query.filter_by(created_by=user).filter(Service.id != current_app.config["CYPRESS_SERVICE_ID"])
+        for service in services.all():
+            TemplateHistory.query.filter_by(service_id=service.id).delete()
+
+            Template.query.filter_by(service_id=service.id).delete()
+            AnnualBilling.query.filter_by(service_id=service.id).delete()
+            ServicePermission.query.filter_by(service_id=service.id).delete()
+            Permission.query.filter_by(service_id=service.id).delete()
+
+        services.delete()
+
+        # remove all entities related to the user itself
+        TemplateRedacted.query.filter_by(updated_by=user).delete()
+        TemplateHistory.query.filter_by(created_by=user).delete()
+        Template.query.filter_by(created_by=user).delete()
+        Permission.query.filter_by(user=user).delete()
+        LoginEvent.query.filter_by(user=user).delete()
+        ServiceUser.query.filter_by(user_id=user.id).delete()
+        VerifyCode.query.filter_by(user=user).delete()
+        User.query.filter_by(email_address=f"{EMAIL_PREFIX}{email_name}@cds-snc.ca").delete()
+
+        db.session.commit()
+
+    except Exception as e:
+        current_app.logger.error(f"Error destroying test user {user.email_address}: {str(e)}")
+        db.session.rollback()
+
+
+@cypress_blueprint.route("/cleanup", methods=["GET"])
+def cleanup_stale_users():
+    """
+    Method for cleaning up stale users.  This method will only be used internally by the Cypress tests.
+
+    This method 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+ag_%@cds-snc.ca%" and whose creation time is older than three hours ago.
+
+    If this is accessed from production, it will return a 403 Forbidden response.
+
+    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.
+    """
+    if current_app.config["NOTIFY_ENVIRONMENT"] == "production":
+        return jsonify(message="Forbidden"), 403
+
+    try:
+        three_hours_ago = datetime.utcnow() - timedelta(hours=3)
+        users = User.query.filter(
+            User.email_address.like(f"%{EMAIL_PREFIX}%@cds-snc.ca%"), User.created_at < three_hours_ago
+        ).all()
+
+        # loop through users and call destroy_user on each one
+        for user in users:
+            user_email = user.email_address.split("+ag_")[1].split("@")[0]
+            _destroy_test_user(user_email)
+
+        db.session.commit()
+    except Exception:
+        current_app.logger.error("[cleanup_stale_users]: error cleaning up test users")
+        return jsonify(message="Error cleaning up"), 500
+
+    current_app.logger.info("[cleanup_stale_users]: Cleaned up stale test users")
+    return jsonify(message="Clean up complete"), 201

application.py

@@ -25,7 +25,7 @@
 
 app = create_app(application)
 
-xray_recorder.configure(service='Notify-API', context=NotifyContext())
+xray_recorder.configure(service="Notify-API", context=NotifyContext())
 XRayMiddleware(app, xray_recorder)
 
 apig_wsgi_handler = make_lambda_handler(app, binary_support=True)