darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-16 20:48:37 -04:00
Code Issues Packages Projects Releases Wiki Activity

merge from main

Browse Source
This commit is contained in:
Kenneth Kehl
2024-03-01 13:50:09 -08:00
parent aef797eb79 18dddf4043
commit b777249b6f
140 changed files with 8031 additions and 5017 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
tests/app/user/test_rest.py
View File

@@ -7,17 +7,9 @@ import pytest
from flask import current_app
from freezegun import freeze_time
from app.dao.permissions_dao import default_service_permissions
from app.dao.service_user_dao import dao_get_service_user, dao_update_service_user
from app.models import (
EMAIL_AUTH_TYPE,
MANAGE_SETTINGS,
MANAGE_TEMPLATES,
SMS_AUTH_TYPE,
Notification,
Permission,
User,
)
from app.enums import AuthType, KeyType, NotificationType, PermissionType
from app.models import Notification, Permission, User
from tests.app.db import (
create_organization,
create_service,
@@ -35,7 +27,7 @@ def test_get_user_list(admin_request, sample_service):
# it may have the notify user in the DB still :weary:
assert len(json_resp["data"]) >= 1
sample_user = sample_service.users[0]
expected_permissions = default_service_permissions
expected_permissions = PermissionType.defaults()
fetched = next(x for x in json_resp["data"] if x["id"] == str(sample_user.id))
assert sample_user.name == fetched["name"]
@@ -63,7 +55,7 @@ def test_get_user(admin_request, sample_service, sample_organization):
sample_user.organizations = [sample_organization]
json_resp = admin_request.get("user.get_user", user_id=sample_user.id)
expected_permissions = default_service_permissions
expected_permissions = PermissionType.defaults()
fetched = json_resp["data"]
assert fetched["id"] == str(sample_user.id)
@@ -71,7 +63,7 @@ def test_get_user(admin_request, sample_service, sample_organization):
assert fetched["mobile_number"] == sample_user.mobile_number
assert fetched["email_address"] == sample_user.email_address
assert fetched["state"] == sample_user.state
assert fetched["auth_type"] == SMS_AUTH_TYPE
assert fetched["auth_type"] == AuthType.SMS
assert fetched["permissions"].keys() == {str(sample_service.id)}
assert fetched["services"] == [str(sample_service.id)]
assert fetched["organizations"] == [str(sample_organization.id)]
@@ -117,7 +109,7 @@ def test_post_user(admin_request, notify_db_session):
"state": "active",
"failed_login_count": 0,
"permissions": {},
"auth_type": EMAIL_AUTH_TYPE,
"auth_type": AuthType.EMAIL,
}
json_resp = admin_request.post("user.create_user", _data=data, _expected_status=201)
@@ -125,7 +117,7 @@ def test_post_user(admin_request, notify_db_session):
assert user.check_password("password")
assert json_resp["data"]["email_address"] == user.email_address
assert json_resp["data"]["id"] == str(user.id)
assert user.auth_type == EMAIL_AUTH_TYPE
assert user.auth_type == AuthType.EMAIL
def test_post_user_without_auth_type(admin_request, notify_db_session):
@@ -142,7 +134,7 @@ def test_post_user_without_auth_type(admin_request, notify_db_session):
user = User.query.filter_by(email_address="user@digital.fake.gov").first()
assert json_resp["data"]["id"] == str(user.id)
assert user.auth_type == SMS_AUTH_TYPE
assert user.auth_type == AuthType.SMS
def test_post_user_missing_attribute_email(admin_request, notify_db_session):
@@ -194,12 +186,12 @@ def test_can_create_user_with_email_auth_and_no_mobile(
"email_address": "user@digital.fake.gov",
"password": "password",
"mobile_number": None,
"auth_type": EMAIL_AUTH_TYPE,
"auth_type": AuthType.EMAIL,
}
json_resp = admin_request.post("user.create_user", _data=data, _expected_status=201)
assert json_resp["data"]["auth_type"] == EMAIL_AUTH_TYPE
assert json_resp["data"]["auth_type"] == AuthType.EMAIL
assert json_resp["data"]["mobile_number"] is None
@@ -211,14 +203,14 @@ def test_cannot_create_user_with_sms_auth_and_no_mobile(
"email_address": "user@digital.fake.gov",
"password": "password",
"mobile_number": None,
"auth_type": SMS_AUTH_TYPE,
"auth_type": AuthType.SMS,
}
json_resp = admin_request.post("user.create_user", _data=data, _expected_status=400)
assert (
json_resp["message"]
== "Mobile number must be set if auth_type is set to sms_auth"
== "Mobile number must be set if auth_type is set to AuthType.SMS"
)
@@ -228,7 +220,7 @@ def test_cannot_create_user_with_empty_strings(admin_request, notify_db_session)
"email_address": "",
"password": "password",
"mobile_number": "",
"auth_type": EMAIL_AUTH_TYPE,
"auth_type": AuthType.EMAIL,
}
resp = admin_request.post("user.create_user", _data=data, _expected_status=400)
assert resp["message"] == {
@@ -269,8 +261,8 @@ def test_post_user_attribute(admin_request, sample_user, user_attribute, user_va
"newuser@mail.com",
dict(
api_key_id=None,
key_type="normal",
notification_type="email",
key_type=KeyType.NORMAL,
notification_type=NotificationType.EMAIL,
personalisation={},
recipient="newuser@mail.com",
reply_to_text="notify@gov.uk",
@@ -284,8 +276,8 @@ def test_post_user_attribute(admin_request, sample_user, user_attribute, user_va
"+4407700900460",
dict(
api_key_id=None,
key_type="normal",
notification_type="sms",
key_type=KeyType.NORMAL,
notification_type=NotificationType.SMS,
personalisation={},
recipient="+4407700900460",
reply_to_text="testing",
@@ -312,7 +304,9 @@ def test_post_user_attribute_with_updated_by(
mock_persist_notification = mocker.patch("app.user.rest.persist_notification")
mocker.patch("app.user.rest.send_notification_to_queue")
json_resp = admin_request.post(
"user.update_user_attribute", user_id=sample_user.id, _data=update_dict
"user.update_user_attribute",
user_id=sample_user.id,
_data=update_dict,
)
assert json_resp["data"][user_attribute] == user_value
if arguments:
@@ -329,7 +323,9 @@ def test_post_user_attribute_with_updated_by_sends_notification_to_international
mocker.patch("app.user.rest.send_notification_to_queue")
admin_request.post(
"user.update_user_attribute", user_id=sample_user.id, _data=update_dict
"user.update_user_attribute",
user_id=sample_user.id,
_data=update_dict,
)
notification = Notification.query.first()
@@ -343,7 +339,9 @@ def test_archive_user(mocker, admin_request, sample_user):
archive_mock = mocker.patch("app.user.rest.dao_archive_user")
admin_request.post(
"user.archive_user", user_id=sample_user.id, _expected_status=204
"user.archive_user",
user_id=sample_user.id,
_expected_status=204,
)
archive_mock.assert_called_once_with(sample_user)
@@ -363,7 +361,9 @@ def test_archive_user_when_user_cannot_be_archived(mocker, admin_request, sample
mocker.patch("app.dao.users_dao.user_can_be_archived", return_value=False)
json_resp = admin_request.post(
"user.archive_user", user_id=sample_user.id, _expected_status=400
"user.archive_user",
user_id=sample_user.id,
_expected_status=400,
)
msg = "User can’t be removed from a service - check all services have another team member with manage_settings"
@@ -375,7 +375,7 @@ def test_get_user_by_email(admin_request, sample_service):
json_resp = admin_request.get("user.get_by_email", email=sample_user.email_address)
expected_permissions = default_service_permissions
expected_permissions = PermissionType.defaults()
fetched = json_resp["data"]
assert str(sample_user.id) == fetched["id"]
@@ -390,7 +390,9 @@ def test_get_user_by_email(admin_request, sample_service):
def test_get_user_by_email_not_found_returns_404(admin_request, sample_user):
json_resp = admin_request.get(
"user.get_by_email", email="no_user@digital.fake.gov", _expected_status=404
"user.get_by_email",
email="no_user@digital.fake.gov",
_expected_status=404,
)
assert json_resp["result"] == "error"
assert json_resp["message"] == "No result found"
@@ -457,21 +459,23 @@ def test_set_user_permissions(admin_request, sample_user, sample_service):
"user.set_permissions",
user_id=str(sample_user.id),
service_id=str(sample_service.id),
_data={"permissions": [{"permission": MANAGE_SETTINGS}]},
_data={"permissions": [{"permission": PermissionType.MANAGE_SETTINGS}]},
_expected_status=204,
)
permission = Permission.query.filter_by(permission=MANAGE_SETTINGS).first()
permission = Permission.query.filter_by(
permission=PermissionType.MANAGE_SETTINGS
).first()
assert permission.user == sample_user
assert permission.service == sample_service
assert permission.permission == MANAGE_SETTINGS
assert permission.permission == PermissionType.MANAGE_SETTINGS
def test_set_user_permissions_multiple(admin_request, sample_user, sample_service):
data = {
"permissions": [
{"permission": MANAGE_SETTINGS},
{"permission": MANAGE_TEMPLATES},
{"permission": PermissionType.MANAGE_SETTINGS},
{"permission": PermissionType.MANAGE_TEMPLATES},
]
}
admin_request.post(
@@ -482,18 +486,22 @@ def test_set_user_permissions_multiple(admin_request, sample_user, sample_servic
_expected_status=204,
)
permission = Permission.query.filter_by(permission=MANAGE_SETTINGS).first()
permission = Permission.query.filter_by(
permission=PermissionType.MANAGE_SETTINGS
).first()
assert permission.user == sample_user
assert permission.service == sample_service
assert permission.permission == MANAGE_SETTINGS
permission = Permission.query.filter_by(permission=MANAGE_TEMPLATES).first()
assert permission.permission == PermissionType.MANAGE_SETTINGS
permission = Permission.query.filter_by(
permission=PermissionType.MANAGE_TEMPLATES
).first()
assert permission.user == sample_user
assert permission.service == sample_service
assert permission.permission == MANAGE_TEMPLATES
assert permission.permission == PermissionType.MANAGE_TEMPLATES
def test_set_user_permissions_remove_old(admin_request, sample_user, sample_service):
data = {"permissions": [{"permission": MANAGE_SETTINGS}]}
data = {"permissions": [{"permission": PermissionType.MANAGE_SETTINGS}]}
admin_request.post(
"user.set_permissions",
@@ -505,7 +513,7 @@ def test_set_user_permissions_remove_old(admin_request, sample_user, sample_serv
query = Permission.query.filter_by(user=sample_user)
assert query.count() == 1
assert query.first().permission == MANAGE_SETTINGS
assert query.first().permission == PermissionType.MANAGE_SETTINGS
def test_set_user_folder_permissions(admin_request, sample_user, sample_service):
@@ -873,35 +881,35 @@ def test_activate_user_fails_if_already_active(admin_request, sample_user):
def test_update_user_auth_type(admin_request, sample_user):
assert sample_user.auth_type == "sms_auth"
assert sample_user.auth_type == AuthType.SMS
resp = admin_request.post(
"user.update_user_attribute",
user_id=sample_user.id,
_data={"auth_type": "email_auth"},
_data={"auth_type": AuthType.EMAIL},
)
assert resp["data"]["id"] == str(sample_user.id)
assert resp["data"]["auth_type"] == "email_auth"
assert resp["data"]["auth_type"] == AuthType.EMAIL
def test_can_set_email_auth_and_remove_mobile_at_same_time(admin_request, sample_user):
sample_user.auth_type = SMS_AUTH_TYPE
sample_user.auth_type = AuthType.SMS
admin_request.post(
"user.update_user_attribute",
user_id=sample_user.id,
_data={
"mobile_number": None,
"auth_type": EMAIL_AUTH_TYPE,
"auth_type": AuthType.EMAIL,
},
)
assert sample_user.mobile_number is None
assert sample_user.auth_type == EMAIL_AUTH_TYPE
assert sample_user.auth_type == AuthType.EMAIL
def test_cannot_remove_mobile_if_sms_auth(admin_request, sample_user):
sample_user.auth_type = SMS_AUTH_TYPE
sample_user.auth_type = AuthType.SMS
json_resp = admin_request.post(
"user.update_user_attribute",
@@ -912,12 +920,12 @@ def test_cannot_remove_mobile_if_sms_auth(admin_request, sample_user):
assert (
json_resp["message"]
== "Mobile number must be set if auth_type is set to sms_auth"
== "Mobile number must be set if auth_type is set to AuthType.SMS"
)
def test_can_remove_mobile_if_email_auth(admin_request, sample_user):
sample_user.auth_type = EMAIL_AUTH_TYPE
sample_user.auth_type = AuthType.EMAIL
admin_request.post(
"user.update_user_attribute",
@@ -931,7 +939,7 @@ def test_can_remove_mobile_if_email_auth(admin_request, sample_user):
def test_cannot_update_user_with_mobile_number_as_empty_string(
admin_request, sample_user
):
sample_user.auth_type = EMAIL_AUTH_TYPE
sample_user.auth_type = AuthType.EMAIL
resp = admin_request.post(
"user.update_user_attribute",

60
tests/app/user/test_rest_verify.py
View File

@@ -10,14 +10,8 @@ import app.celery.tasks
from app import db
from app.dao.services_dao import dao_fetch_service_by_id
from app.dao.users_dao import create_user_code
from app.models import (
EMAIL_TYPE,
SMS_TYPE,
USER_AUTH_TYPES,
Notification,
User,
VerifyCode,
)
from app.enums import AuthType, CodeType
from app.models import Notification, User, VerifyCode
from tests import create_admin_authorization_header
@@ -103,7 +97,7 @@ def test_user_verify_code_rejects_good_code_if_too_many_failed_logins(
@freeze_time("2020-04-01 12:00")
@pytest.mark.parametrize("code_type", [EMAIL_TYPE, SMS_TYPE])
@pytest.mark.parametrize("code_type", [CodeType.EMAIL, CodeType.SMS])
def test_user_verify_code_expired_code_and_increments_failed_login_count(
code_type, admin_request, sample_user
):
@@ -215,7 +209,11 @@ def test_send_user_sms_code(client, sample_user, sms_code_template, mocker):
mocker.patch("app.celery.provider_tasks.deliver_sms.apply_async")
resp = client.post(
url_for("user.send_user_2fa_code", code_type="sms", user_id=sample_user.id),
url_for(
"user.send_user_2fa_code",
code_type=CodeType.SMS,
user_id=sample_user.id,
),
data=json.dumps({}),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -253,7 +251,11 @@ def test_send_user_code_for_sms_with_optional_to_field(
auth_header = create_admin_authorization_header()
resp = client.post(
url_for("user.send_user_2fa_code", code_type="sms", user_id=sample_user.id),
url_for(
"user.send_user_2fa_code",
code_type=CodeType.SMS,
user_id=sample_user.id,
),
data=json.dumps({"to": to_number}),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -271,7 +273,7 @@ def test_send_sms_code_returns_404_for_bad_input_data(client):
uuid_ = uuid.uuid4()
auth_header = create_admin_authorization_header()
resp = client.post(
url_for("user.send_user_2fa_code", code_type="sms", user_id=uuid_),
url_for("user.send_user_2fa_code", code_type=CodeType.SMS, user_id=uuid_),
data=json.dumps({}),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -284,7 +286,7 @@ def test_send_sms_code_returns_204_when_too_many_codes_already_created(
):
for _ in range(5):
verify_code = VerifyCode(
code_type="sms",
code_type=CodeType.SMS,
_code=12345,
created_at=datetime.utcnow() - timedelta(minutes=10),
expiry_datetime=datetime.utcnow() + timedelta(minutes=40),
@@ -295,7 +297,11 @@ def test_send_sms_code_returns_204_when_too_many_codes_already_created(
assert VerifyCode.query.count() == 5
auth_header = create_admin_authorization_header()
resp = client.post(
url_for("user.send_user_2fa_code", code_type="sms", user_id=sample_user.id),
url_for(
"user.send_user_2fa_code",
code_type=CodeType.SMS,
user_id=sample_user.id,
),
data=json.dumps({}),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -431,8 +437,8 @@ def test_reset_failed_login_count_returns_404_when_user_does_not_exist(client):
assert resp.status_code == 404
# we send sms_auth users and webauthn_auth users email code to validate their email access
@pytest.mark.parametrize("auth_type", USER_AUTH_TYPES)
# we send AuthType.SMS users and AuthType.WEBAUTHN users email code to validate their email access
@pytest.mark.parametrize("auth_type", AuthType)
@pytest.mark.parametrize(
"data, expected_auth_url",
(
@@ -469,7 +475,7 @@ def test_send_user_email_code(
admin_request.post(
"user.send_user_2fa_code",
code_type="email",
code_type=CodeType.EMAIL,
user_id=sample_user.id,
_data=data,
_expected_status=204,
@@ -503,7 +509,7 @@ def test_send_user_email_code_with_urlencoded_next_param(
data = {"to": None, "next": "/services"}
admin_request.post(
"user.send_user_2fa_code",
code_type="email",
code_type=CodeType.EMAIL,
user_id=sample_user.id,
_data=data,
_expected_status=204,
@@ -519,7 +525,7 @@ def test_send_user_email_code_with_urlencoded_next_param(
def test_send_email_code_returns_404_for_bad_input_data(admin_request):
resp = admin_request.post(
"user.send_user_2fa_code",
code_type="email",
code_type=CodeType.EMAIL,
user_id=uuid.uuid4(),
_data={},
_expected_status=404,
@@ -528,16 +534,16 @@ def test_send_email_code_returns_404_for_bad_input_data(admin_request):
@freeze_time("2016-01-01T12:00:00")
# we send sms_auth and webauthn_auth users email code to validate their email access
@pytest.mark.parametrize("auth_type", USER_AUTH_TYPES)
# we send iAuthType.SMS and AuthType.WEBAUTHN users email code to validate their email access
@pytest.mark.parametrize("auth_type", AuthType)
def test_user_verify_email_code(admin_request, sample_user, auth_type):
sample_user.logged_in_at = datetime.utcnow() - timedelta(days=1)
sample_user.email_access_validated_at = datetime.utcnow() - timedelta(days=1)
sample_user.auth_type = auth_type
magic_code = str(uuid.uuid4())
verify_code = create_user_code(sample_user, magic_code, EMAIL_TYPE)
verify_code = create_user_code(sample_user, magic_code, CodeType.EMAIL)
data = {"code_type": "email", "code": magic_code}
data = {"code_type": CodeType.EMAIL, "code": magic_code}
admin_request.post(
"user.verify_user_code",
@@ -552,7 +558,7 @@ def test_user_verify_email_code(admin_request, sample_user, auth_type):
assert sample_user.current_session_id is not None
@pytest.mark.parametrize("code_type", [EMAIL_TYPE, SMS_TYPE])
@pytest.mark.parametrize("code_type", [CodeType.EMAIL, CodeType.SMS])
@freeze_time("2016-01-01T12:00:00")
def test_user_verify_email_code_fails_if_code_already_used(
admin_request, sample_user, code_type
@@ -589,7 +595,11 @@ def test_send_user_2fa_code_sends_from_number_for_international_numbers(
mocker.patch("app.user.rest.send_notification_to_queue")
resp = client.post(
url_for("user.send_user_2fa_code", code_type="sms", user_id=sample_user.id),
url_for(
"user.send_user_2fa_code",
code_type=CodeType.SMS,
user_id=sample_user.id,
),
data=json.dumps({}),
headers=[("Content-Type", "application/json"), auth_header],
)