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

Merge pull request #366 from alphagov/email_from_uniqueness

Browse Source 
Email from uniqueness
This commit is contained in:
Rebecca Law
2016-03-31 16:03:40 +01:00
parent 51baa4a210 6394408228
commit 148783023c
11 changed files with 155 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
app/main/forms.py
View File

@@ -1,13 +1,14 @@
import re
from flask_wtf import Form
from utils.recipients import (
validate_phone_number,
InvalidPhoneError
)
from wtforms import (
StringField,
PasswordField,
ValidationError,
TextAreaField,
FileField,
RadioField,
BooleanField,
HiddenField
)
@@ -16,12 +17,6 @@ from wtforms.validators import (DataRequired, Email, Length, Regexp)
from app.main.validators import (Blacklist, CsvFileValidator, ValidEmailDomainRegex)
from utils.recipients import (
validate_phone_number,
format_phone_number,
InvalidPhoneError
)
def email_address(label='Email address'):
return EmailField(label, validators=[
@@ -159,7 +154,9 @@ class AddServiceForm(Form):
)
def validate_name(self, a):
if a.data in self._names_func():
from app.utils import email_safe
# make sure the email_from will be unique to all services
if email_safe(a.data) in self._names_func():
raise ValidationError('This service name is already in use')
@@ -180,7 +177,9 @@ class ServiceNameForm(Form):
])
def validate_name(self, a):
if a.data in self._names_func():
from app.utils import email_safe
# make sure the email_from will be unique to all services
if email_safe(a.data) in self._names_func():
raise ValidationError('This service name is already in use')

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

@@ -16,6 +16,7 @@ from app import (
user_api_client,
service_api_client
)
from app.utils import email_safe
@main.route("/add-service", methods=['GET', 'POST'])
@@ -31,12 +32,17 @@ def add_service():
invite_api_client.accept_invite(service_id, invitation.id)
return redirect(url_for('main.service_dashboard', service_id=service_id))
form = AddServiceForm(service_api_client.find_all_service_names)
form = AddServiceForm(service_api_client.find_all_service_email_from)
heading = 'Which service do you want to set up notifications for?'
if form.validate_on_submit():
session['service_name'] = form.name.data
service_id = service_api_client.create_service(
session['service_name'], False, current_app.config['DEFAULT_SERVICE_LIMIT'], True, session['user_id'])
email_from = email_safe(session['service_name'])
service_id = service_api_client.create_service(service_name=session['service_name'],
active=False,
limit=current_app.config['DEFAULT_SERVICE_LIMIT'],
restricted=True,
user_id=session['user_id'],
email_from=email_from)
return redirect(url_for('main.service_dashboard', service_id=service_id))
else:

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

@@ -15,7 +15,7 @@ from notifications_python_client import HTTPError
from app import service_api_client
from app.main import main
from app.utils import user_has_permissions
from app.utils import user_has_permissions, email_safe
from app.main.forms import ConfirmPasswordForm, ServiceNameForm
from app import user_api_client
@@ -39,7 +39,7 @@ def service_settings(service_id):
def service_name_change(service_id):
service = service_api_client.get_service(service_id)['data']
form = ServiceNameForm(service_api_client.find_all_service_names)
form = ServiceNameForm(service_api_client.find_all_service_email_from)
if form.validate_on_submit():
session['service_name_change'] = form.name.data
@@ -65,6 +65,7 @@ def service_name_change_confirm(service_id):
if form.validate_on_submit():
service['name'] = session['service_name_change']
service['email_from'] = email_safe(session['service_name_change'])
try:
service_api_client.update_service(
service['id'],
@@ -72,7 +73,8 @@ def service_name_change_confirm(service_id):
service['active'],
service['limit'],
service['restricted'],
service['users'])
service['users'],
service['email_from'])
except HTTPError as e:
error_msg = "Duplicate service name '{}'".format(session['service_name_change'])
if e.status_code == 400 and error_msg in e.message['name']:
@@ -144,7 +146,8 @@ def service_status_change_confirm(service_id):
service['active'],
service['limit'],
service['restricted'],
service['users'])
service['users'],
service['email_from'])
return redirect(url_for('.service_settings', service_id=service_id))
return render_template(
'views/service-settings/confirm.html',

15
app/notify_client/api_client.py
View File

@@ -18,7 +18,7 @@ class ServiceAPIClient(NotificationsAPIClient):
self.client_id = application.config['ADMIN_CLIENT_USER_NAME']
self.secret = application.config['ADMIN_CLIENT_SECRET']
def create_service(self, service_name, active, limit, restricted, user_id):
def create_service(self, service_name, active, limit, restricted, user_id, email_from):
"""
Create a service and return the json.
"""
@@ -27,7 +27,8 @@ class ServiceAPIClient(NotificationsAPIClient):
"active": active,
"limit": limit,
"user_id": user_id,
"restricted": restricted
"restricted": restricted,
"email_from": email_from
}
return self.post("/service", data)['data']['id']
@@ -57,7 +58,8 @@ class ServiceAPIClient(NotificationsAPIClient):
active,
limit,
restricted,
users):
users,
email_from):
"""
Update a service.
"""
@@ -67,7 +69,8 @@ class ServiceAPIClient(NotificationsAPIClient):
"active": active,
"limit": limit,
"restricted": restricted,
"users": users
"users": users,
"email_from": email_from
}
endpoint = "/service/{0}".format(service_id)
return self.post(endpoint, data)
@@ -140,9 +143,9 @@ class ServiceAPIClient(NotificationsAPIClient):
endpoint = "/service/{0}/template/{1}".format(service_id, template_id)
return self.delete(endpoint)
def find_all_service_names(self, user_id=None):
def find_all_service_email_from(self, user_id=None):
resp = self.get_services(user_id)
return [x['name'] for x in resp['data']]
return [x['email_from'] for x in resp['data']]
class ServicesBrowsableItem(BrowsableItem):

6
app/utils.py
View File

@@ -101,3 +101,9 @@ def generate_previous_next_dict(view, view_dict, page, title, label):
'title': title,
'label': label
}
def email_safe(string):
return "".join([
character.lower() if character.isalnum() or character == "." else "" for character in re.sub("\s+", ".", string.strip()) # noqa
])

8
tests/__init__.py
View File

@@ -10,8 +10,9 @@ class TestClient(FlaskClient):
with self.session_transaction() as session:
session['user_id'] = user.id
session['_fresh'] = True
if mocker and service:
if mocker:
mocker.patch('app.user_api_client.get_user', return_value=user)
if mocker and service:
mocker.patch('app.service_api_client.get_service', return_value={'data': service})
login_user(user, remember=True)
@@ -22,14 +23,15 @@ class TestClient(FlaskClient):
self.get(url_for("main.logout"))
def service_json(id_, name, users, limit=1000, active=False, restricted=True):
def service_json(id_, name, users, limit=1000, active=False, restricted=True, email_from=None):
return {
'id': id_,
'name': name,
'users': users,
'limit': limit,
'active': active,
'restricted': restricted
'restricted': restricted,
'email_from': email_from
}

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

@@ -4,7 +4,7 @@ from werkzeug.datastructures import MultiDict
def test_form_should_have_errors_when_duplicate_service_is_added(app_):
def _get_form_names():
return ['some service', 'more names']
return ['some.service', 'more.names']
with app_.test_request_context():
form = AddServiceForm(_get_form_names,
formdata=MultiDict([('name', 'some service')]))

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

@@ -1,64 +1,65 @@
from flask import url_for
import app
def test_get_should_render_add_service_template(app_,
api_user_active,
mock_login,
mock_get_service,
mock_get_services,
mock_get_user_by_email):
mocker):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
client.login(api_user_active, mocker)
response = client.get(url_for('main.add_service'))
assert response.status_code == 200
assert 'Which service do you want to set up notifications for?' in response.get_data(as_text=True)
def test_should_add_service_and_redirect_to_next_page(app_,
mock_login,
mocker,
mock_create_service,
mock_get_services,
api_user_active):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
client.login(api_user_active, mocker)
response = client.post(
url_for('main.add_service'),
data={'name': 'testing the post'})
assert response.status_code == 302
assert response.location == url_for('main.service_dashboard', service_id=101, _external=True)
mock_create_service.asset_called_once_with('testing the post', False,
app_.config['DEFAULT_SERVICE_LIMIT'],
True, api_user_active.id)
assert mock_get_services.called
mock_create_service.asset_called_once_with(service_name='testing the post',
active=False,
limit=app_.config['DEFAULT_SERVICE_LIMIT'],
restricted=True,
user_id=api_user_active.id,
email_from='testing.the.post')
def test_should_return_form_errors_when_service_name_is_empty(app_,
api_user_active,
mock_get_service,
mock_get_services,
mock_get_user,
mock_get_user_by_email,
mock_login):
mocker,
api_user_active):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
client.login(api_user_active, mocker)
response = client.post(url_for('main.add_service'), data={})
assert response.status_code == 200
assert 'Service name can’t be empty' in response.get_data(as_text=True)
def test_should_return_form_errors_with_duplicate_service_name(app_,
mock_login,
mock_get_services,
mock_get_user,
api_user_active,
mock_get_user_by_email):
def test_should_return_form_errors_with_duplicate_service_name_regardless_of_case(app_,
mocker,
service_one,
api_user_active,
mock_create_service):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
response = client.post(
url_for('main.add_service'), data={'name': 'service_one'})
client.login(api_user_active, mocker, service_one)
mocker.patch('app.service_api_client.find_all_service_email_from',
return_value=['service_one', 'service.two'])
response = client.post(url_for('main.add_service'), data={'name': 'SERVICE TWO'})
assert response.status_code == 200
assert 'This service name is already in use' in response.get_data(as_text=True)
assert mock_get_services.called
app.service_api_client.find_all_service_email_from.assert_called_once_with()
assert not mock_create_service.called

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

@@ -1,124 +1,107 @@
from flask import url_for
import app
from app.utils import email_safe
from tests import validate_route_permission
from bs4 import BeautifulSoup
def test_should_show_overview(app_,
api_user_active,
mock_get_service,
mock_get_user,
mock_get_user_by_email,
mock_login,
mock_has_permissions):
active_user_with_permissions,
mocker,
service_one):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
service_id = 123
client.login(active_user_with_permissions, mocker, service_one)
response = client.get(url_for(
'main.service_settings', service_id=service_id))
'main.service_settings', service_id=service_one['id']))
assert response.status_code == 200
resp_data = response.get_data(as_text=True)
assert 'Service settings' in resp_data
service = mock_get_service.side_effect(service_id)['data']
assert mock_get_service.called
app.service_api_client.get_service.assert_called_with(service_one['id'])
def test_should_show_service_name(app_,
api_user_active,
mock_get_service,
mock_get_user,
mock_get_user_by_email,
mock_login,
mock_has_permissions):
active_user_with_permissions,
mocker,
service_one):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
service_id = 123
client.login(active_user_with_permissions, mocker, service_one)
response = client.get(url_for(
'main.service_name_change', service_id=service_id))
'main.service_name_change', service_id=service_one['id']))
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
service = mock_get_service.side_effect(service_id)['data']
app.service_api_client.get_service.assert_called_with(service_one['id'])
def test_should_redirect_after_change_service_name(app_,
api_user_active,
mock_get_service,
mock_get_user,
mock_get_user_by_email,
mock_login,
mock_has_permissions):
active_user_with_permissions,
service_one,
mocker,
mock_get_services):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
service_id = 123
client.login(active_user_with_permissions, mocker, service_one)
response = client.post(
url_for('main.service_name_change', service_id=service_id),
url_for('main.service_name_change', service_id=service_one['id']),
data={'name': "new name"})
assert response.status_code == 302
settings_url = url_for(
'main.service_name_change_confirm', service_id=service_id, _external=True)
'main.service_name_change_confirm', service_id=service_one['id'], _external=True)
assert settings_url == response.location
assert mock_get_service.called
assert mock_get_services.called
def test_should_not_allow_duplicate_names(app_,
api_user_active,
mock_get_service,
mock_get_user,
mock_get_user_by_email,
mock_login,
mock_has_permissions,
mock_get_services):
active_user_with_permissions,
mocker,
service_one):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
service_id = 123
client.login(active_user_with_permissions, mocker, service_one)
mocker.patch('app.service_api_client.find_all_service_email_from',
return_value=['service_one', 'service.two'])
service_id = service_one['id']
response = client.post(
url_for('main.service_name_change', service_id=service_id),
data={'name': "service_one"})
data={'name': "SErvICE TWO"})
assert response.status_code == 200
resp_data = response.get_data(as_text=True)
assert 'This service name is already in use' in resp_data
app.service_api_client.find_all_service_email_from.assert_called_once_with()
def test_should_show_service_name_confirmation(app_,
api_user_active,
mock_get_service,
mock_get_user,
mock_get_user_by_email,
mock_login,
mock_has_permissions):
active_user_with_permissions,
mocker,
service_one):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
service_id = 123
client.login(active_user_with_permissions, mocker, service_one)
response = client.get(url_for(
'main.service_name_change_confirm', service_id=service_id))
'main.service_name_change_confirm', service_id=service_one['id']))
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
app.service_api_client.get_service.assert_called_with(service_one['id'])
def test_should_redirect_after_service_name_confirmation(app_,
api_user_active,
mock_get_service,
active_user_with_permissions,
service_one,
mocker,
mock_update_service,
mock_get_user,
mock_get_user_by_email,
mock_login,
mock_verify_password,
mock_has_permissions):
mock_verify_password):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
service_id = 123
client.login(active_user_with_permissions, mocker, service_one)
service_id = service_one['id']
service_new_name = 'New Name'
with client.session_transaction() as session:
session['service_name_change'] = service_new_name
@@ -126,38 +109,40 @@ def test_should_redirect_after_service_name_confirmation(app_,
'main.service_name_change_confirm', service_id=service_id))
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)
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
mock_update_service.assert_called_once_with(service_id,
service_new_name,
service_one['active'],
service_one['limit'],
service_one['restricted'],
service_one['users'],
email_safe(service_new_name))
assert mock_verify_password.called
def test_should_raise_duplicate_name_handled(app_,
api_user_active,
mock_get_service,
active_user_with_permissions,
service_one,
mocker,
mock_get_services,
mock_update_service_raise_httperror_duplicate_name,
mock_get_user,
mock_get_user_by_email,
mock_login,
mock_verify_password,
mock_has_permissions):
mock_verify_password):
with app_.test_request_context():
with app_.test_client() as client:
client.login(api_user_active)
service_id = 123
client.login(active_user_with_permissions, mocker, service_one)
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))
'main.service_name_change_confirm', service_id=service_one['id']))
assert response.status_code == 302
name_change_url = url_for(
'main.service_name_change', service_id=service_id, _external=True)
resp_data = response.get_data(as_text=True)
'main.service_name_change', service_id=service_one['id'], _external=True)
assert name_change_url == response.location
assert mock_update_service_raise_httperror_duplicate_name.called
assert mock_verify_password.called
def test_should_show_request_to_go_live(app_,

8
tests/app/test_utils.py Normal file
View File

@@ -0,0 +1,8 @@
from app.utils import email_safe
def test_email_safe_return_dot_separated_email_domain():
test_name = 'SOME service with+stuff+ b123'
expected = 'some.service.withstuff.b123'
actual = email_safe(test_name)
assert actual == expected

14
tests/conftest.py
View File

@@ -60,10 +60,10 @@ def mock_get_service(mocker, api_user_active):
@pytest.fixture(scope='function')
def mock_create_service(mocker):
def _create(service_name, active, limit, restricted, user_id):
def _create(service_name, active, limit, restricted, user_id, email_from):
service = service_json(
101, service_name, [user_id], limit=limit,
active=active, restricted=restricted)
active=active, restricted=restricted, email_from=email_from)
return service['id']
return mocker.patch(
@@ -77,14 +77,15 @@ def mock_update_service(mocker):
active,
limit,
restricted,
users):
users,
email_from):
service = service_json(
service_id, service_name, users, limit=limit,
active=active, restricted=restricted)
active=active, restricted=restricted, email_from=email_from)
return {'data': service}
return mocker.patch(
'app.service_api_client.update_service', side_effect=_update)
'app.service_api_client.update_service', side_effect=_update, autospec=True)
@pytest.fixture(scope='function')
@@ -94,7 +95,8 @@ def mock_update_service_raise_httperror_duplicate_name(mocker):
active,
limit,
restricted,
users):
users,
email_from):
json_mock = Mock(return_value={'message': {'name': ["Duplicate service name '{}'".format(service_name)]}})
resp_mock = Mock(status_code=400, json=json_mock)
http_error = HTTPError(response=resp_mock, message="Default message")