more tests

app/commands.py

@@ -288,9 +288,7 @@ def populate_organizations_from_file(file_name):
     # The expectation is that the organization, organization_to_service
     # and user_to_organization will be cleared before running this command.
     # Ignoring duplicates allows us to run the command again with the same file or same file with new rows.
-    msg("enter")
     with open(file_name, 'r') as f:
-        msg(f"\nfile is open {file_name}")
 
         def boolean_or_none(field):
             if field == '1':
@@ -301,13 +299,10 @@ def populate_organizations_from_file(file_name):
                 return None
 
         for line in itertools.islice(f, 1, None):
-            msg(f"XXX line {line}")
             columns = line.split('|')
-            msg(f"XXX columns {columns}")
             print(columns)
             email_branding = None
             email_branding_column = columns[5].strip()
-            msg(f"XXX email_branding_column {email_branding_column}")
             if len(email_branding_column) > 0:
                 email_branding = EmailBranding.query.filter(EmailBranding.name == email_branding_column).one()
             data = {
@@ -317,14 +312,12 @@ def populate_organizations_from_file(file_name):
                 'organization_type': columns[1].lower(),
                 'email_branding_id': email_branding.id if email_branding else None
             }
-            msg(f"XXX data {data}")
             org = Organization(**data)
             try:
                 db.session.add(org)
                 db.session.commit()
             except IntegrityError:
                 print("duplicate org", org.name)
-                msg("XXX duplicate org")
                 db.session.rollback()
             domains = columns[4].split(',')
             for d in domains:
@@ -334,17 +327,10 @@ def populate_organizations_from_file(file_name):
                         db.session.add(domain)
                         db.session.commit()
                     except IntegrityError:
-                        msg("XXX duplicate domain")
                         print("duplicate domain", d.strip())
                         db.session.rollback()
 
 
-def msg(msg):
-    f = open("huhhh.txt", "a")
-    f.write(msg)
-    f.close()
-
-
 @notify_command(name='populate-organization-agreement-details-from-file')
 @click.option('-f', '--file_name', required=True,
               help="CSV file containing id, agreement_signed_version, "

app/utils.py

@@ -105,12 +105,6 @@ def escape_special_characters(string):
     return string
 
 
-def email_address_is_nhs(email_address):
-    return email_address.lower().endswith((
-        '@nhs.uk', '@nhs.net', '.nhs.uk', '.nhs.net',
-    ))
-
-
 def get_archived_db_column_value(column):
     date = datetime.utcnow().strftime("%Y-%m-%d")
     return f'_archived_{date}_{column}'

tests/app/celery/test_nightly_tasks.py

@@ -3,6 +3,7 @@ from unittest.mock import ANY, call
 
 import pytest
 from freezegun import freeze_time
+from sqlalchemy.exc import SQLAlchemyError
 
 from app.celery import nightly_tasks
 from app.celery.nightly_tasks import (
@@ -148,6 +149,14 @@ def test_delete_inbound_sms_calls_child_task(notify_api, mocker):
     assert nightly_tasks.delete_inbound_sms_older_than_retention.call_count == 1
 
 
+def test_delete_inbound_sms_calls_child_task_db_error(notify_api, mocker):
+    mock_delete = mocker.patch('app.celery.nightly_tasks.delete_inbound_sms_older_than_retention')
+    mock_delete.side_effect = SQLAlchemyError
+
+    with pytest.raises(expected_exception=SQLAlchemyError):
+        delete_inbound_sms()
+
+
 @freeze_time('2021-01-18T02:00')
 @pytest.mark.parametrize('date_provided', [None, '2021-1-17'])
 def test_save_daily_notification_processing_time(mocker, sample_template, date_provided):
@@ -316,6 +325,10 @@ def test_delete_notifications_task_calls_task_for_services_that_have_sent_notifi
     ])
 
 
+def delete_notifications_by_service_and_type(id, param, param1):
+    pass
+
+
 def test_cleanup_unfinished_jobs(mocker):
     mock_s3 = mocker.patch('app.celery.nightly_tasks.remove_csv_object')
     mock_dao_archive = mocker.patch('app.celery.nightly_tasks.dao_archive_job')

tests/app/clients/test_aws_ses.py

@@ -1,9 +1,11 @@
+import json
+from unittest import mock
 from unittest.mock import ANY, Mock
 
 import botocore
 import pytest
 
-from app import aws_ses_client
+from app import AwsSesStubClient, aws_ses_client
 from app.clients.email import EmailClientNonRetryableException
 from app.clients.email.aws_ses import (
     AwsSesClientException,
@@ -176,3 +178,27 @@ def test_send_email_raises_other_errs_as_AwsSesClientException(mocker):
         )
 
     assert 'some error message from amazon' in str(excinfo.value)
+
+
+@mock.patch('app.clients.email.aws_ses_stub.request')
+def test_send_email_stub(mock_request):
+    mock_request.return_value = FakeResponse()
+    stub = AwsSesStubClient()
+    stub.init_app("fake")
+    answer = stub.send_email(
+                   'fake@fake.gov',
+                   'recipient@wherever.com',
+                   'TestTest',
+                   'TestBody')
+    print(answer)
+    assert answer == 'SomeId'
+
+
+class FakeResponse:
+    def __init__(self):
+
+        t = {"MessageId": "SomeId"}
+        self.text = json.dumps(t)
+
+    def raise_for_status(self):
+        print("raised for status")

tests/app/clients/test_performance_platform.py

@@ -58,3 +58,14 @@ def test_should_raise_for_status(perf_client):
     with pytest.raises(requests.HTTPError), requests_mock.Mocker() as request_mock:
         request_mock.post('https://performance-platform-url/foo', json={}, status_code=403)
         perf_client.send_stats_to_performance_platform({'dataType': 'foo'})
+
+
+def generate_payload_id(payload, param):
+    pass
+
+
+def test_generate_payload_id():
+    payload = {'_timestamp': '2023-01-01 00:00:00', 'service': 'my_service', 'group_name': 'group_name',
+               'dataType': 'dataType', 'period': 'period'}
+    result = PerformancePlatformClient.generate_payload_id(payload, "group_name")
+    assert result == 'MjAyMy0wMS0wMSAwMDowMDowMG15X3NlcnZpY2Vncm91cF9uYW1lZGF0YVR5cGVwZXJpb2Q='

tests/app/dao/test_services_dao.py

@@ -310,6 +310,31 @@ def test_should_remove_user_from_service(notify_db_session):
     assert new_user not in Service.query.first().users
 
 
+def test_should_remove_user_from_service_exception(notify_db_session):
+    user = create_user()
+    service = Service(name="service_name",
+                      email_from="email_from",
+                      message_limit=1000,
+                      restricted=False,
+                      created_by=user)
+    dao_create_service(service, user)
+    new_user = User(
+        name='Test User',
+        email_address='new_user@digital.cabinet-office.gov.uk',
+        password='password',
+        mobile_number='+12028675309'
+    )
+    save_model_user(new_user, validated_email_access=True)
+    wrong_user = User(
+        name='Wrong User',
+        email_address='wrong_user@digital.cabinet-office.gov.uk',
+        password='password',
+        mobile_number='+12028675309'
+    )
+    with pytest.raises(expected_exception=Exception):
+        dao_remove_user_from_service(service, wrong_user)
+
+
 def test_removing_a_user_from_a_service_deletes_their_permissions(sample_user, sample_service):
     assert len(Permission.query.all()) == 7
 

tests/app/notifications/test_receive_notification.py

@@ -271,7 +271,6 @@ def test_sns_inbound_sms_auth(notify_db_session, notify_api, client, mocker, aut
         assert response.status_code == status_code
 
 
-@pytest.mark.skip(reason="Need to implement inbound SNS tests. Body here from MMG")
 def test_create_inbound_sms_object_works_with_alphanumeric_sender(sample_service_full_permissions):
     data = {
         'Message': 'hello',

tests/app/test_commands.py

@@ -6,6 +6,7 @@ import pytest
 from app.commands import (
     _update_template,
     create_test_user,
+    fix_billable_units,
     insert_inbound_numbers_from_file,
     populate_annual_billing_with_defaults,
     populate_annual_billing_with_the_previous_years_allowance,
@@ -15,25 +16,84 @@ from app.commands import (
     update_jobs_archived_flag,
 )
 from app.dao.inbound_numbers_dao import dao_get_available_inbound_numbers
-from app.models import AnnualBilling, Job, Organization, Service, Template, User
+from app.models import (
+    AnnualBilling,
+    Job,
+    Notification,
+    Organization,
+    Service,
+    Template,
+    User,
+)
 from tests.app.db import (
     create_annual_billing,
     create_job,
+    create_notification,
     create_organization,
     create_service,
     create_template,
 )
 
 
-def test_purge_functional_test_data(notify_db_session, notify_api):
+def test_create_test_user_no_dupes(notify_db_session, notify_api):
-    print("ENTER test_purge_functional_test_data")
 
     user_count = User.query.count()
-    print(f"INITIAL user count {user_count}")
+    if user_count > 0:
-    # run the command
+        users = User.query.all()
+        for user in users:
+            notify_db_session.delete(user)
+        notify_db_session.commit()
+    user_count = User.query.count()
+    assert user_count == 0
+
     notify_api.test_cli_runner().invoke(
         create_test_user, [
-            '--email', 'somebody+7af2cdb0-7cbc-44dc-a5d0-f817fc6ee94e@fake.gov',
+            '--email', 'somebody@fake.gov',
+            '--mobile_number', '202-555-5555',
+            '--password', 'correct horse battery staple',
+            '--name', 'Fake Personson',
+            # '--auth_type', 'sms_auth',  # this is the default
+            # '--state', 'active',  # this is the default
+            # '--admin', 'False',  # this is the default
+        ]
+    )
+
+    user_count = User.query.count()
+    print(f"AFTER create_test_user {user_count}")
+    assert user_count == 1
+
+    notify_api.test_cli_runner().invoke(
+        create_test_user, [
+            '--email', 'somebody@fake.gov',
+            '--mobile_number', '202-555-5555',
+            '--password', 'correct horse battery staple',
+            '--name', 'Fake Personson',
+            # '--auth_type', 'sms_auth',  # this is the default
+            # '--state', 'active',  # this is the default
+            # '--admin', 'False',  # this is the default
+        ]
+    )
+
+    user_count = User.query.count()
+    assert user_count == 1
+
+
+@pytest.mark.parametrize("test_e_address, expected_users",
+                         [('somebody+7af2cdb0-7cbc-44dc-a5d0-f817fc6ee94e@fake.gov', 0),
+                          ('somebody@fake.gov', 1)])
+def test_purge_functional_test_data(notify_db_session, notify_api, test_e_address, expected_users):
+
+    user_count = User.query.count()
+    if user_count > 0:
+        users = User.query.all()
+        for user in users:
+            notify_db_session.delete(user)
+        notify_db_session.commit()
+    user_count = User.query.count()
+    assert user_count == 0
+    notify_api.test_cli_runner().invoke(
+        create_test_user, [
+            '--email', test_e_address,
             '--mobile_number', '202-555-5555',
             '--password', 'correct horse battery staple',
             '--name', 'Fake Personson',
@@ -55,8 +115,9 @@ def test_purge_functional_test_data(notify_db_session, notify_api):
     print("INVOKING purge_functional_test_data")
     notify_api.test_cli_runner().invoke(purge_functional_test_data, ['-u', 'somebody'])
     print("IT WAS INVOKED")
-    # there should be one more user
+    # if the email address has a uuid, it is test data so it should be purged and there should be
-    assert User.query.count() == 0
+    # zero users.  Otherwise, it is real data so there should be one user.
+    assert User.query.count() == expected_users
 
 
 def test_purge_functional_test_data_bad_mobile(notify_db_session, notify_api):
@@ -270,6 +331,20 @@ def test_populate_annual_billing_with_the_previous_years_allowance(
     assert results[0].free_sms_fragment_limit == expected_allowance
 
 
+def test_fix_billable_units(notify_db_session, notify_api, sample_template):
+
+    create_notification(template=sample_template)
+    notification = Notification.query.one()
+    assert notification.billable_units == 1
+
+    notify_api.test_cli_runner().invoke(
+        fix_billable_units, []
+    )
+
+    notification = Notification.query.one()
+    assert notification.billable_units == 1
+
+
 def test_populate_annual_billing_with_defaults_sets_free_allowance_to_zero_if_previous_year_is_zero(
         notify_db_session, notify_api
 ):

tests/app/test_utils.py

@@ -1,12 +1,17 @@
+import uuid
 from datetime import date, datetime
 
 import pytest
 from freezegun import freeze_time
 
+from app.models import UPLOAD_DOCUMENT
 from app.utils import (
     format_sequential_number,
     get_midnight_for_day_before,
     get_midnight_in_utc,
+    get_public_notify_type_text,
+    get_reference_from_personalisation,
+    get_uuid_string_or_none,
     midnight_n_days_ago,
 )
 
@@ -56,3 +61,23 @@ def test_midnight_n_days_ago(current_time, arg, expected_datetime):
 
 def test_format_sequential_number():
     assert format_sequential_number(123) == '0000007b'
+
+
+@pytest.mark.parametrize('personalisation, expected_response', [
+    ({"nothing": "interesting"}, None),
+    ({"reference": "something"}, "something"),
+    (None, None)
+])
+def test_get_reference_from_personalisation(personalisation, expected_response):
+    assert get_reference_from_personalisation(personalisation) == expected_response
+
+
+def test_get_uuid_string_or_none():
+    my_uuid = uuid.uuid4()
+    assert str(my_uuid) == get_uuid_string_or_none(my_uuid)
+
+    assert get_uuid_string_or_none(None) is None
+
+
+def test_get_public_notify_type_text():
+    assert get_public_notify_type_text(UPLOAD_DOCUMENT) == 'document'

tests/app/v2/test_errors.py

@@ -2,6 +2,8 @@ import pytest
 from flask import url_for
 from sqlalchemy.exc import DataError
 
+from app.v2.errors import ValidationError
+
 
 @pytest.fixture(scope='function')
 def app_for_test():
@@ -97,6 +99,9 @@ def test_validation_error(app_for_test):
                     'message': "phone_number is a required property"} in error['errors']
             assert {'error': 'ValidationError',
                     'message': "template_id is not a valid UUID"} in error['errors']
+            ve = ValidationError("phone_number is a required property")
+            assert ve.message == "Your notification has failed validation"
+            assert ve.status_code == 400
 
 
 def test_data_errors(app_for_test):