mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-02-06 03:13:42 -05:00
2792bece54
when visited sends sms code for second step of account verification. At that second step user enters just sms code sent to users mobile number. Also moved dao calls that simply proxied calls to client to calling client directly. There is still a place where a user will be a sent a code for verification to their email namely if they update email address.
122 lines
4.7 KiB
Python
122 lines
4.7 KiB
Python
from notifications_python_client.notifications import BaseAPIClient
|
|
from notifications_python_client.errors import HTTPError
|
|
|
|
from app.notify_client.models import User
|
|
|
|
|
|
class UserApiClient(BaseAPIClient):
|
|
def __init__(self, base_url=None, client_id=None, secret=None):
|
|
super(self.__class__, self).__init__(base_url=base_url or 'base_url',
|
|
client_id=client_id or 'client_id',
|
|
secret=secret or 'secret')
|
|
|
|
def init_app(self, app):
|
|
self.base_url = app.config['API_HOST_NAME']
|
|
self.client_id = app.config['ADMIN_CLIENT_USER_NAME']
|
|
self.secret = app.config['ADMIN_CLIENT_SECRET']
|
|
self.max_failed_login_count = app.config["MAX_FAILED_LOGIN_COUNT"]
|
|
|
|
def register_user(self, name, email_address, mobile_number, password):
|
|
data = {
|
|
"name": name,
|
|
"email_address": email_address,
|
|
"mobile_number": mobile_number,
|
|
"password": password
|
|
}
|
|
user_data = self.post("/user", data)
|
|
return User(user_data['data'], max_failed_login_count=self.max_failed_login_count)
|
|
|
|
def get_user(self, id):
|
|
url = "/user/{}".format(id)
|
|
user_data = self.get(url)
|
|
return User(user_data['data'], max_failed_login_count=self.max_failed_login_count)
|
|
|
|
def get_user_by_email(self, email_address):
|
|
user_data = self.get('/user/email', params={'email': email_address})
|
|
return User(user_data['data'], max_failed_login_count=self.max_failed_login_count)
|
|
|
|
def get_users(self):
|
|
users_data = self.get("/user")['data']
|
|
users = []
|
|
for user in users_data:
|
|
users.append(User(user, max_failed_login_count=self.max_failed_login_count))
|
|
return users
|
|
|
|
def update_user(self, user):
|
|
data = user.serialize()
|
|
url = "/user/{}".format(user.id)
|
|
user_data = self.put(url, data=data)
|
|
return User(user_data['data'], max_failed_login_count=self.max_failed_login_count)
|
|
|
|
def verify_password(self, user_id, password):
|
|
try:
|
|
url = "/user/{}/verify/password".format(user_id)
|
|
data = {"password": password}
|
|
self.post(url, data=data)
|
|
return True
|
|
except HTTPError as e:
|
|
if e.status_code == 400 or e.status_code == 404:
|
|
return False
|
|
|
|
def send_verify_code(self, user_id, code_type, to):
|
|
data = {'to': to}
|
|
endpoint = '/user/{0}/{1}-code'.format(user_id, code_type)
|
|
self.post(endpoint, data=data)
|
|
|
|
def send_verify_email(self, user_id, to):
|
|
data = {'to': to}
|
|
endpoint = '/user/{0}/email-verification'.format(user_id)
|
|
self.post(endpoint, data=data)
|
|
|
|
def check_verify_code(self, user_id, code, code_type):
|
|
data = {'code_type': code_type, 'code': code}
|
|
endpoint = '/user/{}/verify/code'.format(user_id)
|
|
try:
|
|
resp = self.post(endpoint, data=data)
|
|
return True, ''
|
|
except HTTPError as e:
|
|
if e.status_code == 400 or e.status_code == 404:
|
|
if 'Code not found' in e.message:
|
|
return False, 'Code not found'
|
|
elif 'Code has expired' in e.message:
|
|
return False, 'Code has expired'
|
|
else:
|
|
# TODO what is the default message?
|
|
return False, 'Code not found'
|
|
raise e
|
|
|
|
def get_users_for_service(self, service_id):
|
|
endpoint = '/service/{}/users'.format(service_id)
|
|
resp = self.get(endpoint)
|
|
return [User(data) for data in resp['data']]
|
|
|
|
def add_user_to_service(self, service_id, user_id, permissions):
|
|
endpoint = '/service/{}/users/{}'.format(service_id, user_id)
|
|
resp = self.post(endpoint, data={'permissions': permissions})
|
|
return User(resp['data'], max_failed_login_count=self.max_failed_login_count)
|
|
|
|
def set_user_permissions(self, user_id, service_id, permissions):
|
|
data = [{'permission': x} for x in permissions]
|
|
endpoint = '/user/{}/service/{}/permission'.format(user_id, service_id)
|
|
self.post(endpoint, data=data)
|
|
|
|
def send_reset_password_url(self, email_address):
|
|
endpoint = '/user/reset-password'
|
|
data = {'email': email_address}
|
|
self.post(endpoint, data=data)
|
|
|
|
def is_email_unique(self, email_address):
|
|
try:
|
|
if self.get_user_by_email(email_address):
|
|
return False
|
|
return True
|
|
except HTTPError as ex:
|
|
if ex.status_code == 404:
|
|
return True
|
|
else:
|
|
raise ex
|
|
|
|
def activate_user(self, user):
|
|
user.state = 'active'
|
|
return self.update_user(user)
|