more tests

tests/app/dao/test_permissions_dao.py

@@ -1,3 +1,4 @@
+from app.dao import DAOClass
 from app.dao.permissions_dao import permission_dao
 from tests.app.db import create_service
 
@@ -22,3 +23,17 @@ def test_get_permissions_by_user_id_returns_only_active_service(sample_user):
     assert len(permissions) == 7
     assert active_service in [i.service for i in permissions]
     assert inactive_service not in [i.service for i in permissions]
+
+
+def test_dao_class(sample_user):
+    create_service(user=sample_user, service_name="Active service")
+    create_service(user=sample_user, service_name="Inactive service", active=False)
+
+    permissions_orig = permission_dao.get_permissions_by_user_id(user_id=sample_user.id)
+    assert len(permissions_orig) == 7
+    dao = DAOClass()
+
+    for permission in permissions_orig:
+        dao.delete_instance(permission, True)
+    permissions = permission_dao.get_permissions_by_user_id(user_id=sample_user.id)
+    assert len(permissions) == 0

tests/app/dao/test_users_dao.py

@@ -12,6 +12,7 @@ from app.dao.service_user_dao import (
     dao_update_service_user,
 )
 from app.dao.users_dao import (
+    _remove_values_for_keys_if_present,
     count_user_verify_codes,
     create_secret_code,
     dao_archive_user,
@@ -294,3 +295,16 @@ def test_user_cannot_be_archived_if_the_other_service_members_do_not_have_the_ma
 
     assert len(sample_service.users) == 3
     assert not user_can_be_archived(active_user)
+
+
+def test_remove_values_for_keys_if_present():
+    keys = {'a', 'b', 'c'}
+    my_dict = {
+        'a': 1,
+        'b': 2,
+        'c': 3,
+        'd': 4,
+    }
+    _remove_values_for_keys_if_present(my_dict, keys)
+
+    assert my_dict == {'d': 4}

tests/app/notifications/test_receive_notification.py

@@ -1,5 +1,6 @@
 from base64 import b64encode
 from datetime import datetime
+from unittest import mock
 
 import pytest
 from flask import json
@@ -7,6 +8,7 @@ from flask import json
 from app.models import EMAIL_TYPE, INBOUND_SMS_TYPE, SMS_TYPE, InboundSms
 from app.notifications.receive_notifications import (
     create_inbound_sms_object,
+    fetch_potential_service,
     has_inbound_sms_permissions,
     unescape_string,
 )
@@ -290,3 +292,15 @@ def test_create_inbound_sms_object_works_with_alphanumeric_sender(sample_service
     )
 
     assert inbound_sms.user_number == 'ALPHANUM3R1C'
+
+
+@mock.patch('app.notifications.receive_notifications.dao_fetch_service_by_inbound_number')
+def test_fetch_potential_service_cant_find_it(mock_dao):
+    mock_dao.return_value = None
+    found_service = fetch_potential_service(234, 'sns')
+    assert found_service is False
+
+    # Permissions will not be set so it will still return false
+    mock_dao.return_value = create_service()
+    found_service = fetch_potential_service(234, 'sns')
+    assert found_service is False

tests/app/notifications/test_validators.py

@@ -9,6 +9,11 @@ from app.models import EMAIL_TYPE, SMS_TYPE
 from app.notifications.process_notifications import (
     create_content_for_notification,
 )
+from app.notifications.sns_cert_validator import (
+    VALID_SNS_TOPICS,
+    get_string_to_sign,
+    validate_sns_cert,
+)
 from app.notifications.validators import (
     check_application_over_retention_limit,
     check_if_service_can_send_files_by_email,
@@ -535,3 +540,34 @@ def test_check_if_service_can_send_files_by_email_passes_if_contact_link_set(sam
         service_contact_link=sample_service.contact_link,
         service_id=sample_service.id
     )
+
+
+def test_get_string_to_sign():
+    VALID_SNS_TOPICS.append("arn:aws:sns:us-west-2:009969138378:connector-svc-test")
+    sns_payload = {
+        "Type": "Notification",
+        "MessageId": "ccccccccc-cccc-cccc-cccc-ccccccccccccc",
+        "TopicArn": "arn:aws:sns:us-west-2:009969138378:connector-svc-test",
+        "Message":  "{\"AbsoluteTime\":\"2021-09-08T13:28:24.656Z\",\"Content\":\"help\",\"ContentType\":\"text/plain\",\"Id\":\"333333333-be0d-4a44-889d-d2a86fc06f0c\",\"Type\":\"MESSAGE\",\"ParticipantId\":\"bbbbbbbb-c562-4d95-b76c-dcbca8b4b5f7\",\"DisplayName\":\"Jane\",\"ParticipantRole\":\"CUSTOMER\",\"InitialContactId\":\"33333333-abc5-46db-9ad5-d772559ab556\",\"ContactId\":\"33333333-abc5-46db-9ad5-d772559ab556\"}",  # noqa
+        "Timestamp": "2021-09-08T13:28:24.860Z",
+        "SignatureVersion": "1",
+        "Signature": "examplegggggg/1tEBYdiVDgJgBoJUniUFcArLFGfg5JCvpOr/v6LPCHiD7A0BWy8+ZOnGTmOjBMn80U9jSzYhKbHDbQHaNYTo9sRyQA31JtHHiIseQeMfTDpcaAXqfs8hdIXq4XZaJYqDFqosfbvh56VPh5QgmeHTltTc7eOZBUwnt/177eOTLTt2yB0ItMV3NAYuE1Tdxya1lLYZQUIMxETTVcRAZkDIu8TbRZC9a00q2RQVjXhDaU3k+tL+kk85syW/2ryjjkDYoUb+dyRGkqMy4aKA22UpfidOtdAZ/GGtXaXSKBqazZTEUuSEzt0duLtFntQiYJanU05gtDig==", # noqa
+        "SigningCertURL": "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-11111111111111111111111111111111.pem", # noqa
+        "UnsubscribeURL": "https://sns.us-west-2.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-west-2:000000000000:connector-svc-test:22222222-aaaa-bbbb-cccc-333333333333", # noqa
+        "MessageAttributes": {
+            "InitialContactId": {"Type": "String", "Value": "33333333-abc5-46db-9ad5-d772559ab556"},
+            "MessageVisibility": {"Type": "String", "Value": "ALL"},
+            "Type": {"Type": "String", "Value": "MESSAGE"},
+            "AccountId": {"Type": "String", "Value": "999999999999"},
+            "ContentType": {"Type": "String", "Value": "text/plain"},
+            "InstanceId": {"Type": "String", "Value": "dddddddd-b64e-40c5-921b-109fd92499ae"},
+            "ContactId": {"Type": "String", "Value": "33333333-abc5-46db-9ad5-d772559ab556"},
+            "ParticipantRole": {"Type": "String", "Value": "CUSTOMER"}
+        }
+    }
+    str = get_string_to_sign(sns_payload)
+    assert str == b'Message\n{"AbsoluteTime":"2021-09-08T13:28:24.656Z","Content":"help","ContentType":"text/plain","Id":"333333333-be0d-4a44-889d-d2a86fc06f0c","Type":"MESSAGE","ParticipantId":"bbbbbbbb-c562-4d95-b76c-dcbca8b4b5f7","DisplayName":"Jane","ParticipantRole":"CUSTOMER","InitialContactId":"33333333-abc5-46db-9ad5-d772559ab556","ContactId":"33333333-abc5-46db-9ad5-d772559ab556"}\nMessageId\nccccccccc-cccc-cccc-cccc-ccccccccccccc\nTimestamp\n2021-09-08T13:28:24.860Z\nTopicArn\narn:aws:sns:us-west-2:009969138378:connector-svc-test\nType\nNotification\n' # noqa
+
+    # This is a test payload with no valid cert, so it should raise a ValueError
+    with pytest.raises(ValueError):
+        validate_sns_cert(sns_payload)

tests/app/test_errors.py

@@ -0,0 +1,6 @@
+from app.errors import VirusScanError
+
+
+def test_virus_scan_error():
+    vse = VirusScanError("a message")
+    assert "a message" in vse.args

tests/app/test_exceptions.py

@@ -0,0 +1,6 @@
+from app.exceptions import DVLAException
+
+
+def test_dvla_exception():
+    dvla = DVLAException("a message")
+    assert dvla.message == "a message"

tests/app/user/test_rest.py

@@ -1,3 +1,4 @@
+import json
 import uuid
 from datetime import datetime
 from unittest import mock
@@ -47,6 +48,14 @@ def test_get_user_list(admin_request, sample_service):
     assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
 
 
+def test_get_all_users(admin_request):
+    create_user()
+    json_resp = admin_request.get('user.get_all_users')
+    json_resp_str = json.dumps(json_resp)
+    assert 'Test User' in json_resp_str
+    assert '+12028675309' in json_resp_str
+
+
 def test_get_user(admin_request, sample_service, sample_organization):
     """
     Tests GET endpoint '/<user_id>' to retrieve a single service.