darkhelm/notifications-admin
1
0
Fork 0
mirror of https://github.com/GSA/notifications-admin.git synced 2026-02-06 11:23:48 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #61 from alphagov/csv-upload

Browse Source 
First slice of csv upload of phone numbers for sending messages.
This commit is contained in:
Rebecca Law
2016-01-12 10:50:25 +00:00
parent 23c4cbbf76 bfd7291426
commit 7190513dc2
7 changed files with 237 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
app/main/forms.py
View File

@@ -1,7 +1,17 @@
from flask_wtf import Form
from wtforms import StringField, PasswordField, ValidationError, TextAreaField
from wtforms import (
StringField,
PasswordField,
ValidationError,
TextAreaField,
FileField
)
from wtforms.validators import DataRequired, Email, Length, Regexp
from app.main.validators import Blacklist, ValidateUserCodes
from app.main.validators import Blacklist, ValidateUserCodes, CsvFileValidator
from app.main.dao import verify_codes_dao
from app.main.encryption import check_hash
def email_address():
@@ -140,3 +150,8 @@ class ForgotPasswordForm(Form):
class NewPasswordForm(Form):
new_password = password()
class CsvUploadForm(Form):
file = FileField('File to upload', validators=[DataRequired(
message='Please pick a file'), CsvFileValidator()])

10
app/main/validators.py
View File

@@ -38,3 +38,13 @@ class ValidateUserCodes(object):
break
if not valid_code:
raise ValidationError(self.invalid_msg)
class CsvFileValidator(object):
def __init__(self, message='Not a csv file'):
self.message = message
def __call__(self, form, field):
if not form.file.data.mimetype == 'text/csv':
raise ValidationError(self.message)

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

@@ -1,7 +1,20 @@
from flask import request, render_template, redirect, url_for
import csv
import re
from flask import (
request,
render_template,
redirect,
url_for,
session,
flash,
current_app
)
from flask_login import login_required
from app.main import main
from app.main.forms import CsvUploadForm
# TODO move this to the templates directory
message_templates = [
@@ -23,45 +36,64 @@ message_templates = [
@main.route("/sms/send", methods=['GET', 'POST'])
@login_required
def sendsms():
if request.method == 'GET':
return render_template(
'views/send-sms.html',
message_templates=message_templates
)
elif request.method == 'POST':
return redirect(url_for('.checksms'))
form = CsvUploadForm()
if form.validate_on_submit():
csv_file = form.file.data
# in memory handing is temporary until next story to save csv file
try:
results = _build_upload_result(csv_file)
except Exception:
message = 'There was a problem with the file: {}'.format(
csv_file.filename)
flash(message)
return redirect(url_for('.sendsms'))
if not results['valid'] and not results['rejects']:
message = "The file {} contained no data".format(csv_file.filename)
flash(message, 'error')
return redirect(url_for('.sendsms'))
session[csv_file.filename] = results
return redirect(url_for('.checksms', recipients=csv_file.filename))
return render_template('views/send-sms.html',
message_templates=message_templates,
form=form)
@main.route("/sms/check", methods=['GET', 'POST'])
@login_required
def checksms():
recipients = [
{'phone': "+44 7700 900989", 'registration number': 'LC12 BFL', 'date': '24 December 2015'},
{'phone': "+44 7700 900479", 'registration number': 'DU04 AOM', 'date': '25 December 2015'},
{'phone': "+44 7700 900964", 'registration number': 'M91 MJB', 'date': '26 December 2015'},
{'phone': "+44 7700 900703", 'registration number': 'Y249 NPU', 'date': '31 December 2015'},
{'phone': "+44 7700 900730", 'registration number': 'LG55 UGB', 'date': '1 January 2016'},
{'phone': "+44 7700 900989", 'registration number': 'LC12 BFL', 'date': '24 December 2015'},
{'phone': "+44 7700 900479", 'registration number': 'DU04 AOM', 'date': '25 December 2015'},
{'phone': "+44 7700 900964", 'registration number': 'M91 MJB', 'date': '26 December 2015'},
{'phone': "+44 7700 900703", 'registration number': 'Y249 NPU', 'date': '31 December 2015'},
{'phone': "+44 7700 900730", 'registration number': 'LG55 UGB', 'date': '1 January 2016'},
]
number_of_recipients = len(recipients)
too_many_recipients_to_display = number_of_recipients > 7
if request.method == 'GET':
recipients_filename = request.args.get('recipients')
# upload results in session until file is persisted in next story
upload_result = session.get(recipients_filename)
if upload_result.get('rejects'):
flash('There was a problem with some of the numbers')
return render_template(
'views/check-sms.html',
number_of_recipients=number_of_recipients,
recipients={
"first_three": recipients[:3] if too_many_recipients_to_display else [],
"last_three": recipients[number_of_recipients - 3:] if too_many_recipients_to_display else [],
"all": recipients if not too_many_recipients_to_display else []
},
upload_result=upload_result,
message_template=message_templates[0]['body']
)
elif request.method == 'POST':
return redirect(url_for('.showjob'))
def _build_upload_result(csv_file):
pattern = re.compile(r'^\+44\s?\d{4}\s?\d{6}$')
reader = csv.DictReader(
csv_file.read().decode('utf-8').splitlines(),
lineterminator='\n',
quoting=csv.QUOTE_NONE)
valid, rejects = [], []
for i, row in enumerate(reader):
if pattern.match(row['phone']):
valid.append(row)
else:
rejects.append({"line_number": i+2, "phone": row['phone']})
return {"valid": valid, "rejects": rejects}

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

@@ -10,57 +10,55 @@
{% block maincolumn_content %}
<h1 class="heading-xlarge">Send text messages</h1>
<h1 class="heading-xlarge">Send text messages</h1>
{% if upload_result.rejects %}
<h3 class="heading-small">The following numbers are invalid</h3>
{% for rejected in upload_result.rejects %}
<p>Line {{rejected.line_number}}: {{rejected.phone }}</a>
{% endfor %}
<p><a href="{{url_for('.sendsms')}}" class="button">Go back and resolve errors</a></p>
<h2 class="heading-medium">Check and confirm</h2>
{% else %}
<form method="POST" enctype="multipart/form-data">
<h2 class="heading-medium">Check and confirm</h2>
<form method="POST" enctype="multipart/form-data">
{{ page_footer(
"Send {} text messages".format(number_of_recipients)
) }}
<h3 class="heading-small">First 3 messages</h2>
{% if recipients.first_three and recipients.last_three %}
{% for recipient in recipients.first_three %}
{{ sms_message(
message_template|replace_placeholders(recipient),
"{}".format(recipient['phone'])
) }}
{% endfor %}
<h3 class="heading-small">Last 3 messages</h2>
{% for recipient in recipients.last_three %}
{{ sms_message(
message_template|replace_placeholders(recipient),
"{}".format(recipient['phone'])
) }}
{% endfor %}
{% else %}
{% for recipient in recipients.all %}
{{ sms_message(
message_template|replace_placeholders(recipient),
"{}".format(recipient['phone'])
) }}
{% endfor %}
{% endif %}
<p>
<a href="#">See all</a>
</p>
{{ page_footer(
button_text = "Send {} text messages".format(number_of_recipients),
{{ page_footer(
button_text = "Send {} text messages".format(upload_result.valid|count),
back_link = url_for(".sendsms")
) }}
)}}
</form>
{% if upload_result.valid | count > 6 %}
<h3 class="heading-small">First three message in file</h3>
{% for recipient in upload_result.valid[:3] %}
{{ sms_message(message_template|replace_placeholders(
recipient),
'{}'.format(recipient['phone'])
)}}
{% endfor %}
<h3 class="heading-small">Last three messages in file</h3>
{% for recipient in upload_result.valid[-3:] %}
{{ sms_message(message_template|replace_placeholders(
recipient),
'{}'.format(recipient['phone'])
)}}
{% endfor %}
{% else %}
<h3 class="heading-small">All messages in file</h3>
{% for recipient in upload_result.valid %}
{{ sms_message(message_template|replace_placeholders(
recipient),
'{}'.format(recipient['phone'])
)}}
{% endfor %}
{% endif %}
{{ page_footer(
button_text = "Send {} text messages".format(upload_result.valid|count),
back_link = url_for(".sendsms")
)}}
</form>
{% endif %}
{% endblock %}

3
app/templates/views/send-sms.html
View File

@@ -1,6 +1,7 @@
{% extends "withnav_template.html" %}
{% from "components/sms-message.html" import sms_message %}
{% from "components/page-footer.html" import page_footer %}
{% from "components/textbox.html" import textbox %}
{% block page_title %}
GOV.UK Notify | Send text messages
@@ -37,7 +38,7 @@
You can also <a href="#">download an example CSV</a>.
</p>
<p>
<input type="file" />
{{textbox(form.file)}}
</p>
{{ page_footer("Continue") }}

2
config.py
View File

@@ -31,6 +31,8 @@ class Config(object):
DANGEROUS_SALT = 'itsdangeroussalt'
TOKEN_MAX_AGE_SECONDS = 3600
MAX_CONTENT_LENGTH = 10 * 1024 * 1024 # 10mb
class Development(Config):
DEBUG = True

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

@@ -1,27 +1,108 @@
def test_should_return_sms_template_picker(notifications_admin):
response = notifications_admin.test_client().get('/sms/send')
assert response.status_code == 200
assert 'Choose text message template' in response.get_data(as_text=True)
from io import BytesIO
from tests.app.main import create_test_user
def test_should_redirect_to_sms_check_page(notifications_admin):
response = notifications_admin.test_client().post('/sms/send')
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)
assert response.status_code == 302
assert response.location == 'http://localhost/sms/check'
upload_data = {'file': (BytesIO(''.encode('utf-8')), 'emtpy.csv')}
response = client.post('/sms/send', data=upload_data,
follow_redirects=True)
assert response.status_code == 200
content = response.get_data(as_text=True)
assert 'The file emtpy.csv contained no data' in content
def test_should_return_check_sms_page(notifications_admin):
response = notifications_admin.test_client().get('/sms/check')
def test_upload_csvfile_with_invalid_phone_shows_check_page_with_errors(
notifications_admin, notifications_admin_db, notify_db_session):
assert response.status_code == 200
assert 'Check and confirm' in response.get_data(as_text=True)
assert 'Send 10 text messages' in response.get_data(as_text=True)
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
file_contents = 'phone\n+44 123\n+44 456'.encode('utf-8')
upload_data = {'file': (BytesIO(file_contents), 'invalid.csv')}
response = client.post('/sms/send', data=upload_data,
follow_redirects=True)
assert response.status_code == 200
content = response.get_data(as_text=True)
assert 'There was a problem with some of the numbers' in content
assert 'The following numbers are invalid' in content
assert '+44 123' in content
assert '+44 456' in content
assert 'Go back and resolve errors' in content
def test_should_redirect_to_job(notifications_admin):
response = notifications_admin.test_client().post('/sms/check')
def test_upload_csvfile_with_valid_phone_shows_first3_and_last3_numbers(
notifications_admin, notifications_admin_db, notify_db_session):
assert response.status_code == 302
assert response.location == 'http://localhost/jobs/job'
with notifications_admin.test_request_context():
with notifications_admin.test_client() as client:
user = create_test_user('active')
client.login(user)
file_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'.encode('utf-8') # noqa
upload_data = {'file': (BytesIO(file_contents), 'valid.csv')}
response = client.post('/sms/send', data=upload_data,
follow_redirects=True)
content = response.get_data(as_text=True)
assert response.status_code == 200
assert 'Check and confirm' in content
assert 'First three message in file' in content
assert 'Last three messages in file' in content
assert '+44 7700 900981' in content
assert '+44 7700 900982' in content
assert '+44 7700 900983' in content
assert '+44 7700 900984' not in content
assert '+44 7700 900985' not in content
assert '+44 7700 900986' not in content
assert '+44 7700 900987' in content
assert '+44 7700 900988' in content
assert '+44 7700 900989' in content
def test_upload_csvfile_with_valid_phone_shows_all_if_6_or_less_numbers(
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)
file_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'.encode('utf-8') # noqa
upload_data = {'file': (BytesIO(file_contents), 'valid.csv')}
response = client.post('/sms/send', data=upload_data,
follow_redirects=True)
content = response.get_data(as_text=True)
assert response.status_code == 200
assert 'Check and confirm' in content
assert 'All messages in file' in content
assert '+44 7700 900981' in content
assert '+44 7700 900982' in content
assert '+44 7700 900983' in content
assert '+44 7700 900984' in content
assert '+44 7700 900985' in content
assert '+44 7700 900986' in content
def test_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('/sms/check')
assert response.status_code == 302
assert response.location == 'http://localhost/jobs/job'