diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 1ca6b7d279..037f12dbb5 100644 --- a/.github/workflows/test.yaml +++ b/.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: diff --git a/app/__init__.py b/app/__init__.py index 69c265132e..e06e5914fe 100644 --- a/app/__init__.py +++ b/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) diff --git a/app/authentication/auth.py b/app/authentication/auth.py index 8b627f8919..23e708691c 100644 --- a/app/authentication/auth.py +++ b/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() diff --git a/app/config.py b/app/config.py index aa6361f56e..fcc54ef641 100644 --- a/app/config.py +++ b/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") diff --git a/app/cypress/__init__.py b/app/cypress/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/app/cypress/rest.py b/app/cypress/rest.py new file mode 100644 index 0000000000..956c684b31 --- /dev/null +++ b/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/", 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}_admin@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, + "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 diff --git a/application.py b/application.py index 7235426b1e..7f554ecd3f 100644 --- a/application.py +++ b/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) diff --git a/migrations/versions/0466_add_cypress_data.py b/migrations/versions/0466_add_cypress_data.py new file mode 100644 index 0000000000..a5a3bb2749 --- /dev/null +++ b/migrations/versions/0466_add_cypress_data.py @@ -0,0 +1,140 @@ +"""empty message + +Revision ID: 0466_add_cypress_data +Revises: 0465_add_constraints +Create Date: 2016-06-01 14:17:01.963181 + +""" + +import hashlib +import uuid + +# revision identifiers, used by Alembic. +from datetime import datetime + +from alembic import op +from flask import current_app + +from app.dao.date_util import get_current_financial_year_start_year +from app.encryption import hashpw +from app.models import PERMISSION_LIST + +revision = "0466_add_cypress_data" +down_revision = "0465_add_constraints" + +user_id = current_app.config["CYPRESS_TEST_USER_ID"] +admin_user_id = current_app.config["CYPRESS_TEST_USER_ADMIN_ID"] +service_id = current_app.config["CYPRESS_SERVICE_ID"] +email_template_id = current_app.config["CYPRESS_SMOKE_TEST_EMAIL_TEMPLATE_ID"] +sms_template_id = current_app.config["CYPRESS_SMOKE_TEST_SMS_TEMPLATE_ID"] +default_category_id = current_app.config["DEFAULT_TEMPLATE_CATEGORY_LOW"] + + +def upgrade(): + password = hashpw( + hashlib.sha256( + (current_app.config["CYPRESS_USER_PW_SECRET"] + current_app.config["DANGEROUS_SALT"]).encode("utf-8") + ).hexdigest() + ) + current_year = get_current_financial_year_start_year() + default_limit = 250000 + + op.get_bind() + + # insert test user + user_insert = """INSERT INTO users (id, name, email_address, created_at, failed_login_count, _password, mobile_number, password_changed_at, state, platform_admin, auth_type) + VALUES ('{}', 'Notify UI test user', 'notify-ui-tests+regular_user@cds-snc.ca', '{}', 0,'{}', '+441234123412', '{}', 'active', False, 'email_auth') + """ + op.execute(user_insert.format(user_id, datetime.utcnow(), password, datetime.utcnow())) + # insert test user thats platform admin + user_insert = """INSERT INTO users (id, name, email_address, created_at, failed_login_count, _password, mobile_number, password_changed_at, state, platform_admin, auth_type) + VALUES ('{}', 'Notify UI test user', 'notify-ui-tests+platform_admin@cds-snc.ca', '{}', 0,'{}', '+441234123412', '{}', 'active', True, 'email_auth') + """ + op.execute(user_insert.format(admin_user_id, datetime.utcnow(), password, datetime.utcnow())) + + # insert test service + service_history_insert = """INSERT INTO services_history (id, name, created_at, active, message_limit, restricted, research_mode, email_from, created_by_id, sms_daily_limit, prefix_sms, organisation_type, version) + VALUES ('{}', 'Cypress UI Testing Service', '{}', True, 10000, False, False, 'notify@digital.cabinet-office.gov.uk', + '{}', 10000, True, 'central', 1) + """ + op.execute(service_history_insert.format(service_id, datetime.utcnow(), user_id)) + service_insert = """INSERT INTO services (id, name, created_at, active, message_limit, restricted, research_mode, email_from, created_by_id, sms_daily_limit, prefix_sms, organisation_type, version) + VALUES ('{}', 'Cypress UI Testing Service', '{}', True, 10000, False, False, 'notify@digital.cabinet-office.gov.uk', + '{}', 10000, True, 'central', 1) + """ + op.execute(service_insert.format(service_id, datetime.utcnow(), user_id)) + + for send_type in ("sms", "email"): + service_perms_insert = """INSERT INTO service_permissions (service_id, permission, created_at) + VALUES ('{}', '{}', '{}')""" + op.execute(service_perms_insert.format(service_id, send_type, datetime.utcnow())) + + insert_row_if_not_exist = """INSERT INTO annual_billing (id, service_id, financial_year_start, free_sms_fragment_limit, created_at, updated_at) + VALUES ('{}', '{}', {}, {}, '{}', '{}') + """ + op.execute( + insert_row_if_not_exist.format( + uuid.uuid4(), service_id, current_year, default_limit, datetime.utcnow(), datetime.utcnow() + ) + ) + + user_to_service_insert = """INSERT INTO user_to_service (user_id, service_id) VALUES ('{}', '{}')""" + op.execute(user_to_service_insert.format(user_id, service_id)) + + for permission in PERMISSION_LIST: + perms_insert = ( + """INSERT INTO permissions (id, service_id, user_id, permission, created_at) VALUES ('{}', '{}', '{}', '{}', '{}')""" + ) + op.execute(perms_insert.format(uuid.uuid4(), service_id, user_id, permission, datetime.utcnow())) + + # insert test email template + _insert_template(email_template_id, "SMOKE_TEST_EMAIL", "SMOKE_TEST_EMAIL", "email", "SMOKE_TEST_EMAIL", default_category_id) + + # insert test SMS template + _insert_template(sms_template_id, "SMOKE_TEST_SMS", "SMOKE_TEST_SMS", "sms", None, default_category_id) + + # insert 10 random email templates + for i in range(10): + _insert_template( + uuid.uuid4(), "Template {}".format(i), "Template {}".format(i), "email", "Template {}".format(i), default_category_id + ) + + # insert 1 random sms template + _insert_template(uuid.uuid4(), "Template 11", "Template 11", "sms", "Template 11", "b6c42a7e-2a26-4a07-802b-123a5c3198a9") + + +def downgrade(): + op.get_bind() + op.execute("delete from permissions where service_id = '{}'".format(service_id)) + op.execute("delete from annual_billing where service_id = '{}'".format(service_id)) + op.execute("delete from service_permissions where service_id = '{}'".format(service_id)) + op.execute("delete from login_events where user_id = '{}'".format(user_id)) + op.execute("delete from verify_codes where user_id = '{}'".format(user_id)) + op.execute("delete from login_events where user_id = '{}'".format(admin_user_id)) + op.execute("delete from verify_codes where user_id = '{}'".format(admin_user_id)) + op.execute("delete from templates where service_id = '{}'".format(service_id)) + op.execute("delete from templates_history where service_id = '{}'".format(service_id)) + op.execute("delete from user_to_service where service_id = '{}'".format(service_id)) + op.execute("delete from services_history where id = '{}'".format(service_id)) + op.execute("delete from services where id = '{}'".format(service_id)) + op.execute("delete from users where id = '{}'".format(user_id)) + op.execute("delete from users where id = '{}'".format(admin_user_id)) + + +def _insert_template(id, name, content, type, subject, category_id): + template_history_insert = """INSERT INTO templates_history (id, name, template_type, created_at, + content, archived, service_id, + subject, created_by_id, hidden, template_category_id, version) + VALUES ('{}', '{}', '{}', '{}', '{}', False, '{}', '{}', '{}', False, '{}', 1) + """ + template_insert = """INSERT INTO templates (id, name, template_type, created_at, + content, archived, service_id, subject, created_by_id, hidden, template_category_id, version) + VALUES ('{}', '{}', '{}', '{}', '{}', False, '{}', '{}', '{}', False, '{}', 1) + """ + + op.execute( + template_history_insert.format( + uuid.uuid4(), name, type, datetime.utcnow(), content, service_id, subject, user_id, category_id + ) + ) + op.execute(template_insert.format(id, name, type, datetime.utcnow(), content, service_id, subject, user_id, category_id)) diff --git a/run_celery.py b/run_celery.py index c8381752d4..ce470fc06b 100644 --- a/run_celery.py +++ b/run_celery.py @@ -17,7 +17,7 @@ application = Flask("celery") create_app(application) -xray_recorder.configure(service='Notify', context=NotifyContext()) +xray_recorder.configure(service="Notify", context=NotifyContext()) XRayMiddleware(application, xray_recorder) application.app_context().push() diff --git a/tests/__init__.py b/tests/__init__.py index af1cad00e9..457d133e2c 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -50,6 +50,14 @@ def create_cache_clear_authorization_header(): return "Authorization", "Bearer {}".format(token) +def create_cypress_authorization_header(): + client_id = current_app.config["CYPRESS_AUTH_USER_NAME"] + secret = current_app.config["CYPRESS_AUTH_CLIENT_SECRET"] + + token = create_jwt_token(secret=secret, client_id=client_id) + return "Authorization", "Bearer {}".format(token) + + def unwrap_function(fn): """ Given a function, returns its undecorated original. diff --git a/tests/app/authentication/test_authentication.py b/tests/app/authentication/test_authentication.py index 1b2e83c3d2..913a7ef0f6 100644 --- a/tests/app/authentication/test_authentication.py +++ b/tests/app/authentication/test_authentication.py @@ -44,7 +44,6 @@ def test_should_not_allow_request_with_incorrect_header(client, auth_fn): + "GC Notify supports the following authentication methods. " + f"{AUTH_TYPES[0][0]}: {AUTH_TYPES[0][2]}" + f", {AUTH_TYPES[1][0]}: {AUTH_TYPES[1][2]}" - + f", {AUTH_TYPES[2][0]}: {AUTH_TYPES[2][2]}" ) diff --git a/tests/app/conftest.py b/tests/app/conftest.py index cb9f1a4601..8dbfd4c5b7 100644 --- a/tests/app/conftest.py +++ b/tests/app/conftest.py @@ -124,6 +124,14 @@ def notify_user(notify_db_session): ) +@pytest.fixture(scope="function") +def cypress_user(notify_db_session): + return create_user( + email="cypress-service-user@cds-snc.ca", + id_=current_app.config["CYPRESS_TEST_USER_ID"], + ) + + def create_code(notify_db, notify_db_session, code_type, usr=None, code=None): if code is None: code = create_secret_code() @@ -215,6 +223,42 @@ def sample_service( ) +@pytest.fixture(scope="function") +def sample_service_cypress( + notify_db, + notify_db_session, +): + user = create_user() + service = Service.query.get(current_app.config["CYPRESS_SERVICE_ID"]) + if not service: + service = Service( + name="Cypress Service", + message_limit=1000, + sms_daily_limit=1000, + restricted=False, + email_from="notify.service", + created_by=user, + prefix_sms=False, + ) + dao_create_service( + service=service, + service_id=current_app.config["CYPRESS_SERVICE_ID"], + user=user, + ) + + data = { + "service": service, + "email_address": "notify@gov.uk", + "is_default": True, + } + reply_to = ServiceEmailReplyTo(**data) + + db.session.add(reply_to) + db.session.commit() + + return service, user + + @pytest.fixture(scope="function", name="sample_service_full_permissions") def _sample_service_full_permissions(notify_db_session): service = create_service( diff --git a/tests/app/cypress/__init__.py b/tests/app/cypress/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/app/cypress/test_rest.py b/tests/app/cypress/test_rest.py new file mode 100644 index 0000000000..079d1de05d --- /dev/null +++ b/tests/app/cypress/test_rest.py @@ -0,0 +1,99 @@ +import json +from datetime import datetime, timedelta + +from app.models import User +from tests import create_cypress_authorization_header +from tests.conftest import set_config_values + +EMAIL_PREFIX = "notify-ui-tests+ag_" + + +def test_create_test_user(client, sample_service_cypress): + auth_header = create_cypress_authorization_header() + + resp = client.post( + "/cypress/create_user/{}".format("emailsuffix"), + headers=[auth_header], + content_type="application/json", + ) + + data = json.loads(resp.data) + + assert resp.status_code == 201 + assert "regular" in data + assert "admin" in data + assert data["regular"]["email_address"] == f"{EMAIL_PREFIX}emailsuffix@cds-snc.ca" + assert data["admin"]["email_address"] == f"{EMAIL_PREFIX}emailsuffix_admin@cds-snc.ca" + + # verify users were created in the DB + user = User.query.filter_by(email_address=f"{EMAIL_PREFIX}emailsuffix@cds-snc.ca").first() + assert user is not None + + user = User.query.filter_by(email_address=f"{EMAIL_PREFIX}emailsuffix_admin@cds-snc.ca").first() + assert user is not None + + +def test_create_test_user_fails_bad_chars(client, sample_service_cypress): + auth_header = create_cypress_authorization_header() + + resp = client.post( + "/cypress/create_user/{}".format("email-suffix"), + headers=[auth_header], + content_type="application/json", + ) + + assert resp.status_code == 400 + + +def test_create_test_user_fails_in_prod(client, notify_api, sample_service_cypress): + with set_config_values(notify_api, {"NOTIFY_ENVIRONMENT": "production"}): + auth_header = create_cypress_authorization_header() + + resp = client.post( + "/cypress/create_user/{}".format("email-suffix"), + headers=[auth_header], + content_type="application/json", + ) + + assert resp.status_code == 403 + + +def test_cleanup_stale_users(client, sample_service_cypress, cypress_user, notify_db): + auth_header = create_cypress_authorization_header() + resp = client.post( + "/cypress/create_user/{}".format("emailsuffix"), + headers=[auth_header], + content_type="application/json", + ) + + assert resp.status_code == 201 + # verify users were created in the DB + user = User.query.filter_by(email_address=f"{EMAIL_PREFIX}emailsuffix@cds-snc.ca").first() + assert user is not None + user2 = User.query.filter_by(email_address=f"{EMAIL_PREFIX}emailsuffix_admin@cds-snc.ca").first() + assert user2 is not None + # update created_at time so they can be cleaned up + user.created_at = datetime.utcnow() - timedelta(days=30) + user2.created_at = datetime.utcnow() - timedelta(days=30) + notify_db.session.add(user) + notify_db.session.add(user2) + notify_db.session.commit() + + # clean up users + auth_header = create_cypress_authorization_header() + resp = client.get( + "/cypress/cleanup", + headers=[auth_header], + content_type="application/json", + ) + data = json.loads(resp.data) + + assert resp.status_code == 201 + assert data["message"] == "Clean up complete" + + # Verify the stale user has been deleted + user = User.query.filter_by(email_address=f"{EMAIL_PREFIX}emailsuffix@cds-snc.ca").first() + assert user is None + + user = User.query.filter_by(email_address=f"{EMAIL_PREFIX}emailsuffix_admin@cds-snc.ca").first() + assert user is None