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

Merge branch 'master' into gsm

Browse Source
This commit is contained in:
Chris Hill-Scott
2017-02-20 10:13:53 +00:00
committed by GitHub
parent c30b4d00e7 87a794b471
commit 22aeb5455c
11 changed files with 379 additions and 627 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
app/dao/services_dao.py
View File

@@ -224,25 +224,6 @@ def _stats_for_service_query(service_id):
)
@statsd(namespace="dao")
def dao_fetch_weekly_historical_stats_for_service(service_id):
monday_of_notification_week = func.date_trunc('week', NotificationHistory.created_at).label('week_start')
return db.session.query(
NotificationHistory.notification_type,
NotificationHistory.status,
monday_of_notification_week,
func.count(NotificationHistory.id).label('count')
).filter(
NotificationHistory.service_id == service_id
).group_by(
NotificationHistory.notification_type,
NotificationHistory.status,
monday_of_notification_week
).order_by(
asc(monday_of_notification_week), NotificationHistory.status
).all()
@statsd(namespace="dao")
def dao_fetch_monthly_historical_stats_for_service(service_id, year):
monday_of_notification_week = func.date_trunc('week', NotificationHistory.created_at).label('week_start')

1
app/dao/users_dao.py
View File

@@ -113,6 +113,7 @@ def reset_failed_login_count(user):
def update_user_password(user, password):
# reset failed login count - they've just reset their password so should be fine
user.password = password
user.password_changed_at = datetime.utcnow()
db.session.add(user)

9
app/service/rest.py
View File

@@ -25,7 +25,6 @@ from app.dao.services_dao import (
dao_remove_user_from_service,
dao_fetch_stats_for_service,
dao_fetch_todays_stats_for_service,
dao_fetch_weekly_historical_stats_for_service,
dao_fetch_todays_stats_for_all_services,
dao_archive_service,
fetch_stats_by_date_range_for_all_services,
@@ -270,14 +269,6 @@ def get_all_notifications_for_service(service_id):
), 200
@service_blueprint.route('/<uuid:service_id>/notifications/weekly', methods=['GET'])
def get_weekly_notification_stats(service_id):
service = dao_fetch_service_by_id(service_id)
stats = dao_fetch_weekly_historical_stats_for_service(service_id)
stats = statistics.format_weekly_notification_stats(stats, service.created_at)
return jsonify(data={week.date().isoformat(): statistics for week, statistics in stats.items()})
@service_blueprint.route('/<uuid:service_id>/notifications/monthly', methods=['GET'])
def get_monthly_notification_stats(service_id):
service = dao_fetch_service_by_id(service_id)

22
app/service/statistics.py
View File

@@ -15,20 +15,6 @@ def format_statistics(statistics):
return counts
def format_weekly_notification_stats(statistics, service_created_at):
preceeding_monday = (service_created_at - timedelta(days=service_created_at.weekday()))
# turn a datetime into midnight that day http://stackoverflow.com/a/1937636
preceeding_monday_midnight = datetime.combine(preceeding_monday.date(), datetime.min.time())
week_dict = {
week: create_zeroed_stats_dicts()
for week in _weeks_for_range(preceeding_monday_midnight, datetime.utcnow())
}
for row in statistics:
_update_statuses_from_row(week_dict[row.week_start][row.notification_type], row)
return week_dict
def create_zeroed_stats_dicts():
return {
template_type: {
@@ -43,11 +29,3 @@ def _update_statuses_from_row(update_dict, row):
update_dict['delivered'] += row.count
elif row.status in ('failed', 'technical-failure', 'temporary-failure', 'permanent-failure'):
update_dict['failed'] += row.count
def _weeks_for_range(start, end):
"""
Generator that yields dates from `start` to `end`, in 7 day intervals. End is inclusive.
"""
infinite_date_generator = (start + timedelta(days=i) for i in itertools.count(step=7))
return itertools.takewhile(lambda x: x <= end, infinite_date_generator)

14
app/user/rest.py
View File

@@ -59,12 +59,14 @@ def update_user(user_id):
user_to_update = get_user_by_id(user_id=user_id)
req_json = request.get_json()
update_dct, errors = user_schema_load_json.load(req_json)
# TODO don't let password be updated in this PUT method (currently used by the forgot password flow)
pwd = req_json.get('password', None)
# TODO password validation, it is already done on the admin app
# but would be good to have the same validation here.
if pwd is not None and not pwd:
errors.update({'password': ['Invalid data for field']})
raise InvalidRequest(errors, status_code=400)
if pwd is not None:
if not pwd:
errors.update({'password': ['Invalid data for field']})
raise InvalidRequest(errors, status_code=400)
else:
reset_failed_login_count(user_to_update)
save_model_user(user_to_update, update_dict=update_dct, pwd=pwd)
return jsonify(data=user_schema.dump(user_to_update).data), 200
@@ -130,6 +132,7 @@ def verify_user_code(user_id):
increment_failed_login_count(user_to_verify)
raise InvalidRequest("Code has expired", status_code=400)
use_user_code(code.id)
reset_failed_login_count(user_to_verify)
return jsonify({}), 204
@@ -323,6 +326,7 @@ def update_password(user_id):
if errors:
raise InvalidRequest(errors, status_code=400)
reset_failed_login_count(user)
update_user_password(user, pwd)
return jsonify(data=user_schema.dump(user).data), 200

2
requirements.txt
View File

@@ -29,6 +29,6 @@ notifications-python-client>=3.1,<3.2
awscli>=1.11,<1.12
awscli-cwlogs>=1.4,<1.5
git+https://github.com/alphagov/notifications-utils.git@13.6.0#egg=notifications-utils==13.6.0
git+https://github.com/alphagov/notifications-utils.git@13.7.0#egg=notifications-utils==13.7.0
git+https://github.com/alphagov/boto.git@2.43.0-patch3#egg=boto==2.43.0-patch3

79
tests/app/dao/test_services_dao.py
View File

@@ -19,7 +19,6 @@ from app.dao.services_dao import (
delete_service_and_all_associated_db_objects,
dao_fetch_stats_for_service,
dao_fetch_todays_stats_for_service,
dao_fetch_weekly_historical_stats_for_service,
dao_fetch_monthly_historical_stats_for_service,
fetch_todays_total_message_count,
dao_fetch_todays_stats_for_all_services,
@@ -56,7 +55,7 @@ from tests.app.conftest import (
def test_should_have_decorated_services_dao_functions():
assert dao_fetch_weekly_historical_stats_for_service.__wrapped__.__name__ == 'dao_fetch_weekly_historical_stats_for_service' # noqa
assert dao_fetch_monthly_historical_stats_for_service.__wrapped__.__name__ == 'dao_fetch_monthly_historical_stats_for_service' # noqa
assert dao_fetch_todays_stats_for_service.__wrapped__.__name__ == 'dao_fetch_todays_stats_for_service' # noqa
assert dao_fetch_stats_for_service.__wrapped__.__name__ == 'dao_fetch_stats_for_service' # noqa
@@ -480,35 +479,7 @@ def test_fetch_stats_for_today_only_includes_today(notify_db, notify_db_session,
assert stats['created'] == 1
def test_fetch_weekly_historical_stats_separates_weeks(notify_db, notify_db_session, sample_template):
notification_history = functools.partial(
create_notification_history,
notify_db,
notify_db_session,
sample_template
)
week_53_last_yr = notification_history(created_at=datetime(2016, 1, 1))
week_1_last_yr = notification_history(created_at=datetime(2016, 1, 5))
last_sunday = notification_history(created_at=datetime(2016, 7, 24, 23, 59))
last_monday_morning = notification_history(created_at=datetime(2016, 7, 25, 0, 0))
last_monday_evening = notification_history(created_at=datetime(2016, 7, 25, 23, 59))
with freeze_time('Wed 27th July 2016'):
today = notification_history(created_at=datetime.now(), status='delivered')
ret = dao_fetch_weekly_historical_stats_for_service(sample_template.service_id)
assert [(row.week_start, row.status) for row in ret] == [
(datetime(2015, 12, 28), 'created'),
(datetime(2016, 1, 4), 'created'),
(datetime(2016, 7, 18), 'created'),
(datetime(2016, 7, 25), 'created'),
(datetime(2016, 7, 25), 'delivered')
]
assert ret[-2].count == 2
assert ret[-1].count == 1
def test_fetch_monthly_historical_stats_separates_weeks(notify_db, notify_db_session, sample_template):
def test_fetch_monthly_historical_stats_separates_months(notify_db, notify_db_session, sample_template):
notification_history = functools.partial(
create_notification_history,
notify_db,
@@ -556,52 +527,6 @@ def test_fetch_monthly_historical_stats_separates_weeks(notify_db, notify_db_ses
}
def test_fetch_weekly_historical_stats_ignores_second_service(notify_db, notify_db_session, service_factory):
template_1 = service_factory.get('1').templates[0]
template_2 = service_factory.get('2').templates[0]
notification_history = functools.partial(
create_notification_history,
notify_db,
notify_db_session
)
last_sunday = notification_history(template_1, created_at=datetime(2016, 7, 24, 23, 59))
last_monday_morning = notification_history(template_2, created_at=datetime(2016, 7, 25, 0, 0))
with freeze_time('Wed 27th July 2016'):
ret = dao_fetch_weekly_historical_stats_for_service(template_1.service_id)
assert len(ret) == 1
assert ret[0].week_start == datetime(2016, 7, 18)
assert ret[0].count == 1
def test_fetch_weekly_historical_stats_separates_types(notify_db,
notify_db_session,
sample_template,
sample_email_template):
notification_history = functools.partial(
create_notification_history,
notify_db,
notify_db_session,
created_at=datetime(2016, 7, 25)
)
notification_history(sample_template)
notification_history(sample_email_template)
with freeze_time('Wed 27th July 2016'):
ret = dao_fetch_weekly_historical_stats_for_service(sample_template.service_id)
assert len(ret) == 2
assert ret[0].week_start == datetime(2016, 7, 25)
assert ret[0].count == 1
assert ret[0].notification_type == 'email'
assert ret[1].week_start == datetime(2016, 7, 25)
assert ret[1].count == 1
assert ret[1].notification_type == 'sms'
def test_dao_fetch_todays_total_message_count_returns_count_for_today(notify_db,
notify_db_session,
sample_notification):

20
tests/app/service/test_rest.py
View File

@@ -1202,26 +1202,6 @@ def test_get_detailed_service(notify_db, notify_db_session, notify_api, sample_s
assert service['statistics']['sms'] == stats
def test_get_weekly_notification_stats(notify_api, notify_db, notify_db_session):
with freeze_time('2000-01-01T12:00:00'):
noti = create_sample_notification(notify_db, notify_db_session)
with notify_api.test_request_context(), notify_api.test_client() as client, freeze_time('2000-01-02T12:00:00'):
resp = client.get(
'/service/{}/notifications/weekly'.format(noti.service_id),
headers=[create_authorization_header()]
)
assert resp.status_code == 200
data = json.loads(resp.get_data(as_text=True))['data']
assert data == {
'1999-12-27': {
'sms': {'requested': 1, 'delivered': 0, 'failed': 0},
'email': {'requested': 0, 'delivered': 0, 'failed': 0},
'letter': {'delivered': 0, 'failed': 0, 'requested': 0}
}
}
@pytest.mark.parametrize(
'url, expected_status, expected_json', [
(

90
tests/app/service/test_statistics.py
View File

@@ -6,13 +6,10 @@ from freezegun import freeze_time
from app.service.statistics import (
format_statistics,
_weeks_for_range,
create_zeroed_stats_dicts,
format_weekly_notification_stats
)
StatsRow = collections.namedtuple('row', ('notification_type', 'status', 'count'))
WeeklyStatsRow = collections.namedtuple('row', ('notification_type', 'status', 'week_start', 'count'))
# email_counts and sms_counts are 3-tuple of requested, delivered, failed
@@ -57,18 +54,6 @@ def test_format_statistics(stats, email_counts, sms_counts, letter_counts):
}
@pytest.mark.parametrize('start,end,dates', [
(datetime(2016, 7, 25), datetime(2016, 7, 25), [datetime(2016, 7, 25)]),
(datetime(2016, 7, 25), datetime(2016, 7, 28), [datetime(2016, 7, 25)]),
(datetime(2016, 7, 25), datetime(2016, 8, 1), [datetime(2016, 7, 25), datetime(2016, 8, 1)]),
(datetime(2016, 7, 25), datetime(2016, 8, 10), [
datetime(2016, 7, 25), datetime(2016, 8, 1), datetime(2016, 8, 8)
])
])
def test_weeks_for_range(start, end, dates):
assert list(_weeks_for_range(start, end)) == dates
def test_create_zeroed_stats_dicts():
assert create_zeroed_stats_dicts() == {
'sms': {'requested': 0, 'delivered': 0, 'failed': 0},
@@ -79,78 +64,3 @@ def test_create_zeroed_stats_dicts():
def _stats(requested, delivered, failed):
return {'requested': requested, 'delivered': delivered, 'failed': failed}
@freeze_time('2016-07-28T12:00:00')
@pytest.mark.parametrize('created_at, statistics, expected_results', [
# with no stats and just today, return this week's stats
(datetime(2016, 7, 28), [], {
datetime(2016, 7, 25): {
'sms': _stats(0, 0, 0),
'email': _stats(0, 0, 0),
'letter': _stats(0, 0, 0)
}
}),
# with a random created time, still create the dict for midnight
(datetime(2016, 7, 28, 12, 13, 14), [], {
datetime(2016, 7, 25, 0, 0, 0): {
'sms': _stats(0, 0, 0),
'email': _stats(0, 0, 0),
'letter': _stats(0, 0, 0)
}
}),
# with no stats but a service
(datetime(2016, 7, 14), [], {
datetime(2016, 7, 11): {
'sms': _stats(0, 0, 0),
'email': _stats(0, 0, 0),
'letter': _stats(0, 0, 0)
},
datetime(2016, 7, 18): {
'sms': _stats(0, 0, 0),
'email': _stats(0, 0, 0),
'letter': _stats(0, 0, 0)
},
datetime(2016, 7, 25): {
'sms': _stats(0, 0, 0),
'email': _stats(0, 0, 0),
'letter': _stats(0, 0, 0)
}
}),
# two stats for same week dont re-zero each other
(datetime(2016, 7, 21), [
WeeklyStatsRow('email', 'created', datetime(2016, 7, 18), 1),
WeeklyStatsRow('sms', 'created', datetime(2016, 7, 18), 1),
WeeklyStatsRow('letter', 'created', datetime(2016, 7, 18), 1),
], {
datetime(2016, 7, 18): {
'sms': _stats(1, 0, 0),
'email': _stats(1, 0, 0),
'letter': _stats(1, 0, 0)
},
datetime(2016, 7, 25): {
'sms': _stats(0, 0, 0),
'email': _stats(0, 0, 0),
'letter': _stats(0, 0, 0)
}
}),
# two stats for same type are added together
(datetime(2016, 7, 21), [
WeeklyStatsRow('sms', 'created', datetime(2016, 7, 18), 1),
WeeklyStatsRow('sms', 'delivered', datetime(2016, 7, 18), 1),
WeeklyStatsRow('sms', 'created', datetime(2016, 7, 25), 1),
], {
datetime(2016, 7, 18): {
'sms': _stats(2, 1, 0),
'email': _stats(0, 0, 0),
'letter': _stats(0, 0, 0)
},
datetime(2016, 7, 25): {
'sms': _stats(1, 0, 0),
'email': _stats(0, 0, 0),
'letter': _stats(0, 0, 0)
}
})
])
def test_format_weekly_notification_stats(statistics, created_at, expected_results):
assert format_weekly_notification_stats(statistics, created_at) == expected_results

633
tests/app/user/test_rest.py
View File

@@ -10,174 +10,165 @@ from app.dao.permissions_dao import default_service_permissions
from tests import create_authorization_header
def test_get_user_list(notify_api, notify_db, notify_db_session, sample_service):
def test_get_user_list(client, sample_service):
"""
Tests GET endpoint '/' to retrieve entire user list.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
header = create_authorization_header()
response = client.get(url_for('user.get_user'),
headers=[header])
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 1
sample_user = sample_service.users[0]
expected_permissions = default_service_permissions
fetched = json_resp['data'][0]
header = create_authorization_header()
response = client.get(url_for('user.get_user'),
headers=[header])
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 1
sample_user = sample_service.users[0]
expected_permissions = default_service_permissions
fetched = json_resp['data'][0]
assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address']
assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address']
assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
def test_get_user(notify_api, notify_db, notify_db_session, sample_service):
def test_get_user(client, sample_service):
"""
Tests GET endpoint '/<user_id>' to retrieve a single service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
sample_user = sample_service.users[0]
header = create_authorization_header()
resp = client.get(url_for('user.get_user',
user_id=sample_user.id),
headers=[header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
sample_user = sample_service.users[0]
header = create_authorization_header()
resp = client.get(url_for('user.get_user',
user_id=sample_user.id),
headers=[header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
expected_permissions = default_service_permissions
fetched = json_resp['data']
expected_permissions = default_service_permissions
fetched = json_resp['data']
assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address']
assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address']
assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
def test_post_user(notify_api, notify_db, notify_db_session):
def test_post_user(client, notify_db, notify_db_session):
"""
Tests POST endpoint '/' to create a user.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert User.query.count() == 0
data = {
"name": "Test User",
"email_address": "user@digital.cabinet-office.gov.uk",
"password": "password",
"mobile_number": "+447700900986",
"logged_in_at": None,
"state": "active",
"failed_login_count": 0,
"permissions": {}
}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('user.create_user'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 201
user = User.query.filter_by(email_address='user@digital.cabinet-office.gov.uk').first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['email_address'] == user.email_address
assert json_resp['data']['id'] == str(user.id)
assert User.query.count() == 0
data = {
"name": "Test User",
"email_address": "user@digital.cabinet-office.gov.uk",
"password": "password",
"mobile_number": "+447700900986",
"logged_in_at": None,
"state": "active",
"failed_login_count": 0,
"permissions": {}
}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('user.create_user'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 201
user = User.query.filter_by(email_address='user@digital.cabinet-office.gov.uk').first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['email_address'] == user.email_address
assert json_resp['data']['id'] == str(user.id)
def test_post_user_missing_attribute_email(notify_api, notify_db, notify_db_session):
def test_post_user_missing_attribute_email(client, notify_db, notify_db_session):
"""
Tests POST endpoint '/' missing attribute email.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert User.query.count() == 0
data = {
"name": "Test User",
"password": "password",
"mobile_number": "+447700900986",
"logged_in_at": None,
"state": "active",
"failed_login_count": 0,
"permissions": {}
}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('user.create_user'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 400
assert User.query.count() == 0
json_resp = json.loads(resp.get_data(as_text=True))
assert {'email_address': ['Missing data for required field.']} == json_resp['message']
assert User.query.count() == 0
data = {
"name": "Test User",
"password": "password",
"mobile_number": "+447700900986",
"logged_in_at": None,
"state": "active",
"failed_login_count": 0,
"permissions": {}
}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('user.create_user'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 400
assert User.query.count() == 0
json_resp = json.loads(resp.get_data(as_text=True))
assert {'email_address': ['Missing data for required field.']} == json_resp['message']
def test_create_user_missing_attribute_password(notify_api, notify_db, notify_db_session):
def test_create_user_missing_attribute_password(client, notify_db, notify_db_session):
"""
Tests POST endpoint '/' missing attribute password.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert User.query.count() == 0
data = {
"name": "Test User",
"email_address": "user@digital.cabinet-office.gov.uk",
"mobile_number": "+447700900986",
"logged_in_at": None,
"state": "active",
"failed_login_count": 0,
"permissions": {}
}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('user.create_user'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 400
assert User.query.count() == 0
json_resp = json.loads(resp.get_data(as_text=True))
assert {'password': ['Missing data for required field.']} == json_resp['message']
assert User.query.count() == 0
data = {
"name": "Test User",
"email_address": "user@digital.cabinet-office.gov.uk",
"mobile_number": "+447700900986",
"logged_in_at": None,
"state": "active",
"failed_login_count": 0,
"permissions": {}
}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('user.create_user'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 400
assert User.query.count() == 0
json_resp = json.loads(resp.get_data(as_text=True))
assert {'password': ['Missing data for required field.']} == json_resp['message']
def test_put_user(notify_api, notify_db, notify_db_session, sample_service):
def test_put_user(client, sample_service):
"""
Tests PUT endpoint '/' to update a user.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert User.query.count() == 1
sample_user = sample_service.users[0]
new_email = 'new@digital.cabinet-office.gov.uk'
data = {
'name': sample_user.name,
'email_address': new_email,
'mobile_number': sample_user.mobile_number
}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('user.update_user', user_id=sample_user.id),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 200
assert User.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['email_address'] == new_email
expected_permissions = default_service_permissions
fetched = json_resp['data']
assert User.query.count() == 1
sample_user = sample_service.users[0]
sample_user.failed_login_count = 1
new_email = 'new@digital.cabinet-office.gov.uk'
data = {
'name': sample_user.name,
'email_address': new_email,
'mobile_number': sample_user.mobile_number
}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('user.update_user', user_id=sample_user.id),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 200
assert User.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['email_address'] == new_email
expected_permissions = default_service_permissions
fetched = json_resp['data']
assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number']
assert new_email == fetched['email_address']
assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number']
assert new_email == fetched['email_address']
assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
# password wasn't updated, so failed_login_count stays the same
assert sample_user.failed_login_count == 1
@pytest.mark.parametrize('user_attribute, user_value', [
@@ -203,214 +194,169 @@ def test_post_user_attribute(client, sample_user, user_attribute, user_value):
assert json_resp['data'][user_attribute] == user_value
def test_put_user_update_password(notify_api,
notify_db,
notify_db_session,
sample_service):
def test_put_user_update_password(client, sample_service):
"""
Tests PUT endpoint '/' to update a user including their password.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert User.query.count() == 1
sample_user = sample_service.users[0]
new_password = '1234567890'
data = {
'name': sample_user.name,
'email_address': sample_user.email_address,
'mobile_number': sample_user.mobile_number,
'password': new_password
}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('user.update_user', user_id=sample_user.id),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 200
assert User.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['password_changed_at'] is not None
data = {'password': new_password}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('user.verify_user_password', user_id=str(sample_user.id)),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 204
assert User.query.count() == 1
sample_user = sample_service.users[0]
new_password = '1234567890'
data = {
'name': sample_user.name,
'email_address': sample_user.email_address,
'mobile_number': sample_user.mobile_number,
'password': new_password
}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('user.update_user', user_id=sample_user.id),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 200
assert User.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['password_changed_at'] is not None
data = {'password': new_password}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('user.verify_user_password', user_id=str(sample_user.id)),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 204
def test_put_user_not_exists(notify_api, notify_db, notify_db_session, sample_user, fake_uuid):
def test_put_user_not_exists(client, sample_user, fake_uuid):
"""
Tests PUT endpoint '/' to update a user doesn't exist.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert User.query.count() == 1
new_email = 'new@digital.cabinet-office.gov.uk'
data = {'email_address': new_email}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('user.update_user', user_id=fake_uuid),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 404
assert User.query.count() == 1
user = User.query.filter_by(id=str(sample_user.id)).first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['result'] == "error"
assert json_resp['message'] == 'No result found'
assert User.query.count() == 1
new_email = 'new@digital.cabinet-office.gov.uk'
data = {'email_address': new_email}
auth_header = create_authorization_header()
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('user.update_user', user_id=fake_uuid),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 404
assert User.query.count() == 1
user = User.query.filter_by(id=str(sample_user.id)).first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['result'] == "error"
assert json_resp['message'] == 'No result found'
assert user == sample_user
assert user.email_address != new_email
assert user == sample_user
assert user.email_address != new_email
def test_get_user_by_email(notify_api, notify_db, notify_db_session, sample_service):
def test_get_user_by_email(client, sample_service):
sample_user = sample_service.users[0]
header = create_authorization_header()
url = url_for('user.get_by_email', email=sample_user.email_address)
resp = client.get(url, headers=[header])
assert resp.status_code == 200
with notify_api.test_request_context():
with notify_api.test_client() as client:
sample_user = sample_service.users[0]
header = create_authorization_header()
url = url_for('user.get_by_email', email=sample_user.email_address)
resp = client.get(url, headers=[header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
expected_permissions = default_service_permissions
fetched = json_resp['data']
json_resp = json.loads(resp.get_data(as_text=True))
expected_permissions = default_service_permissions
fetched = json_resp['data']
assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address']
assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
assert str(sample_user.id) == fetched['id']
assert sample_user.name == fetched['name']
assert sample_user.mobile_number == fetched['mobile_number']
assert sample_user.email_address == fetched['email_address']
assert sample_user.state == fetched['state']
assert sorted(expected_permissions) == sorted(fetched['permissions'][str(sample_service.id)])
def test_get_user_by_email_not_found_returns_404(notify_api,
notify_db,
notify_db_session,
sample_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
header = create_authorization_header()
url = url_for('user.get_by_email', email='no_user@digital.gov.uk')
resp = client.get(url, headers=[header])
assert resp.status_code == 404
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'No result found'
def test_get_user_by_email_not_found_returns_404(client, sample_user):
header = create_authorization_header()
url = url_for('user.get_by_email', email='no_user@digital.gov.uk')
resp = client.get(url, headers=[header])
assert resp.status_code == 404
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'No result found'
def test_get_user_by_email_bad_url_returns_404(notify_api,
notify_db,
notify_db_session,
sample_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
header = create_authorization_header()
url = '/user/email'
resp = client.get(url, headers=[header])
assert resp.status_code == 400
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'Invalid request. Email query string param required'
def test_get_user_by_email_bad_url_returns_404(client, sample_user):
header = create_authorization_header()
url = '/user/email'
resp = client.get(url, headers=[header])
assert resp.status_code == 400
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'Invalid request. Email query string param required'
def test_get_user_with_permissions(notify_api,
notify_db,
notify_db_session,
sample_service_permission):
with notify_api.test_request_context():
with notify_api.test_client() as client:
header = create_authorization_header()
response = client.get(url_for('user.get_user', user_id=str(sample_service_permission.user.id)),
headers=[header])
assert response.status_code == 200
permissions = json.loads(response.get_data(as_text=True))['data']['permissions']
assert sample_service_permission.permission in permissions[str(sample_service_permission.service.id)]
def test_get_user_with_permissions(client, sample_service_permission):
header = create_authorization_header()
response = client.get(url_for('user.get_user', user_id=str(sample_service_permission.user.id)),
headers=[header])
assert response.status_code == 200
permissions = json.loads(response.get_data(as_text=True))['data']['permissions']
assert sample_service_permission.permission in permissions[str(sample_service_permission.service.id)]
def test_set_user_permissions(notify_api,
notify_db,
notify_db_session,
sample_user,
sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps([{'permission': MANAGE_SETTINGS}])
header = create_authorization_header()
headers = [('Content-Type', 'application/json'), header]
response = client.post(
url_for(
'user.set_permissions',
user_id=str(sample_user.id),
service_id=str(sample_service.id)),
headers=headers,
data=data)
def test_set_user_permissions(client, sample_user, sample_service):
data = json.dumps([{'permission': MANAGE_SETTINGS}])
header = create_authorization_header()
headers = [('Content-Type', 'application/json'), header]
response = client.post(
url_for(
'user.set_permissions',
user_id=str(sample_user.id),
service_id=str(sample_service.id)),
headers=headers,
data=data)
assert response.status_code == 204
permission = Permission.query.filter_by(permission=MANAGE_SETTINGS).first()
assert permission.user == sample_user
assert permission.service == sample_service
assert permission.permission == MANAGE_SETTINGS
assert response.status_code == 204
permission = Permission.query.filter_by(permission=MANAGE_SETTINGS).first()
assert permission.user == sample_user
assert permission.service == sample_service
assert permission.permission == MANAGE_SETTINGS
def test_set_user_permissions_multiple(notify_api,
notify_db,
notify_db_session,
sample_user,
sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps([{'permission': MANAGE_SETTINGS}, {'permission': MANAGE_TEMPLATES}])
header = create_authorization_header()
headers = [('Content-Type', 'application/json'), header]
response = client.post(
url_for(
'user.set_permissions',
user_id=str(sample_user.id),
service_id=str(sample_service.id)),
headers=headers,
data=data)
def test_set_user_permissions_multiple(client, sample_user, sample_service):
data = json.dumps([{'permission': MANAGE_SETTINGS}, {'permission': MANAGE_TEMPLATES}])
header = create_authorization_header()
headers = [('Content-Type', 'application/json'), header]
response = client.post(
url_for(
'user.set_permissions',
user_id=str(sample_user.id),
service_id=str(sample_service.id)),
headers=headers,
data=data)
assert response.status_code == 204
permission = Permission.query.filter_by(permission=MANAGE_SETTINGS).first()
assert permission.user == sample_user
assert permission.service == sample_service
assert permission.permission == MANAGE_SETTINGS
permission = Permission.query.filter_by(permission=MANAGE_TEMPLATES).first()
assert permission.user == sample_user
assert permission.service == sample_service
assert permission.permission == MANAGE_TEMPLATES
assert response.status_code == 204
permission = Permission.query.filter_by(permission=MANAGE_SETTINGS).first()
assert permission.user == sample_user
assert permission.service == sample_service
assert permission.permission == MANAGE_SETTINGS
permission = Permission.query.filter_by(permission=MANAGE_TEMPLATES).first()
assert permission.user == sample_user
assert permission.service == sample_service
assert permission.permission == MANAGE_TEMPLATES
def test_set_user_permissions_remove_old(notify_api,
notify_db,
notify_db_session,
sample_user,
sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = json.dumps([{'permission': MANAGE_SETTINGS}])
header = create_authorization_header()
headers = [('Content-Type', 'application/json'), header]
response = client.post(
url_for(
'user.set_permissions',
user_id=str(sample_user.id),
service_id=str(sample_service.id)),
headers=headers,
data=data)
def test_set_user_permissions_remove_old(client, sample_user, sample_service):
data = json.dumps([{'permission': MANAGE_SETTINGS}])
header = create_authorization_header()
headers = [('Content-Type', 'application/json'), header]
response = client.post(
url_for(
'user.set_permissions',
user_id=str(sample_user.id),
service_id=str(sample_service.id)),
headers=headers,
data=data)
assert response.status_code == 204
query = Permission.query.filter_by(user=sample_user)
assert query.count() == 1
assert query.first().permission == MANAGE_SETTINGS
assert response.status_code == 204
query = Permission.query.filter_by(user=sample_user)
assert query.count() == 1
assert query.first().permission == MANAGE_SETTINGS
@freeze_time("2016-01-01 11:09:00.061258")
@@ -446,8 +392,7 @@ def test_send_user_reset_password_should_return_400_when_email_is_missing(client
assert mocked.call_count == 0
def test_send_user_reset_password_should_return_400_when_user_doesnot_exist(client,
mocker):
def test_send_user_reset_password_should_return_400_when_user_doesnot_exist(client, mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
bad_email_address = 'bad@email.gov.uk'
data = json.dumps({'email': bad_email_address})
@@ -495,15 +440,15 @@ def test_send_already_registered_email(client, sample_user, already_registered_t
def test_send_already_registered_email_returns_400_when_data_is_missing(client, sample_user):
data = json.dumps({})
auth_header = create_authorization_header()
data = json.dumps({})
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_already_registered_email', user_id=str(sample_user.id)),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 400
assert json.loads(resp.get_data(as_text=True))['message'] == {'email': ['Missing data for required field.']}
resp = client.post(
url_for('user.send_already_registered_email', user_id=str(sample_user.id)),
data=data,
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 400
assert json.loads(resp.get_data(as_text=True))['message'] == {'email': ['Missing data for required field.']}
def test_send_user_confirm_new_email_returns_204(client, sample_user, change_email_confirmation_template, mocker):
@@ -535,7 +480,6 @@ def test_send_user_confirm_new_email_returns_400_when_email_missing(client, samp
def test_update_user_password_saves_correctly(client, sample_service):
assert User.query.count() == 1
sample_user = sample_service.users[0]
new_password = '1234567890'
data = {
@@ -548,7 +492,7 @@ def test_update_user_password_saves_correctly(client, sample_service):
data=json.dumps(data),
headers=headers)
assert resp.status_code == 200
assert User.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['password_changed_at'] is not None
data = {'password': new_password}
@@ -559,3 +503,36 @@ def test_update_user_password_saves_correctly(client, sample_service):
data=json.dumps(data),
headers=headers)
assert resp.status_code == 204
def test_update_user_password_resets_failed_login_count(client, sample_service):
user = sample_service.users[0]
user.failed_login_count = 1
resp = client.post(
url_for('user.update_password', user_id=user.id),
data=json.dumps({'_password': 'foo'}),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
)
assert resp.status_code == 200
assert user.failed_login_count == 0
def test_update_user_resets_failed_login_count_if_updating_password(client, sample_service):
user = sample_service.users[0]
user.failed_login_count = 1
resp = client.put(
url_for('user.update_user', user_id=user.id),
data=json.dumps({
'name': user.name,
'email_address': user.email_address,
'mobile_number': user.mobile_number,
'password': 'foo'
}),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
)
assert resp.status_code == 200
assert user.failed_login_count == 0

117
tests/app/user/test_rest_verify.py
View File

@@ -1,27 +1,24 @@
import json
import uuid
import pytest
from datetime import (
datetime,
timedelta
)
import pytest
from flask import url_for, current_app
from freezegun import freeze_time
from app.dao.services_dao import dao_update_service, dao_fetch_service_by_id
from app.models import (
VerifyCode,
User,
Notification
)
from app import db
import app.celery.tasks
from tests import create_authorization_header
from freezegun import freeze_time
import app.celery.tasks
def test_user_verify_code(client,
@@ -163,7 +160,7 @@ def test_user_verify_password_missing_password(client,
@pytest.mark.parametrize('research_mode', [True, False])
@freeze_time("2016-01-01 11:09:00.061258")
def test_send_user_sms_code(notify_api,
def test_send_user_sms_code(client,
sample_user,
sms_code_template,
mocker,
@@ -171,68 +168,63 @@ def test_send_user_sms_code(notify_api,
"""
Tests POST endpoint /user/<user_id>/sms-code
"""
if research_mode:
notify_service = dao_fetch_service_by_id(current_app.config['NOTIFY_SERVICE_ID'])
notify_service.research_mode = True
dao_update_service(notify_service)
with notify_api.test_request_context():
with notify_api.test_client() as client:
if research_mode:
notify_service = dao_fetch_service_by_id(current_app.config['NOTIFY_SERVICE_ID'])
notify_service.research_mode = True
dao_update_service(notify_service)
auth_header = create_authorization_header()
mocked = mocker.patch('app.user.rest.create_secret_code', return_value='11111')
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
auth_header = create_authorization_header()
mocked = mocker.patch('app.user.rest.create_secret_code', return_value='11111')
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
resp = client.post(
url_for('user.send_user_sms_code', user_id=sample_user.id),
data=json.dumps({}),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
resp = client.post(
url_for('user.send_user_sms_code', user_id=sample_user.id),
data=json.dumps({}),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
assert mocked.call_count == 1
assert VerifyCode.query.count() == 1
assert VerifyCode.query.first().check_code('11111')
assert mocked.call_count == 1
assert VerifyCode.query.count() == 1
assert VerifyCode.query.first().check_code('11111')
assert Notification.query.count() == 1
notification = Notification.query.first()
assert notification.personalisation == {'verify_code': '11111'}
assert notification.to == sample_user.mobile_number
assert str(notification.service_id) == current_app.config['NOTIFY_SERVICE_ID']
assert Notification.query.count() == 1
notification = Notification.query.first()
assert notification.personalisation == {'verify_code': '11111'}
assert notification.to == sample_user.mobile_number
assert str(notification.service_id) == current_app.config['NOTIFY_SERVICE_ID']
app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
([str(notification.id)]),
queue="notify"
)
app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
([str(notification.id)]),
queue="notify"
)
@freeze_time("2016-01-01 11:09:00.061258")
def test_send_user_code_for_sms_with_optional_to_field(notify_api,
def test_send_user_code_for_sms_with_optional_to_field(client,
sample_user,
sms_code_template,
mocker):
"""
Tests POST endpoint /user/<user_id>/sms-code with optional to field
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
to_number = '+441119876757'
mocked = mocker.patch('app.user.rest.create_secret_code', return_value='11111')
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
auth_header = create_authorization_header()
to_number = '+441119876757'
mocked = mocker.patch('app.user.rest.create_secret_code', return_value='11111')
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
auth_header = create_authorization_header()
resp = client.post(
url_for('user.send_user_sms_code', user_id=sample_user.id),
data=json.dumps({'to': to_number}),
headers=[('Content-Type', 'application/json'), auth_header])
resp = client.post(
url_for('user.send_user_sms_code', user_id=sample_user.id),
data=json.dumps({'to': to_number}),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 204
assert mocked.call_count == 1
notification = Notification.query.first()
assert notification.to == to_number
app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
([str(notification.id)]),
queue="notify"
)
assert resp.status_code == 204
assert mocked.call_count == 1
notification = Notification.query.first()
assert notification.to == to_number
app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
([str(notification.id)]),
queue="notify"
)
def test_send_sms_code_returns_404_for_bad_input_data(client):
@@ -282,12 +274,11 @@ def test_send_user_email_verification(client,
mocked.assert_called_once_with(([str(notification.id)]), queue="notify")
def test_send_email_verification_returns_404_for_bad_input_data(client, notify_db, notify_db_session, mocker):
def test_send_email_verification_returns_404_for_bad_input_data(client, notify_db_session, mocker):
"""
Tests POST endpoint /user/<user_id>/sms-code return 404 for bad input data
"""
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
import uuid
uuid_ = uuid.uuid4()
auth_header = create_authorization_header()
resp = client.post(
@@ -297,3 +288,17 @@ def test_send_email_verification_returns_404_for_bad_input_data(client, notify_d
assert resp.status_code == 404
assert json.loads(resp.get_data(as_text=True))['message'] == 'No result found'
assert mocked.call_count == 0
def test_user_verify_user_code_valid_code_resets_failed_login_count(client, sample_sms_code):
sample_sms_code.user.failed_login_count = 1
data = json.dumps({
'code_type': sample_sms_code.code_type,
'code': sample_sms_code.txt_code})
resp = client.post(
url_for('user.verify_user_code', user_id=sample_sms_code.user.id),
data=data,
headers=[('Content-Type', 'application/json'), create_authorization_header()])
assert resp.status_code == 204
assert sample_sms_code.user.failed_login_count == 0
assert sample_sms_code.code_used