mirror of
https://github.com/GSA/notifications-api.git
synced 2026-09-11 18:38:14 -04:00
Merge pull request #1352 from alphagov/auth-type
Add auth_type column to user and invited user
This commit is contained in:
@@ -50,6 +50,11 @@ PRIORITY = 'priority'
|
||||
TEMPLATE_PROCESS_TYPE = [NORMAL, PRIORITY]
|
||||
|
||||
|
||||
SMS_AUTH_TYPE = 'sms_auth'
|
||||
EMAIL_AUTH_TYPE = 'email_auth'
|
||||
USER_AUTH_TYPE = [SMS_AUTH_TYPE, EMAIL_AUTH_TYPE]
|
||||
|
||||
|
||||
def filter_null_value_fields(obj):
|
||||
return dict(
|
||||
filter(lambda x: x[1] is not None, obj.items())
|
||||
@@ -100,6 +105,7 @@ class User(db.Model):
|
||||
state = db.Column(db.String, nullable=False, default='pending')
|
||||
platform_admin = db.Column(db.Boolean, nullable=False, default=False)
|
||||
current_session_id = db.Column(UUID(as_uuid=True), nullable=True)
|
||||
auth_type = db.Column(db.String, db.ForeignKey('auth_type.name'), index=True, nullable=False, default=SMS_AUTH_TYPE)
|
||||
|
||||
services = db.relationship(
|
||||
'Service',
|
||||
@@ -1245,6 +1251,13 @@ class InvitedUser(db.Model):
|
||||
status = db.Column(
|
||||
db.Enum(*INVITED_USER_STATUS_TYPES, name='invited_users_status_types'), nullable=False, default='pending')
|
||||
permissions = db.Column(db.String, nullable=False)
|
||||
auth_type = db.Column(
|
||||
db.String,
|
||||
db.ForeignKey('auth_type.name'),
|
||||
index=True,
|
||||
nullable=False,
|
||||
default=SMS_AUTH_TYPE
|
||||
)
|
||||
|
||||
# would like to have used properties for this but haven't found a way to make them
|
||||
# play nice with marshmallow yet
|
||||
@@ -1549,3 +1562,9 @@ class NotificationSmsSender(db.Model):
|
||||
nullable=False,
|
||||
primary_key=True
|
||||
)
|
||||
|
||||
|
||||
class AuthType(db.Model):
|
||||
__tablename__ = 'auth_type'
|
||||
|
||||
name = db.Column(db.String, primary_key=True)
|
||||
|
||||
@@ -87,6 +87,7 @@ class UserSchema(BaseSchema):
|
||||
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')
|
||||
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):
|
||||
retval = {}
|
||||
@@ -505,6 +506,7 @@ class NotificationWithPersonalisationSchema(NotificationWithTemplateSchema):
|
||||
|
||||
|
||||
class InvitedUserSchema(BaseSchema):
|
||||
auth_type = field_for(models.InvitedUser, 'auth_type')
|
||||
|
||||
class Meta:
|
||||
model = models.InvitedUser
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
"""
|
||||
|
||||
Revision ID: 0131_user_auth_types
|
||||
Revises: 0130_service_email_reply_to_row
|
||||
Create Date: 2017-10-27 16:19:51.458863
|
||||
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
revision = '0131_user_auth_types'
|
||||
down_revision = '0130_service_email_reply_to_row'
|
||||
|
||||
|
||||
def upgrade():
|
||||
op.create_table(
|
||||
'auth_type',
|
||||
sa.Column('name', sa.String(), nullable=False),
|
||||
sa.PrimaryKeyConstraint('name')
|
||||
)
|
||||
op.execute("INSERT INTO auth_type VALUES ('email_auth'), ('sms_auth')")
|
||||
|
||||
op.add_column('users', sa.Column('auth_type', sa.String(), nullable=False, server_default='sms_auth'))
|
||||
|
||||
op.create_index(op.f('ix_users_auth_type'), 'users', ['auth_type'], unique=False)
|
||||
op.create_foreign_key(None, 'users', 'auth_type', ['auth_type'], ['name'])
|
||||
|
||||
op.add_column('invited_users', sa.Column('auth_type', sa.String(), nullable=False, server_default='sms_auth'))
|
||||
|
||||
op.create_index(op.f('ix_invited_users_auth_type'), 'invited_users', ['auth_type'], unique=False)
|
||||
op.create_foreign_key(None, 'invited_users', 'auth_type', ['auth_type'], ['name'])
|
||||
|
||||
|
||||
def downgrade():
|
||||
op.drop_column('users', 'auth_type')
|
||||
op.drop_column('invited_users', 'auth_type')
|
||||
op.drop_table('auth_type')
|
||||
@@ -1,11 +1,11 @@
|
||||
import json
|
||||
import uuid
|
||||
|
||||
from app.models import Notification
|
||||
from app.models import Notification, SMS_AUTH_TYPE, EMAIL_AUTH_TYPE
|
||||
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')
|
||||
email_address = 'invited_user@service.gov.uk'
|
||||
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),
|
||||
'email_address': email_address,
|
||||
'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(
|
||||
'/service/{}/invite'.format(sample_service.id),
|
||||
headers=[('Content-Type', 'application/json'), auth_header],
|
||||
data=json.dumps(data)
|
||||
json_resp = admin_request.post(
|
||||
'invite.create_invited_user',
|
||||
service_id=sample_service.id,
|
||||
_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']['email_address'] == email_address
|
||||
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']['auth_type'] == EMAIL_AUTH_TYPE
|
||||
assert json_resp['data']['id']
|
||||
|
||||
notification = Notification.query.first()
|
||||
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):
|
||||
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
|
||||
email_address = 'notanemail'
|
||||
|
||||
+37
-13
@@ -1,30 +1,34 @@
|
||||
import json
|
||||
import pytest
|
||||
|
||||
from flask import url_for, current_app
|
||||
from flask import url_for
|
||||
from freezegun import freeze_time
|
||||
|
||||
import app
|
||||
from app.models import (User, Permission, MANAGE_SETTINGS, MANAGE_TEMPLATES, Notification)
|
||||
from app.models import (
|
||||
User,
|
||||
Permission,
|
||||
MANAGE_SETTINGS,
|
||||
MANAGE_TEMPLATES,
|
||||
Notification,
|
||||
SMS_AUTH_TYPE,
|
||||
EMAIL_AUTH_TYPE
|
||||
)
|
||||
from app.dao.permissions_dao import default_service_permissions
|
||||
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.
|
||||
"""
|
||||
header = create_authorization_header()
|
||||
response = client.get(url_for('user.get_user'),
|
||||
headers=[header])
|
||||
assert response.status_code == 200
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
assert len(json_resp['data']) == 1
|
||||
json_resp = admin_request.get('user.get_user')
|
||||
|
||||
# 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
|
||||
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.mobile_number == fetched['mobile_number']
|
||||
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.email_address == fetched['email_address']
|
||||
assert sample_user.state == fetched['state']
|
||||
assert fetched['auth_type'] == SMS_AUTH_TYPE
|
||||
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,
|
||||
"state": "active",
|
||||
"failed_login_count": 0,
|
||||
"permissions": {}
|
||||
"permissions": {},
|
||||
"auth_type": EMAIL_AUTH_TYPE
|
||||
}
|
||||
auth_header = create_authorization_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))
|
||||
assert json_resp['data']['email_address'] == user.email_address
|
||||
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):
|
||||
|
||||
+2
-1
@@ -106,7 +106,8 @@ def notify_db_session(notify_db):
|
||||
"template_process_type",
|
||||
"dvla_organisation",
|
||||
"notification_status_types",
|
||||
"service_permission_types"]:
|
||||
"service_permission_types",
|
||||
"auth_type"]:
|
||||
notify_db.engine.execute(tbl.delete())
|
||||
notify_db.session.commit()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user