mirror of
https://github.com/GSA/notifications-api.git
synced 2026-09-11 18:38:14 -04:00
add auth_type to schemas to enable updating in endpoints
This commit is contained in:
@@ -87,6 +87,7 @@ class UserSchema(BaseSchema):
|
|||||||
permissions = fields.Method("user_permissions", dump_only=True)
|
permissions = fields.Method("user_permissions", dump_only=True)
|
||||||
password_changed_at = field_for(models.User, 'password_changed_at', format='%Y-%m-%d %H:%M:%S.%f')
|
password_changed_at = field_for(models.User, 'password_changed_at', format='%Y-%m-%d %H:%M:%S.%f')
|
||||||
created_at = field_for(models.User, 'created_at', format='%Y-%m-%d %H:%M:%S.%f')
|
created_at = field_for(models.User, 'created_at', format='%Y-%m-%d %H:%M:%S.%f')
|
||||||
|
auth_type = field_for(models.User, 'auth_type')
|
||||||
|
|
||||||
def user_permissions(self, usr):
|
def user_permissions(self, usr):
|
||||||
retval = {}
|
retval = {}
|
||||||
@@ -505,6 +506,7 @@ class NotificationWithPersonalisationSchema(NotificationWithTemplateSchema):
|
|||||||
|
|
||||||
|
|
||||||
class InvitedUserSchema(BaseSchema):
|
class InvitedUserSchema(BaseSchema):
|
||||||
|
auth_type = field_for(models.InvitedUser, 'auth_type')
|
||||||
|
|
||||||
class Meta:
|
class Meta:
|
||||||
model = models.InvitedUser
|
model = models.InvitedUser
|
||||||
|
|||||||
@@ -1,11 +1,11 @@
|
|||||||
import json
|
import json
|
||||||
import uuid
|
import uuid
|
||||||
|
|
||||||
from app.models import Notification
|
from app.models import Notification, SMS_AUTH_TYPE, EMAIL_AUTH_TYPE
|
||||||
from tests import create_authorization_header
|
from tests import create_authorization_header
|
||||||
|
|
||||||
|
|
||||||
def test_create_invited_user(client, sample_service, mocker, invitation_email_template):
|
def test_create_invited_user(admin_request, sample_service, mocker, invitation_email_template):
|
||||||
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
|
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
|
||||||
email_address = 'invited_user@service.gov.uk'
|
email_address = 'invited_user@service.gov.uk'
|
||||||
invite_from = sample_service.users[0]
|
invite_from = sample_service.users[0]
|
||||||
@@ -14,28 +14,50 @@ def test_create_invited_user(client, sample_service, mocker, invitation_email_te
|
|||||||
'service': str(sample_service.id),
|
'service': str(sample_service.id),
|
||||||
'email_address': email_address,
|
'email_address': email_address,
|
||||||
'from_user': str(invite_from.id),
|
'from_user': str(invite_from.id),
|
||||||
'permissions': 'send_messages,manage_service,manage_api_keys'
|
'permissions': 'send_messages,manage_service,manage_api_keys',
|
||||||
|
'auth_type': EMAIL_AUTH_TYPE
|
||||||
}
|
}
|
||||||
auth_header = create_authorization_header()
|
|
||||||
|
|
||||||
response = client.post(
|
json_resp = admin_request.post(
|
||||||
'/service/{}/invite'.format(sample_service.id),
|
'invite.create_invited_user',
|
||||||
headers=[('Content-Type', 'application/json'), auth_header],
|
service_id=sample_service.id,
|
||||||
data=json.dumps(data)
|
_data=data,
|
||||||
|
_expected_status=201
|
||||||
)
|
)
|
||||||
assert response.status_code == 201
|
|
||||||
json_resp = json.loads(response.get_data(as_text=True))
|
|
||||||
|
|
||||||
assert json_resp['data']['service'] == str(sample_service.id)
|
assert json_resp['data']['service'] == str(sample_service.id)
|
||||||
assert json_resp['data']['email_address'] == email_address
|
assert json_resp['data']['email_address'] == email_address
|
||||||
assert json_resp['data']['from_user'] == str(invite_from.id)
|
assert json_resp['data']['from_user'] == str(invite_from.id)
|
||||||
assert json_resp['data']['permissions'] == 'send_messages,manage_service,manage_api_keys'
|
assert json_resp['data']['permissions'] == 'send_messages,manage_service,manage_api_keys'
|
||||||
|
assert json_resp['data']['auth_type'] == EMAIL_AUTH_TYPE
|
||||||
assert json_resp['data']['id']
|
assert json_resp['data']['id']
|
||||||
|
|
||||||
notification = Notification.query.first()
|
notification = Notification.query.first()
|
||||||
mocked.assert_called_once_with([(str(notification.id))], queue="notify-internal-tasks")
|
mocked.assert_called_once_with([(str(notification.id))], queue="notify-internal-tasks")
|
||||||
|
|
||||||
|
|
||||||
|
def test_create_invited_user_without_auth_type(admin_request, sample_service, mocker, invitation_email_template):
|
||||||
|
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
|
||||||
|
email_address = 'invited_user@service.gov.uk'
|
||||||
|
invite_from = sample_service.users[0]
|
||||||
|
|
||||||
|
data = {
|
||||||
|
'service': str(sample_service.id),
|
||||||
|
'email_address': email_address,
|
||||||
|
'from_user': str(invite_from.id),
|
||||||
|
'permissions': 'send_messages,manage_service,manage_api_keys',
|
||||||
|
}
|
||||||
|
|
||||||
|
json_resp = admin_request.post(
|
||||||
|
'invite.create_invited_user',
|
||||||
|
service_id=sample_service.id,
|
||||||
|
_data=data,
|
||||||
|
_expected_status=201
|
||||||
|
)
|
||||||
|
|
||||||
|
assert json_resp['data']['auth_type'] == SMS_AUTH_TYPE
|
||||||
|
|
||||||
|
|
||||||
def test_create_invited_user_invalid_email(client, sample_service, mocker):
|
def test_create_invited_user_invalid_email(client, sample_service, mocker):
|
||||||
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
|
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
|
||||||
email_address = 'notanemail'
|
email_address = 'notanemail'
|
||||||
|
|||||||
+37
-13
@@ -1,30 +1,34 @@
|
|||||||
import json
|
import json
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from flask import url_for, current_app
|
from flask import url_for
|
||||||
from freezegun import freeze_time
|
from freezegun import freeze_time
|
||||||
|
|
||||||
import app
|
from app.models import (
|
||||||
from app.models import (User, Permission, MANAGE_SETTINGS, MANAGE_TEMPLATES, Notification)
|
User,
|
||||||
|
Permission,
|
||||||
|
MANAGE_SETTINGS,
|
||||||
|
MANAGE_TEMPLATES,
|
||||||
|
Notification,
|
||||||
|
SMS_AUTH_TYPE,
|
||||||
|
EMAIL_AUTH_TYPE
|
||||||
|
)
|
||||||
from app.dao.permissions_dao import default_service_permissions
|
from app.dao.permissions_dao import default_service_permissions
|
||||||
from tests import create_authorization_header
|
from tests import create_authorization_header
|
||||||
|
|
||||||
|
|
||||||
def test_get_user_list(client, sample_service):
|
def test_get_user_list(admin_request, sample_service):
|
||||||
"""
|
"""
|
||||||
Tests GET endpoint '/' to retrieve entire user list.
|
Tests GET endpoint '/' to retrieve entire user list.
|
||||||
"""
|
"""
|
||||||
header = create_authorization_header()
|
json_resp = admin_request.get('user.get_user')
|
||||||
response = client.get(url_for('user.get_user'),
|
|
||||||
headers=[header])
|
# it may have the notify user in the DB still :weary:
|
||||||
assert response.status_code == 200
|
assert len(json_resp['data']) >= 1
|
||||||
json_resp = json.loads(response.get_data(as_text=True))
|
|
||||||
assert len(json_resp['data']) == 1
|
|
||||||
sample_user = sample_service.users[0]
|
sample_user = sample_service.users[0]
|
||||||
expected_permissions = default_service_permissions
|
expected_permissions = default_service_permissions
|
||||||
fetched = json_resp['data'][0]
|
fetched = next(x for x in json_resp['data'] if x['id'] == str(sample_user.id))
|
||||||
|
|
||||||
assert str(sample_user.id) == fetched['id']
|
|
||||||
assert sample_user.name == fetched['name']
|
assert sample_user.name == fetched['name']
|
||||||
assert sample_user.mobile_number == fetched['mobile_number']
|
assert sample_user.mobile_number == fetched['mobile_number']
|
||||||
assert sample_user.email_address == fetched['email_address']
|
assert sample_user.email_address == fetched['email_address']
|
||||||
@@ -52,6 +56,7 @@ def test_get_user(client, sample_service):
|
|||||||
assert sample_user.mobile_number == fetched['mobile_number']
|
assert sample_user.mobile_number == fetched['mobile_number']
|
||||||
assert sample_user.email_address == fetched['email_address']
|
assert sample_user.email_address == fetched['email_address']
|
||||||
assert sample_user.state == fetched['state']
|
assert sample_user.state == fetched['state']
|
||||||
|
assert fetched['auth_type'] == SMS_AUTH_TYPE
|
||||||
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
|
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
|
||||||
|
|
||||||
|
|
||||||
@@ -68,7 +73,8 @@ def test_post_user(client, notify_db, notify_db_session):
|
|||||||
"logged_in_at": None,
|
"logged_in_at": None,
|
||||||
"state": "active",
|
"state": "active",
|
||||||
"failed_login_count": 0,
|
"failed_login_count": 0,
|
||||||
"permissions": {}
|
"permissions": {},
|
||||||
|
"auth_type": EMAIL_AUTH_TYPE
|
||||||
}
|
}
|
||||||
auth_header = create_authorization_header()
|
auth_header = create_authorization_header()
|
||||||
headers = [('Content-Type', 'application/json'), auth_header]
|
headers = [('Content-Type', 'application/json'), auth_header]
|
||||||
@@ -81,6 +87,24 @@ def test_post_user(client, notify_db, notify_db_session):
|
|||||||
json_resp = json.loads(resp.get_data(as_text=True))
|
json_resp = json.loads(resp.get_data(as_text=True))
|
||||||
assert json_resp['data']['email_address'] == user.email_address
|
assert json_resp['data']['email_address'] == user.email_address
|
||||||
assert json_resp['data']['id'] == str(user.id)
|
assert json_resp['data']['id'] == str(user.id)
|
||||||
|
assert user.auth_type == EMAIL_AUTH_TYPE
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_user_without_auth_type(admin_request, notify_db_session):
|
||||||
|
assert User.query.count() == 0
|
||||||
|
data = {
|
||||||
|
"name": "Test User",
|
||||||
|
"email_address": "user@digital.cabinet-office.gov.uk",
|
||||||
|
"password": "password",
|
||||||
|
"mobile_number": "+447700900986",
|
||||||
|
"permissions": {},
|
||||||
|
}
|
||||||
|
|
||||||
|
json_resp = admin_request.post('user.create_user', _data=data, _expected_status=201)
|
||||||
|
|
||||||
|
user = User.query.filter_by(email_address='user@digital.cabinet-office.gov.uk').first()
|
||||||
|
assert json_resp['data']['id'] == str(user.id)
|
||||||
|
assert user.auth_type == SMS_AUTH_TYPE
|
||||||
|
|
||||||
|
|
||||||
def test_post_user_missing_attribute_email(client, notify_db, notify_db_session):
|
def test_post_user_missing_attribute_email(client, notify_db, notify_db_session):
|
||||||
|
|||||||
+2
-1
@@ -106,7 +106,8 @@ def notify_db_session(notify_db):
|
|||||||
"template_process_type",
|
"template_process_type",
|
||||||
"dvla_organisation",
|
"dvla_organisation",
|
||||||
"notification_status_types",
|
"notification_status_types",
|
||||||
"service_permission_types"]:
|
"service_permission_types",
|
||||||
|
"auth_type"]:
|
||||||
notify_db.engine.execute(tbl.delete())
|
notify_db.engine.execute(tbl.delete())
|
||||||
notify_db.session.commit()
|
notify_db.session.commit()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user