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

Merge pull request #84 from alphagov/integrate-python-client

Browse Source 
Integrate python client
This commit is contained in:
Adam Shimali
2016-01-19 11:57:33 +00:00
parent a90f575671 e2f5d64695
commit e74e591ca6
66 changed files with 1497 additions and 1017 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
app/__init__.py
View File

@@ -7,8 +7,7 @@ from flask.ext.sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flask_wtf import CsrfProtect
from werkzeug.exceptions import abort
from app.notify_client.api_client import AdminAPIClient
from app.notify_client.api_client import NotificationsAdminAPIClient
from app.its_dangerous_session import ItsdangerousSessionInterface
import app.proxy_fix
from config import configs
@@ -18,7 +17,7 @@ db = SQLAlchemy()
login_manager = LoginManager()
csrf = CsrfProtect()
admin_api_client = AdminAPIClient()
notifications_api_client = NotificationsAdminAPIClient()
def create_app(config_name, config_overrides=None):
@@ -31,6 +30,8 @@ def create_app(config_name, config_overrides=None):
init_csrf(application)
logging.init_app(application)
notifications_api_client.init_app(application)
login_manager.init_app(application)
login_manager.login_view = 'main.sign_in'
@@ -43,7 +44,6 @@ def create_app(config_name, config_overrides=None):
proxy_fix.init_app(application)
application.session_interface = ItsdangerousSessionInterface()
admin_api_client.init_app(application)
application.add_template_filter(placeholders)
application.add_template_filter(replace_placeholders)

108
app/main/dao/services_dao.py
View File

@@ -1,47 +1,97 @@
from flask import url_for
from datetime import datetime
from client.errors import HTTPError, InvalidResponse
from sqlalchemy.orm import load_only
from app import db
from app.models import Service
from flask.ext.login import current_user
from app import (db, notifications_api_client)
from app.main.utils import BrowsableItem
def insert_new_service(service_name, user):
service = Service(name=service_name,
created_at=datetime.now(),
limit=1000,
active=False,
restricted=True)
add_service(service)
service.users.append(user)
db.session.commit()
return service.id
def insert_new_service(service_name, user_id):
# Add a service with default attributes
# Should we try and handle exception here
resp = notifications_api_client.create_service(
service_name, False, 1000, True, user_id)
return resp['data']['id']
def get_service_by_id(id):
return Service.query.get(id)
def update_service(service):
return notifications_api_client.update_service(
service['id'],
service['name'],
service['active'],
service['limit'],
service['restricted'],
service['users'])
def get_service_by_id(id_):
return notifications_api_client.get_service(id_)
def get_services():
return notifications_api_client.get_services()
def unrestrict_service(service_id):
service = get_service_by_id(service_id)
service.restricted = False
add_service(service)
resp = notifications_api_client.get_service(service_id)
if resp['data']['restricted']:
resp = notifications_api_client.update_service(
service_id,
resp['data']['name'],
resp['data']['active'],
resp['data']['limit'],
False,
resp['data']['users'])
def activate_service(service_id):
service = get_service_by_id(service_id)
service.active = True
add_service(service)
def add_service(service):
db.session.add(service)
db.session.commit()
resp = notifications_api_client.get_service(service_id)
if not resp['data']['active']:
resp = notifications_api_client.update_service(
service_id,
resp['data']['name'],
True,
resp['data']['limit'],
resp['data']['restricted'],
resp['data']['users'])
# TODO Fix when functionality is added to the api.
def find_service_by_service_name(service_name):
return Service.query.filter_by(name=service_name).first()
resp = notifications_api_client.get_services()
retval = None
for srv_json in resp['data']:
if srv_json['name'] == service_name:
retval = srv_json
break
return retval
def delete_service(id_):
return notifications_api_client.delete_service(id_)
def find_all_service_names():
return [x.name for x in Service.query.options(load_only("name")).all()]
resp = notifications_api_client.get_services()
return [x['name'] for x in resp['data']]
class ServicesBrowsableItem(BrowsableItem):
@property
def title(self):
return self._item['name']
@property
def link(self):
return url_for('main.service_dashboard', service_id=self._item['id'])
@property
def destructive(self):
return False
@property
def hint(self):
return "Some service hint here"

17
app/main/forms.py
View File

@@ -157,24 +157,29 @@ class TextNotReceivedForm(Form):
class AddServiceForm(Form):
def __init__(self, service_names, *args, **kwargs):
self.service_names = service_names
def __init__(self, names_func, *args, **kwargs):
"""
Keyword arguments:
names_func -- Returns a list of unique service_names already registered
on the system.
"""
self._names_func = names_func
super(AddServiceForm, self).__init__(*args, **kwargs)
service_name = StringField(
name = StringField(
'Service name',
validators=[
DataRequired(message='Service name can not be empty')
]
)
def validate_service_name(self, a):
if self.service_name.data in self.service_names:
def validate_name(self, a):
if a.data in self._names_func():
raise ValidationError('Service name already exists')
class ServiceNameForm(Form):
service_name = StringField(u'New name')
name = StringField(u'New name')
class ConfirmPasswordForm(Form):

26
app/main/utils.py Normal file
View File

@@ -0,0 +1,26 @@
class BrowsableItem(object):
"""
Maps for the template browse-list.
"""
def __init__(self, item, *args, **kwargs):
self._item = item
super(BrowsableItem, self).__init__()
@property
def title(self):
pass
@property
def link(self):
pass
@property
def hint(self):
pass
@property
def destructive(self):
pass

18
app/main/views/add_service.py
View File

@@ -6,22 +6,18 @@ from app.main.forms import AddServiceForm
@main.route("/add-service", methods=['GET', 'POST'])
@main.route("/add-service/<string:first>", methods=['GET', 'POST'])
@login_required
def add_service(first=False):
if first:
if first == 'first':
heading = 'Set up notifications for your service'
else:
abort(404)
def add_service():
form = AddServiceForm(services_dao.find_all_service_names)
services = services_dao.get_services()
if len(services) > 0:
heading = 'Set up notifications for your service'
else:
heading = 'Add a new service'
form = AddServiceForm(services_dao.find_all_service_names())
if form.validate_on_submit():
user = users_dao.get_user_by_id(session['user_id'])
services_dao.insert_new_service(form.service_name.data, user)
return redirect(url_for('.dashboard', service_id=123))
service_id = services_dao.insert_new_service(form.name.data, user)
return redirect(url_for('main.service_dashboard', service_id=service_id))
else:
return render_template(
'views/add-service.html',

13
app/main/views/choose_service.py
View File

@@ -1,9 +1,18 @@
from flask import render_template
from flask import (render_template, redirect, url_for)
from flask_login import login_required
from app.main.dao import services_dao
from app.main import main
@main.route("/services")
@login_required
def choose_service():
return render_template('views/choose-service.html')
services = services_dao.get_services()
# If there is only one service redirect
# to the service dashboard.
if len(services['data']) == 1:
return redirect(url_for(
'main.service_dashboard', service_id=services['data'][0]['id']))
return render_template(
'views/choose-service.html',
services=[services_dao.ServicesBrowsableItem(x) for x in services['data']])

20
app/main/views/dashboard.py
View File

@@ -1,18 +1,24 @@
from flask import render_template
from flask import (abort, render_template)
from flask_login import login_required
from app.main import main
from app.main.dao.services_dao import get_service_by_id
from client.errors import HTTPError
from ._jobs import jobs
@main.route("/services/<int:service_id>/dashboard")
@login_required
def dashboard(service_id):
def service_dashboard(service_id):
try:
service = get_service_by_id(service_id)
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
return render_template(
'views/dashboard.html',
'views/service_dashboard.html',
jobs=jobs,
free_text_messages_remaining=560,
spent_this_month='0.00',
service_id=service_id
)
service_id=service_id)

10
app/main/views/index.py
View File

@@ -10,29 +10,29 @@ def index():
@main.route("/register-from-invite")
@login_required
def registerfrominvite():
def register_from_invite():
return render_template('views/register-from-invite.html')
@main.route("/verify-mobile")
@login_required
def verifymobile():
def verify_mobile():
return render_template('views/verify-mobile.html')
@main.route("/services/<int:service_id>/send-email")
@login_required
def sendemail(service_id):
def send_email(service_id):
return render_template('views/send-email.html')
@main.route("/services/<int:service_id>/check-email")
@login_required
def checkemail(service_id):
def check_email(service_id):
return render_template('views/check-email.html')
@main.route("/services/<int:service_id>/manage-users")
@login_required
def manageusers(service_id):
def manage_users(service_id):
return render_template('views/manage-users.html', service_id=service_id)

6
app/main/views/jobs.py
View File

@@ -47,7 +47,7 @@ messages = [
@main.route("/services/<int:service_id>/jobs")
@login_required
def showjobs(service_id):
def view_jobs(service_id):
return render_template(
'views/jobs.html',
jobs=jobs,
@@ -57,7 +57,7 @@ def showjobs(service_id):
@main.route("/services/<int:service_id>/jobs/<job_id>")
@login_required
def showjob(service_id, job_id):
def view_job(service_id, job_id):
# TODO the uploaded file name could be part of job definition
# so won't need to be passed on from last view via session
@@ -86,7 +86,7 @@ def showjob(service_id, job_id):
@main.route("/services/<int:service_id>/jobs/<job_id>/notification/<string:notification_id>")
@login_required
def shownotification(service_id, job_id, notification_id):
def view_notification(service_id, job_id, notification_id):
return render_template(
'views/notification.html',
message=[

174
app/main/views/service_settings.py
View File

@@ -1,19 +1,24 @@
from flask import render_template, redirect, request, url_for, abort
from flask import (
render_template, redirect, request, url_for, abort, session)
from flask_login import login_required
from app.main import main
from app.main.dao.services_dao import (
get_service_by_id, delete_service, update_service)
from app.main.forms import ConfirmPasswordForm, ServiceNameForm
service = {
'name': 'Service name',
'live': False,
'active': True
}
from client.errors import HTTPError
@main.route("/services/<int:service_id>/service-settings")
@login_required
def service_settings(service_id):
try:
service = get_service_by_id(service_id)['data']
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
return render_template(
'views/service-settings.html',
service=service,
@@ -23,42 +28,62 @@ def service_settings(service_id):
@main.route("/services/<int:service_id>/service-settings/name", methods=['GET', 'POST'])
@login_required
def name(service_id):
def service_name_change(service_id):
try:
service = get_service_by_id(service_id)['data']
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
form = ServiceNameForm()
form.service_name.data = 'Service name'
if request.method == 'GET':
return render_template(
'views/service-settings/name.html',
service=service,
form=form,
service_id=service_id
)
elif request.method == 'POST':
return redirect(url_for('.confirm_name_change', service_id=service_id))
if form.validate_on_submit():
session['service_name_change'] = form.name.data
return redirect(url_for('.service_name_change_confirm', service_id=service_id))
return render_template(
'views/service-settings/name.html',
service=service,
form=form,
service_id=service_id)
@main.route("/services/<int:service_id>/service-settings/name/confirm", methods=['GET', 'POST'])
@login_required
def confirm_name_change(service_id):
def service_name_change_confirm(service_id):
try:
service = get_service_by_id(service_id)['data']
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
form = ConfirmPasswordForm()
if request.method == 'GET':
return render_template(
'views/service-settings/confirm.html',
heading='Change your service name',
form=form,
service_id=service_id
)
elif request.method == 'POST':
if form.validate_on_submit():
service['name'] = session['service_name_change']
update_service(service)
return redirect(url_for('.service_settings', service_id=service_id))
return render_template(
'views/service-settings/confirm.html',
heading='Change your service name',
form=form,
service_id=service_id)
@main.route("/services/<int:service_id>/service-settings/request-to-go-live", methods=['GET', 'POST'])
@login_required
def request_to_go_live(service_id):
def service_request_to_go_live(service_id):
try:
service = get_service_by_id(service_id)['data']
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
if request.method == 'GET':
return render_template(
'views/service-settings/request-to-go-live.html',
@@ -66,12 +91,22 @@ def request_to_go_live(service_id):
service_id=service_id
)
elif request.method == 'POST':
service['restricted']
update_service(service)
return redirect(url_for('.service_settings', service_id=service_id))
@main.route("/services/<int:service_id>/service-settings/status", methods=['GET', 'POST'])
@login_required
def status(service_id):
def service_status_change(service_id):
try:
service = get_service_by_id(service_id)['data']
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
if request.method == 'GET':
return render_template(
'views/service-settings/status.html',
@@ -79,30 +114,47 @@ def status(service_id):
service_id=service_id
)
elif request.method == 'POST':
return redirect(url_for('.confirm_status_change', service_id=service_id))
return redirect(url_for('.service_status_change_confirm', service_id=service_id))
@main.route("/services/<int:service_id>/service-settings/status/confirm", methods=['GET', 'POST'])
@login_required
def confirm_status_change(service_id):
def service_status_change_confirm(service_id):
try:
service = get_service_by_id(service_id)['data']
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
# TODO validate password, will leave until
# user management has been moved to the api.
form = ConfirmPasswordForm()
if request.method == 'GET':
return render_template(
'views/service-settings/confirm.html',
heading='Turn off all outgoing notifications',
destructive=True,
form=form,
service_id=service_id
)
elif request.method == 'POST':
if form.validate_on_submit():
service['active'] = True
update_service(service)
return redirect(url_for('.service_settings', service_id=service_id))
return render_template(
'views/service-settings/confirm.html',
heading='Turn off all outgoing notifications',
destructive=True,
form=form,
service_id=service_id)
@main.route("/services/<int:service_id>/service-settings/delete", methods=['GET', 'POST'])
@login_required
def delete(service_id):
def service_delete(service_id):
try:
service = get_service_by_id(service_id)['data']
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
if request.method == 'GET':
return render_template(
'views/service-settings/delete.html',
@@ -110,22 +162,36 @@ def delete(service_id):
service_id=service_id
)
elif request.method == 'POST':
return redirect(url_for('.confirm_delete', service_id=service_id))
return redirect(url_for('.service_delete_confirm', service_id=service_id))
@main.route("/services/<int:service_id>/service-settings/delete/confirm", methods=['GET', 'POST'])
@login_required
def confirm_delete(service_id):
def service_delete_confirm(service_id):
try:
service = get_service_by_id(service_id)['data']
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
# TODO validate password, will leave until
# user management has been moved to the api.
form = ConfirmPasswordForm()
if request.method == 'GET':
return render_template(
'views/service-settings/confirm.html',
heading='Delete this service from Notify',
destructive=True,
form=form,
service_id=service_id
)
elif request.method == 'POST':
return redirect(url_for('.dashboard', service_id=service_id))
if form.validate_on_submit():
try:
service = delete_service(service_id)
except HTTPError as e:
if e.status_code == 404:
abort(404)
else:
raise e
return redirect(url_for('.choose_service'))
return render_template(
'views/service-settings/confirm.html',
heading='Delete this service from Notify',
destructive=True,
form=form,
service_id=service_id)

10
app/main/views/sms.py
View File

@@ -34,14 +34,14 @@ sms_templates = [
@main.route("/services/<int:service_id>/sms/send", methods=['GET', 'POST'])
@login_required
def sendsms(service_id):
def send_sms(service_id):
form = CsvUploadForm()
if form.validate_on_submit():
try:
csv_file = form.file.data
filedata = _get_filedata(csv_file)
upload_id = s3upload(service_id, filedata)
return redirect(url_for('.checksms',
return redirect(url_for('.check_sms',
service_id=service_id,
upload_id=upload_id))
except ValueError as e:
@@ -49,7 +49,7 @@ def sendsms(service_id):
csv_file.filename)
flash(message)
flash(str(e))
return redirect(url_for('.sendsms', service_id=service_id))
return redirect(url_for('.send_sms', service_id=service_id))
return render_template('views/send-sms.html',
message_templates=sms_templates,
@@ -60,7 +60,7 @@ def sendsms(service_id):
@main.route("/services/<int:service_id>/sms/check/<upload_id>",
methods=['GET', 'POST'])
@login_required
def checksms(service_id, upload_id):
def check_sms(service_id, upload_id):
if request.method == 'GET':
contents = s3download(service_id, upload_id)
upload_result = _get_numbers(contents)
@@ -74,7 +74,7 @@ def checksms(service_id, upload_id):
)
elif request.method == 'POST':
# TODO create the job with template, file location etc.
return redirect(url_for('main.showjob',
return redirect(url_for('main.view_job',
service_id=service_id,
job_id=upload_id))

38
app/main/views/user_profile.py
View File

@@ -8,12 +8,12 @@ from app.main.forms import (
@main.route("/user-profile")
def userprofile():
def user_profile():
return render_template('views/user-profile.html')
@main.route("/user-profile/name", methods=['GET', 'POST'])
def userprofile_name():
def user_profile_name():
form = ChangeNameForm()
@@ -26,11 +26,11 @@ def userprofile_name():
form_field=form.new_name
)
elif request.method == 'POST':
return redirect(url_for('.userprofile'))
return redirect(url_for('.user_profile'))
@main.route("/user-profile/email", methods=['GET', 'POST'])
def userprofile_email():
def user_profile_email():
form = ChangeEmailForm()
@@ -43,11 +43,11 @@ def userprofile_email():
form_field=form.email_address
)
elif request.method == 'POST':
return redirect(url_for('.userprofile_email_authenticate'))
return redirect(url_for('.user_profile_email_authenticate'))
@main.route("/user-profile/email/authenticate", methods=['GET', 'POST'])
def userprofile_email_authenticate():
def user_profile_email_authenticate():
form = ConfirmPasswordForm()
@@ -56,14 +56,14 @@ def userprofile_email_authenticate():
'views/user-profile/authenticate.html',
thing='email address',
form=form,
back_link=url_for('.userprofile_email')
back_link=url_for('.user_profile_email')
)
elif request.method == 'POST':
return redirect(url_for('.userprofile_email_confirm'))
return redirect(url_for('.user_profile_email_confirm'))
@main.route("/user-profile/email/confirm", methods=['GET', 'POST'])
def userprofile_email_confirm():
def user_profile_email_confirm():
form = ConfirmEmailForm()
@@ -74,11 +74,11 @@ def userprofile_email_confirm():
thing='email address'
)
elif request.method == 'POST':
return redirect(url_for('.userprofile'))
return redirect(url_for('.user_profile'))
@main.route("/user-profile/mobile-number", methods=['GET', 'POST'])
def userprofile_mobile_number():
def user_profile_mobile_number():
form = ChangeMobileNumberForm()
@@ -91,11 +91,11 @@ def userprofile_mobile_number():
form_field=form.mobile_number
)
elif request.method == 'POST':
return redirect(url_for('.userprofile_mobile_number_authenticate'))
return redirect(url_for('.user_profile_mobile_number_authenticate'))
@main.route("/user-profile/mobile-number/authenticate", methods=['GET', 'POST'])
def userprofile_mobile_number_authenticate():
def user_profile_mobile_number_authenticate():
form = ConfirmPasswordForm()
@@ -104,14 +104,14 @@ def userprofile_mobile_number_authenticate():
'views/user-profile/authenticate.html',
thing='mobile number',
form=form,
back_link=url_for('.userprofile_mobile_number_confirm')
back_link=url_for('.user_profile_mobile_number_confirm')
)
elif request.method == 'POST':
return redirect(url_for('.userprofile_mobile_number_confirm'))
return redirect(url_for('.user_profile_mobile_number_confirm'))
@main.route("/user-profile/mobile-number/confirm", methods=['GET', 'POST'])
def userprofile_mobile_number_confirm():
def user_profile_mobile_number_confirm():
form = ConfirmMobileNumberForm()
@@ -122,11 +122,11 @@ def userprofile_mobile_number_confirm():
thing='mobile number'
)
elif request.method == 'POST':
return redirect(url_for('.userprofile'))
return redirect(url_for('.user_profile'))
@main.route("/user-profile/password", methods=['GET', 'POST'])
def userprofile_password():
def user_profile_password():
form = ChangePasswordForm()
@@ -136,4 +136,4 @@ def userprofile_password():
form=form
)
elif request.method == 'POST':
return redirect(url_for('.userprofile'))
return redirect(url_for('.user_profile'))

66
app/models.py
View File

@@ -1,3 +1,4 @@
import datetime
from app import db
from flask import current_app
@@ -33,12 +34,21 @@ class User(db.Model):
email_address = db.Column(db.String(255), nullable=False, index=True, unique=True)
password = db.Column(db.String, index=False, unique=False, nullable=False)
mobile_number = db.Column(db.String, index=False, unique=False, nullable=False)
created_at = db.Column(db.DateTime, index=False, unique=False, nullable=False)
updated_at = db.Column(db.DateTime, index=False, unique=False, nullable=True)
created_at = db.Column(db.DateTime,
index=False,
unique=False,
nullable=False,
default=datetime.datetime.now)
updated_at = db.Column(db.DateTime,
index=False,
unique=False,
nullable=True,
onupdate=datetime.datetime.now)
password_changed_at = db.Column(db.DateTime, index=False, unique=False, nullable=True)
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'), index=True, unique=False, nullable=False)
logged_in_at = db.Column(db.DateTime, nullable=True)
failed_login_count = db.Column(db.Integer, nullable=False, default=0)
# TODO should this be an enum?
state = db.Column(db.String, nullable=False, default='pending')
def serialize(self):
@@ -78,37 +88,37 @@ class User(db.Model):
return True
user_to_service = db.Table(
'user_to_service',
db.Model.metadata,
db.Column('user_id', db.Integer, db.ForeignKey('users.id')),
db.Column('service_id', db.Integer, db.ForeignKey('services.id'))
)
# user_to_service = db.Table(
# 'user_to_service',
# db.Model.metadata,
# db.Column('user_id', db.Integer, db.ForeignKey('users.id')),
# db.Column('service_id', db.Integer, db.ForeignKey('services.id'))
# )
class Service(db.Model):
__tablename__ = 'services'
# class Service(db.Model):
# __tablename__ = 'services'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(255), nullable=False, unique=True)
created_at = db.Column(db.DateTime, index=False, unique=False, nullable=False)
active = db.Column(db.Boolean, index=False, unique=False, nullable=False)
limit = db.Column(db.BigInteger, index=False, unique=False, nullable=False)
users = db.relationship('User', secondary=user_to_service, backref=db.backref('user_to_service', lazy='dynamic'))
restricted = db.Column(db.Boolean, index=False, unique=False, nullable=False)
# id = db.Column(db.Integer, primary_key=True)
# name = db.Column(db.String(255), nullable=False, unique=True)
# created_at = db.Column(db.DateTime, index=False, unique=False, nullable=False)
# active = db.Column(db.Boolean, index=False, unique=False, nullable=False)
# limit = db.Column(db.BigInteger, index=False, unique=False, nullable=False)
# users = db.relationship('User', secondary=user_to_service, backref=db.backref('user_to_service', lazy='dynamic'))
# restricted = db.Column(db.Boolean, index=False, unique=False, nullable=False)
def serialize(self):
serialized = {
'id': self.id,
'name': self.name,
'createdAt': self.created_at.strftime(DATETIME_FORMAT),
'active': self.active,
'restricted': self.restricted,
'limit': self.limit,
'user': self.users.serialize()
}
# def serialize(self):
# serialized = {
# 'id': self.id,
# 'name': self.name,
# 'createdAt': self.created_at.strftime(DATETIME_FORMAT),
# 'active': self.active,
# 'restricted': self.restricted,
# 'limit': self.limit,
# 'user': self.users.serialize()
# }
return filter_null_value_fields(serialized)
# return filter_null_value_fields(serialized)
def filter_null_value_fields(obj):

128
app/notify_client/api_client.py
View File

@@ -1,8 +1,126 @@
from __future__ import unicode_literals
from notify_client import NotifyAPIClient
from client.notifications import NotificationsAPIClient
class AdminAPIClient(NotifyAPIClient):
def init_app(self, app):
self.base_url = app.config['NOTIFY_DATA_API_URL']
self.auth_token = app.config['NOTIFY_DATA_API_AUTH_TOKEN']
class NotificationsAdminAPIClient(NotificationsAPIClient):
# Fudge assert in the super __init__ so
# we can set those variables later.
def __init__(self):
super(NotificationsAdminAPIClient, self).__init__("api_url",
"client",
"secret")
def init_app(self, application):
self.base_url = application.config['NOTIFY_API_URL']
self.client_id = application.config['NOTIFY_API_CLIENT']
self.secret = application.config['NOTIFY_API_SECRET']
def create_service(self, service_name, active, limit, restricted, user_id):
"""
Create a service and return the json.
"""
data = {
"name": service_name,
"active": active,
"limit": limit,
"users": [user_id],
"restricted": restricted
}
return self.post("/service", data)
def delete_service(self, service_id):
"""
Delete a service.
"""
endpoint = "/service/{0}".format(service_id)
return self.delete(endpoint)
def get_service(self, service_id, *params):
"""
Retrieve a service.
"""
return self.get(
'/service/{0}'.format(service_id))
def get_services(self, *params):
"""
Retrieve a list of services.
"""
return self.get('/service', *params)
def update_service(self,
service_id,
service_name,
active,
limit,
restricted,
users):
"""
Update a service.
"""
data = {
"id": service_id,
"name": service_name,
"active": active,
"limit": limit,
"restricted": restricted,
"users": users
}
endpoint = "/service/{0}".format(service_id)
return self.put(endpoint, update_dict)
def create_service_template(self, name, type_, content, service_id):
"""
Create a service template.
"""
data = {
"name": name,
"template_type": type_,
"content": content,
"service": service_id
}
endpoint = "/service/{0}/template".format(service_id)
return self.post(endpoint, data)
def get_service_template(self, service_id, template_id, *params):
"""
Retrieve a service template.
"""
endpoint = '/service/{service_id}/template/{template_id}'.format(
service_id=service_id,
template_id=template_id)
return self.get(endpoint, *params)
def get_service_templates(self, service_id, *params):
"""
Retrieve all templates for service.
"""
endpoint = '/service/{service_id}/template'.format(
service_id=service_id)
return self.get(endpoint, *params)
def delete_service_template(self, service_id, template_id):
"""
Delete a service template.
"""
endpoint = "/service/{0}/template/{1}".format(service_id, template_id)
return self.delete(endpoint)
# The implementation of these will change after the notifications-api
# functionality updates to include the ability to send notifications.
def send_sms(self,
mobile_number,
message,
job_id=None,
description=None):
pass
def send_email(self,
email_address,
message,
from_address,
subject,
job_id=None,
description=None):
pass

27
app/notify_client/sender.py
View File

@@ -1,8 +1,8 @@
from random import randint
from flask import url_for, current_app
from itsdangerous import URLSafeTimedSerializer, SignatureExpired
from app import admin_api_client
from app.main.dao import verify_codes_dao
from app import notifications_api_client
def create_verify_code():
@@ -12,32 +12,27 @@ def create_verify_code():
def send_sms_code(user_id, mobile_number):
sms_code = create_verify_code()
verify_codes_dao.add_code(user_id=user_id, code=sms_code, code_type='sms')
admin_api_client.send_sms(
mobile_number=mobile_number.replace(" ", ""),
message=sms_code,
token=admin_api_client.auth_token)
notifications_api_client.send_sms(mobile_number=mobile_number,
message=sms_code)
return sms_code
def send_email_code(user_id, email):
email_code = create_verify_code()
verify_codes_dao.add_code(user_id=user_id, code=email_code, code_type='email')
admin_api_client.send_email(email_address=email,
from_str='notify@digital.cabinet-office.gov.uk',
message=email_code,
subject='Verification code',
token=admin_api_client.auth_token)
notifications_api_client.send_email(email_address=email,
from_address='notify@digital.cabinet-office.gov.uk',
message=email_code,
subject='Verification code')
return email_code
def send_change_password_email(email):
link_to_change_password = url_for('.new_password', token=generate_token(email), _external=True)
admin_api_client.send_email(email_address=email,
from_str='notify@digital.cabinet-office.gov.uk',
message=link_to_change_password,
subject='Reset password for GOV.UK Notify',
token=admin_api_client.auth_token)
notifications_api_client.send_email(email_address=email,
from_address='notify@digital.cabinet-office.gov.uk',
message=link_to_change_password,
subject='Reset password for GOV.UK Notify')
def generate_token(email):

4
app/templates/admin_template.html
View File

@@ -42,7 +42,7 @@
{% if not current_user.is_authenticated() %}
{% set homepage_url = url_for('main.index') %}
{% else %}
{% set homepage_url = url_for('main.dashboard', service_id=123) %}
{% set homepage_url = url_for('main.choose_service') %}
{% endif %}
{% block content %}
@@ -60,7 +60,7 @@
</div>
</div>
<div class="column-half management-navigation-account">
<a href="{{ url_for('main.userprofile') }}">{{ current_user.name }}</a>
<a href="{{ url_for('main.user_profile') }}">{{ current_user.name }}</a>
<a href="{{ url_for('main.sign_out')}}">Sign out</a>
</div>
</div>

16
app/templates/main_nav.html
View File

@@ -1,18 +1,18 @@
<nav class="navigation">
<ul>
<li><a href="{{ url_for('.dashboard', service_id=123) }}">Dashboard</a></li>
<li><a href="{{ url_for('.service_dashboard', service_id=service_id) }}">Dashboard</a></li>
</ul>
<ul>
<li><a href="{{ url_for('.sendsms', service_id=123) }}">Send text messages</a></li>
<li><a href="{{ url_for('.sendemail', service_id=123) }}">Send emails</a></li>
<li><a href="{{ url_for('.showjobs', service_id=123) }}">Activity</a></li>
<li><a href="{{ url_for('.manage_templates', service_id=123) }}">Templates</a></li>
<li><a href="{{ url_for('.send_sms', service_id=service_id) }}">Send text messages</a></li>
<li><a href="{{ url_for('.send_email', service_id=service_id) }}">Send emails</a></li>
<li><a href="{{ url_for('.view_jobs', service_id=service_id) }}">Activity</a></li>
<li><a href="{{ url_for('.manage_templates', service_id=service_id) }}">Templates</a></li>
</ul>
<ul>
<li><a href="{{ url_for('.api_keys', service_id=123) }}">API keys and documentation</a></li>
<li><a href="{{ url_for('.api_keys', service_id=service_id) }}">API keys and documentation</a></li>
</ul>
<ul>
<li><a href="{{ url_for('.manageusers', service_id=123) }}">Manage users</a></li>
<li><a href="{{ url_for('.service_settings', service_id=123) }}">Service settings</a></li>
<li><a href="{{ url_for('.manage_users', service_id=service_id) }}">Manage users</a></li>
<li><a href="{{ url_for('.service_settings', service_id=service_id) }}">Service settings</a></li>
</ul>
</nav>

2
app/templates/views/add-service.html
View File

@@ -30,7 +30,7 @@ GOV.UK Notify | Set up service
</ul>
<form autocomplete="off" method="post">
{{ textbox(form.service_name) }}
{{ textbox(form.name) }}
{{ page_footer(
'Continue'
) }}

2
app/templates/views/api-keys.html
View File

@@ -49,7 +49,7 @@
</div>
{{ page_footer(
back_link=url_for('.dashboard', service_id=service_id),
back_link=url_for('.service_dashboard', service_id=service_id),
back_link_text='Back to dashboard'
) }}

6
app/templates/views/check-sms.html
View File

@@ -17,7 +17,7 @@
{% for rejected in upload_result.rejects %}
<p>Line {{rejected.line_number}}: {{rejected.phone }}</a>
{% endfor %}
<p><a href="{{url_for('.sendsms', service_id=service_id)}}" class="button">Go back and resolve errors</a></p>
<p><a href="{{url_for('.send_sms', service_id=service_id)}}" class="button">Go back and resolve errors</a></p>
{% else %}
@@ -26,7 +26,7 @@
{{ page_footer(
button_text = "Send {} text messages".format(upload_result.valid|count),
back_link = url_for(".sendsms", service_id=service_id)
back_link = url_for(".send_sms", service_id=service_id)
)}}
{% if upload_result.valid | count > 6 %}
@@ -56,7 +56,7 @@
{{ page_footer(
button_text = "Send {} text messages".format(upload_result.valid|count),
back_link = url_for(".sendsms", service_id=service_id)
back_link = url_for(".send_sms", service_id=service_id)
)}}
</form>

13
app/templates/views/choose-service.html
View File

@@ -1,4 +1,4 @@
{% extends "withnav_template.html" %}
{% extends "admin_template.html" %}
{% from "components/browse-list.html" import browse_list %}
{% block page_title %}
@@ -11,16 +11,7 @@
Choose service
</h1>
{{ browse_list([
{
'title': 'MOT Reminders',
'link': url_for('.dashboard', service_id=123)
},
{
'title': 'Vehicle Tax',
'link': url_for('.dashboard', service_id=123)
},
]) }}
{{ browse_list(services) }}
{{ browse_list([
{
'title': 'Add a new service…',

4
app/templates/views/job.html
View File

@@ -52,10 +52,10 @@ GOV.UK Notify | Notifications activity
]
) %}
{% call field() %}
<a href="{{ url_for('.shownotification', service_id=service_id, job_id=456, notification_id=item.id) }}">{{item.phone}}</a>
<a href="{{ url_for('.view_notification', service_id=service_id, job_id=456, notification_id=item.id) }}">{{item.phone}}</a>
{% endcall %}
{% call field() %}
<a href="{{ url_for('.shownotification', service_id=service_id, job_id=456, notification_id=item.id) }}">{{item.message[:50]}}…</a>
<a href="{{ url_for('.view_notification', service_id=service_id, job_id=456, notification_id=item.id) }}">{{item.message[:50]}}…</a>
{% endcall %}
{% call field(
align='right',

4
app/templates/views/jobs.html
View File

@@ -16,10 +16,10 @@ GOV.UK Notify | Notifications activity
field_headings=['Job', 'File', 'Time', 'Status']
) %}
{% call field() %}
<a href="{{ url_for('.showjob', service_id=service_id, job_id=456) }}">{{ item.file }}</a>
<a href="{{ url_for('.view_job', service_id=service_id, job_id=456) }}">{{ item.file }}</a>
{% endcall %}
{% call field() %}
<a href="{{ url_for('.showjob', service_id=service_id, job_id=456) }}">{{ item.job }}</a>
<a href="{{ url_for('.view_job', service_id=service_id, job_id=456) }}">{{ item.job }}</a>
{% endcall %}
{% call field() %}
{{ item.time }}

2
app/templates/views/manage-users.html
View File

@@ -12,7 +12,7 @@ GOV.UK Notify | Manage users
<p>Here's where you can add or remove users of a service.</p>
{{ page_footer(
back_link = url_for('.dashboard', service_id=service_id),
back_link = url_for('.service_dashboard', service_id=service_id),
back_link_text = 'Back to dashboard'
) }}

2
app/templates/views/notification.html
View File

@@ -26,7 +26,7 @@ GOV.UK Notify | Notifications activity
</div>
{{ page_footer(
back_link = url_for('.showjob', service_id=service_id, job_id=job_id),
back_link = url_for('.view_job', service_id=service_id, job_id=job_id),
back_link_text = 'View other messages in this job'
) }}

10
app/templates/views/service-settings.html
View File

@@ -12,26 +12,26 @@
{{ browse_list([
{
'title': 'Change your service name',
'link': url_for('.name', service_id=service_id),
'link': url_for('.service_name_change', service_id=service_id),
'hint': 'Your service name ({}) is included in every sent notification'.format(service.name)
},
{
'title': 'Request to go live and turn off sending restrictions',
'link': url_for('.request_to_go_live', service_id=service_id),
'link': url_for('.service_request_to_go_live', service_id=service_id),
'hint': 'A live service can send notifications to any phone number or email address',
} if not service.live else {
},
{
'title': 'Turn off all outgoing notifications',
'link': url_for('.status', service_id=service_id),
'link': url_for('.service_status_change', service_id=service_id),
'destructive': True
} if service.active else {
'title': 'Restart sending notifications',
'link': url_for('.status', service_id=service_id)
'link': url_for('.service_status_change', service_id=service_id)
},
{
'title': 'Delete this service from Notify',
'link': url_for('.delete', service_id=service_id),
'link': url_for('.service_delete', service_id=service_id),
'destructive': True
},
]) }}

2
app/templates/views/service-settings/name.html
View File

@@ -21,7 +21,7 @@ GOV.UK Notify | Service settings
</ul>
<form method="post">
{{ textbox(form.service_name) }}
{{ textbox(form.name) }}
{{ page_footer(
'Save',
back_link=url_for('.service_settings', service_id=service_id)

6
app/templates/views/dashboard.html → app/templates/views/service_dashboard.html
View File

@@ -30,10 +30,10 @@
field_headings=['Job', 'File', 'Time', 'Status']
) %}
{% call field() %}
<a href="{{ url_for('.showjob', service_id=service_id, job_id=456) }}">{{ item.file }}</a>
<a href="{{ url_for('.view_job', service_id=service_id, job_id=456) }}">{{ item.file }}</a>
{% endcall %}
{% call field() %}
<a href="{{ url_for('.showjob', service_id=service_id, job_id=456) }}">{{ item.job }}</a>
<a href="{{ url_for('.view_job', service_id=service_id, job_id=456) }}">{{ item.job }}</a>
{% endcall %}
{% call field() %}
{{ item.time }}
@@ -43,7 +43,7 @@
{% endcall %}
{% endcall %}
<p>
<a href={{ url_for('.showjobs', service_id=service_id) }}>See all notifications activity</a>
<a href={{ url_for('.view_jobs', service_id=service_id) }}>See all notifications activity</a>
</p>

8
app/templates/views/user-profile.html
View File

@@ -11,10 +11,10 @@
{% call(item) list_table(
[
{'label': 'Name', 'value': current_user.name, 'url': url_for('.userprofile_name')},
{'label': 'Email address', 'value': current_user.email_address, 'url': url_for('.userprofile_email')},
{'label': 'Mobile number', 'value': current_user.mobile_number, 'url': url_for('.userprofile_mobile_number')},
{'label': 'Password', 'value': 'Last changed 1 January 2016, 10:00AM', 'url': url_for('.userprofile_password')},
{'label': 'Name', 'value': current_user.name, 'url': url_for('.user_profile_name')},
{'label': 'Email address', 'value': current_user.email_address, 'url': url_for('.user_profile_email')},
{'label': 'Mobile number', 'value': current_user.mobile_number, 'url': url_for('.user_profile_mobile_number')},
{'label': 'Password', 'value': 'Last changed 1 January 2016, 10:00AM', 'url': url_for('.user_profile_password')},
],
caption='Account settings',
field_headings=['Setting', 'Value', 'Link to change'],

2
app/templates/views/user-profile/change-password.html
View File

@@ -17,7 +17,7 @@ GOV.UK Notify | Service settings
{{ textbox(form.new_password) }}
{{ page_footer(
'Save',
back_link=url_for('.userprofile'),
back_link=url_for('.user_profile'),
back_link_text="Back to your profile"
) }}
</form>

2
app/templates/views/user-profile/change.html
View File

@@ -21,7 +21,7 @@ GOV.UK Notify | Service settings
{{ textbox(form_field) }}
{{ page_footer(
'Save',
back_link=url_for('.userprofile'),
back_link=url_for('.user_profile'),
back_link_text="Back to your profile"
) }}
</form>

2
app/templates/views/user-profile/confirm.html
View File

@@ -20,7 +20,7 @@ GOV.UK Notify | Service settings
{{ page_footer(
'Confirm',
destructive=destructive,
back_link=url_for('.userprofile')
back_link=url_for('.user_profile')
) }}
</form>
</div>

5
config.py
View File

@@ -22,8 +22,9 @@ class Config(object):
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SECURE = True
NOTIFY_DATA_API_URL = os.getenv('NOTIFY_API_URL', "http://localhost:6001")
NOTIFY_DATA_API_AUTH_TOKEN = os.getenv('NOTIFY_API_TOKEN', "dev-token")
NOTIFY_API_URL = os.getenv('NOTIFY_API_URL', "http://localhost:6001")
NOTIFY_API_SECRET = os.getenv('NOTIFY_API_SECRET', "dev-secret")
NOTIFY_API_CLIENT = os.getenv('NOTIFY_API_CLIENT', "admin")
WTF_CSRF_ENABLED = True
SECRET_KEY = 'secret-key'

43
migrations/versions/80_remove_services.py Normal file
View File

@@ -0,0 +1,43 @@
"""empty message
Revision ID: 80_remove_services
Revises: 70_unique_email
Create Date: 2016-01-14 17:48:42.426979
"""
# revision identifiers, used by Alembic.
revision = '80_remove_services'
down_revision = '70_unique_email'
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.drop_table('user_to_service')
op.drop_table('services')
### end Alembic commands ###
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_table('services',
sa.Column('id', sa.INTEGER(), server_default=sa.text("nextval('services_id_seq'::regclass)"), nullable=False),
sa.Column('name', sa.VARCHAR(length=255), autoincrement=False, nullable=False),
sa.Column('created_at', postgresql.TIMESTAMP(), autoincrement=False, nullable=False),
sa.Column('active', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.Column('limit', sa.BIGINT(), autoincrement=False, nullable=False),
sa.Column('restricted', sa.BOOLEAN(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name='services_pkey'),
sa.UniqueConstraint('name', name='services_name_key'),
postgresql_ignore_search_path=False
)
op.create_table('user_to_service',
sa.Column('user_id', sa.INTEGER(), autoincrement=False, nullable=True),
sa.Column('service_id', sa.INTEGER(), autoincrement=False, nullable=True),
sa.ForeignKeyConstraint(['service_id'], ['services.id'], name='user_to_service_service_id_fkey'),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], name='user_to_service_user_id_fkey')
)
### end Alembic commands ###

2
requirements.txt
View File

@@ -11,6 +11,6 @@ Flask-Bcrypt==0.6.2
credstash==1.8.0
boto3==1.2.3
git+https://github.com/alphagov/notify-api-client.git@0.1.4#egg=notify-api-client==0.1.4
git+https://github.com/alphagov/notifications-python-client.git@0.1.6#egg=notifications-python-client==0.1.6
git+https://github.com/alphagov/notifications-utils.git@0.0.3#egg=notifications-utils==0.0.3

58
tests/__init__.py
View File

@@ -0,0 +1,58 @@
from flask.testing import FlaskClient
from flask import url_for
from app.models import User
from app.main.dao import (users_dao, verify_codes_dao)
class TestClient(FlaskClient):
def login(self, user):
# Skipping authentication here and just log them in
with self.session_transaction() as session:
session['user_email'] = user.email_address
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='sms')
response = self.post(
url_for('main.two_factor'), data={'sms_code': '12345'})
assert response.status_code == 302
def logout(self, user):
self.get(url_for("main.logout"))
def service_json(id_, name, users, limit=1000, active=False, restricted=True):
return {
'id': id_,
'name': name,
'users': users,
'limit': limit,
'active': active,
'restricted': restricted
}
TEST_USER_EMAIL = 'test@user.gov.uk'
def create_test_user(state):
user = User(name='Test User',
password='somepassword',
email_address=TEST_USER_EMAIL,
mobile_number='+441234123412',
role_id=1,
state=state)
users_dao.insert_user(user)
return user
def create_another_test_user(state):
user = User(name='Another Test User',
password='someOtherpassword',
email_address='another_test@user.gov.uk',
mobile_number='+442233123412',
role_id=1,
state=state)
users_dao.insert_user(user)
return user
def get_test_user():
return users_dao.get_user_by_email(TEST_USER_EMAIL)

28
tests/app/main/__init__.py
View File

@@ -1,28 +0,0 @@
from datetime import datetime
from app.main.dao import users_dao
from app.models import User
def create_test_user(state):
user = User(name='Test User',
password='somepassword',
email_address='test@user.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1,
state=state)
users_dao.insert_user(user)
return user
def create_another_test_user(state):
user = User(name='Another Test User',
password='someOtherpassword',
email_address='another_test@user.gov.uk',
mobile_number='+442233123412',
created_at=datetime.now(),
role_id=1,
state=state)
users_dao.insert_user(user)
return user

8
tests/app/main/dao/test_roles_dao.py
View File

@@ -5,7 +5,7 @@ from app.models import Roles
from app.main.dao import roles_dao
def test_insert_role_should_be_able_to_get_role(notifications_admin, notifications_admin_db, notify_db_session):
def test_insert_role_should_be_able_to_get_role(app_, db_, db_session):
role = Roles(id=1000, role='some role for test')
roles_dao.insert_role(role)
@@ -13,9 +13,9 @@ def test_insert_role_should_be_able_to_get_role(notifications_admin, notificatio
assert saved_role == role
def test_insert_role_will_throw_error_if_role_already_exists(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_insert_role_will_throw_error_if_role_already_exists(app_,
db_,
db_session):
role1 = roles_dao.get_role_by_id(1)
assert role1.id == 1

97
tests/app/main/dao/test_service_dao.py
View File

@@ -1,73 +1,62 @@
import pytest
import sqlalchemy
from app.main.dao import services_dao
from tests.app.main import create_test_user
def test_can_insert_and_retrieve_new_service(notifications_admin, notifications_admin_db, notify_db_session):
user = create_test_user('active')
id = services_dao.insert_new_service('testing service', user)
saved_service = services_dao.get_service_by_id(id)
assert id == saved_service.id
assert saved_service.users == [user]
assert saved_service.name == 'testing service'
def test_can_insert_new_service(db_,
db_session,
active_user,
mock_create_service):
service_name = 'testing service'
id_ = services_dao.insert_new_service(service_name, active_user.id)
mock_create_service.assert_called_once_with(
service_name, False, 1000, True, active_user.id)
def test_unrestrict_service_updates_the_service(notifications_admin, notifications_admin_db, notify_db_session):
user = create_test_user('active')
id = services_dao.insert_new_service('unrestricted service', user)
saved_service = services_dao.get_service_by_id(id)
assert saved_service.restricted is True
services_dao.unrestrict_service(id)
unrestricted_service = services_dao.get_service_by_id(id)
assert unrestricted_service.restricted is False
def test_unrestrict_service_updates_the_service(db_,
db_session,
mock_get_service,
mock_update_service):
service_one = mock_get_service.side_effect(123)['data']
services_dao.unrestrict_service(service_one['id'])
mock_update_service.assert_called_once_with(service_one['id'],
service_one['name'],
service_one['active'],
service_one['limit'],
False,
service_one['users'])
def test_activate_service_update_service(notifications_admin, notifications_admin_db, notify_db_session):
user = create_test_user('active')
id = services_dao.insert_new_service('activated service', user)
service = services_dao.get_service_by_id(id)
assert service.active is False
services_dao.activate_service(id)
activated_service = services_dao.get_service_by_id(id)
assert activated_service.active is True
def test_activate_service_update_service(db_,
db_session,
active_user,
mock_get_service,
mock_update_service):
service_one = mock_get_service.side_effect(123)['data']
services_dao.activate_service(service_one['id'])
mock_update_service.assert_called_once_with(service_one['id'],
service_one['name'],
True,
service_one['limit'],
service_one['restricted'],
service_one['users'])
def test_get_service_returns_none_if_service_does_not_exist(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_get_service_returns_none_if_service_does_not_exist(db_, db_session, mock_get_service):
mock_get_service.side_effect = lambda x: None
service = services_dao.get_service_by_id(1)
assert service is None
def test_find_by_service_name_returns_right_service(notifications_admin,
notifications_admin_db,
notify_db_session):
user = create_test_user('active')
id = services_dao.insert_new_service('testing service', user)
another = services_dao.insert_new_service('Testing the Service', user)
found = services_dao.find_service_by_service_name('testing service')
assert found.id == id
assert found.name == 'testing service'
found_another = services_dao.find_service_by_service_name('Testing the Service')
assert found_another == services_dao.get_service_by_id(another)
def test_find_by_service_name_returns_right_service(db_, db_session, mock_get_services):
service_name = "service_one"
service = services_dao.find_service_by_service_name(service_name)
assert mock_get_services.called
assert service['name'] == service_name
def test_should_not_allow_two_services_of_the_same_name(notifications_admin, notifications_admin_db, notify_db_session):
user = create_test_user('active')
services_dao.insert_new_service('duplicate service', user)
with pytest.raises(sqlalchemy.exc.IntegrityError) as error:
services_dao.insert_new_service('duplicate service', user)
assert 'duplicate key value violates unique constraint "services_name_key' in error.value
def test_should_return_list_of_service_names(notifications_admin, notifications_admin_db, notify_db_session):
user = create_test_user('active')
services_dao.insert_new_service('first service', user)
services_dao.insert_new_service('second service', user)
services_dao.insert_new_service('third service', user)
expected = ['first service', 'second service', 'third service']
def test_should_return_list_of_service_names(db_, db_session, mock_get_services):
expected = ['service_one', 'service_two']
actual = services_dao.find_all_service_names()
assert mock_get_services.called
assert actual == expected

46
tests/app/main/dao/test_users_dao.py
View File

@@ -6,12 +6,11 @@ from app.models import User
from app.main.dao import users_dao
def test_insert_user_should_add_user(notifications_admin, notifications_admin_db, notify_db_session):
def test_insert_user_should_add_user(db_, db_session):
user = User(name='test insert',
password='somepassword',
email_address='test@insert.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1)
users_dao.insert_user(user)
@@ -19,26 +18,22 @@ def test_insert_user_should_add_user(notifications_admin, notifications_admin_db
assert saved_user == user
def test_insert_user_with_role_that_does_not_exist_fails(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_insert_user_with_role_that_does_not_exist_fails(db_, db_session):
user = User(name='role does not exist',
password='somepassword',
email_address='test@insert.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=100)
with pytest.raises(sqlalchemy.exc.IntegrityError) as error:
users_dao.insert_user(user)
assert 'insert or update on table "users" violates foreign key constraint "users_role_id_fkey"' in str(error.value)
def test_get_user_by_email(notifications_admin, notifications_admin_db, notify_db_session):
def test_get_user_by_email(db_, db_session):
user = User(name='test_get_by_email',
password='somepassword',
email_address='email@example.gov.uk',
mobile_number='+441234153412',
created_at=datetime.now(),
role_id=1)
users_dao.insert_user(user)
@@ -46,24 +41,21 @@ def test_get_user_by_email(notifications_admin, notifications_admin_db, notify_d
assert retrieved == user
def test_get_all_users_returns_all_users(notifications_admin, notifications_admin_db, notify_db_session):
def test_get_all_users_returns_all_users(db_, db_session):
user1 = User(name='test one',
password='somepassword',
email_address='test1@get_all.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1)
user2 = User(name='test two',
password='some2ndpassword',
email_address='test2@get_all.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1)
user3 = User(name='test three',
password='some2ndpassword',
email_address='test3@get_all.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1)
users_dao.insert_user(user1)
@@ -74,14 +66,11 @@ def test_get_all_users_returns_all_users(notifications_admin, notifications_admi
assert users == [user1, user2, user3]
def test_increment_failed_lockout_count_should_increade_count_by_1(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_increment_failed_lockout_count_should_increade_count_by_1(db_, db_session):
user = User(name='cannot remember password',
password='somepassword',
email_address='test1@get_all.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1)
users_dao.insert_user(user)
@@ -91,13 +80,11 @@ def test_increment_failed_lockout_count_should_increade_count_by_1(notifications
assert users_dao.get_user_by_id(user.id).failed_login_count == 1
def test_user_is_locked_if_failed_login_count_is_10_or_greater(notifications_admin,
notifications_admin_db, notify_db_session):
def test_user_is_locked_if_failed_login_count_is_10_or_greater(db_, db_session):
user = User(name='cannot remember password',
password='somepassword',
email_address='test1@get_all.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1)
users_dao.insert_user(user)
saved_user = users_dao.get_user_by_id(user.id)
@@ -111,12 +98,11 @@ def test_user_is_locked_if_failed_login_count_is_10_or_greater(notifications_adm
assert saved_user.is_locked() is True
def test_user_is_active_is_false_if_state_is_inactive(notifications_admin, notifications_admin_db, notify_db_session):
def test_user_is_active_is_false_if_state_is_inactive(db_, db_session):
user = User(name='inactive user',
password='somepassword',
email_address='test1@get_all.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1,
state='inactive')
users_dao.insert_user(user)
@@ -125,12 +111,11 @@ def test_user_is_active_is_false_if_state_is_inactive(notifications_admin, notif
assert saved_user.is_active() is False
def test_should_update_user_to_active(notifications_admin, notifications_admin_db, notify_db_session):
def test_should_update_user_to_active(db_, db_session):
user = User(name='Make user active',
password='somepassword',
email_address='activate@user.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1,
state='pending')
users_dao.insert_user(user)
@@ -139,18 +124,17 @@ def test_should_update_user_to_active(notifications_admin, notifications_admin_d
assert updated_user.state == 'active'
def test_should_throws_error_when_id_does_not_exist(notifications_admin, notifications_admin_db, notify_db_session):
def test_should_throws_error_when_id_does_not_exist(db_, db_session):
with pytest.raises(AttributeError) as error:
users_dao.activate_user(123)
assert '''object has no attribute 'state''''' in str(error.value)
def test_should_update_email_address(notifications_admin, notifications_admin_db, notify_db_session):
def test_should_update_email_address(db_, db_session):
user = User(name='Update Email',
password='somepassword',
email_address='test@it.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1,
state='inactive')
users_dao.insert_user(user)
@@ -162,12 +146,11 @@ def test_should_update_email_address(notifications_admin, notifications_admin_db
assert updated.email_address == 'new_email@testit.gov.uk'
def test_should_update_password(notifications_admin, notifications_admin_db, notify_db_session):
def test_should_update_password(db_, db_session):
user = User(name='Update Email',
password='somepassword',
email_address='test@it.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1,
state='active')
start = datetime.now()
@@ -183,19 +166,17 @@ def test_should_update_password(notifications_admin, notifications_admin_db, not
assert updated.password_changed_at > start
def test_should_return_list_of_all_email_addresses(notifications_admin, notifications_admin_db, notify_db_session):
def test_should_return_list_of_all_email_addresses(db_, db_session):
first = User(name='First Person',
password='somepassword',
email_address='first@it.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1,
state='active')
second = User(name='Second Person',
password='somepassword',
email_address='second@it.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1,
state='active')
users_dao.insert_user(first)
@@ -206,12 +187,11 @@ def test_should_return_list_of_all_email_addresses(notifications_admin, notifica
assert expected == [x.email_address for x in email_addresses]
def test_should_update_state_to_request_password_reset(notifications_admin, notifications_admin_db, notify_db_session):
def test_should_update_state_to_request_password_reset(db_, db_session):
user = User(name='Requesting Password Resest',
password='somepassword',
email_address='request@new_password.gov.uk',
mobile_number='+441234123412',
created_at=datetime.now(),
role_id=1,
state='active')
users_dao.insert_user(user)

34
tests/app/main/dao/test_verify_codes_dao.py
View File

@@ -3,10 +3,10 @@ from pytest import fail
from app.main.dao import verify_codes_dao
from app.main.encryption import check_hash
from tests.app.main import create_test_user, create_another_test_user
from tests import create_test_user, create_another_test_user
def test_insert_new_code_and_get_it_back(notifications_admin, notifications_admin_db, notify_db_session):
def test_insert_new_code_and_get_it_back(app_, db_, db_session):
user = create_test_user('pending')
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='email')
@@ -19,9 +19,9 @@ def test_insert_new_code_and_get_it_back(notifications_admin, notifications_admi
assert saved_code.code_used is False
def test_insert_new_code_should_thrw_exception_when_type_does_not_exist(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_insert_new_code_should_thrw_exception_when_type_does_not_exist(app_,
db_,
db_session):
user = create_test_user('pending')
try:
verify_codes_dao.add_code(user_id=user.id, code='23545', code_type='not_real')
@@ -30,9 +30,9 @@ def test_insert_new_code_should_thrw_exception_when_type_does_not_exist(notifica
assert 'invalid input value for enum verify_code_types: "not_real"' in e.orig.pgerror
def test_should_throw_exception_when_user_does_not_exist(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_should_throw_exception_when_user_does_not_exist(app_,
db_,
db_session):
try:
verify_codes_dao.add_code(user_id=1, code='12345', code_type='email')
fail('Should throw exception')
@@ -41,9 +41,9 @@ def test_should_throw_exception_when_user_does_not_exist(notifications_admin,
'foreign key constraint "verify_codes_user_id_fkey"' in e.orig.pgerror
def test_should_return_none_if_code_is_used(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_should_return_none_if_code_is_used(app_,
db_,
db_session):
user = create_test_user('pending')
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='email')
@@ -52,9 +52,9 @@ def test_should_return_none_if_code_is_used(notifications_admin,
assert saved_code.code_used is True
def test_should_return_none_if_code_is_used(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_should_return_none_if_code_is_used(app_,
db_,
db_session):
user = create_test_user('pending')
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='sms')
@@ -64,9 +64,9 @@ def test_should_return_none_if_code_is_used(notifications_admin,
assert used_code == []
def test_should_return_all_unused_code_when_there_are_many(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_should_return_all_unused_code_when_there_are_many(app_,
db_,
db_session):
user = create_test_user('pending')
another_user = create_another_test_user('active')
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='sms')

24
tests/app/main/notify_client/test_sender.py
View File

@@ -4,17 +4,17 @@ from pytest import fail
from app.notify_client.sender import generate_token, check_token
def test_should_return_email_from_signed_token(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_should_return_email_from_signed_token(app_,
db_,
db_session):
email = 'email@something.com'
token = generate_token(email)
assert email == check_token(token)
def test_should_throw_exception_when_token_is_tampered_with(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_should_throw_exception_when_token_is_tampered_with(app_,
db_,
db_session):
email = 'email@something.com'
token = generate_token(email)
try:
@@ -24,12 +24,12 @@ def test_should_throw_exception_when_token_is_tampered_with(notifications_admin,
pass
def test_return_none_when_token_is_expired(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
notifications_admin.config['TOKEN_MAX_AGE_SECONDS'] = -1000
def test_return_none_when_token_is_expired(app_,
db_,
db_session):
with app_.test_request_context():
app_.config['TOKEN_MAX_AGE_SECONDS'] = -1000
email = 'email@something.com'
token = generate_token(email)
assert check_token(token) is None
notifications_admin.config['TOKEN_MAX_AGE_SECONDS'] = 120000
app_.config['TOKEN_MAX_AGE_SECONDS'] = 120000

16
tests/app/main/test_add_service_form.py
View File

@@ -2,11 +2,13 @@ from app.main.forms import AddServiceForm
from werkzeug.datastructures import MultiDict
def test_form_should_have_errors_when_duplicate_service_is_added(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
form = AddServiceForm(['some service', 'more names'],
formdata=MultiDict([('service_name', 'some service')]))
def test_form_should_have_errors_when_duplicate_service_is_added(app_,
db_,
db_session):
def _get_form_names():
return ['some service', 'more names']
with app_.test_request_context():
form = AddServiceForm(_get_form_names,
formdata=MultiDict([('name', 'some service')]))
form.validate()
assert {'service_name': ['Service name already exists']} == form.errors
assert {'name': ['Service name already exists']} == form.errors

32
tests/app/main/test_two_factor_form.py
View File

@@ -2,12 +2,12 @@ from datetime import datetime, timedelta
from app.main.dao import verify_codes_dao
from app.main.forms import TwoFactorForm
from tests.app.main import create_test_user
from tests import create_test_user
def test_form_is_valid_returns_no_errors(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={'sms_code': '12345'}) as req:
def test_form_is_valid_returns_no_errors(app_, db_, db_session):
with app_.test_request_context(method='POST',
data={'sms_code': '12345'}) as req:
user = set_up_test_data()
codes = verify_codes_dao.get_codes(user.id)
form = TwoFactorForm(codes)
@@ -15,9 +15,9 @@ def test_form_is_valid_returns_no_errors(notifications_admin, notifications_admi
assert len(form.errors) == 0
def test_returns_errors_when_code_is_too_short(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={'sms_code': '145'}) as req:
def test_returns_errors_when_code_is_too_short(app_, db_, db_session):
with app_.test_request_context(method='POST',
data={'sms_code': '145'}) as req:
user = set_up_test_data()
codes = verify_codes_dao.get_codes(user.id)
form = TwoFactorForm(codes)
@@ -26,9 +26,9 @@ def test_returns_errors_when_code_is_too_short(notifications_admin, notification
assert set(form.errors) == set({'sms_code': ['Code must be 5 digits', 'Code does not match']})
def test_returns_errors_when_code_is_missing(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={}) as req:
def test_returns_errors_when_code_is_missing(app_, db_, db_session):
with app_.test_request_context(method='POST',
data={}) as req:
user = set_up_test_data()
codes = verify_codes_dao.get_codes(user.id)
form = TwoFactorForm(codes)
@@ -37,9 +37,9 @@ def test_returns_errors_when_code_is_missing(notifications_admin, notifications_
assert set(form.errors) == set({'sms_code': ['Code must not be empty']})
def test_returns_errors_when_code_contains_letters(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={'sms_code': 'asdfg'}) as req:
def test_returns_errors_when_code_contains_letters(app_, db_, db_session):
with app_.test_request_context(method='POST',
data={'sms_code': 'asdfg'}) as req:
user = set_up_test_data()
codes = verify_codes_dao.get_codes(user.id)
form = TwoFactorForm(codes)
@@ -48,9 +48,9 @@ def test_returns_errors_when_code_contains_letters(notifications_admin, notifica
assert set(form.errors) == set({'sms_code': ['Code must be 5 digits', 'Code does not match']})
def test_should_return_errors_when_code_is_expired(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={'sms_code': '23456'}) as req:
def test_should_return_errors_when_code_is_expired(app_, db_, db_session):
with app_.test_request_context(method='POST',
data={'sms_code': '23456'}) as req:
user = create_test_user('active')
verify_codes_dao.add_code_with_expiry(user_id=user.id,
code='23456',

2
tests/app/main/test_validators.py
View File

@@ -2,7 +2,7 @@ from app.main.dao import users_dao
from app.main.forms import RegisterUserForm
def test_should_raise_validation_error_for_password(notifications_admin):
def test_should_raise_validation_error_for_password(app_):
form = RegisterUserForm(users_dao.get_user_by_email)
form.name.data = 'test'
form.email_address.data = 'teset@example.gov.uk'

46
tests/app/main/test_verify_form.py
View File

@@ -1,12 +1,12 @@
from datetime import datetime, timedelta
from app.main.dao import verify_codes_dao
from app.main.forms import VerifyForm
from tests.app.main import create_test_user
from tests import create_test_user
def test_form_should_have_error_when_code_is_not_valid(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={'sms_code': '12345aa', 'email_code': 'abcde'}) as req:
def test_form_should_have_error_when_code_is_not_valid(app_, db_, db_session):
with app_.test_request_context(method='POST',
data={'sms_code': '12345aa', 'email_code': 'abcde'}) as req:
user = set_up_test_data()
codes = verify_codes_dao.get_codes(user.id)
form = VerifyForm(codes)
@@ -19,9 +19,9 @@ def test_form_should_have_error_when_code_is_not_valid(notifications_admin, noti
assert set(errors) == set(expected)
def test_should_return_errors_when_code_missing(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={}) as req:
def test_should_return_errors_when_code_missing(app_, db_, db_session):
with app_.test_request_context(method='POST',
data={}) as req:
user = set_up_test_data()
codes = verify_codes_dao.get_codes(user.id)
form = VerifyForm(codes)
@@ -33,9 +33,9 @@ def test_should_return_errors_when_code_missing(notifications_admin, notificatio
assert set(errors) == set(expected)
def test_should_return_errors_when_code_is_too_short(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={'sms_code': '123', 'email_code': '123'}) as req:
def test_should_return_errors_when_code_is_too_short(app_, db_, db_session):
with app_.test_request_context(method='POST',
data={'sms_code': '123', 'email_code': '123'}) as req:
user = set_up_test_data()
codes = verify_codes_dao.get_codes(user.id)
form = VerifyForm(codes)
@@ -47,9 +47,9 @@ def test_should_return_errors_when_code_is_too_short(notifications_admin, notifi
assert set(errors) == set(expected)
def test_should_return_errors_when_code_does_not_match(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={'sms_code': '34567', 'email_code': '34567'}) as req:
def test_should_return_errors_when_code_does_not_match(app_, db_, db_session):
with app_.test_request_context(method='POST',
data={'sms_code': '34567', 'email_code': '34567'}) as req:
user = set_up_test_data()
codes = verify_codes_dao.get_codes(user.id)
form = VerifyForm(codes)
@@ -61,10 +61,10 @@ def test_should_return_errors_when_code_does_not_match(notifications_admin, noti
assert set(errors) == set(expected)
def test_should_return_errors_when_code_is_expired(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={'sms_code': '23456',
'email_code': '23456'}) as req:
def test_should_return_errors_when_code_is_expired(app_, db_, db_session):
with app_.test_request_context(method='POST',
data={'sms_code': '23456',
'email_code': '23456'}) as req:
user = create_test_user('pending')
verify_codes_dao.add_code_with_expiry(user_id=user.id,
code='23456',
@@ -85,12 +85,12 @@ def test_should_return_errors_when_code_is_expired(notifications_admin, notifica
assert set(errors) == set(expected)
def test_should_return_valid_form_when_many_codes_exist(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context(method='POST',
data={'sms_code': '23456',
'email_code': '23456'}) as req:
def test_should_return_valid_form_when_many_codes_exist(app_,
db_,
db_session):
with app_.test_request_context(method='POST',
data={'sms_code': '23456',
'email_code': '23456'}) as req:
user = set_up_test_data()
verify_codes_dao.add_code(user_id=user.id, code='23456', code_type='email')
verify_codes_dao.add_code(user_id=user.id, code='23456', code_type='sms')

85
tests/app/main/views/test_add_service.py
View File

@@ -1,58 +1,65 @@
from flask import url_for
from app.main.dao import verify_codes_dao, services_dao
from tests.app.main import create_test_user
from tests import create_test_user
from app.models import User
def test_get_should_render_add_service_template(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='sms')
client.post(url_for('.two_factor'), data={'sms_code': '12345'})
response = client.get(url_for('.add_service', first='first'))
def test_get_should_render_add_service_template(app_,
db_,
db_session,
active_user,
mock_get_service,
mock_get_services):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
response = client.get(url_for('main.add_service'))
assert response.status_code == 200
assert 'Set up notifications for your service' in response.get_data(as_text=True)
def test_should_add_service_and_redirect_to_next_page(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
def test_should_add_service_and_redirect_to_next_page(app_,
db_,
db_session,
mock_create_service,
mock_get_services):
with app_.test_request_context():
with app_.test_client() as client:
user = User.query.first()
client.login(user)
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='sms')
client.post(url_for('.two_factor'), data={'sms_code': '12345'})
response = client.post(
url_for('.add_service', first='first'),
data={'service_name': 'testing the post'}
)
url_for('main.add_service'),
data={'name': 'testing the post'})
assert response.status_code == 302
assert response.location == url_for('.dashboard', service_id=123, _external=True)
saved_service = services_dao.find_service_by_service_name('testing the post')
assert saved_service is not None
assert response.location == url_for('main.service_dashboard', service_id=101, _external=True)
assert mock_create_service.called
assert mock_get_services.called
def test_should_return_form_errors_when_service_name_is_empty(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='sms')
client.post(url_for('.two_factor'), data={'sms_code': '12345'})
response = client.post(url_for('.add_service', first='first'), data={})
def test_should_return_form_errors_when_service_name_is_empty(app_,
db_,
db_session,
active_user,
mock_get_service,
mock_get_services):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
response = client.post(url_for('main.add_service'), data={})
assert response.status_code == 200
assert 'Service name can not be empty' in response.get_data(as_text=True)
def test_should_show_page_for_adding_another_service(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
def test_should_return_form_errors_with_duplicate_service_name(app_,
db_,
db_session,
mock_get_services):
with app_.test_request_context():
with app_.test_client() as client:
user = User.query.first()
client.login(user)
response = client.get(url_for('.add_service'))
response = client.post(
url_for('main.add_service'), data={'name': 'service_one'})
assert response.status_code == 200
assert 'Add a new service' in response.get_data(as_text=True)
assert 'Service name already exists' in response.get_data(as_text=True)
assert mock_get_services.called

17
tests/app/main/views/test_api_keys.py
View File

@@ -1,14 +1,13 @@
from tests.app.main import create_test_user
from flask import url_for
def test_should_show_api_keys_and_documentation_page(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.get(url_for('.api_keys', service_id=123))
def test_should_show_api_keys_and_documentation_page(app_,
db_,
db_session,
active_user):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
response = client.get(url_for('main.api_keys', service_id=123))
assert response.status_code == 200

26
tests/app/main/views/test_choose_services.py
View File

@@ -1,15 +1,23 @@
from tests.app.main import create_test_user
from tests import create_test_user
from flask import url_for
from app.models import User
def test_should_show_choose_services_page(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
def test_should_show_choose_services_page(app_,
db_,
db_session,
active_user,
mock_get_services):
with app_.test_request_context():
with app_.test_client() as client:
user = User.query.first()
client.login(user)
response = client.get('/services')
response = client.get(url_for('main.choose_service'))
assert response.status_code == 200
assert 'Choose service' in response.get_data(as_text=True)
resp_data = response.get_data(as_text=True)
assert 'Choose service' in resp_data
services = mock_get_services.side_effect()
assert mock_get_services.called
assert services['data'][0]['name'] in resp_data
assert services['data'][1]['name'] in resp_data

145
tests/app/main/views/test_code_not_received.py
View File

@@ -1,16 +1,16 @@
from app.main.dao import verify_codes_dao, users_dao
from tests.app.main import create_test_user
from tests import create_test_user
from flask import url_for
def test_should_render_email_code_not_received_template_and_populate_email_address(notifications_admin,
notifications_admin_db,
notify_db_session,
mocker):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_render_email_code_not_received_template_and_populate_email_address(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
_set_up_mocker(mocker)
user = create_test_user('pending')
session['user_email'] = user.email_address
response = client.get(url_for('main.check_and_resend_email_code'))
@@ -20,14 +20,14 @@ def test_should_render_email_code_not_received_template_and_populate_email_addre
assert 'value="test@user.gov.uk"' in response.get_data(as_text=True)
def test_should_check_and_resend_email_code_redirect_to_verify(notifications_admin,
notifications_admin_db,
notify_db_session,
mocker):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_check_and_resend_email_code_redirect_to_verify(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
_set_up_mocker(mocker)
user = create_test_user('pending')
session['user_email'] = user.email_address
verify_codes_dao.add_code(user.id, code='12345', code_type='email')
@@ -37,14 +37,14 @@ def test_should_check_and_resend_email_code_redirect_to_verify(notifications_adm
assert response.location == url_for('main.verify', _external=True)
def test_should_render_text_code_not_received_template(notifications_admin,
notifications_admin_db,
notify_db_session,
mocker):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_render_text_code_not_received_template(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
_set_up_mocker(mocker)
user = create_test_user('pending')
session['user_email'] = user.email_address
verify_codes_dao.add_code(user.id, code='12345', code_type='sms')
@@ -55,14 +55,14 @@ def test_should_render_text_code_not_received_template(notifications_admin,
assert 'value="+441234123412"'
def test_should_check_and_redirect_to_verify(notifications_admin,
notifications_admin_db,
notify_db_session,
mocker):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_check_and_redirect_to_verify(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
_set_up_mocker(mocker)
user = create_test_user('pending')
session['user_email'] = user.email_address
verify_codes_dao.add_code(user.id, code='12345', code_type='sms')
@@ -72,14 +72,14 @@ def test_should_check_and_redirect_to_verify(notifications_admin,
assert response.location == url_for('main.verify', _external=True)
def test_should_update_email_address_resend_code(notifications_admin,
notifications_admin_db,
notify_db_session,
mocker):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_update_email_address_resend_code(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
_set_up_mocker(mocker)
user = create_test_user('pending')
session['user_email'] = user.email_address
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='email')
@@ -91,14 +91,14 @@ def test_should_update_email_address_resend_code(notifications_admin,
assert updated_user.email_address == 'new@address.gov.uk'
def test_should_update_mobile_number_resend_code(notifications_admin,
notifications_admin_db,
notify_db_session,
mocker):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_update_mobile_number_resend_code(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
_set_up_mocker(mocker)
user = create_test_user('pending')
session['user_email'] = user.email_address
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='sms')
@@ -110,14 +110,14 @@ def test_should_update_mobile_number_resend_code(notifications_admin,
assert updated_user.mobile_number == '+44 7700 900 460'
def test_should_render_verification_code_not_received(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_render_verification_code_not_received(app_,
db_,
db_session,
active_user):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
user = create_test_user('active')
session['user_email'] = user.email_address
session['user_email'] = active_user.email_address
response = client.get(url_for('main.verification_code_not_received'))
assert response.status_code == 200
assert 'Resend verification code' in response.get_data(as_text=True)
@@ -125,41 +125,36 @@ def test_should_render_verification_code_not_received(notifications_admin,
'speak to your service manager to reset the number.' in response.get_data(as_text=True)
def test_check_and_redirect_to_two_factor(notifications_admin,
notifications_admin_db,
notify_db_session,
mocker):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_check_and_redirect_to_two_factor(app_,
db_,
db_session,
active_user,
mock_send_sms,
mock_send_email):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
user = create_test_user('active')
session['user_email'] = user.email_address
_set_up_mocker(mocker)
session['user_email'] = active_user.email_address
response = client.get(url_for('main.check_and_resend_verification_code'))
assert response.status_code == 302
assert response.location == url_for('main.two_factor', _external=True)
def test_should_create_new_code_for_user(notifications_admin,
notifications_admin_db,
notify_db_session,
mocker):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_create_new_code_for_user(app_,
db_,
db_session,
active_user,
mock_send_sms,
mock_send_email):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
user = create_test_user('active')
session['user_email'] = user.email_address
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='sms')
_set_up_mocker(mocker)
session['user_email'] = active_user.email_address
verify_codes_dao.add_code(user_id=active_user.id, code='12345', code_type='sms')
response = client.get(url_for('main.check_and_resend_verification_code'))
assert response.status_code == 302
assert response.location == url_for('main.two_factor', _external=True)
codes = verify_codes_dao.get_codes(user_id=user.id, code_type='sms')
codes = verify_codes_dao.get_codes(user_id=active_user.id, code_type='sms')
assert len(codes) == 2
for x in ([used.code_used for used in codes]):
assert x is False
def _set_up_mocker(mocker):
mocker.patch("app.admin_api_client.send_sms")
mocker.patch("app.admin_api_client.send_email")

19
tests/app/main/views/test_dashboard.py
View File

@@ -1,15 +1,16 @@
from tests.app.main import create_test_user
from app.models import User
from flask import url_for
def test_should_show_recent_jobs_on_dashboard(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.get('/services/123/dashboard')
def test_should_show_recent_jobs_on_dashboard(app_,
db_,
db_session,
active_user,
mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
response = client.get(url_for('main.service_dashboard', service_id=123))
assert response.status_code == 200
assert 'Test message 1' in response.get_data(as_text=True)

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

@@ -1,26 +1,27 @@
from flask import url_for
from app.main.dao import users_dao
from tests.app.main import create_test_user
from tests import create_test_user
def test_should_render_forgot_password(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
response = notifications_admin.test_client().get(url_for('.forgot_password'))
def test_should_render_forgot_password(app_, db_, db_session):
with app_.test_request_context():
response = app_.test_client().get(url_for('.forgot_password'))
assert response.status_code == 200
assert 'If you have forgotten your password, we can send you an email to create a new password.' \
in response.get_data(as_text=True)
def test_should_redirect_to_password_reset_sent_and_state_updated(notifications_admin,
notifications_admin_db,
mocker,
notify_db_session):
mocker.patch("app.admin_api_client.send_email")
with notifications_admin.test_request_context():
user = create_test_user('active')
response = notifications_admin.test_client().post(url_for('.forgot_password'),
data={'email_address': user.email_address})
def test_should_redirect_to_password_reset_sent_and_state_updated(app_,
db_,
db_session,
active_user,
mock_send_email):
with app_.test_request_context():
response = app_.test_client().post(
url_for('.forgot_password'),
data={'email_address': active_user.email_address})
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)
assert users_dao.get_user_by_id(user.id).state == 'request_password_reset'
assert (
'You have been sent an email containing a link'
' to reset your password.') in response.get_data(as_text=True)
assert users_dao.get_user_by_id(active_user.id).state == 'request_password_reset'

6
tests/app/main/views/test_headers.py
View File

@@ -1,7 +1,7 @@
def test_owasp_useful_headers_set(notifications_admin):
with notifications_admin.test_request_context():
response = notifications_admin.test_client().get('/')
def test_owasp_useful_headers_set(app_):
with app_.test_request_context():
response = app_.test_client().get('/')
assert response.status_code == 200
assert response.headers['X-Frame-Options'] == 'deny'
assert response.headers['X-Content-Type-Options'] == 'nosniff'

38
tests/app/main/views/test_jobs.py
View File

@@ -1,39 +1,45 @@
from tests.app.main import create_test_user
from flask import url_for
from app.models import User
from tests import create_test_user
def test_should_return_list_of_all_jobs(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
def test_should_return_list_of_all_jobs(app_, db_, db_session, service_one):
with app_.test_request_context():
with app_.test_client() as client:
user = User.query.first()
client.login(user)
response = client.get('/services/123/jobs')
response = client.get(url_for('main.view_jobs', service_id=101))
assert response.status_code == 200
assert 'Test message 1' in response.get_data(as_text=True)
assert 'Final reminder' in response.get_data(as_text=True)
def test_should_show_page_for_one_job(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_show_page_for_one_job(app_, db_, db_session, service_one):
with app_.test_request_context():
with app_.test_client() as client:
# TODO filename will be part of job metadata not in session
with client.session_transaction() as s:
s[456] = 'dispatch_20151114.csv'
user = create_test_user('active')
user = User.query.first()
client.login(user)
response = client.get('/services/123/jobs/456')
response = client.get(url_for('main.view_job', service_id=123, job_id=456))
assert response.status_code == 200
assert 'dispatch_20151114.csv' in response.get_data(as_text=True)
assert 'Test message 1' in response.get_data(as_text=True)
def test_should_show_page_for_one_notification(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
def test_should_show_page_for_one_notification(app_, db_, db_session, service_one):
with app_.test_request_context():
with app_.test_client() as client:
user = User.query.first()
client.login(user)
response = client.get('/services/123/jobs/456/notification/3')
response = client.get(url_for(
'main.view_notification',
service_id=101,
job_id=123,
notification_id=3))
assert response.status_code == 200
assert 'Text message' in response.get_data(as_text=True)

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

@@ -3,12 +3,12 @@ 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
from tests.app.main import create_test_user
from tests import create_test_user
def test_should_render_new_password_template(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_render_new_password_template(app_, db_, db_session):
with app_.test_request_context():
with app_.test_client() as client:
user = create_test_user('request_password_reset')
token = generate_token(user.email_address)
response = client.get(url_for('.new_password', token=token))
@@ -16,10 +16,9 @@ def test_should_render_new_password_template(notifications_admin, notifications_
assert ' You can now create a new password for your account.' in response.get_data(as_text=True)
def test_should_render_new_password_template_with_message_of_bad_token(notifications_admin, notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_render_new_password_template_with_message_of_bad_token(app_, db_, db_session):
with app_.test_request_context():
with app_.test_client() as client:
create_test_user('request_password_reset')
token = generate_token('no_user@d.gov.uk')
response = client.get(url_for('.new_password', token=token))
@@ -28,13 +27,12 @@ def test_should_render_new_password_template_with_message_of_bad_token(notificat
response.get_data(as_text=True)
def test_should_redirect_to_two_factor_when_password_reset_is_successful(notifications_admin,
notifications_admin_db,
notify_db_session,
mocker):
_set_up_mocker(mocker)
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_redirect_to_two_factor_when_password_reset_is_successful(app_,
db_,
db_session,
mock_send_sms):
with app_.test_request_context():
with app_.test_client() as client:
user = create_test_user('request_password_reset')
token = generate_token(user.email_address)
response = client.post(url_for('.new_password', token=token), data={'new_password': 'a-new_password'})
@@ -45,31 +43,27 @@ def test_should_redirect_to_two_factor_when_password_reset_is_successful(notific
assert saved_user.state == 'active'
def test_should_redirect_to_forgot_password_with_flash_message_when_token_is_expired(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
notifications_admin.config['TOKEN_MAX_AGE_SECONDS'] = -1000
def test_should_redirect_to_forgot_password_with_flash_message_when_token_is_expired(app_,
db_,
db_session):
with app_.test_request_context():
with app_.test_client() as client:
app_.config['TOKEN_MAX_AGE_SECONDS'] = -1000
user = create_test_user('request_password_reset')
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('.forgot_password', _external=True)
notifications_admin.config['TOKEN_MAX_AGE_SECONDS'] = 3600
app_.config['TOKEN_MAX_AGE_SECONDS'] = 3600
def test_should_redirect_to_forgot_password_when_user_is_active_should_be_request_password_reset(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_redirect_to_forgot_password_when_user_is_active_should_be_request_password_reset(app_,
db_,
db_session):
with app_.test_request_context():
with app_.test_client() as client:
user = create_test_user('active')
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)
def _set_up_mocker(mocker):
mocker.patch("app.admin_api_client.send_sms")

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

@@ -1,57 +1,64 @@
from flask import url_for
def test_render_register_returns_template_with_form(notifications_admin, notifications_admin_db, notify_db_session):
response = notifications_admin.test_client().get('/register')
def test_render_register_returns_template_with_form(app_, db_, db_session):
response = app_.test_client().get('/register')
assert response.status_code == 200
assert 'Create an account' in response.get_data(as_text=True)
def test_process_register_creates_new_user(notifications_admin, notifications_admin_db, mocker, notify_db_session):
_set_up_mocker(mocker)
response = notifications_admin.test_client().post('/register',
data={'name': 'Some One Valid',
'email_address': 'someone@example.gov.uk',
'mobile_number': '+4407700900460',
'password': 'validPassword!'})
assert response.status_code == 302
assert response.location == 'http://localhost/verify'
def test_process_register_creates_new_user(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
with app_.test_request_context():
response = app_.test_client().post('/register',
data={'name': 'Some One Valid',
'email_address': 'someone@example.gov.uk',
'mobile_number': '+4407700900460',
'password': 'validPassword!'})
assert response.status_code == 302
assert response.location == url_for('main.verify', _external=True)
def test_process_register_returns_400_when_mobile_number_is_invalid(notifications_admin,
notifications_admin_db,
mocker,
notify_db_session):
_set_up_mocker(mocker)
response = notifications_admin.test_client().post('/register',
data={'name': 'Bad Mobile',
'email_address': 'bad_mobile@example.gov.uk',
'mobile_number': 'not good',
'password': 'validPassword!'})
def test_process_register_returns_400_when_mobile_number_is_invalid(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
response = app_.test_client().post('/register',
data={'name': 'Bad Mobile',
'email_address': 'bad_mobile@example.gov.uk',
'mobile_number': 'not good',
'password': 'validPassword!'})
assert response.status_code == 200
assert 'Must be a UK mobile number (eg 07700 900460)' in response.get_data(as_text=True)
def test_should_return_400_when_email_is_not_gov_uk(notifications_admin,
notifications_admin_db,
mocker,
notify_db_session):
_set_up_mocker(mocker)
response = notifications_admin.test_client().post('/register',
data={'name': 'Bad Mobile',
'email_address': 'bad_mobile@example.not.right',
'mobile_number': '+44123412345',
'password': 'validPassword!'})
def test_should_return_400_when_email_is_not_gov_uk(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
response = app_.test_client().post('/register',
data={'name': 'Bad Mobile',
'email_address': 'bad_mobile@example.not.right',
'mobile_number': '+44123412345',
'password': 'validPassword!'})
assert response.status_code == 200
assert 'Enter a gov.uk email address' in response.get_data(as_text=True)
def test_should_add_verify_codes_on_session(notifications_admin, notifications_admin_db, mocker, notify_db_session):
_set_up_mocker(mocker)
with notifications_admin.test_client() as client:
def test_should_add_verify_codes_on_session(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
with app_.test_client() as client:
response = client.post('/register',
data={'name': 'Test Codes',
'email_address': 'test_codes@example.gov.uk',
@@ -61,17 +68,12 @@ def test_should_add_verify_codes_on_session(notifications_admin, notifications_a
assert 'notify_admin_session' in response.headers.get('Set-Cookie')
def _set_up_mocker(mocker):
mocker.patch("app.admin_api_client.send_sms")
mocker.patch("app.admin_api_client.send_email")
def test_should_return_400_if_password_is_blacklisted(notifications_admin, notifications_admin_db, notify_db_session):
response = notifications_admin.test_client().post('/register',
data={'name': 'Bad Mobile',
'email_address': 'bad_mobile@example.not.right',
'mobile_number': '+44123412345',
'password': 'password1234'})
def test_should_return_400_if_password_is_blacklisted(app_, db_, db_session):
response = app_.test_client().post('/register',
data={'name': 'Bad Mobile',
'email_address': 'bad_mobile@example.not.right',
'mobile_number': '+44123412345',
'password': 'password1234'})
response.status_code == 200
assert 'That password is blacklisted, too common' in response.get_data(as_text=True)

329
tests/app/main/views/test_service_settings.py
View File

@@ -1,165 +1,252 @@
from tests.app.main import create_test_user
from flask import (url_for, session)
def test_should_show_overview(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.get('/services/123/service-settings')
def test_should_show_overview(app_, db_, db_session, active_user, mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.get(url_for(
'main.service_settings', service_id=service_id))
assert response.status_code == 200
assert 'Service settings' in response.get_data(as_text=True)
resp_data = response.get_data(as_text=True)
assert 'Service settings' in resp_data
service = mock_get_service.side_effect(service_id)['data']
assert service['name'] in resp_data
assert mock_get_service.called
def test_should_show_service_name(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.get('/services/123/service-settings/name')
def test_should_show_service_name(app_, db_, db_session, active_user, mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.get(url_for(
'main.service_name_change', service_id=service_id))
assert response.status_code == 200
assert 'Change your service name' in response.get_data(as_text=True)
resp_data = response.get_data(as_text=True)
assert 'Change your service name' in resp_data
assert mock_get_service.called
service = mock_get_service.side_effect(service_id)['data']
def test_should_redirect_after_change_service_name(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.post('/services/123/service-settings/request-to-go-live')
def test_should_redirect_after_change_service_name(app_, db_, db_session, active_user, mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.post(url_for(
'main.service_name_change', service_id=service_id))
assert response.status_code == 302
assert 'http://localhost/services/123/service-settings' == response.location
assert response.status_code == 302
settings_url = url_for(
'main.service_name_change_confirm', service_id=service_id, _external=True)
assert settings_url == response.location
assert mock_get_service.called
def test_should_show_service_name_confirmation(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.get('/services/123/service-settings/name/confirm')
def test_should_show_service_name_confirmation(app_,
db_,
db_session,
active_user,
mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.get(url_for(
'main.service_name_change_confirm', service_id=service_id))
assert response.status_code == 200
assert 'Change your service name' in response.get_data(as_text=True)
assert response.status_code == 200
resp_data = response.get_data(as_text=True)
assert 'Change your service name' in resp_data
assert mock_get_service.called
def test_should_redirect_after_service_name_confirmation(notifications_admin, notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.post('/services/123/service-settings/name/confirm')
def test_should_redirect_after_service_name_confirmation(app_,
db_,
db_session,
active_user,
mock_get_service,
mock_update_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
service_new_name = 'New Name'
with client.session_transaction() as session:
session['service_name_change'] = service_new_name
response = client.post(url_for(
'main.service_name_change_confirm', service_id=service_id))
assert response.status_code == 302
assert 'http://localhost/services/123/service-settings' == response.location
assert response.status_code == 302
settings_url = url_for(
'main.service_settings', service_id=service_id, _external=True)
resp_data = response.get_data(as_text=True)
assert settings_url == response.location
assert mock_get_service.called
assert mock_update_service.called
def test_should_show_request_to_go_live(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.get('/services/123/service-settings/request-to-go-live')
assert response.status_code == 200
assert 'Request to go live' in response.get_data(as_text=True)
def test_should_show_request_to_go_live(app_, db_, db_session, active_user, mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.get(
url_for('main.service_request_to_go_live', service_id=service_id))
service = mock_get_service.side_effect(service_id)['data']
assert response.status_code == 200
resp_data = response.get_data(as_text=True)
assert 'Request to go live' in resp_data
assert mock_get_service.called
def test_should_redirect_after_request_to_go_live(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.post('/services/123/service-settings/request-to-go-live')
def test_should_redirect_after_request_to_go_live(app_,
db_,
db_session,
active_user,
mock_get_service,
mock_update_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.post(url_for(
'main.service_request_to_go_live', service_id=service_id))
assert response.status_code == 302
assert 'http://localhost/services/123/service-settings' == response.location
assert response.status_code == 302
settings_url = url_for(
'main.service_settings', service_id=service_id, _external=True)
assert settings_url == response.location
assert mock_get_service.called
assert mock_update_service.called
def test_should_show_status_page(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.get('/services/123/service-settings/status')
def test_should_show_status_page(app_, db_, db_session, active_user, mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.get(url_for(
'main.service_status_change', service_id=service_id))
assert response.status_code == 200
assert 'Turn off all outgoing notifications' in response.get_data(as_text=True)
assert response.status_code == 200
resp_data = response.get_data(as_text=True)
assert 'Turn off all outgoing notifications' in resp_data
assert mock_get_service.called
def test_should_show_redirect_after_status_change(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.post('/services/123/service-settings/status')
def test_should_show_redirect_after_status_change(app_,
db_,
db_session,
active_user,
mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.post(url_for(
'main.service_status_change', service_id=service_id))
assert response.status_code == 302
assert 'http://localhost/services/123/service-settings/status/confirm' == response.location
assert response.status_code == 302
redirect_url = url_for(
'main.service_status_change_confirm', service_id=service_id, _external=True)
assert redirect_url == response.location
assert mock_get_service.called
def test_should_show_status_confirmation(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.get('/services/123/service-settings/status/confirm')
def test_should_show_status_confirmation(app_, db_, db_session, active_user, mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.get(url_for(
'main.service_status_change_confirm', service_id=service_id))
assert response.status_code == 200
assert 'Turn off all outgoing notifications' in response.get_data(as_text=True)
assert response.status_code == 200
resp_data = response.get_data(as_text=True)
assert 'Turn off all outgoing notifications' in resp_data
assert mock_get_service.called
def test_should_redirect_after_status_confirmation(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.post('/services/123/service-settings/status/confirm')
def test_should_redirect_after_status_confirmation(app_,
db_,
db_session,
active_user,
mock_get_service,
mock_update_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.post(url_for(
'main.service_status_change_confirm', service_id=service_id))
assert response.status_code == 302
assert 'http://localhost/services/123/service-settings' == response.location
assert response.status_code == 302
settings_url = url_for(
'main.service_settings', service_id=service_id, _external=True)
assert settings_url == response.location
assert mock_get_service.called
assert mock_update_service.called
def test_should_show_delete_page(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.get('/services/123/service-settings/delete')
def test_should_show_delete_page(app_, db_, db_session, active_user, mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.get(url_for(
'main.service_delete', service_id=service_id))
assert response.status_code == 200
assert 'Delete this service from Notify' in response.get_data(as_text=True)
assert response.status_code == 200
assert 'Delete this service from Notify' in response.get_data(as_text=True)
assert mock_get_service.called
def test_should_show_redirect_after_deleting_service(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.post('/services/123/service-settings/delete')
def test_should_show_redirect_after_deleting_service(app_, db_, db_session, active_user, mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.post(url_for(
'main.service_delete', service_id=service_id))
assert response.status_code == 302
assert 'http://localhost/services/123/service-settings/delete/confirm' == response.location
assert response.status_code == 302
delete_url = url_for(
'main.service_delete_confirm', service_id=service_id, _external=True)
assert delete_url == response.location
def test_should_show_delete_confirmation(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.get('/services/123/service-settings/delete/confirm')
def test_should_show_delete_confirmation(app_, db_, db_session, active_user, mock_get_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.get(url_for(
'main.service_delete_confirm', service_id=service_id))
assert response.status_code == 200
assert 'Delete this service from Notify' in response.get_data(as_text=True)
assert response.status_code == 200
assert 'Delete this service from Notify' in response.get_data(as_text=True)
assert mock_get_service.called
def test_should_redirect_delete_confirmation(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.post('/services/123/service-settings/delete/confirm')
def test_should_redirect_delete_confirmation(app_,
db_,
db_session,
active_user,
mock_get_service,
mock_delete_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
service_id = 123
response = client.post(url_for(
'main.service_delete_confirm', service_id=service_id))
assert response.status_code == 302
assert 'http://localhost/services/123/dashboard' == response.location
assert response.status_code == 302
choose_url = url_for(
'main.choose_service', _external=True)
assert choose_url == response.location
assert mock_get_service.called
assert mock_delete_service.called

58
tests/app/main/views/test_sign_in.py
View File

@@ -5,9 +5,9 @@ from app.models import User
from flask import url_for
def test_render_sign_in_returns_sign_in_template(notifications_admin):
with notifications_admin.test_request_context():
response = notifications_admin.test_client().get(url_for('main.sign_in'))
def test_render_sign_in_returns_sign_in_template(app_):
with app_.test_request_context():
response = app_.test_client().get(url_for('main.sign_in'))
assert response.status_code == 200
assert 'Sign in' in response.get_data(as_text=True)
assert 'Email address' in response.get_data(as_text=True)
@@ -15,8 +15,11 @@ def test_render_sign_in_returns_sign_in_template(notifications_admin):
assert 'Forgotten password?' in response.get_data(as_text=True)
def test_process_sign_in_return_2fa_template(notifications_admin, notifications_admin_db, mocker, notify_db_session):
_set_up_mocker(mocker)
def test_process_sign_in_return_2fa_template(app_,
db_,
db_session,
mock_send_sms,
mock_send_email):
user = User(email_address='valid@example.gov.uk',
password='val1dPassw0rd!',
mobile_number='+441234123123',
@@ -25,8 +28,8 @@ def test_process_sign_in_return_2fa_template(notifications_admin, notifications_
role_id=1,
state='active')
users_dao.insert_user(user)
with notifications_admin.test_request_context():
response = notifications_admin.test_client().post(
with app_.test_request_context():
response = app_.test_client().post(
url_for('main.sign_in'), data={
'email_address': 'valid@example.gov.uk',
'password': 'val1dPassw0rd!'})
@@ -34,9 +37,9 @@ def test_process_sign_in_return_2fa_template(notifications_admin, notifications_
assert response.location == 'http://localhost/two-factor'
def test_should_return_locked_out_true_when_user_is_locked(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_should_return_locked_out_true_when_user_is_locked(app_,
db_,
db_session):
user = User(email_address='valid@example.gov.uk',
password='val1dPassw0rd!',
mobile_number='+441234123123',
@@ -45,14 +48,14 @@ def test_should_return_locked_out_true_when_user_is_locked(notifications_admin,
role_id=1,
state='active')
users_dao.insert_user(user)
with notifications_admin.test_request_context():
with app_.test_request_context():
for _ in range(10):
notifications_admin.test_client().post(
app_.test_client().post(
url_for('main.sign_in'), data={
'email_address': 'valid@example.gov.uk',
'password': 'whatIsMyPassword!'})
response = notifications_admin.test_client().post(
response = app_.test_client().post(
url_for('main.sign_in'), data={
'email_address': 'valid@example.gov.uk',
'password': 'val1dPassw0rd!'})
@@ -60,7 +63,7 @@ def test_should_return_locked_out_true_when_user_is_locked(notifications_admin,
assert response.status_code == 200
assert 'Username or password is incorrect' in response.get_data(as_text=True)
another_bad_attempt = notifications_admin.test_client().post(
another_bad_attempt = app_.test_client().post(
url_for('main.sign_in'), data={
'email_address': 'valid@example.gov.uk',
'password': 'whatIsMyPassword!'})
@@ -68,9 +71,9 @@ def test_should_return_locked_out_true_when_user_is_locked(notifications_admin,
assert 'Username or password is incorrect' in response.get_data(as_text=True)
def test_should_return_active_user_is_false_if_user_is_inactive(notifications_admin,
notifications_admin_db,
notify_db_session):
def test_should_return_active_user_is_false_if_user_is_inactive(app_,
db_,
db_session):
user = User(email_address='inactive_user@example.gov.uk',
password='val1dPassw0rd!',
mobile_number='+441234123123',
@@ -80,8 +83,8 @@ def test_should_return_active_user_is_false_if_user_is_inactive(notifications_ad
state='inactive')
users_dao.insert_user(user)
with notifications_admin.test_request_context():
response = notifications_admin.test_client().post(
with app_.test_request_context():
response = app_.test_client().post(
url_for('main.sign_in'), data={
'email_address': 'inactive_user@example.gov.uk',
'password': 'val1dPassw0rd!'})
@@ -90,9 +93,9 @@ def test_should_return_active_user_is_false_if_user_is_inactive(notifications_ad
assert 'Username or password is incorrect' in response.get_data(as_text=True)
def test_should_return_200_when_user_does_not_exist(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
response = notifications_admin.test_client().post(
def test_should_return_200_when_user_does_not_exist(app_, db_, db_session):
with app_.test_request_context():
response = app_.test_client().post(
url_for('main.sign_in'), data={
'email_address': 'does_not_exist@gov.uk',
'password': 'doesNotExist!'})
@@ -100,7 +103,7 @@ def test_should_return_200_when_user_does_not_exist(notifications_admin, notific
assert 'Username or password is incorrect' in response.get_data(as_text=True)
def test_should_return_200_when_user_is_not_active(notifications_admin, notifications_admin_db, notify_db_session):
def test_should_return_200_when_user_is_not_active(app_, db_, db_session):
user = User(email_address='PendingUser@example.gov.uk',
password='val1dPassw0rd!',
mobile_number='+441234123123',
@@ -109,15 +112,10 @@ def test_should_return_200_when_user_is_not_active(notifications_admin, notifica
role_id=1,
state='pending')
users_dao.insert_user(user)
with notifications_admin.test_request_context():
response = notifications_admin.test_client().post(
with app_.test_request_context():
response = app_.test_client().post(
url_for('main.sign_in'), data={
'email_address': 'PendingUser@example.gov.uk',
'password': 'val1dPassw0rd!'})
assert response.status_code == 200
assert 'Username or password is incorrect' in response.get_data(as_text=True)
def _set_up_mocker(mocker):
mocker.patch("app.admin_api_client.send_sms")
mocker.patch("app.admin_api_client.send_email")

26
tests/app/main/views/test_sign_out.py
View File

@@ -3,24 +3,23 @@ from flask import url_for
from app.main.dao import users_dao
from app.models import User
from .test_sign_in import _set_up_mocker
def test_render_sign_out_redirects_to_sign_in(notifications_admin):
with notifications_admin.test_request_context():
response = notifications_admin.test_client().get(
def test_render_sign_out_redirects_to_sign_in(app_):
with app_.test_request_context():
response = app_.test_client().get(
url_for('main.sign_out'))
assert response.status_code == 302
assert response.location == url_for(
'main.sign_in', _external=True, next=url_for('main.sign_out'))
def test_sign_out_user(notifications_admin,
notifications_admin_db,
notify_db_session,
mocker):
with notifications_admin.test_request_context():
_set_up_mocker(mocker)
def test_sign_out_user(app_,
db_,
db_session,
mock_send_sms,
mock_send_email,
mock_get_service):
with app_.test_request_context():
email = 'valid@example.gov.uk'
password = 'val1dPassw0rd!'
user = User(email_address=email,
@@ -31,10 +30,11 @@ def test_sign_out_user(notifications_admin,
role_id=1,
state='active')
users_dao.insert_user(user)
with notifications_admin.test_client() as client:
with app_.test_client() as client:
client.login(user)
# Check we are logged in
response = client.get('/services/123/dashboard')
response = client.get(
url_for('main.service_dashboard', service_id="123"))
assert response.status_code == 200
response = client.get(url_for('main.sign_out'))
assert response.status_code == 302

74
tests/app/main/views/test_sms.py
View File

@@ -2,17 +2,14 @@ from io import BytesIO
from flask import url_for
import moto
from tests.app.main import create_test_user
def test_upload_empty_csvfile_returns_to_upload_page(
notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
def test_upload_empty_csvfile_returns_to_upload_page(app_, db_, db_session, active_user):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
upload_data = {'file': (BytesIO(''.encode('utf-8')), 'emtpy.csv')}
response = client.post(url_for('main.sendsms', service_id=123),
response = client.post(url_for('main.send_sms', service_id=123),
data=upload_data, follow_redirects=True)
assert response.status_code == 200
@@ -21,18 +18,20 @@ def test_upload_empty_csvfile_returns_to_upload_page(
@moto.mock_s3
def test_upload_csvfile_with_invalid_phone_shows_check_page_with_errors(
notifications_admin, notifications_admin_db, notify_db_session):
def test_upload_csvfile_with_invalid_phone_shows_check_page_with_errors(app_,
db_,
db_session,
mocker,
active_user):
contents = 'phone\n+44 123\n+44 456'
file_data = (BytesIO(contents.encode('utf-8')), 'invalid.csv')
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
upload_data = {'file': file_data}
response = client.post(url_for('main.sendsms', service_id=123),
response = client.post(url_for('main.send_sms', service_id=123),
data=upload_data,
follow_redirects=True)
assert response.status_code == 200
@@ -44,20 +43,21 @@ def test_upload_csvfile_with_invalid_phone_shows_check_page_with_errors(
@moto.mock_s3
def test_upload_csvfile_with_valid_phone_shows_first3_and_last3_numbers(
notifications_admin, notifications_admin_db, notify_db_session,
mocker):
def test_upload_csvfile_with_valid_phone_shows_first3_and_last3_numbers(app_,
db_,
db_session,
mocker,
active_user):
contents = 'phone\n+44 7700 900981\n+44 7700 900982\n+44 7700 900983\n+44 7700 900984\n+44 7700 900985\n+44 7700 900986\n+44 7700 900987\n+44 7700 900988\n+44 7700 900989' # noqa
file_data = (BytesIO(contents.encode('utf-8')), 'valid.csv')
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
upload_data = {'file': file_data}
response = client.post(url_for('main.sendsms', service_id=123),
response = client.post(url_for('main.send_sms', service_id=123),
data=upload_data,
follow_redirects=True)
@@ -79,18 +79,21 @@ def test_upload_csvfile_with_valid_phone_shows_first3_and_last3_numbers(
@moto.mock_s3
def test_upload_csvfile_with_valid_phone_shows_all_if_6_or_less_numbers(
notifications_admin, notifications_admin_db, notify_db_session):
def test_upload_csvfile_with_valid_phone_shows_all_if_6_or_less_numbers(app_,
db_,
db_session,
mocker,
active_user):
contents = 'phone\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')
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
upload_data = {'file': file_data}
response = client.post(url_for('main.sendsms', service_id=123),
response = client.post(url_for('main.send_sms', service_id=123),
data=upload_data,
follow_redirects=True)
@@ -108,12 +111,11 @@ def test_upload_csvfile_with_valid_phone_shows_all_if_6_or_less_numbers(
@moto.mock_s3
def test_post_to_check_should_redirect_to_job(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
response = client.post(url_for('main.checksms',
def test_should_redirect_to_job(app_, db_, db_session, mocker, active_user):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
response = client.post(url_for('main.check_sms',
service_id=123,
upload_id='someid'))
assert response.status_code == 302

4
tests/app/main/views/test_styleguide.py
View File

@@ -1,4 +1,4 @@
def test_styleguide_can_render(notifications_admin):
response = notifications_admin.test_client().get('/_styleguide')
def test_styleguide_can_render(app_):
response = app_.test_client().get('/_styleguide')
assert response.status_code == 200

46
tests/app/main/views/test_templates.py
View File

@@ -1,54 +1,48 @@
from flask import url_for
from tests.app.main import create_test_user
def test_should_return_list_of_all_templates(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
def test_should_return_list_of_all_templates(app_, db_, db_session, active_user):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
response = client.get(url_for('.manage_templates', service_id=123))
assert response.status_code == 200
def test_should_show_page_for_one_template(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
def test_should_show_page_for_one_templates(app_, db_, db_session, active_user):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
response = client.get(url_for('.edit_template', service_id=123, template_id=1))
assert response.status_code == 200
def test_should_redirect_when_saving_a_template(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
def test_should_redirect_when_saving_a_template(app_, db_, db_session, active_user):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
response = client.post(url_for('.edit_template', service_id=123, template_id=1))
assert response.status_code == 302
assert response.location == url_for('.manage_templates', service_id=123, _external=True)
def test_should_show_delete_template_page(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
def test_should_show_delete_template_page(app_, db_, db_session, active_user):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
response = client.get(url_for('.delete_template', service_id=123, template_id=1))
assert response.status_code == 200
assert 'Are you sure' in response.get_data(as_text=True)
def test_should_redirect_when_deleting_a_template(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
def test_should_redirect_when_deleting_a_template(app_, db_, db_session, active_user):
with app_.test_request_context():
with app_.test_client() as client:
client.login(active_user)
response = client.post(url_for('.delete_template', service_id=123, template_id=1))
assert response.status_code == 302

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

@@ -1,12 +1,12 @@
from flask import json, url_for
from app.main.dao import verify_codes_dao
from tests.app.main import create_test_user
from tests import create_test_user
def test_should_render_two_factor_page(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_render_two_factor_page(app_, db_, db_session):
with app_.test_request_context():
with app_.test_client() as client:
# TODO this lives here until we work out how to
# reassign the session after it is lost mid register process
with client.session_transaction() as session:
@@ -17,9 +17,9 @@ def test_should_render_two_factor_page(notifications_admin, notifications_admin_
assert '''We've sent you a text message with a verification code.''' in response.get_data(as_text=True)
def test_should_login_user_and_redirect_to_dashboard(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_login_user_and_redirect_to_dashboard(app_, db_, db_session):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
user = create_test_user('active')
session['user_email'] = user.email_address
@@ -31,11 +31,11 @@ def test_should_login_user_and_redirect_to_dashboard(notifications_admin, notifi
assert response.location == url_for('main.choose_service', _external=True)
def test_should_return_200_with_sms_code_error_when_sms_code_is_wrong(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_return_200_with_sms_code_error_when_sms_code_is_wrong(app_,
db_,
db_session):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
user = create_test_user('active')
session['user_email'] = user.email_address
@@ -46,11 +46,11 @@ def test_should_return_200_with_sms_code_error_when_sms_code_is_wrong(notificati
assert 'Code does not match' in response.get_data(as_text=True)
def test_should_login_user_when_multiple_valid_codes_exist(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_login_user_when_multiple_valid_codes_exist(app_,
db_,
db_session):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
user = create_test_user('active')
session['user_email'] = user.email_address

68
tests/app/main/views/test_user_profile.py
View File

@@ -1,121 +1,121 @@
def test_should_show_overview_page(notifications_admin):
response = notifications_admin.test_client().get('/user-profile')
def test_should_show_overview_page(app_):
response = app_.test_client().get('/user-profile')
assert 'Your profile' in response.get_data(as_text=True)
assert response.status_code == 200
def test_should_show_name_page(notifications_admin):
response = notifications_admin.test_client().get('/user-profile/name')
def test_should_show_name_page(app_):
response = app_.test_client().get('/user-profile/name')
assert 'Change your name' in response.get_data(as_text=True)
assert response.status_code == 200
def test_should_redirect_after_name_change(notifications_admin):
response = notifications_admin.test_client().post('/user-profile/name')
def test_should_redirect_after_name_change(app_):
response = app_.test_client().post('/user-profile/name')
assert response.status_code == 302
assert response.location == 'http://localhost/user-profile'
def test_should_show_email_page(notifications_admin):
response = notifications_admin.test_client().get('/user-profile/email')
def test_should_show_email_page(app_):
response = app_.test_client().get('/user-profile/email')
assert 'Change your email address' in response.get_data(as_text=True)
assert response.status_code == 200
def test_should_redirect_after_email_change(notifications_admin):
response = notifications_admin.test_client().post('/user-profile/email')
def test_should_redirect_after_email_change(app_):
response = app_.test_client().post('/user-profile/email')
assert response.status_code == 302
assert response.location == 'http://localhost/user-profile/email/authenticate'
def test_should_show_authenticate_after_email_change(notifications_admin):
response = notifications_admin.test_client().get('/user-profile/email/authenticate')
def test_should_show_authenticate_after_email_change(app_):
response = app_.test_client().get('/user-profile/email/authenticate')
assert 'Change your email address' in response.get_data(as_text=True)
assert 'Confirm' in response.get_data(as_text=True)
assert response.status_code == 200
def test_should_redirect_after_email_change_confirm(notifications_admin):
response = notifications_admin.test_client().post('/user-profile/email/authenticate')
def test_should_redirect_after_email_change_confirm(app_):
response = app_.test_client().post('/user-profile/email/authenticate')
assert response.status_code == 302
assert response.location == 'http://localhost/user-profile/email/confirm'
def test_should_show_confirm_after_email_change(notifications_admin):
response = notifications_admin.test_client().get('/user-profile/email/confirm')
def test_should_show_confirm_after_email_change(app_):
response = app_.test_client().get('/user-profile/email/confirm')
assert 'Change your email address' in response.get_data(as_text=True)
assert 'Confirm' in response.get_data(as_text=True)
assert response.status_code == 200
def test_should_redirect_after_email_change_confirm(notifications_admin):
response = notifications_admin.test_client().post('/user-profile/email/confirm')
def test_should_redirect_after_email_change_confirm(app_):
response = app_.test_client().post('/user-profile/email/confirm')
assert response.status_code == 302
assert response.location == 'http://localhost/user-profile'
def test_should_show_mobile_number_page(notifications_admin):
response = notifications_admin.test_client().get('/user-profile/mobile-number')
def test_should_show_mobile_number_page(app_):
response = app_.test_client().get('/user-profile/mobile-number')
assert 'Change your mobile number' in response.get_data(as_text=True)
assert response.status_code == 200
def test_should_redirect_after_mobile_number_change(notifications_admin):
response = notifications_admin.test_client().post('/user-profile/email')
def test_should_redirect_after_mobile_number_change(app_):
response = app_.test_client().post('/user-profile/email')
assert response.status_code == 302
assert response.location == 'http://localhost/user-profile/email/authenticate'
def test_should_show_authenticate_after_mobile_number_change(notifications_admin):
response = notifications_admin.test_client().get('/user-profile/mobile-number/authenticate')
def test_should_show_authenticate_after_mobile_number_change(app_):
response = app_.test_client().get('/user-profile/mobile-number/authenticate')
assert 'Change your mobile number' in response.get_data(as_text=True)
assert 'Confirm' in response.get_data(as_text=True)
assert response.status_code == 200
def test_should_redirect_after_mobile_number_authenticate(notifications_admin):
response = notifications_admin.test_client().post('/user-profile/mobile-number/authenticate')
def test_should_redirect_after_mobile_number_authenticate(app_):
response = app_.test_client().post('/user-profile/mobile-number/authenticate')
assert response.status_code == 302
assert response.location == 'http://localhost/user-profile/mobile-number/confirm'
def test_should_show_confirm_after_mobile_number_change(notifications_admin):
response = notifications_admin.test_client().get('/user-profile/mobile-number/confirm')
def test_should_show_confirm_after_mobile_number_change(app_):
response = app_.test_client().get('/user-profile/mobile-number/confirm')
assert 'Change your mobile number' in response.get_data(as_text=True)
assert 'Confirm' in response.get_data(as_text=True)
assert response.status_code == 200
def test_should_redirect_after_mobile_number_confirm(notifications_admin):
response = notifications_admin.test_client().post('/user-profile/mobile-number/confirm')
def test_should_redirect_after_mobile_number_confirm(app_):
response = app_.test_client().post('/user-profile/mobile-number/confirm')
assert response.status_code == 302
assert response.location == 'http://localhost/user-profile'
def test_should_show_password_page(notifications_admin):
response = notifications_admin.test_client().get('/user-profile/password')
def test_should_show_password_page(app_):
response = app_.test_client().get('/user-profile/password')
assert 'Change your password' in response.get_data(as_text=True)
assert response.status_code == 200
def test_should_redirect_after_password_change(notifications_admin):
response = notifications_admin.test_client().post('/user-profile/password')
def test_should_redirect_after_password_change(app_):
response = app_.test_client().post('/user-profile/password')
assert response.status_code == 302
assert response.location == 'http://localhost/user-profile'

40
tests/app/main/views/test_verify.py
View File

@@ -1,11 +1,11 @@
from flask import json, url_for
from app.main.dao import users_dao, verify_codes_dao
from tests.app.main import create_test_user
from tests import create_test_user
def test_should_return_verify_template(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_return_verify_template(app_, db_, db_session):
with app_.test_request_context():
with app_.test_client() as client:
# TODO this lives here until we work out how to
# reassign the session after it is lost mid register process
with client.session_transaction() as session:
@@ -18,11 +18,11 @@ def test_should_return_verify_template(notifications_admin, notifications_admin_
" You need to enter both codes here.") in response.get_data(as_text=True)
def test_should_redirect_to_add_service_when_code_are_correct(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_redirect_to_add_service_when_code_are_correct(app_,
db_,
db_session):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
user = create_test_user('pending')
session['user_email'] = user.email_address
@@ -35,9 +35,9 @@ def test_should_redirect_to_add_service_when_code_are_correct(notifications_admi
assert response.location == url_for('main.add_service', first='first', _external=True)
def test_should_activate_user_after_verify(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_activate_user_after_verify(app_, db_, db_session):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
user = create_test_user('pending')
session['user_email'] = user.email_address
@@ -51,9 +51,9 @@ def test_should_activate_user_after_verify(notifications_admin, notifications_ad
assert after_verify.state == 'active'
def test_should_return_200_when_codes_are_wrong(notifications_admin, notifications_admin_db, notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_return_200_when_codes_are_wrong(app_, db_, db_session):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
user = create_test_user('pending')
session['user_email'] = user.email_address
@@ -68,11 +68,11 @@ def test_should_return_200_when_codes_are_wrong(notifications_admin, notificatio
assert resp_data.count('Code does not match') == 2
def test_should_mark_all_codes_as_used_when_many_codes_exist(notifications_admin,
notifications_admin_db,
notify_db_session):
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
def test_should_mark_all_codes_as_used_when_many_codes_exist(app_,
db_,
db_session):
with app_.test_request_context():
with app_.test_client() as client:
with client.session_transaction() as session:
user = create_test_user('pending')
session['user_email'] = user.email_address

119
tests/conftest.py
View File

@@ -1,34 +1,18 @@
import os
import pytest
from alembic.command import upgrade
from alembic.config import Config
from flask import url_for
from flask.ext.migrate import Migrate, MigrateCommand
from flask.ext.script import Manager
from flask_login import login_user
from sqlalchemy.schema import MetaData
from flask.testing import FlaskClient
from app.main.dao import verify_codes_dao
from . import (
create_test_user, create_another_test_user, service_json, TestClient,
get_test_user)
from app import create_app, db
class TestClient(FlaskClient):
def login(self, user):
# Skipping authentication here and just log them in
with self.session_transaction() as session:
session['user_email'] = user.email_address
verify_codes_dao.add_code(user_id=user.id, code='12345', code_type='sms')
response = self.post(
url_for('main.two_factor'), data={'sms_code': '12345'})
def logout(self, user):
self.get(url_for("main.logout"))
@pytest.fixture(scope='session')
def notifications_admin(request):
def app_(request):
app = create_app('test')
ctx = app.app_context()
@@ -43,28 +27,28 @@ def notifications_admin(request):
@pytest.fixture(scope='session')
def notifications_admin_db(notifications_admin, request):
Migrate(notifications_admin, db)
def db_(app_, request):
Migrate(app_, db)
Manager(db, MigrateCommand)
BASE_DIR = os.path.dirname(os.path.dirname(__file__))
ALEMBIC_CONFIG = os.path.join(BASE_DIR, 'migrations')
config = Config(ALEMBIC_CONFIG + '/alembic.ini')
config.set_main_option("script_location", ALEMBIC_CONFIG)
with notifications_admin.app_context():
with app_.app_context():
upgrade(config, 'head')
def teardown():
db.session.remove()
db.drop_all()
db.engine.execute("drop table alembic_version")
db.get_engine(notifications_admin).dispose()
db.get_engine(app_).dispose()
request.addfinalizer(teardown)
@pytest.fixture(scope='function')
def notify_db_session(request):
def db_session(request):
def teardown():
db.session.remove()
for tbl in reversed(meta.sorted_tables):
@@ -73,3 +57,88 @@ def notify_db_session(request):
meta = MetaData(bind=db.engine, reflect=True)
request.addfinalizer(teardown)
@pytest.fixture(scope='function')
def service_one(request, active_user):
return service_json(1, 'service one', [active_user.id])
@pytest.fixture(scope='function')
def active_user(request, db_, db_session):
usr = get_test_user()
if usr:
return usr
return create_test_user('active')
@pytest.fixture(scope='function')
def mock_send_sms(request, mocker):
return mocker.patch("app.notifications_api_client.send_sms")
@pytest.fixture(scope='function')
def mock_send_email(request, mocker):
return mocker.patch("app.notifications_api_client.send_email")
@pytest.fixture(scope='function')
def mock_get_service(mocker, active_user):
def _create(service_id):
service = service_json(
service_id, "Test Service", [active_user.id], limit=1000,
active=False, restricted=True)
return {'data': service, 'token': 1}
return mocker.patch('app.notifications_api_client.get_service', side_effect=_create)
@pytest.fixture(scope='function')
def mock_create_service(mocker):
# TODO fix token generation
def _create(service_name, active, limit, restricted, user_id):
service = service_json(
101, service_name, [user_id], limit=limit,
active=active, restricted=restricted)
return {'data': service, 'token': 1}
mock_class = mocker.patch(
'app.notifications_api_client.create_service', side_effect=_create)
return mock_class
@pytest.fixture(scope='function')
def mock_update_service(mocker):
def _update(service_id,
service_name,
active,
limit,
restricted,
users):
service = service_json(
service_id, service_name, users, limit=limit,
active=active, restricted=restricted)
return {'data': service}
mock_class = mocker.patch(
'app.notifications_api_client.update_service', side_effect=_update)
return mock_class
@pytest.fixture(scope='function')
def mock_get_services(mocker, active_user):
def _create():
service_one = service_json(
1, "service_one", [active_user.id], 1000, True, False)
service_two = service_json(
2, "service_two", [active_user.id], 1000, True, False)
return {'data': [service_one, service_two]}
mock_class = mocker.patch(
'app.notifications_api_client.get_services', side_effect=_create)
return mock_class
@pytest.fixture(scope='function')
def mock_delete_service(mocker, mock_get_service):
def _delete(service_id):
return mock_get_service.side_effect(service_id)
mock_class = mocker.patch(
'app.notifications_api_client.delete_service', side_effect=_delete)
return mock_class