darkhelm/notifications-admin
1
0
Fork 0
mirror of https://github.com/GSA/notifications-admin.git synced 2026-08-11 09:28:27 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge with master

Browse Source
This commit is contained in:
Nicholas Staples
2016-03-09 12:12:01 +00:00
parent d832f7aa5f 2dbfff4730
commit d0f113b6bb
32 changed files with 504 additions and 595 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/__init__.py
View File

@@ -19,7 +19,7 @@ from app.notify_client.status_api_client import StatusApiClient
from app.notify_client.invite_api_client import InviteApiClient
from app.its_dangerous_session import ItsdangerousSessionInterface
from app.asset_fingerprinter import AssetFingerprinter
from app.utils import validate_phone_number, InvalidPhoneError
from utils.recipients import validate_phone_number, InvalidPhoneError
import app.proxy_fix
from config import configs
from utils import logging

12
app/assets/stylesheets/components/banner.scss
View File

@@ -12,6 +12,18 @@
position: relative;
clear: both;
&-title {
@include bold-24;
}
p {
margin: 10px 0 5px 0;
}
.list-bullet {
@include copy-19;
}
}
.banner-with-tick,

31
app/assets/stylesheets/components/table.scss
View File

@@ -16,10 +16,26 @@
%table-field,
.table-field {
vertical-align: top;
&:last-child {
padding-right: 0;
}
&-error {
border-left: 5px solid $error-colour;
padding-left: 7px;
display: block;
&-label {
display: block;
color: $error-colour;
font-weight: bold;
}
}
&-status {
&-default {
@@ -55,13 +71,6 @@
background-image: file-url('tick.png');
}
&-missing {
color: $error-colour;
font-weight: bold;
border-left: 5px solid $error-colour;
padding-left: 7px;
}
}
}
@@ -92,9 +101,15 @@
}
.table-show-more-link {
@include bold-16;
@include core-16;
color: $secondary-text-colour;
margin-top: -20px;
border-bottom: 1px solid $border-colour;
padding-bottom: 10px;
text-align: center;
}
a.table-show-more-link {
@include bold-16;
color: $link-colour;
}

5
app/main/dao/users_dao.py
View File

@@ -64,11 +64,6 @@ def is_email_unique(email_address):
raise ex
def request_password_reset(user):
user.state = 'request_password_reset'
user_api_client.update_user(user)
def send_verify_code(user_id, code_type, to):
return user_api_client.send_verify_code(user_id, code_type, to)

15
app/main/forms.py
View File

@@ -16,7 +16,7 @@ from wtforms.validators import DataRequired, Email, Length, Regexp
from app.main.validators import Blacklist, CsvFileValidator
from app.utils import (
from utils.recipients import (
validate_phone_number,
format_phone_number,
InvalidPhoneError
@@ -37,19 +37,10 @@ class UKMobileNumber(TelField):
def pre_validate(self, form):
try:
self.data = validate_phone_number(self.data)
validate_phone_number(self.data)
except InvalidPhoneError as e:
raise ValidationError(e.message)
def post_validate(self, form, validation_stopped):
if len(self.data) != 9:
return
# TODO implement in the render field method.
# API's require no spaces in the number
# self.data = '+44 7{} {} {}'.format(*re.findall('...', self.data))
self.data = format_phone_number(self.data)
def mobile_number():
return UKMobileNumber('Mobile phone number',
@@ -104,7 +95,7 @@ class RegisterUserFromInviteForm(Form):
mobile_number = mobile_number()
password = password()
service = HiddenField('service')
email_address = email_address()
email_address = HiddenField('email_address')
class InviteUserForm(Form):

2
app/main/uploader.py
View File

@@ -8,7 +8,7 @@ BUCKET_NAME = 'service-{}-notify'
def s3upload(upload_id, service_id, filedata, region):
s3 = resource('s3')
bucket_name = BUCKET_NAME.format(service_id)
contents = '\n'.join(filedata['data'])
contents = filedata['data']
exists = True
try:

19
app/main/views/forgot_password.py
View File

@@ -1,25 +1,22 @@
from flask import (
render_template,
flash
)
from notifications_python_client.errors import HTTPError
from app.main import main
from app.main.dao import users_dao
from app.main.forms import ForgotPasswordForm
from app.notify_client.sender import send_change_password_email
from app import user_api_client
@main.route('/forgot-password', methods=['GET', 'POST'])
def forgot_password():
form = ForgotPasswordForm()
if form.validate_on_submit():
if not users_dao.is_email_unique(form.email_address.data):
user = users_dao.get_user_by_email(form.email_address.data)
users_dao.request_password_reset(user)
send_change_password_email(form.email_address.data)
return render_template('views/password-reset-sent.html')
else:
return render_template('views/password-reset-sent.html')
try:
user_api_client.send_reset_password_url(form.email_address.data)
except HTTPError as e:
if e.status_code != 404:
raise e
return render_template('views/password-reset-sent.html')
return render_template('views/forgot-password.html', form=form)

1
app/main/views/invites.py
View File

@@ -33,6 +33,7 @@ def accept_invite(token):
session['invited_user'] = invited_user.serialize()
if existing_user:
user_api_client.add_user_to_service(invited_user.service,
existing_user.id,
invited_user.permissions)

26
app/main/views/new_password.py
View File

@@ -1,22 +1,33 @@
from flask import (render_template, url_for, redirect, flash, session)
import json
from flask import (render_template, url_for, redirect, flash, session, current_app, abort)
from itsdangerous import SignatureExpired
from app.main import main
from app.main.dao import users_dao
from app.main.forms import NewPasswordForm
from app.notify_client.sender import check_token
from datetime import datetime
@main.route('/new-password/<path:token>', methods=['GET', 'POST'])
def new_password(token):
email_address = check_token(token)
if not email_address:
from utils.url_safe_token import check_token
try:
token_data = check_token(token, current_app.config['SECRET_KEY'], current_app.config['DANGEROUS_SALT'],
current_app.config['TOKEN_MAX_AGE_SECONDS'])
except SignatureExpired:
flash('The link in the email we sent you has expired. Enter your email address to resend.')
return redirect(url_for('.forgot_password'))
email_address = json.loads(token_data)['email']
user = users_dao.get_user_by_email(email_address=email_address)
if user and user.state != 'request_password_reset':
flash('The link in the email we sent you has already been used.')
return redirect(url_for('.index'))
# TODO: what should this be??
if not user:
abort(404, 'user not found')
if user.password_changed_at and datetime.strptime(user.password_changed_at, '%Y-%m-%d %H:%M:%S.%f') > \
datetime.strptime(json.loads(token_data)['created_at'], '%Y-%m-%d %H:%M:%S.%f'):
flash('The link in the email has already been used')
return redirect(url_for('main.index'))
form = NewPasswordForm()
@@ -26,7 +37,6 @@ def new_password(token):
'id': user.id,
'email': user.email_address,
'password': form.new_password.data}
users_dao.activate_user(user)
return redirect(url_for('main.two_factor'))
else:
return render_template('views/new-password.html', token=token, form=form, user=user)

2
app/main/views/register.py
View File

@@ -51,7 +51,7 @@ def register_from_invite():
form.service.data = invited_user['service']
form.email_address.data = invited_user['email_address']
return render_template('views/register-from-invite.html', form=form)
return render_template('views/register-from-invite.html', email_address=invited_user['email_address'], form=form)
def _do_registration(form, service=None):

187
app/main/views/send.py
View File

@@ -1,6 +1,7 @@
import csv
import io
import uuid
from contextlib import suppress
from flask import (
request,
@@ -15,7 +16,8 @@ from flask import (
from flask_login import login_required, current_user
from notifications_python_client.errors import HTTPError
from utils.template import Template, NeededByTemplateError, NoPlaceholderForDataError
from utils.template import Template
from utils.recipients import RecipientCSV, first_column_heading
from app.main import main
from app.main.forms import CsvUploadForm
@@ -26,10 +28,7 @@ from app.main.uploader import (
from app.main.dao import templates_dao
from app.main.dao import services_dao
from app import job_api_client
from app.utils import (
validate_recipient, validate_header_row, InvalidPhoneError, InvalidEmailError, user_has_permissions,
InvalidHeaderError)
from utils.process_csv import first_column_heading
from app.utils import user_has_permissions, get_errors_for_csv
send_messages_page_headings = {
@@ -100,16 +99,25 @@ def send_messages(service_id, template_id):
form = CsvUploadForm()
if form.validate_on_submit():
try:
csv_file = form.file
filedata = _get_filedata(csv_file)
upload_id = str(uuid.uuid4())
s3upload(upload_id, service_id, filedata, current_app.config['AWS_REGION'])
session['upload_data'] = {"template_id": template_id, "original_file_name": filedata['file_name']}
s3upload(
upload_id,
service_id,
{
'file_name': form.file.data.filename,
'data': form.file.data.getvalue().decode('utf-8')
},
current_app.config['AWS_REGION']
)
session['upload_data'] = {
"template_id": template_id,
"original_file_name": form.file.data.filename
}
return redirect(url_for('.check_messages',
service_id=service_id,
upload_id=upload_id))
except ValueError as e:
flash('There was a problem uploading: {}'.format(csv_file.data.filename))
flash('There was a problem uploading: {}'.format(form.file.data.filename))
flash(str(e))
return redirect(url_for('.send_messages', service_id=service_id, template_id=template_id))
@@ -118,7 +126,6 @@ def send_messages(service_id, template_id):
templates_dao.get_service_template_or_404(service_id, template_id)['data'],
prefix=service['name']
)
recipient_column = first_column_heading[template.template_type]
return render_template(
'views/send.html',
@@ -174,7 +181,7 @@ def send_message_to_self(service_id, template_id):
filedata = {
'file_name': 'Test run',
'data': output.getvalue().splitlines()
'data': output.getvalue()
}
upload_id = str(uuid.uuid4())
s3upload(upload_id, service_id, filedata, current_app.config['AWS_REGION'])
@@ -185,107 +192,75 @@ def send_message_to_self(service_id, template_id):
upload_id=upload_id))
@main.route("/services/<service_id>/check/<upload_id>",
methods=['GET', 'POST'])
@main.route("/services/<service_id>/check/<upload_id>", methods=['GET'])
@login_required
@user_has_permissions('send_texts', 'send_emails', 'send_letters')
def check_messages(service_id, upload_id):
upload_data = session['upload_data']
template_id = upload_data.get('template_id')
service = services_dao.get_service_by_id_or_404(service_id)
if request.method == 'GET':
contents = s3download(service_id, upload_id)
if not contents:
flash('There was a problem reading your upload file')
raw_template = templates_dao.get_service_template_or_404(service_id, template_id)['data']
upload_result = _get_rows(contents, raw_template)
session['upload_data']['notification_count'] = len(upload_result['rows'])
template = Template(
raw_template,
values=upload_result['rows'][0] if upload_result['valid'] else {},
drop_values={first_column_heading[raw_template['template_type']]},
prefix=service['name']
)
return render_template(
'views/check.html',
upload_result=upload_result,
template=template,
page_heading=get_page_headings(template.template_type),
column_headers=[first_column_heading[template.template_type]] + list(template.placeholders_as_markup),
original_file_name=upload_data.get('original_file_name'),
service_id=service_id,
service=service,
form=CsvUploadForm()
)
elif request.method == 'POST':
if request.files:
# The csv was invalid, validate the csv again
return send_messages(service_id, template_id)
contents = s3download(service_id, upload_id)
if not contents:
flash('There was a problem reading your upload file')
original_file_name = upload_data.get('original_file_name')
notification_count = upload_data.get('notification_count')
session.pop('upload_data')
try:
job_api_client.create_job(upload_id, service_id, template_id, original_file_name, notification_count)
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
return redirect(
url_for('main.view_job', service_id=service_id, job_id=upload_id)
)
def _get_filedata(file):
import itertools
reader = csv.reader(
file.data.getvalue().decode('utf-8').splitlines(),
quoting=csv.QUOTE_NONE,
skipinitialspace=True
template = Template(
templates_dao.get_service_template_or_404(
service_id,
session['upload_data'].get('template_id')
)['data'],
prefix=service['name']
)
lines = []
for row in reader:
non_empties = itertools.dropwhile(lambda x: x.strip() == '', row)
has_content = []
for item in non_empties:
has_content.append(item)
if has_content:
lines.append(row)
if len(lines) < 2: # must be header row and at least one data row
message = 'The file {} contained no data'.format(file.data.filename)
raise ValueError(message)
content_lines = []
for row in lines:
content_lines.append(','.join(row).rstrip(','))
return {'file_name': file.data.filename, 'data': content_lines}
def _get_rows(contents, raw_template):
reader = csv.DictReader(
contents.split('\n'),
quoting=csv.QUOTE_NONE,
skipinitialspace=True
recipients = RecipientCSV(
contents,
template_type=template.template_type,
placeholders=template.placeholders,
max_initial_rows_shown=5
)
with suppress(StopIteration):
template.values = next(recipients.rows)
session['upload_data']['notification_count'] = len(list(recipients.rows))
session['upload_data']['valid'] = not recipients.has_errors
return render_template(
'views/check.html',
recipients=recipients,
template=template,
page_heading=get_page_headings(template.template_type),
errors=get_errors_for_csv(recipients, template.template_type),
count_of_recipients=session['upload_data']['notification_count'],
count_of_displayed_recipients=len(list(recipients.rows_annotated_and_truncated)),
original_file_name=session['upload_data'].get('original_file_name'),
service_id=service_id,
service=service,
form=CsvUploadForm()
)
@main.route("/services/<service_id>/check/<upload_id>", methods=['POST'])
@login_required
@user_has_permissions('send_texts', 'send_emails', 'send_letters')
def start_job(service_id, upload_id):
upload_data = session['upload_data']
services_dao.get_service_by_id_or_404(service_id)
if request.files or not upload_data.get('valid'):
# The csv was invalid, validate the csv again
return send_messages(service_id, upload_data.get('template_id'))
session.pop('upload_data')
job_api_client.create_job(
upload_id,
service_id,
upload_data.get('template_id'),
upload_data.get('original_file_name'),
upload_data.get('notification_count')
)
return redirect(
url_for('main.view_job', service_id=service_id, job_id=upload_id)
)
valid = True
rows = []
for row in reader:
rows.append(row)
try:
validate_recipient(
row, template_type=raw_template['template_type']
)
Template(
raw_template,
values=row,
drop_values={first_column_heading[raw_template['template_type']]}
).replaced
except (InvalidEmailError, InvalidPhoneError, NeededByTemplateError,
NoPlaceholderForDataError, InvalidHeaderError):
valid = False
return {"valid": valid, "rows": rows}

5
app/main/views/sign_in.py
View File

@@ -45,6 +45,11 @@ def sign_in():
# Vague error message for login in case of user not known, locked, inactive or password not verified
flash('Username or password is incorrect')
invited_user = session.get('invited_user')
if invited_user:
message = 'You already have an account with GOV.UK Notify. Sign in to your account to accept this invitation.'
flash(message, 'default')
return render_template('views/signin.html', form=form)

2
app/main/views/two_factor.py
View File

@@ -1,3 +1,4 @@
from flask import (
render_template,
redirect,
@@ -29,6 +30,7 @@ def two_factor():
try:
user = users_dao.get_user_by_id(user_id)
services = services_dao.get_services(user_id).get('data', [])
# Check if coming from new password page
if 'password' in session['user_details']:
user.set_password(session['user_details']['password'])
users_dao.update_user(user)

2
app/notify_client/models.py
View File

@@ -92,7 +92,7 @@ class User(UserMixin):
if service_id in self._permissions:
if or_:
return any([x in self._permissions[service_id] for x in permissions])
print(set(self._permissions[service_id]) >= set(permissions))
return set(self._permissions[service_id]) >= set(permissions)
return False

7
app/notify_client/user_api_client.py
View File

@@ -102,4 +102,9 @@ class UserApiClient(BaseAPIClient):
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)
resp = self.post(endpoint, data=data)
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)

4
app/templates/components/banner.html
View File

@@ -22,3 +22,7 @@
{% endif %}
</div>
{% endmacro %}
{% macro banner_wrapper(type=None, with_tick=False, delete_button=None, subhead=None) %}
{{ banner(caller()|safe, type=type, with_tick=with_tick, delete_button=delete_button, subhead=subhead) }}
{% endmacro %}

2
app/templates/components/file-upload.html
View File

@@ -1,5 +1,5 @@
{% macro file_upload(field, button_text="Choose file") %}
<form method="post" enctype="multipart/form-data" class="form-group{% if field.errors %} error{% endif %}" data-module="file-upload">
<form method="post" enctype="multipart/form-data" class="{% if field.errors %}error{% endif %}" data-module="file-upload">
<label class="file-upload-label" for="{{ field.name }}">
<span class="visually-hidden">{{ field.label }}</span>
{% if hint %}

104
app/templates/views/check.html
View File

@@ -1,38 +1,53 @@
{% extends "withnav_template.html" %}
{% from "components/banner.html" import banner_wrapper %}
{% from "components/email-message.html" import email_message %}
{% from "components/sms-message.html" import sms_message %}
{% from "components/table.html" import list_table, field %}
{% from "components/table.html" import list_table, field, text_field, hidden_field_heading %}
{% from "components/placeholder.html" import placeholder %}
{% from "components/file-upload.html" import file_upload %}
{% from "components/page-footer.html" import page_footer %}
{% block page_title %}
{{ "Check and confirm" if upload_result.valid else page_heading }} – GOV.UK Notify
{{ page_heading if errors else "Check and confirm" }} – GOV.UK Notify
{% endblock %}
{% block maincolumn_content %}
{% if template.additional_data %}
{{ banner(
"Remove these columns from your CSV file:" + ", ".join(template.missing_data),
type="dangerous"
) }}
{% elif not upload_result.valid %}
{{ banner(
"Your CSV file contained missing or invalid data",
type="dangerous"
) }}
{% if errors %}
<div class="bottom-gutter">
{% call banner_wrapper(type='dangerous') %}
{% if errors|length == 1 %}
<h1 class='banner-title'>
There was a problem with {{ original_file_name }}
</h1>
<p>
You need to {{ errors[0] }}
</p>
{% else %}
<h1 class='banner-title'>
There were some problems with {{ original_file_name }}
</h1>
<p>
You need to:
</p>
<ul class="list-bullet">
{% for error in errors %}
<li>{{ error }}</li>
{% endfor %}
</ul>
{% endif %}
{% endcall %}
</div>
{% else %}
<h1 class="heading-large">
Check and confirm
</h1>
{% endif %}
<h1 class="heading-large">
{{ "Check and confirm" if upload_result.valid else page_heading }}
</h1>
{% if 'email' == template.template_type %}
{{ email_message(
template.subject,
template.replaced if upload_result.valid else template.formatted_as_markup,
template.formatted_as_markup if errors else template.replaced,
from_address='{}@notifications.service.gov.uk'.format(service.email_from),
from_name=service.name
)}}
@@ -40,51 +55,52 @@
<div class="grid-row">
<div class="column-two-thirds">
{{ sms_message(
template.replaced if upload_result.valid else template.formatted_as_markup
template.formatted_as_markup if errors else template.replaced
)}}
</div>
</div>
{% endif %}
{% if upload_result.valid %}
{% if errors %}
{{file_upload(form.file, button_text='Re-upload your file')}}
{% else %}
<form method="post" enctype="multipart/form-data">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}" />
<input type="submit" class="button" value="{{ "Send {} message{}".format(upload_result.rows|count, '' if upload_result.rows|count == 1 else 's') }}" />
<input type="submit" class="button" value="{{ "Send {} message{}".format(count_of_recipients, '' if count_of_recipients == 1 else 's') }}" />
<a href="{{url_for('.send_messages', service_id=service_id, template_id=template.id)}}" class="page-footer-back-link">Back</a>
</form>
{% else %}
{{file_upload(form.file, button_text='Upload a CSV file')}}
{% endif %}
{% call(item) list_table(
upload_result.rows,
recipients.rows_annotated_and_truncated,
caption=original_file_name,
field_headings=column_headers
field_headings=['Row'] + recipients.column_headers_with_placeholders_highlighted
) %}
{% if item.get('phone number', '')|valid_phone_number %}
{% call field() %}
{{ item['phone number'] }}
{% endcall %}
{% elif item.get('email address') %}
{% call field() %}
{{ item['email address'] }}
{% endcall %}
{% else %}
{% call field(status='missing') %}
{{ item['phone number'] }}
{% endcall %}
{% endif %}
{% for column in template.placeholders %}
{% if item.get(column) %}
{% call field() %}
{{ item.index + 1 }}
{% endcall %}
{% for column in recipients.column_headers %}
{% if item[column].error %}
{% call field() %}
{{ item.get(column) }}
<span class="table-field-error">
<span class="table-field-error-label">{{ item[column].error }}</span>
{{ item[column].data if item[column].data != None }}
</span>
{% endcall %}
{% elif item[column].ignore %}
{% call field(status='default') %}
{{ item[column].data if item[column].data != None }}
{% endcall %}
{% else %}
{% call field(status='missing') %}
missing
{% endcall %}
{{ text_field(item[column].data) }}
{% endif %}
{% endfor %}
{% endcall %}
{% if count_of_displayed_recipients < count_of_recipients %}
<p class="table-show-more-link">
{{ count_of_recipients - count_of_displayed_recipients }} more {{ "row" if 1 == (count_of_recipients - count_of_displayed_recipients) else "rows"}} not shown
</p>
{% endif %}
{% endblock %}

3
app/templates/views/register-from-invite.html
View File

@@ -11,13 +11,14 @@ Create an account – GOV.UK Notify
<div class="grid-row">
<div class="column-two-thirds">
<h1 class="heading-large">Create an account</h1>
<p>Your account will be created with this email: {{email_address}}</p>
<form method="post" autocomplete="nope">
{{ textbox(form.email_address, width='3-4', disabled=True ) }}
{{ textbox(form.name, width='3-4') }}
{{ textbox(form.mobile_number, width='3-4') }}
{{ textbox(form.password, hint="Your password must have at least 10 characters", width='3-4') }}
{{ page_footer("Continue") }}
{{form.service}}
{{form.email_address}}
</form>
</div>
</div>

121
app/utils.py
View File

@@ -3,8 +3,6 @@ import re
from functools import wraps
from flask import (abort, session)
from utils.process_csv import get_recipient_from_row, first_column_heading
class BrowsableItem(object):
"""
@@ -32,84 +30,6 @@ class BrowsableItem(object):
pass
class InvalidEmailError(Exception):
def __init__(self, message):
self.message = message
class InvalidPhoneError(Exception):
def __init__(self, message):
self.message = message
class InvalidHeaderError(Exception):
def __init__(self, message):
self.message = message
def validate_phone_number(number):
sanitised_number = number.replace('(', '')
sanitised_number = sanitised_number.replace(')', '')
sanitised_number = sanitised_number.replace(' ', '')
sanitised_number = sanitised_number.replace('-', '')
if sanitised_number.startswith('+'):
sanitised_number = sanitised_number[1:]
valid_prefixes = ['07', '447', '4407', '00447']
if not sum(sanitised_number.startswith(prefix) for prefix in valid_prefixes):
raise InvalidPhoneError('Must be a UK mobile number (eg 07700 900460)')
for digit in sanitised_number:
try:
int(digit)
except(ValueError):
raise InvalidPhoneError('Must not contain letters or symbols')
# Split number on first 7
sanitised_number = sanitised_number.split('7', 1)[1]
if len(sanitised_number) > 9:
raise InvalidPhoneError('Too many digits')
if len(sanitised_number) < 9:
raise InvalidPhoneError('Not enough digits')
return sanitised_number
def format_phone_number(number):
import re
if len(number) > 9:
raise InvalidPhoneError('Too many digits')
if len(number) < 9:
raise InvalidPhoneError('Not enough digits')
return '+447{}{}{}'.format(*re.findall('...', number))
def validate_email_address(email_address):
if re.match(r"(^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$)", email_address):
return
raise InvalidEmailError('Not a valid email address')
def validate_recipient(row, template_type):
validate_header_row(row, template_type)
return {
'email': validate_email_address,
'sms': validate_phone_number
}[template_type](get_recipient_from_row(row, template_type))
def validate_header_row(row, template_type):
try:
column_heading = first_column_heading[template_type]
row[column_heading]
except KeyError as e:
raise InvalidHeaderError('Invalid header name, should be {}'.format(column_heading))
def user_has_permissions(*permissions, or_=False):
def wrap(func):
@wraps(func)
@@ -121,3 +41,44 @@ def user_has_permissions(*permissions, or_=False):
abort(403)
return wrap_func
return wrap
def get_errors_for_csv(recipients, template_type):
errors = []
missing_column_headers = list(recipients.missing_column_headers)
if len(missing_column_headers) == 1:
errors.append("add a column called ‘{}’".format("".join(missing_column_headers)))
elif len(missing_column_headers) == 2:
errors.append("add 2 columns, ‘{}’".format("’ and ‘".join(missing_column_headers)))
elif len(missing_column_headers) > 2:
errors.append(
"add columns called ‘{}’, and ‘{}’".format(
"’, ‘".join(missing_column_headers[0:-1]),
missing_column_headers[-1]
)
)
if recipients.rows_with_bad_recipients:
number_of_bad_recipients = len(list(recipients.rows_with_bad_recipients))
if 'sms' == template_type:
if 1 == number_of_bad_recipients:
errors.append("fix 1 phone number")
else:
errors.append("fix {} phone numbers".format(number_of_bad_recipients))
elif 'email' == template_type:
if 1 == number_of_bad_recipients:
errors.append("fix 1 email address")
else:
errors.append("fix {} email addresses".format(number_of_bad_recipients))
if recipients.rows_with_missing_data:
number_of_rows_with_missing_data = len(list(recipients.rows_with_missing_data))
if 1 == number_of_rows_with_missing_data:
errors.append("fill in 1 empty cell")
else:
errors.append("fill in {} empty cells".format(number_of_rows_with_missing_data))
return errors

2
requirements.txt
View File

@@ -14,4 +14,4 @@ Pygments==2.0.2
git+https://github.com/alphagov/notifications-python-client.git@0.3.1#egg=notifications-python-client==0.3.1
git+https://github.com/alphagov/notifications-utils.git@1.0.0#egg=notifications-utils==1.0.0
git+https://github.com/alphagov/notifications-utils.git@2.0.0#egg=notifications-utils==2.0.0

79
tests/app/main/test_errors_for_csv.py Normal file
View File

@@ -0,0 +1,79 @@
from collections import namedtuple
import pytest
from app.utils import get_errors_for_csv
MockRecipients = namedtuple(
'RecipientCSV',
[
'missing_column_headers',
'rows_with_bad_recipients',
'rows_with_missing_data'
]
)
@pytest.mark.parametrize(
"missing_column_headers,rows_with_bad_recipients,rows_with_missing_data,template_type,expected_errors",
[
(
[], [], [],
'sms',
[]
),
(
[], {2}, [],
'sms',
['fix 1 phone number']
),
(
[], {2, 4, 6}, [],
'sms',
['fix 3 phone numbers']
),
(
[], {1}, [],
'email',
['fix 1 email address']
),
(
[], {2, 4, 6}, [],
'email',
['fix 3 email addresses']
),
(
['name'], {2}, {3},
'sms',
[
'add a column called ‘name’',
'fix 1 phone number',
'fill in 1 empty cell'
]
),
(
['name', 'date'], [], [],
'sms',
['add 2 columns, ‘name’ and ‘date’']
),
(
['name', 'date', 'time'], {2, 4, 6, 8}, {3, 6, 9, 12},
'sms',
[
'add columns called ‘name’, ‘date’, and ‘time’',
'fix 4 phone numbers',
'fill in 4 empty cells'
]
)
]
)
def test_get_errors_for_csv(
missing_column_headers, rows_with_bad_recipients, rows_with_missing_data,
template_type,
expected_errors
):
assert get_errors_for_csv(
MockRecipients(missing_column_headers, rows_with_bad_recipients, rows_with_missing_data),
template_type
) == expected_errors

18
tests/app/main/test_utils.py → tests/app/main/test_permissions.py
View File

@@ -1,7 +1,7 @@
import pytest
from flask import url_for
from app.utils import user_has_permissions, validate_header_row, validate_recipient, InvalidHeaderError
from app.utils import user_has_permissions
from app.main.views.index import index
from werkzeug.exceptions import Forbidden
@@ -58,9 +58,13 @@ def test_user_has_permissions_multiple(app_,
response = decorated_index()
def test_validate_header_row():
row = {'bad': '+44 7700 900981'}
try:
validate_header_row(row, 'sms')
except InvalidHeaderError as e:
assert e.message == 'Invalid header name, should be phone number'
def test_exact_permissions(app_,
api_user_active,
mock_login,
mock_get_user_with_permissions):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
decorator = user_has_permissions('manage_users', 'manage_templates', 'manage_settings')
decorated_index = decorator(index)
response = decorated_index()

75
tests/app/main/test_phone_number_form_field.py
View File

@@ -1,75 +0,0 @@
import pytest
from wtforms import Form
from app.main.forms import UKMobileNumber
class FormExample(Form):
phone_number = UKMobileNumber()
invalid_phone_numbers = sum([
[
(phone_number, error) for phone_number in group
] for error, group in [
('Too many digits', (
'0712345678910',
'0044712345678910',
'0044712345678910',
'+44 (0)7123 456 789 10',
)),
('Not enough digits', (
'0712345678',
'004471234567',
'00447123456',
'+44 (0)7123 456 78',
)),
('Must be a UK mobile number (eg 07700 900460)', (
'08081 570364',
'+44 8081 570364',
'0117 496 0860',
'+44 117 496 0860',
'020 7946 0991',
'+44 20 7946 0991',
'71234567890',
)),
('Must not contain letters or symbols', (
'07890x32109',
'07123 456789...',
'07123 ☟☜⬇⬆☞☝',
'07123☟☜⬇⬆☞☝',
'07";DROP TABLE;"',
'+44 07ab cde fgh',
))
]
], [])
valid_phone_numbers = [
'07123456789',
'07123 456789',
'07123-456-789',
'00447123456789',
'00 44 7123456789',
'+447123456789',
'+44 7123 456 789',
'+44 (0)7123 456 789'
]
@pytest.mark.parametrize("phone_number", valid_phone_numbers)
def test_phone_number_accepts_valid_values(phone_number):
form = FormExample(phone_number=phone_number)
form.validate()
assert form.errors == {}
@pytest.mark.parametrize("phone_number, error_message", invalid_phone_numbers)
def test_phone_number_rejects_invalid_values(phone_number, error_message):
form = FormExample(phone_number=phone_number)
form.validate()
assert form.phone_number.errors[0] == error_message
@pytest.mark.parametrize("phone_number", valid_phone_numbers)
def test_phone_number_outputs_in_correct_format(phone_number):
form = FormExample(phone_number=phone_number)
form.validate()
assert form.phone_number.data == '+447123456789'

68
tests/app/main/test_phone_validation.py
View File

@@ -1,68 +0,0 @@
from app.utils import (
validate_phone_number,
InvalidPhoneError
)
import pytest
valid_phone_numbers = [
'07123456789',
'07123 456789',
'07123-456-789',
'00447123456789',
'00 44 7123456789',
'+447123456789',
'+44 7123 456 789',
'+44 (0)7123 456 789'
]
invalid_phone_numbers = sum([
[
(phone_number, error) for phone_number in group
] for error, group in [
('Too many digits', (
'0712345678910',
'0044712345678910',
'0044712345678910',
'+44 (0)7123 456 789 10',
)),
('Not enough digits', (
'0712345678',
'004471234567',
'00447123456',
'+44 (0)7123 456 78',
)),
('Must be a UK mobile number (eg 07700 900460)', (
'08081 570364',
'+44 8081 570364',
'0117 496 0860',
'+44 117 496 0860',
'020 7946 0991',
'+44 20 7946 0991',
'71234567890',
)),
('Must not contain letters or symbols', (
'07890x32109',
'07123 456789...',
'07123 ☟☜⬇⬆☞☝',
'07123☟☜⬇⬆☞☝',
'07";DROP TABLE;"',
'+44 07ab cde fgh',
))
]
], [])
@pytest.mark.parametrize("phone_number", valid_phone_numbers)
def test_phone_number_accepts_valid_values(phone_number):
try:
validate_phone_number(phone_number)
except InvalidPhoneError:
pytest.fail('Unexpected InvalidPhoneError')
@pytest.mark.parametrize("phone_number, error_message", invalid_phone_numbers)
def test_phone_number_rejects_invalid_values(phone_number, error_message):
with pytest.raises(InvalidPhoneError) as e:
validate_phone_number(phone_number)
assert error_message == str(e.value)

11
tests/app/main/views/test_accept_invite.py
View File

@@ -58,6 +58,10 @@ def test_existing_signed_out_user_accept_invite_redirects_to_sign_in(app_,
assert response.status_code == 200
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
assert page.h1.string.strip() == 'Sign in'
flash_banners = page.find_all('div', class_='banner-default')
assert len(flash_banners) == 2
assert flash_banners[0].text.strip() == 'Please log in to access this page.'
assert flash_banners[1].text.strip() == 'You already have an account with GOV.UK Notify. Sign in to your account to accept this invitation.' # noqa
def test_new_user_accept_invite_calls_api_and_redirects_to_registration(app_,
@@ -100,14 +104,17 @@ def test_new_user_accept_invite_calls_api_and_views_registration_page(app_,
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
assert page.h1.string.strip() == 'Create an account'
email_in_page = page.find('p')
assert email_in_page.text.strip() == 'Your account will be created with this email: invited_user@test.gov.uk' # noqa
form = page.find('form')
email = form.find('input', id='email_address')
name = form.find('input', id='name')
password = form.find('input', id='password')
service = form.find('input', type='hidden', id='service')
email = form.find('input', type='hidden', id='email_address')
assert email
assert email.attrs['disabled']
assert email.attrs['value'] == 'invited_user@test.gov.uk'
assert name
assert password
assert service

35
tests/app/main/views/test_forgot_password.py
View File

@@ -1,5 +1,7 @@
from flask import url_for
import app
def test_should_render_forgot_password(app_):
with app_.test_request_context():
@@ -9,14 +11,12 @@ def test_should_render_forgot_password(app_):
in response.get_data(as_text=True)
def test_should_redirect_to_password_reset_sent_and_state_updated(
app_,
api_user_active,
mock_get_user_by_email,
mock_update_user,
mock_send_email
):
def test_should_redirect_to_password_reset_sent_for_valid_email(
app_,
api_user_active,
mocker):
with app_.test_request_context():
mocker.patch('app.user_api_client.send_reset_password_url', return_value=None)
response = app_.test_client().post(
url_for('.forgot_password'),
data={'email_address': api_user_active.email_address})
@@ -24,23 +24,4 @@ def test_should_redirect_to_password_reset_sent_and_state_updated(
assert (
'You have been sent an email containing a link'
' to reset your password.') in response.get_data(as_text=True)
assert mock_send_email.call_count == 1
def test_should_redirect_to_password_reset_sent_for_non_existant_email_address(
app_,
api_user_active,
mock_dont_get_user_by_email,
mock_update_user,
mock_send_email
):
with app_.test_request_context():
response = app_.test_client().post(
url_for('.forgot_password'),
data={'email_address': 'nope@example.gov.uk'})
assert response.status_code == 200
assert (
'You have been sent an email containing a link'
' to reset your password.') in response.get_data(as_text=True)
mock_dont_get_user_by_email.assert_called_once_with('nope@example.gov.uk')
assert not mock_send_email.called
app.user_api_client.send_reset_password_url.assert_called_once_with(api_user_active.email_address)

65
tests/app/main/views/test_new_password.py
View File

@@ -1,10 +1,8 @@
import json
from datetime import datetime
from flask import url_for
from app.main.dao import users_dao
from app.main.encryption import check_hash
from app.notify_client.sender import generate_token
import pytest
from utils.url_safe_token import generate_token
def test_should_render_new_password_template(app_,
@@ -14,59 +12,62 @@ def test_should_render_new_password_template(app_,
mock_get_user_by_email_request_password_reset):
with app_.test_request_context():
with app_.test_client() as client:
token = generate_token(api_user_active.email_address)
data = json.dumps({'email': api_user_active.email_address, 'created_at': str(datetime.now())})
token = generate_token(data, app_.config['SECRET_KEY'],
app_.config['DANGEROUS_SALT'])
response = client.get(url_for('.new_password', token=token))
assert response.status_code == 200
assert 'You can now create a new password for your account.' in response.get_data(as_text=True)
@pytest.mark.skipif(True, reason='Password reset no implemented')
def test_should_render_new_password_template_with_message_of_bad_token(app_,
mock_get_user_by_email):
def test_should_return_404_when_email_address_does_not_exist(app_, mock_get_user_by_email_not_found):
with app_.test_request_context():
with app_.test_client() as client:
token = generate_token('no_user@d.gov.uk')
data = json.dumps({'email': 'no_user@d.gov.uk', 'created_at': str(datetime.now())})
token = generate_token(data, app_.config['SECRET_KEY'], app_.config['DANGEROUS_SALT'])
response = client.get(url_for('.new_password', token=token))
assert response.status_code == 200
assert 'Message about email address does not exist. Some one needs to figure out the words here.' in \
response.get_data(as_text=True)
assert response.status_code == 404
@pytest.mark.skipif(True, reason='Password reset no implemented')
def test_should_redirect_to_two_factor_when_password_reset_is_successful(app_,
mock_get_user_by_email_request_password_reset,
mock_login):
mock_login,
mock_send_verify_code):
with app_.test_request_context():
with app_.test_client() as client:
user = mock_get_user_by_email_request_password_reset.return_value
token = generate_token(user.email_address)
data = json.dumps({'email': user.email_address, 'created_at': str(datetime.now())})
token = generate_token(data, app_.config['SECRET_KEY'], app_.config['DANGEROUS_SALT'])
response = client.post(url_for('.new_password', token=token), data={'new_password': 'a-new_password'})
assert response.status_code == 302
assert response.location == url_for('.two_factor', _external=True)
mock_get_user_by_email_request_password_reset.assert_called_once_with(user.email_address)
def test_should_redirect_index_if_user_has_already_changed_password(app_,
mock_get_user_by_email_user_changed_password,
mock_login,
mock_send_verify_code):
with app_.test_request_context():
with app_.test_client() as client:
user = mock_get_user_by_email_user_changed_password.return_value
data = json.dumps({'email': user.email_address, 'created_at': str(datetime.now())})
token = generate_token(data, app_.config['SECRET_KEY'], app_.config['DANGEROUS_SALT'])
response = client.post(url_for('.new_password', token=token), data={'new_password': 'a-new_password'})
assert response.status_code == 302
assert response.location == url_for('.index', _external=True)
mock_get_user_by_email_user_changed_password.assert_called_once_with(user.email_address)
def test_should_redirect_to_forgot_password_with_flash_message_when_token_is_expired(
app_, mock_get_user_by_email_request_password_reset, mock_login
app_, mock_get_user_by_email_request_password_reset, mock_login
):
with app_.test_request_context():
with app_.test_client() as client:
app_.config['TOKEN_MAX_AGE_SECONDS'] = -1000
user = mock_get_user_by_email_request_password_reset.return_value
token = generate_token(user.email_address)
token = generate_token(user.email_address, app_.config['SECRET_KEY'], app_.config['DANGEROUS_SALT'])
response = client.post(url_for('.new_password', token=token), data={'new_password': 'a-new_password'})
assert response.status_code == 302
assert response.location == url_for('.forgot_password', _external=True)
app_.config['TOKEN_MAX_AGE_SECONDS'] = 3600
@pytest.mark.skipif(True, reason='Password reset no implemented')
def test_should_redirect_to_forgot_pass_when_user_active_should_be_request_passw_reset(
app_, mock_get_user_by_email_request_password_reset, mock_login
):
with app_.test_request_context():
with app_.test_client() as client:
user = mock_get_user_by_email_request_password_reset.return_value
token = generate_token(user.email_address)
response = client.post(url_for('.new_password', token=token), data={'new_password': 'a-new_password'})
assert response.status_code == 302
assert response.location == url_for('.index', _external=True)

2
tests/app/main/views/test_register.py
View File

@@ -55,7 +55,7 @@ def test_process_register_returns_400_when_mobile_number_is_invalid(app_,
'password': 'validPassword!'})
assert response.status_code == 200
assert 'Must be a UK mobile number (eg 07700 900460)' in response.get_data(as_text=True)
assert 'Must not contain letters or symbols' in response.get_data(as_text=True)
def test_should_return_400_when_email_is_not_gov_uk(app_,

109
tests/app/main/views/test_send.py
View File

@@ -34,33 +34,7 @@ def test_choose_template(
assert '{} template two content'.format(template_type) in content
def test_upload_empty_csvfile_returns_to_upload_page(
app_,
api_user_active,
mock_login,
mock_get_user,
mock_get_service,
mock_get_service_templates,
mock_check_verify_code,
mock_get_service_template,
mock_has_permissions
):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
upload_data = {'file': (BytesIO(''.encode('utf-8')), 'emtpy.csv')}
response = client.post(
url_for('main.send_messages', service_id=12345, template_id=54321),
data=upload_data,
follow_redirects=True
)
assert response.status_code == 200
content = response.get_data(as_text=True)
assert 'The file emtpy.csv contained no data' in content
def test_upload_csvfile_with_invalid_phone_shows_check_page_with_errors(
def test_upload_csvfile_with_errors_shows_check_page_with_errors(
app_,
api_user_active,
mocker,
@@ -86,43 +60,11 @@ def test_upload_csvfile_with_invalid_phone_shows_check_page_with_errors(
)
assert response.status_code == 200
content = response.get_data(as_text=True)
assert 'Your CSV file contained missing or invalid data' in content
assert 'There was a problem with invalid.csv' in content
assert '+44 123' in content
assert '+44 456' in content
assert 'Upload a CSV file' in content
def test_upload_csvfile_removes_empty_lines_and_trailing_commas(
app_,
api_user_active,
mocker,
mock_login,
mock_get_service,
mock_get_service_template,
mock_s3_upload,
mock_has_permissions
):
contents = 'phone number,name,,,\n++44 7700 900981,test1,,,\n+44 7700 900981,test2,,,\n ,,, \n ,,, \t \t \n'
file_data = (BytesIO(contents.encode('utf-8')), 'invalid.csv')
expected_data = {'data': ['phone number,name', '++44 7700 900981,test1', '+44 7700 900981,test2'],
'file_name': 'invalid.csv'}
mocker.patch('app.main.views.send.s3download',
return_value='phone number,name\n++44 7700 900981,test1\n+44 7700 900981,test2')
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
upload_data = {'file': file_data}
response = client.post(
url_for('main.send_messages', service_id=12345, template_id=54321),
data=upload_data,
follow_redirects=True
)
assert response.status_code == 200
mock_s3_upload.assert_called_with(ANY, '12345', expected_data, 'eu-west-1')
assert 'Not a UK mobile number' in content
assert 'Re-upload your file' in content
def test_send_test_message_to_self(
@@ -161,7 +103,7 @@ def test_send_test_message_to_self(
mock_has_permissions
):
expected_data = {'data': ['email address', 'test@user.gov.uk'], 'file_name': 'Test run'}
expected_data = {'data': 'email address\r\ntest@user.gov.uk\r\n', 'file_name': 'Test run'}
mocker.patch('app.main.views.send.s3download', return_value='email address\r\ntest@user.gov.uk')
with app_.test_request_context():
@@ -208,31 +150,32 @@ def test_upload_csvfile_with_valid_phone_shows_all_numbers(
mock_has_permissions
):
contents = 'phone number\n+44 7700 900981\n+44 7700 900982\n+44 7700 900983\n+44 7700 900984\n+44 7700 900985\n+44 7700 900986' # noqa
file_data = (BytesIO(contents.encode('utf-8')), 'valid.csv')
mocker.patch('app.main.views.send.s3download', return_value=contents)
mocker.patch(
'app.main.views.send.s3download',
return_value='phone number\n+44 7700 900981\n+44 7700 900982\n+44 7700 900983\n+44 7700 900984\n+44 7700 900985\n+44 7700 900986' # noqa
)
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
upload_data = {'file': file_data}
response = client.post(url_for('main.send_messages', service_id=12345, template_id=54321),
data=upload_data,
follow_redirects=True)
response = client.post(
url_for('main.send_messages', service_id=12345, template_id=54321),
data={'file': (BytesIO(), 'valid.csv')},
follow_redirects=True
)
with client.session_transaction() as sess:
assert int(sess['upload_data']['template_id']) == 54321
assert sess['upload_data']['original_file_name'] == 'valid.csv'
assert sess['upload_data']['notification_count'] == 6
content = response.get_data(as_text=True)
assert response.status_code == 200
assert '+44 7700 900981' in content
assert '+44 7700 900982' in content
assert '+44 7700 900983' in content
assert '+44 7700 900984' in content
assert '+44 7700 900985' in content
assert '+44 7700 900986' in content
assert '1 more row not shown' in content
def test_create_job_should_call_api(
@@ -261,8 +204,9 @@ def test_create_job_should_call_api(
with client.session_transaction() as session:
session['upload_data'] = {'original_file_name': original_file_name,
'template_id': template_id,
'notification_count': notification_count}
url = url_for('main.check_messages', service_id=service_one['id'], upload_id=job_id)
'notification_count': notification_count,
'valid': True}
url = url_for('main.start_job', service_id=service_one['id'], upload_id=job_id)
response = client.post(url, data=job_data, follow_redirects=True)
assert response.status_code == 200
@@ -285,11 +229,11 @@ def test_check_messages_should_revalidate_file_when_uploading_file(
):
service_id = service_one['id']
contents = 'phone number,name,,,\n++44 7700 900981,test1,,,\n+44 7700 900981,test2,,,\n ,,, \n ,,, \t \t \n'
file_data = (BytesIO(contents.encode('utf-8')), 'invalid.csv')
upload_data = {'file': file_data}
mocker.patch('app.main.views.send.s3download', return_value=contents)
mocker.patch(
'app.main.views.send.s3download',
return_value='phone number,name,,,\n++44 7700 900981,test1,,,\n+44 7700 900981,test2,,,\n ,,, \n ,,, \t \t \n'
)
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
@@ -297,10 +241,13 @@ def test_check_messages_should_revalidate_file_when_uploading_file(
session['upload_data'] = {'original_file_name': 'invalid.csv',
'template_id': job_data['template'],
'notification_count': job_data['notification_count']}
url = url_for('main.check_messages', service_id=service_id, upload_id=job_data['id'])
response = client.post(url, data=upload_data, follow_redirects=True)
response = client.post(
url_for('main.check_messages', service_id=service_id, upload_id=job_data['id']),
data={'file': (BytesIO(), 'invalid.csv')},
follow_redirects=True
)
assert response.status_code == 200
assert 'Your CSV file contained missing or invalid data' in response.get_data(as_text=True)
assert 'There was a problem with invalid.csv' in response.get_data(as_text=True)
def test_route_permissions(mocker,

32
tests/app/main/views/test_two_factor.py
View File

@@ -1,6 +1,5 @@
from flask import url_for
from tests import create_test_user
from tests.conftest import SERVICE_ONE_ID
def test_should_render_two_factor_page(app_,
@@ -33,11 +32,10 @@ def test_should_login_user_and_redirect_to_service_dashboard(app_,
'email': api_user_active.email_address}
response = client.post(url_for('main.two_factor'),
data={'sms_code': '12345'})
assert response.status_code == 302
assert response.location == url_for(
'main.service_dashboard',
service_id="596364a0-858e-42c8-9062-a8fe822260eb",
service_id=SERVICE_ONE_ID,
_external=True
)
@@ -109,3 +107,29 @@ def test_remember_me_set(app_,
response = client.post(url_for('main.two_factor'),
data={'sms_code': '23456', 'remember_me': True})
assert response.status_code == 302
def test_two_factor_should_set_password_when_new_password_exists_in_session(app_,
api_user_active,
mock_get_user,
mock_check_verify_code,
mock_get_services_with_one_service,
mock_update_user):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
session['user_details'] = {
'id': api_user_active.id,
'email': api_user_active.email_address,
'password': 'changedpassword'}
response = client.post(url_for('main.two_factor'),
data={'sms_code': '12345'})
assert response.status_code == 302
assert response.location == url_for(
'main.service_dashboard',
service_id=SERVICE_ONE_ID,
_external=True
)
api_user_active.password = 'changedpassword'
mock_update_user.assert_called_once_with(api_user_active)

51
tests/conftest.py
View File

@@ -1,5 +1,5 @@
import uuid
from datetime import date
from datetime import date, datetime, timedelta
import pytest
from app import create_app
@@ -90,17 +90,20 @@ def mock_update_service(mocker):
'app.notifications_api_client.update_service', side_effect=_update)
SERVICE_ONE_ID = "596364a0-858e-42c8-9062-a8fe822260eb"
SERVICE_TWO_ID = "147ad62a-2951-4fa1-9ca0-093cd1a52c52"
@pytest.fixture(scope='function')
def mock_get_services(mocker, user=None):
if user is None:
user = api_user_active()
def _create(user_id=None):
import uuid
service_one = service_json(
"596364a0-858e-42c8-9062-a8fe822260eb", "service_one", [user.id], 1000, True, False)
SERVICE_ONE_ID, "service_one", [user.id], 1000, True, False)
service_two = service_json(
"147ad62a-2951-4fa1-9ca0-093cd1a52c52", "service_two", [user.id], 1000, True, False)
SERVICE_TWO_ID, "service_two", [user.id], 1000, True, False)
return {'data': [service_one, service_two]}
return mocker.patch(
@@ -113,9 +116,8 @@ def mock_get_services_with_one_service(mocker, user=None):
user = api_user_active()
def _create(user_id=None):
import uuid
return {'data': [service_json(
"596364a0-858e-42c8-9062-a8fe822260eb", "service_one", [user.id], 1000, True, False
SERVICE_ONE_ID, "service_one", [user.id], 1000, True, False
)]}
return mocker.patch(
@@ -268,9 +270,27 @@ def api_user_request_password_reset():
'password': 'somepassword',
'email_address': 'test@user.gov.uk',
'mobile_number': '+4412341234',
'state': 'request_password_reset',
'state': 'active',
'failed_login_count': 5,
'permissions': {}
'permissions': {},
'password_changed_at': None
}
user = User(user_data)
return user
@pytest.fixture(scope='function')
def api_user_changed_password():
from app.notify_client.user_api_client import User
user_data = {'id': 555,
'name': 'Test User',
'password': 'somepassword',
'email_address': 'test@user.gov.uk',
'mobile_number': '+4412341234',
'state': 'active',
'failed_login_count': 5,
'permissions': {},
'password_changed_at': str(datetime.now() + timedelta(minutes=1))
}
user = User(user_data)
return user
@@ -345,6 +365,13 @@ def mock_get_user_by_email_request_password_reset(mocker, api_user_request_passw
return_value=api_user_request_password_reset)
@pytest.fixture(scope='function')
def mock_get_user_by_email_user_changed_password(mocker, api_user_changed_password):
return mocker.patch(
'app.user_api_client.get_user_by_email',
return_value=api_user_changed_password)
@pytest.fixture(scope='function')
def mock_get_user_by_email_locked(mocker, api_user_locked):
return mocker.patch(
@@ -382,14 +409,6 @@ def mock_verify_password(mocker):
side_effect=_verify_password)
@pytest.fixture(scope='function')
def mock_password_reset(mocker, api_user_active):
def _reset(email):
api_user_active.state = 'request_password_reset'
return mocker.patch('app.main.dao.users_dao.request_password_reset', side_effect=_reset)
@pytest.fixture(scope='function')
def mock_update_user(mocker):