Tweaks to make the fixtures work with a sanitised database

app/commands.py

@@ -864,7 +864,11 @@ def functional_test_fixtures():
 
     """
     if current_app.config["REGISTER_FUNCTIONAL_TESTING_BLUEPRINT"]:
-        apply_fixtures()
+        try:
+            apply_fixtures()
+        except Exception:
+            current_app.logger.exception("Functional test fixtures failed")
+            raise
     else:
         print("Functional test fixtures are disabled. Set REGISTER_FUNCTIONAL_TESTING_BLUEPRINT to True in config.")
         raise SystemExit(1)

app/functional_tests_fixtures/__init__.py

@@ -4,8 +4,10 @@
 
 import boto3
 from flask import current_app
+from itsdangerous.exc import BadSignature
 from sqlalchemy.exc import NoResultFound
 
+from app import db
 from app.constants import (
     EDIT_FOLDER_PERMISSIONS,
     EMAIL_AUTH,
@@ -20,6 +22,7 @@
     set_default_free_allowance_for_service,
 )
 from app.dao.api_key_dao import (
+    expire_api_key,
     get_model_api_keys,
     save_model_api_key,
 )
@@ -32,7 +35,6 @@
     dao_create_organisation,
     dao_get_organisation_by_id,
     dao_get_organisations_by_partial_name,
-    dao_update_organisation,
 )
 from app.dao.permissions_dao import permission_dao
 from app.dao.service_callback_api_dao import get_service_callback_api_by_callback_type, save_service_callback_api
@@ -63,13 +65,15 @@
 )
 from app.dao.users_dao import get_user_by_email, save_model_user
 from app.models import (
+    Domain,
     InboundNumber,
     InboundSms,
     Organisation,
     Permission,
     Service,
     ServiceCallbackApi,
     ServiceEmailReplyTo,
+    ServiceSmsSender,
     User,
 )
 from app.schemas import api_key_schema, template_schema
@@ -153,8 +157,11 @@ def _create_db_objects(
 ) -> dict[str, str]:
     current_app.logger.info("Creating functional test fixtures for %s:", environment)
 
+    service_name_with_environment = f"Functional Tests ({environment})"
+    org_name_with_environment = f"{org_name} ({environment})"
+
     current_app.logger.info("--> Ensure organisation exists")
-    org = _create_organiation(email_domain, org_name)
+    org = _create_organiation(email_domain, org_name_with_environment)
 
     current_app.logger.info("--> Ensure users exists")
     func_test_user = _create_user(
@@ -178,7 +185,7 @@ def _create_db_objects(
     )
 
     current_app.logger.info("--> Ensure service exists")
-    service = _create_service(org.id, service_admin_user)
+    service = _create_service(org.id, service_admin_user, service_name_with_environment)
 
     current_app.logger.info("--> Ensure users are added to service")
     dao_add_user_to_service(service, service_admin_user)
@@ -392,10 +399,31 @@ def _create_organiation(email_domain, org_name):
 
     if org is None:
         org = Organisation(name=org_name, active=True, crown=False, organisation_type="central")
-
         dao_create_organisation(org)
 
-    dao_update_organisation(org.id, domains=[email_domain], can_approve_own_go_live_requests=True)
+    org.name = org_name
+    org.active = True
+    org.crown = False
+    org.organisation_type = "central"
+    org.can_approve_own_go_live_requests = True
+    db.session.add(org)
+
+    normalised_email_domain = email_domain.lower()
+    existing_domain = Domain.query.filter_by(domain=normalised_email_domain).one_or_none()
+    if existing_domain is None:
+        db.session.add(Domain(domain=normalised_email_domain, organisation_id=org.id))
+    elif existing_domain.organisation_id != org.id:
+        previous_org_id = existing_domain.organisation_id
+        existing_domain.organisation_id = org.id
+        db.session.add(existing_domain)
+        current_app.logger.info(
+            "Reassigned domain %s from organisation %s to fixture organisation %s",
+            normalised_email_domain,
+            previous_org_id,
+            org.id,
+        )
+
+    db.session.commit()
 
     return org
 
@@ -444,6 +472,20 @@ def _create_api_key(name, service_id, user_id, key_type="normal"):
     api_keys = get_model_api_keys(service_id=service_id)
     for key in api_keys:
         if key.name == name:
+            if key.expiry_date is not None:
+                continue
+
+            try:
+                _ = key.secret
+            except BadSignature:
+                current_app.logger.warning(
+                    "Fixture api key %s for service %s had invalid signature; expiring and recreating",
+                    name,
+                    service_id,
+                )
+                expire_api_key(service_id=service_id, api_key_id=key.id)
+                continue
+
             return key
 
     request = {"created_by": user_id, "key_type": key_type, "name": name}
@@ -459,7 +501,37 @@ def _create_api_key(name, service_id, user_id, key_type="normal"):
 def _create_inbound_numbers(service_id, user_id, number="07700900500", provider="mmg"):
     inbound_number = dao_get_inbound_number_for_service(service_id=service_id)
 
+    if inbound_number is not None and inbound_number.number == number:
+        return inbound_number.id
+
+    inbound_number_by_number = InboundNumber.query.filter_by(number=number).one_or_none()
+    if inbound_number_by_number is not None:
+        if inbound_number is not None and inbound_number.id != inbound_number_by_number.id:
+            inbound_number.service_id = None
+            db.session.add(inbound_number)
+
+        previous_service_id = inbound_number_by_number.service_id
+        inbound_number_by_number.service_id = service_id
+        inbound_number_by_number.active = True
+        db.session.add(inbound_number_by_number)
+        db.session.commit()
+
+        if previous_service_id and previous_service_id != service_id:
+            current_app.logger.info(
+                "Reassigned inbound number %s from service %s to fixture service %s",
+                number,
+                previous_service_id,
+                service_id,
+            )
+
+        return inbound_number_by_number.id
+
     if inbound_number is not None:
+        inbound_number.number = number
+        inbound_number.provider = provider
+        inbound_number.active = True
+        db.session.add(inbound_number)
+        db.session.commit()
         return inbound_number.id
 
     inbound_number = InboundNumber()
@@ -629,6 +701,35 @@ def _create_service_sms_senders(service_id, sms_sender, is_default, inbound_numb
     for service_sms_sender in service_sms_senders:
         if service_sms_sender.sms_sender == sms_sender:
             return service_sms_sender
+        if inbound_number_id and service_sms_sender.inbound_number_id == inbound_number_id:
+            return service_sms_sender
+
+    if inbound_number_id:
+        existing_inbound_sender = ServiceSmsSender.query.filter_by(inbound_number_id=inbound_number_id).one_or_none()
+        if existing_inbound_sender is not None:
+            if is_default:
+                for service_sms_sender in service_sms_senders:
+                    if service_sms_sender.id != existing_inbound_sender.id and service_sms_sender.is_default:
+                        service_sms_sender.is_default = False
+                        db.session.add(service_sms_sender)
+
+            previous_service_id = existing_inbound_sender.service_id
+            existing_inbound_sender.service_id = service_id
+            existing_inbound_sender.sms_sender = sms_sender
+            existing_inbound_sender.is_default = is_default
+            existing_inbound_sender.archived = False
+            db.session.add(existing_inbound_sender)
+            db.session.commit()
+
+            if previous_service_id != service_id:
+                current_app.logger.info(
+                    "Reassigned inbound sms sender for number %s from service %s to fixture service %s",
+                    sms_sender,
+                    previous_service_id,
+                    service_id,
+                )
+
+            return existing_inbound_sender
 
     return dao_add_sms_sender_for_service(service_id, sms_sender, is_default, inbound_number_id)
 

tests/app/functional_test_fixtures/test_functional_test_fixtures.py

@@ -7,7 +7,21 @@
 from moto import mock_aws
 
 from app import db
-from app.functional_tests_fixtures import _create_db_objects, _create_user, apply_fixtures
+from app.functional_tests_fixtures import (
+    _create_api_key,
+    _create_db_objects,
+    _create_service_sms_senders,
+    _create_user,
+    apply_fixtures,
+)
+from app.models import ApiKey, Domain, InboundNumber, Organisation
+from tests.app.db import (
+    create_api_key,
+    create_inbound_number,
+    create_organisation,
+    create_service,
+    create_service_sms_sender,
+)
 from tests.conftest import set_config_values
 
 
@@ -73,8 +87,10 @@ def test_create_db_objects_sets_db_up(notify_api, notify_service):
     assert "FUNCTIONAL_TESTS_SERVICE_EMAIL_PASSWORD" in variables[0]
     assert variables[0]["FUNCTIONAL_TESTS_SERVICE_NUMBER"] == "07700900501"
     assert "FUNCTIONAL_TESTS_SERVICE_ID" in variables[0]
-    assert variables[0]["FUNCTIONAL_TESTS_SERVICE_NAME"] == "Functional Tests"
+    assert variables[0]["FUNCTIONAL_TESTS_SERVICE_NAME"] == "Functional Tests (dev-env)"
     assert "FUNCTIONAL_TESTS_ORGANISATION_ID" in variables[0]
+    fixture_org = Organisation.query.get(variables[0]["FUNCTIONAL_TESTS_ORGANISATION_ID"])
+    assert fixture_org.name == "Functional Tests Org (dev-env)"
     assert variables[0]["FUNCTIONAL_TESTS_SERVICE_API_KEY"].startswith("functional_tests_service_live_key-")
     assert variables[0]["FUNCTIONAL_TESTS_SERVICE_API_TEST_KEY"].startswith("functional_tests_service_test_key-")
     assert variables[0]["FUNCTIONAL_TESTS_API_AUTH_SECRET"] == "functional-tests-secret-key"
@@ -112,6 +128,136 @@ def test_create_user_revalidates_email():
     assert (datetime.utcnow() - test_user.email_access_validated_at).total_seconds() < 60
 
 
+def test_create_db_objects_reassigns_existing_inbound_number_from_another_service(notify_api, notify_service):
+    existing_service = create_service(service_name="Existing inbound owner")
+    existing_inbound = create_inbound_number(number="07700900500", service_id=existing_service.id)
+
+    with set_config_values(
+        notify_api,
+        {
+            "MMG_INBOUND_SMS_USERNAME": ["test_mmg_username"],
+            "MMG_INBOUND_SMS_AUTH": ["test_mmg_password"],
+            "INTERNAL_CLIENT_API_KEYS": {"notify-functional-tests": ["functional-tests-secret-key"]},
+            "ADMIN_BASE_URL": "http://localhost:6012",
+            "API_HOST_NAME": "http://localhost:6011",
+        },
+    ):
+        variables = _create_db_objects(
+            "fake password",
+            "test_request_bin_token",
+            "dev-env",
+            "notify-tests-preview",
+            "digital.cabinet-office.gov.uk",
+            "govuk_notify",
+            "functional_tests_service_live_key",
+            "functional_tests_service_test_key",
+            str(notify_service.id),
+            "Functional Tests Org",
+            "07700900500",
+        )
+
+    reassigned_inbound = InboundNumber.query.filter_by(number="07700900500").one()
+
+    assert str(reassigned_inbound.id) == str(existing_inbound.id)
+    assert str(reassigned_inbound.service_id) == str(variables["FUNCTIONAL_TESTS_SERVICE_ID"])
+
+
+def test_create_db_objects_reassigns_domain_to_environment_org(notify_api, notify_service):
+    previous_owner = create_organisation(name="Shared Org", domains=["digital.cabinet-office.gov.uk"])
+
+    with set_config_values(
+        notify_api,
+        {
+            "MMG_INBOUND_SMS_USERNAME": ["test_mmg_username"],
+            "MMG_INBOUND_SMS_AUTH": ["test_mmg_password"],
+            "INTERNAL_CLIENT_API_KEYS": {"notify-functional-tests": ["functional-tests-secret-key"]},
+            "ADMIN_BASE_URL": "http://localhost:6012",
+            "API_HOST_NAME": "http://localhost:6011",
+        },
+    ):
+        variables = _create_db_objects(
+            "fake password",
+            "test_request_bin_token",
+            "dev-env",
+            "notify-tests-preview",
+            "digital.cabinet-office.gov.uk",
+            "govuk_notify",
+            "functional_tests_service_live_key",
+            "functional_tests_service_test_key",
+            str(notify_service.id),
+            "Functional Tests Org",
+            "07700900500",
+        )
+
+    fixture_org = Organisation.query.get(variables["FUNCTIONAL_TESTS_ORGANISATION_ID"])
+    domain = Domain.query.filter_by(domain="digital.cabinet-office.gov.uk").one()
+
+    assert fixture_org.name == "Functional Tests Org (dev-env)"
+    assert str(domain.organisation_id) == str(fixture_org.id)
+    assert str(domain.organisation_id) != str(previous_owner.id)
+
+
+def test_create_service_sms_senders_reuses_existing_sender_with_same_inbound_number_id(notify_service):
+    inbound = create_inbound_number(number="07700900888", service_id=notify_service.id)
+    existing_sender = create_service_sms_sender(
+        service=notify_service,
+        sms_sender="existing",
+        is_default=True,
+        inbound_number_id=inbound.id,
+    )
+
+    sender = _create_service_sms_senders(
+        notify_service.id,
+        "07700900888",
+        True,
+        inbound.id,
+    )
+
+    assert sender.id == existing_sender.id
+
+
+def test_create_service_sms_senders_reclaims_existing_sender_from_another_service(notify_service):
+    other_service = create_service(service_name="other sender owner")
+    inbound = create_inbound_number(number="07700900999", service_id=notify_service.id)
+    existing_sender = create_service_sms_sender(
+        service=other_service,
+        sms_sender="old",
+        is_default=True,
+        inbound_number_id=inbound.id,
+    )
+
+    sender = _create_service_sms_senders(
+        notify_service.id,
+        "07700900999",
+        True,
+        inbound.id,
+    )
+
+    assert sender.id == existing_sender.id
+    assert sender.service_id == notify_service.id
+    assert sender.sms_sender == "07700900999"
+
+
+def test_create_api_key_recreates_when_existing_key_secret_is_invalid(notify_service, sample_user):
+    broken_key = create_api_key(notify_service, key_name="functional_tests_service_live_key")
+    broken_key._secret = "broken-signature-value"
+    db.session.add(broken_key)
+    db.session.commit()
+
+    recreated_key = _create_api_key(
+        "functional_tests_service_live_key",
+        notify_service.id,
+        sample_user.id,
+        "normal",
+    )
+
+    refreshed_broken_key = ApiKey.query.get(broken_key.id)
+    assert refreshed_broken_key.expiry_date is not None
+    assert recreated_key.id != broken_key.id
+    assert recreated_key.expiry_date is None
+    assert recreated_key.secret is not None
+
+
 @mock_aws
 def test_function_test_fixtures_saves_to_disk_and_ssm(notify_api, os_environ, mocker):
     mocker.patch("app.functional_tests_fixtures._create_db_objects", return_value={"FOO": "BAR", "BAZ": "WAZ"})