darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-20 23:41:17 -05:00
Code Issues Packages Projects Releases Wiki Activity

Added authorization headers for all requests

Browse Source
This commit is contained in:
Rebecca Law
2016-01-15 16:22:03 +00:00
parent 5f59b631e1
commit 0ad292300d
10 changed files with 400 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
app/authentication/auth.py
View File

@@ -2,6 +2,8 @@ from flask import request, jsonify, _request_ctx_stack
from client.authentication import decode_jwt_token, get_token_issuer
from client.errors import TokenDecodeError, TokenRequestError, TokenExpiredError, TokenPayloadError
from app.dao.tokens_dao import get_unsigned_token
def authentication_response(message, code):
return jsonify(
@@ -47,5 +49,5 @@ def requires_auth():
def fetch_client(client):
return {
"client": client,
"secret": "secret"
"secret": get_unsigned_token(client)
}

2
app/service/rest.py
View File

@@ -1,6 +1,6 @@
from datetime import datetime
from flask import (jsonify, request)
from flask import (jsonify, request, current_app)
from sqlalchemy.exc import DataError
from sqlalchemy.orm.exc import NoResultFound

21
tests/__init__.py
View File

@@ -0,0 +1,21 @@
from client.authentication import create_jwt_token
from app.dao.tokens_dao import get_unsigned_token
def create_authorization_header(service_id, path, method, request_body=None):
if request_body:
token = create_jwt_token(
request_method=method,
request_path=path,
secret=get_unsigned_token(service_id),
client_id=service_id,
request_body=request_body)
else:
token = create_jwt_token(request_method=method,
request_path=path,
secret=get_unsigned_token(service_id),
client_id=service_id)
return 'Authorization', 'Bearer {}'.format(token)

83
tests/app/authentication/test_authentication.py
View File

@@ -1,6 +1,7 @@
import pytest
from flask import json, url_for
from client.authentication import create_jwt_token
from flask import json, url_for
from app.dao.tokens_dao import get_unsigned_token
def test_should_not_allow_request_with_no_token(notify_api):
@@ -32,10 +33,13 @@ def test_should_not_allow_request_with_incorrect_token(notify_api):
assert data['error'] == 'Invalid token: signature'
def test_should_not_allow_incorrect_path(notify_api):
def test_should_not_allow_incorrect_path(notify_api, notify_db, notify_db_session, sample_token):
with notify_api.test_request_context():
with notify_api.test_client() as client:
token = create_jwt_token(request_method="GET", request_path="/bad", secret="secret", client_id="client_id")
token = create_jwt_token(request_method="GET",
request_path="/bad",
secret=get_unsigned_token(sample_token.service_id),
client_id=sample_token.service_id)
response = client.get(url_for('status.show_status'),
headers={'Authorization': "Bearer {}".format(token)})
assert response.status_code == 403
@@ -43,10 +47,10 @@ def test_should_not_allow_incorrect_path(notify_api):
assert data['error'] == 'Invalid token: request'
def test_should_not_allow_incorrect_method(notify_api):
def test_should_not_allow_incorrect_method(notify_api, notify_db, notify_db_session, sample_token):
with notify_api.test_request_context():
with notify_api.test_client() as client:
token = create_jwt_token(request_method="POST", request_path="/", secret="secret", client_id="client_id")
token = __create_post_token(sample_token.service_id, {})
response = client.get(url_for('status.show_status'),
headers={'Authorization': "Bearer {}".format(token)})
assert response.status_code == 403
@@ -54,11 +58,11 @@ def test_should_not_allow_incorrect_method(notify_api):
assert data['error'] == 'Invalid token: request'
def test_should_not_allow_invalid_secret(notify_api):
def test_should_not_allow_invalid_secret(notify_api, notify_db, notify_db_session, sample_token):
with notify_api.test_request_context():
with notify_api.test_client() as client:
token = create_jwt_token(request_method="POST", request_path="/", secret="not-so-secret",
client_id="client_id")
client_id=sample_token.service_id)
response = client.get(url_for('status.show_status'),
headers={'Authorization': "Bearer {}".format(token)})
assert response.status_code == 403
@@ -66,57 +70,56 @@ def test_should_not_allow_invalid_secret(notify_api):
assert data['error'] == 'Invalid token: signature'
def test_should_allow_valid_token(notify_api):
def test_should_allow_valid_token(notify_api, notify_db, notify_db_session, sample_token):
with notify_api.test_request_context():
with notify_api.test_client() as client:
token = create_jwt_token(request_method="GET",
request_path=url_for('status.show_status'),
secret="secret",
client_id="client_id")
token = __create_get_token(sample_token.service_id)
response = client.get(url_for('status.show_status'),
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
def test_should_allow_valid_token_with_post_body(notify_api):
JSON_BODY = json.dumps({
"key1": "value1",
"key2": "value2",
"key3": "value3"
})
def test_should_allow_valid_token_with_post_body(notify_api, notify_db, notify_db_session, sample_token):
with notify_api.test_request_context():
with notify_api.test_client() as client:
json_body = json.dumps({
"key1": "value1",
"key2": "value2",
"key3": "value3"
})
token = create_jwt_token(
request_method="POST",
request_path=url_for('status.show_status'),
secret="secret",
client_id="client_id",
request_body=json_body
)
token = __create_post_token(sample_token.service_id, JSON_BODY)
response = client.post(url_for('status.show_status'),
data=json_body,
data=JSON_BODY,
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
def test_should_not_allow_valid_token_with_invalid_post_body(notify_api):
def test_should_not_allow_valid_token_with_invalid_post_body(notify_api, notify_db, notify_db_session, sample_token):
with notify_api.test_request_context():
with notify_api.test_client() as client:
json_body = json.dumps({
"key1": "value1",
"key2": "value2",
"key3": "value3"
})
token = create_jwt_token(
request_method="POST",
request_path=url_for('status.show_status'),
secret="secret",
client_id="client_id",
request_body=json_body
)
token = __create_post_token(sample_token.service_id, JSON_BODY)
response = client.post(url_for('status.show_status'),
data="spurious",
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 403
data = json.loads(response.get_data())
assert data['error'] == 'Invalid token: payload'
def __create_get_token(service_id):
return create_jwt_token(request_method="GET",
request_path=url_for('status.show_status'),
secret=get_unsigned_token(service_id),
client_id=service_id)
def __create_post_token(service_id, request_body):
return create_jwt_token(
request_method="POST",
request_path=url_for('status.show_status'),
secret=get_unsigned_token(service_id),
client_id=service_id,
request_body=request_body
)

10
tests/app/conftest.py
View File

@@ -44,6 +44,7 @@ def sample_template(notify_db,
service=None):
if service is None:
service = sample_service(notify_db, notify_db_session)
sample_token(notify_db, notify_db_session, service=service)
data = {
'name': template_name,
'template_type': template_type,
@@ -86,3 +87,12 @@ def sample_job(notify_db,
job = Job(**data)
save_job(job)
return job
@pytest.fixture(scope='function')
def sample_admin_service_id(notify_db, notify_db_session):
admin_user = sample_user(notify_db, notify_db_session, email="notify_admin@digital.cabinet-office.gov.uk")
admin_service = sample_service(notify_db, notify_db_session, service_name="Sample Admin Service", user=admin_user)
data = {'service_id': admin_service.id}
token = Token(**data)
save_model_token(token)
return admin_service.id

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

0
tests/app/main/views/__init__.py
View File

290
tests/app/service/views/test_rest.py
View File

@@ -1,68 +1,79 @@
import json
from flask import url_for
from app.dao.services_dao import save_model_service
from app.models import (Service, User, Token, Template)
from app.models import (Service, Token, Template)
from tests import create_authorization_header
from tests.app.conftest import sample_user as create_sample_user
def test_get_service_list(notify_api, notify_db, notify_db_session, sample_service):
def test_get_service_list(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests GET endpoint '/' to retrieve entire service list.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
response = client.get(url_for('service.get_service'))
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.get_service'),
method='GET')
response = client.get(url_for('service.get_service'),
headers=[auth_header])
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
# TODO assert correct json returned
assert len(json_resp) == 1
assert len(json_resp['data']) == 2
assert json_resp['data'][0]['name'] == sample_service.name
assert json_resp['data'][0]['id'] == sample_service.id
def test_get_service(notify_api, notify_db, notify_db_session, sample_service):
def test_get_service(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests GET endpoint '/<service_id>' to retrieve a single service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.get_service', service_id=sample_service.id),
method='GET')
resp = client.get(url_for('service.get_service',
service_id=sample_service.id))
service_id=sample_service.id),
headers=[auth_header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['name'] == sample_service.name
assert json_resp['data']['id'] == sample_service.id
def test_post_service(notify_api, notify_db, notify_db_session, sample_user):
def test_post_service(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
"""
Tests POST endpoint '/' to create a service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert Service.query.count() == 0
assert Service.query.count() == 1
data = {
'name': 'created service',
'users': [sample_user.id],
'limit': 1000,
'restricted': False,
'active': False}
headers = [('Content-Type', 'application/json')]
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.create_service'),
method='POST',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('service.create_service'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 201
service = Service.query.first()
service = Service.query.filter_by(name='created service').first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['name'] == service.name
assert json_resp['data']['limit'] == service.limit
assert json_resp['token'] is not None
def test_post_service_multiple_users(notify_api, notify_db, notify_db_session, sample_user):
def test_post_service_multiple_users(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
"""
Tests POST endpoint '/' to create a service with multiple users.
"""
@@ -72,80 +83,91 @@ def test_post_service_multiple_users(notify_api, notify_db, notify_db_session, s
notify_db,
notify_db_session,
"new@digital.cabinet-office.gov.uk")
assert Service.query.count() == 0
assert Service.query.count() == 1
data = {
'name': 'created service',
'users': [sample_user.id, another_user.id],
'limit': 1000,
'restricted': False,
'active': False}
headers = [('Content-Type', 'application/json')]
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.create_service'),
method='POST',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('service.create_service'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 201
service = Service.query.first()
service = Service.query.filter_by(name='created service').first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['name'] == service.name
assert json_resp['data']['limit'] == service.limit
assert len(service.users) == 2
def test_post_service_without_users_attribute(notify_api, notify_db, notify_db_session):
def test_post_service_without_users_attribute(notify_api, notify_db, notify_db_session, sample_admin_service_id):
"""
Tests POST endpoint '/' to create a service without 'users' attribute.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert Service.query.count() == 0
assert Service.query.count() == 1
data = {
'name': 'created service',
'limit': 1000,
'restricted': False,
'active': False}
headers = [('Content-Type', 'application/json')]
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.create_service'),
method='POST',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.post(
url_for('service.create_service'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 400
assert Service.query.count() == 0
assert Service.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['message'] == '{"users": ["Missing data for required attribute"]}'
def test_put_service(notify_api, notify_db, notify_db_session, sample_service):
def test_put_service(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests PUT endpoint '/<service_id>' to edit a service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert Service.query.count() == 1
sample_user = User.query.first()
old_service = Service.query.first()
assert Service.query.count() == 2
new_name = 'updated service'
data = {
'name': new_name,
'users': [sample_user.id],
'users': [sample_service.users[0].id],
'limit': 1000,
'restricted': False,
'active': False}
headers = [('Content-Type', 'application/json')]
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.update_service',
service_id=sample_service.id),
method='PUT',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('service.update_service', service_id=old_service.id),
url_for('service.update_service', service_id=sample_service.id),
data=json.dumps(data),
headers=headers)
assert Service.query.count() == 1
assert Service.query.count() == 2
assert resp.status_code == 200
updated_service = Service.query.first()
updated_service = Service.query.get(sample_service.id)
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['name'] == updated_service.name
assert json_resp['data']['limit'] == updated_service.limit
assert updated_service.name == new_name
def test_put_service_not_exists(notify_api, notify_db, notify_db_session, sample_service):
def test_put_service_not_exists(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests PUT endpoint '/<service_id>' service doesn't exist.
"""
@@ -159,56 +181,65 @@ def test_put_service_not_exists(notify_api, notify_db, notify_db_session, sample
'limit': 1000,
'restricted': False,
'active': False}
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.update_service', service_id="123"),
method='PUT',
request_body=json.dumps(data))
resp = client.put(
url_for('service.update_service', service_id="123"),
data=data,
headers=[('Content-Type', 'application/json')])
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
assert Service.query.first().name == sample_service.name
assert Service.query.first().name != new_name
def test_put_service_add_user(notify_api, notify_db, notify_db_session, sample_service):
def test_put_service_add_user(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests PUT endpoint '/<service_id>' add user to the service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert Service.query.count() == 1
assert Service.query.count() == 2
another_user = create_sample_user(
notify_db,
notify_db_session,
"new@digital.cabinet-office.gov.uk")
sample_user = User.query.first()
old_service = Service.query.first()
new_name = 'updated service'
sample_user = sample_service.users[0]
data = {
'name': new_name,
'users': [sample_user.id, another_user.id],
'limit': 1000,
'restricted': False,
'active': False}
headers = [('Content-Type', 'application/json')]
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.update_service',
service_id=sample_service.id),
method='PUT',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('service.update_service', service_id=old_service.id),
url_for('service.update_service', service_id=sample_service.id),
data=json.dumps(data),
headers=headers)
assert Service.query.count() == 1
assert Service.query.count() == 2
assert resp.status_code == 200
updated_service = Service.query.first()
updated_service = Service.query.get(sample_service.id)
json_resp = json.loads(resp.get_data(as_text=True))
assert len(json_resp['data']['users']) == 2
assert sample_user.id in json_resp['data']['users']
assert another_user.id in json_resp['data']['users']
assert updated_service.users == [sample_user, another_user]
def test_put_service_remove_user(notify_api, notify_db, notify_db_session, sample_service):
def test_put_service_remove_user(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests PUT endpoint '/<service_id>' add user to the service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
sample_user = User.query.first()
sample_user = sample_service.users[0]
another_user = create_sample_user(
notify_db,
notify_db_session,
@@ -220,108 +251,136 @@ def test_put_service_remove_user(notify_api, notify_db, notify_db_session, sampl
'restricted': sample_service.restricted,
'active': sample_service.active}
save_model_service(sample_service, update_dict=data)
assert Service.query.count() == 1
sample_user = User.query.first()
assert Service.query.count() == 2
data['users'] = [another_user.id]
headers = [('Content-Type', 'application/json')]
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.update_service',
service_id=sample_service.id),
method='PUT',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('service.update_service', service_id=sample_service.id),
data=json.dumps(data),
headers=headers)
assert Service.query.count() == 1
assert Service.query.count() == 2
assert resp.status_code == 200
updated_service = Service.query.first()
updated_service = Service.query.get(sample_service.id)
json_resp = json.loads(resp.get_data(as_text=True))
assert len(json_resp['data']['users']) == 1
assert sample_user.id not in json_resp['data']['users']
assert another_user.id in json_resp['data']['users']
assert sample_user not in updated_service.users
assert another_user in updated_service.users
def test_delete_service(notify_api, notify_db, notify_db_session, sample_service):
def test_delete_service(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests DELETE endpoint '/<service_id>' delete service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
service = Service.query.first()
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.update_service',
service_id=sample_service.id),
method='DELETE')
resp = client.delete(
url_for('service.update_service', service_id=service.id),
headers=[('Content-Type', 'application/json')])
url_for('service.update_service', service_id=sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 202
json_resp = json.loads(resp.get_data(as_text=True))
json_resp['data']['name'] == sample_service.name
assert Service.query.count() == 0
assert Service.query.count() == 1
def test_delete_service_not_exists(notify_api, notify_db, notify_db_session, sample_service):
def test_delete_service_not_exists(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests DELETE endpoint '/<service_id>' delete service doesn't exist.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert Service.query.count() == 2
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.update_service', service_id="123"),
method='DELETE')
resp = client.delete(
url_for('service.update_service', service_id="123"),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
assert Service.query.count() == 1
assert Service.query.count() == 2
def test_renew_token_should_return_token_when_service_does_not_have_a_valid_token(notify_api, notify_db,
notify_db_session, sample_service):
notify_db_session,
sample_service,
sample_admin_service_id):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.renew_token', service_id=sample_service.id),
method='POST')
response = client.post(url_for('service.renew_token', service_id=sample_service.id),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201
assert response.get_data is not None
saved_token = Token.query.first()
saved_token = Token.query.filter_by(service_id=sample_service.id).first()
assert saved_token.service_id == sample_service.id
def test_renew_token_should_expire_the_old_token_and_create_a_new_token(notify_api, notify_db, notify_db_session,
sample_service):
sample_token, sample_admin_service_id):
with notify_api.test_request_context():
with notify_api.test_client() as client:
response = client.post(url_for('service.renew_token', service_id=sample_service.id),
headers=[('Content-Type', 'application/json')])
assert response.status_code == 201
assert len(Token.query.all()) == 1
saved_token = Token.query.first()
assert Token.query.count() == 2
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.renew_token',
service_id=sample_token.service_id),
method='POST')
response = client.post(url_for('service.renew_token', service_id=sample_service.id),
headers=[('Content-Type', 'application/json')])
response = client.post(url_for('service.renew_token', service_id=sample_token.service_id),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201
all_tokens = Token.query.all()
assert len(all_tokens) == 2
assert Token.query.count() == 3
all_tokens = Token.query.filter_by(service_id=sample_token.service_id).all()
for x in all_tokens:
if x.id == saved_token.id:
if x.id == sample_token.id:
assert x.expiry_date is not None
else:
assert x.expiry_date is None
assert x.token is not saved_token.token
assert x.token is not sample_token.token
def test_create_token_should_return_error_when_service_does_not_exist(notify_api, notify_db, notify_db_session,
sample_service):
sample_service, sample_admin_service_id):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.renew_token', service_id="123"),
method='POST')
response = client.post(url_for('service.renew_token', service_id=123),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 404
def test_revoke_token_should_expire_token_for_service(notify_api, notify_db, notify_db_session, sample_token):
def test_revoke_token_should_expire_token_for_service(notify_api, notify_db, notify_db_session,
sample_token, sample_admin_service_id):
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert len(Token.query.all()) == 1
response = client.post(url_for('service.revoke_token', service_id=sample_token.service_id))
assert Token.query.count() == 2
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.revoke_token',
service_id=sample_token.service_id),
method='POST')
response = client.post(url_for('service.revoke_token', service_id=sample_token.service_id),
headers=[auth_header])
assert response.status_code == 202
all_tokens = Token.query.all()
assert len(all_tokens) == 1
assert all_tokens[0].expiry_date is not None
tokens_for_service = Token.query.filter_by(service_id=sample_token.service_id).first()
assert tokens_for_service.expiry_date is not None
def test_create_service_should_create_new_token_for_service(notify_api, notify_db, notify_db_session, sample_user):
def test_create_service_should_create_new_token_for_service(notify_api, notify_db, notify_db_session, sample_user,
sample_admin_service_id):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
@@ -330,16 +389,20 @@ def test_create_service_should_create_new_token_for_service(notify_api, notify_d
'limit': 1000,
'restricted': False,
'active': False}
headers = [('Content-Type', 'application/json')]
assert len(Token.query.all()) == 0
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.create_service'),
method='POST',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
assert Token.query.count() == 1
resp = client.post(url_for('service.create_service'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 201
assert len(Token.query.all()) == 1
assert Token.query.count() == 2
def test_create_template(notify_api, notify_db, notify_db_session, sample_service):
def test_create_template(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests POST endpoint '/<service_id>/template' a template can be created
from a service.
@@ -356,10 +419,15 @@ def test_create_template(notify_api, notify_db, notify_db_session, sample_servic
'content': template_content,
'service': sample_service.id
}
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.create_template',
service_id=sample_service.id),
method='POST',
request_body=json.dumps(data))
resp = client.post(
url_for('service.create_template', service_id=sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 201
assert Template.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))
@@ -368,7 +436,8 @@ def test_create_template(notify_api, notify_db, notify_db_session, sample_servic
assert json_resp['data']['content'] == template_content
def test_create_template_service_not_exists(notify_api, notify_db, notify_db_session, sample_service):
def test_create_template_service_not_exists(notify_api, notify_db, notify_db_session, sample_service,
sample_admin_service_id):
"""
Tests POST endpoint '/<service_id>/template' a template can be created
from a service.
@@ -385,17 +454,21 @@ def test_create_template_service_not_exists(notify_api, notify_db, notify_db_ses
'content': template_content,
'service': sample_service.id
}
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.create_template', service_id="123"),
method='POST',
request_body=json.dumps(data))
resp = client.post(
url_for('service.create_template', service_id="123"),
data=json.dumps(data),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
assert Template.query.count() == 0
json_resp = json.loads(resp.get_data(as_text=True))
assert "Service not found" in json_resp['message']
def test_update_template(notify_api, notify_db, notify_db_session, sample_template):
def test_update_template(notify_api, notify_db, notify_db_session, sample_template, sample_admin_service_id):
"""
Tests PUT endpoint '/<service_id>/template/<template_id>' a template can be
updated.
@@ -414,12 +487,18 @@ def test_update_template(notify_api, notify_db, notify_db_session, sample_templa
'content': template_content,
'service': sample_service.id
}
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.update_template',
service_id=sample_service.id,
template_id=sample_template.id),
method='PUT',
request_body=json.dumps(data))
resp = client.put(
url_for('service.update_template',
service_id=sample_service.id,
template_id=sample_template.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 200
assert Template.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))
@@ -429,7 +508,8 @@ def test_update_template(notify_api, notify_db, notify_db_session, sample_templa
assert old_name != template_name
def test_update_template_service_not_exists(notify_api, notify_db, notify_db_session, sample_template):
def test_update_template_service_not_exists(notify_api, notify_db, notify_db_session,
sample_template, sample_admin_service_id):
"""
Tests PUT endpoint '/<service_id>/template/<template_id>' a 404 if service
doesn't exist.
@@ -437,7 +517,6 @@ def test_update_template_service_not_exists(notify_api, notify_db, notify_db_ses
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert Template.query.count() == 1
sample_service = Service.query.first()
template_name = "new name"
template_type = "sms"
template_content = "content has been changed"
@@ -445,21 +524,28 @@ def test_update_template_service_not_exists(notify_api, notify_db, notify_db_ses
'name': template_name,
'template_type': template_type,
'content': template_content,
'service': sample_service.id
'service': sample_template.service_id
}
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.update_template',
service_id="123",
template_id=sample_template.id),
method='PUT',
request_body=json.dumps(data))
resp = client.put(
url_for('service.update_template',
service_id="123",
template_id=sample_template.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
json_resp = json.loads(resp.get_data(as_text=True))
assert "Service not found" in json_resp['message']
assert template_name != sample_template.name
def test_update_template_template_not_exists(notify_api, notify_db, notify_db_session, sample_template):
def test_update_template_template_not_exists(notify_api, notify_db, notify_db_session,
sample_template, sample_admin_service_id):
"""
Tests PUT endpoint '/<service_id>/template/<template_id>' a 404 if template
doesn't exist.
@@ -477,19 +563,26 @@ def test_update_template_template_not_exists(notify_api, notify_db, notify_db_se
'content': template_content,
'service': sample_service.id
}
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.update_template',
service_id=sample_service.id,
template_id="123"),
method='PUT',
request_body=json.dumps(data))
resp = client.put(
url_for('service.update_template',
service_id=sample_service.id,
template_id="123"),
data=json.dumps(data),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
json_resp = json.loads(resp.get_data(as_text=True))
assert "Template not found" in json_resp['message']
assert template_name != sample_template.name
def test_create_template_unicode_content(notify_api, notify_db, notify_db_session, sample_service):
def test_create_template_unicode_content(notify_api, notify_db, notify_db_session, sample_service,
sample_admin_service_id):
"""
Tests POST endpoint '/<service_id>/template/<template_id>' a template is
created and the content encoding is respected after saving and loading
@@ -507,10 +600,15 @@ def test_create_template_unicode_content(notify_api, notify_db, notify_db_sessio
'content': template_content,
'service': sample_service.id
}
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.create_template',
service_id=sample_service.id),
method='POST',
request_body=json.dumps(data))
resp = client.post(
url_for('service.create_template', service_id=sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 201
assert Template.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))

16
tests/app/template/test_rest.py
View File

@@ -1,7 +1,8 @@
import json
from app.models import Template
from flask import url_for
from tests import create_authorization_header
def test_get_template_list(notify_api, notify_db, notify_db_session, sample_template):
"""
@@ -9,7 +10,11 @@ def test_get_template_list(notify_api, notify_db, notify_db_session, sample_temp
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
response = client.get(url_for('template.get_template'))
auth_header = create_authorization_header(service_id=sample_template.service_id,
path=url_for('template.get_template'),
method='GET')
response = client.get(url_for('template.get_template'),
headers=[auth_header])
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 1
@@ -23,8 +28,13 @@ def test_get_template(notify_api, notify_db, notify_db_session, sample_template)
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(service_id=sample_template.service_id,
path=url_for('template.get_template',
template_id=sample_template.id),
method='GET')
resp = client.get(url_for(
'template.get_template', template_id=sample_template.id))
'template.get_template', template_id=sample_template.id),
headers=[auth_header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['name'] == sample_template.name

178
tests/app/user/views/test_rest.py
View File

@@ -1,170 +1,202 @@
import json
from app.models import (User, Service)
from tests.app.conftest import sample_service as create_sample_service
from flask import url_for
from app.models import (User, Service)
from tests import create_authorization_header
from tests.app.conftest import sample_service as create_sample_service
def test_get_user_list(notify_api, notify_db, notify_db_session, sample_user):
def test_get_user_list(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
"""
Tests GET endpoint '/' to retrieve entire user list.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
response = client.get(url_for('user.get_user'))
header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.get_user'),
method='GET')
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))
# TODO assert correct json returned
assert len(json_resp['data']) == 1
assert json_resp['data'][0]['email_address'] == sample_user.email_address
assert json_resp['data'][0]['id'] == sample_user.id
assert len(json_resp['data']) == 2
assert {"email_address": sample_user.email_address, "id": sample_user.id} in json_resp['data']
def test_get_user(notify_api, notify_db, notify_db_session, sample_user):
def test_get_user(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
"""
Tests GET endpoint '/<user_id>' to retrieve a single service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.get_user', user_id=sample_user.id),
method='GET')
resp = client.get(url_for('user.get_user',
user_id=sample_user.id))
user_id=sample_user.id),
headers=[header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['email_address'] == sample_user.email_address
assert json_resp['data']['id'] == sample_user.id
assert json_resp['data'] == {"email_address": sample_user.email_address, "id": sample_user.id}
def test_post_user(notify_api, notify_db, notify_db_session):
def test_post_user(notify_api, notify_db, notify_db_session, sample_admin_service_id):
"""
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 = {
'email_address': 'user@digital.cabinet-office.gov.uk'}
headers = [('Content-Type', 'application/json')]
assert User.query.count() == 1
data = {'email_address': 'user@digital.cabinet-office.gov.uk'}
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.create_user'),
method='POST',
request_body=json.dumps(data))
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.first()
user = User.query.filter_by(email_address='user@digital.cabinet-office.gov.uk').first()
json_resp = json.loads(resp.get_data(as_text=True))
json_resp['data'] == {"email_address": user.email_address, "id": user.id}
assert json_resp['data']['email_address'] == user.email_address
assert json_resp['data']['id'] == user.id
def test_post_user_missing_attribute_email(notify_api, notify_db, notify_db_session):
def test_post_user_missing_attribute_email(notify_api, notify_db, notify_db_session, sample_admin_service_id):
"""
Tests POST endpoint '/' missing attribute email.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert User.query.count() == 0
assert User.query.count() == 1
data = {
'blah': 'blah.blah'}
headers = [('Content-Type', 'application/json')]
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.create_user'),
method='POST',
request_body=json.dumps(data))
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
assert User.query.count() == 1
json_resp = json.loads(resp.get_data(as_text=True))
assert {'email_address': ['Missing data for required field.']} == json_resp['message']
def test_put_user(notify_api, notify_db, notify_db_session, sample_user):
def test_put_user(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
"""
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
assert User.query.count() == 2
new_email = 'new@digital.cabinet-office.gov.uk'
data = {
'email_address': new_email}
headers = [('Content-Type', 'application/json')]
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.update_user', user_id=sample_user.id),
method='PUT',
request_body=json.dumps(data))
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
user = User.query.first()
assert User.query.count() == 2
user = User.query.filter_by(email_address=new_email).first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data'] == {'email_address': new_email, 'id': user.id}
assert json_resp['data']['email_address'] == new_email
assert json_resp['data']['id'] == user.id
def test_put_user_not_exists(notify_api, notify_db, notify_db_session, sample_user):
def test_put_user_not_exists(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
"""
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
assert User.query.count() == 2
new_email = 'new@digital.cabinet-office.gov.uk'
data = {
'email_address': new_email}
headers = [('Content-Type', 'application/json')]
data = {'email_address': new_email}
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.update_user', user_id="123"),
method='PUT',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
resp = client.put(
url_for('user.update_user', user_id="123"),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 404
assert User.query.count() == 1
user = User.query.first()
assert User.query.count() == 2
user = User.query.filter_by(id=sample_user.id).first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp == {'result': 'error', 'message': 'User not found'}
assert user == sample_user
assert user.email_address != new_email
def test_put_user_missing_email(notify_api, notify_db, notify_db_session, sample_user):
def test_put_user_missing_email(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
"""
Tests PUT endpoint '/' missing attribute email.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert User.query.count() == 1
assert User.query.count() == 2
new_email = 'new@digital.cabinet-office.gov.uk'
data = {
'blah': new_email}
headers = [('Content-Type', 'application/json')]
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.update_user', user_id=sample_user.id),
method='PUT',
request_body=json.dumps(data))
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 == 400
assert User.query.count() == 1
user = User.query.first()
assert User.query.count() == 2
user = User.query.get(sample_user.id)
json_resp = json.loads(resp.get_data(as_text=True))
assert user.email_address == sample_user.email_address
assert {'email_address': ['Missing data for required field.']} == json_resp['message']
def test_get_user_services(notify_api, notify_db, notify_db_session, sample_service):
def test_get_user_services(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests GET endpoint "/<user_id>/service/<service_id>" to retrieve services for a user.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
user = User.query.first()
user = User.query.get(sample_service.users[0].id)
another_name = "another name"
another_service = create_sample_service(
create_sample_service(
notify_db,
notify_db_session,
service_name=another_name,
user=user)
assert Service.query.count() == 2
assert Service.query.count() == 3
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.get_service_by_user_id', user_id=user.id),
method='GET')
resp = client.get(
url_for('user.get_service_by_user_id', user_id=user.id),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
assert len(json_resp['data']) == 2
def test_get_user_service(notify_api, notify_db, notify_db_session, sample_service):
def test_get_user_service(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
"""
Tests GET endpoint "/<user_id>/service/<service_id>" to retrieve a service for a user.
"""
@@ -177,72 +209,92 @@ def test_get_user_service(notify_api, notify_db, notify_db_session, sample_servi
notify_db_session,
service_name=another_name,
user=user)
assert Service.query.count() == 2
assert Service.query.count() == 3
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.get_service_by_user_id', user_id=user.id,
service_id=another_service.id),
method='GET')
resp = client.get(
url_for('user.get_service_by_user_id', user_id=user.id, service_id=another_service.id),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['name'] == another_name
assert json_resp['data']['id'] == another_service.id
def test_get_user_service_user_not_exists(notify_api, notify_db, notify_db_session, sample_service):
def test_get_user_service_user_not_exists(notify_api, notify_db, notify_db_session, sample_service,
sample_admin_service_id):
"""
Tests GET endpoint "/<user_id>/service/<service_id>" 404 is returned for invalid user.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
user = User.query.first()
assert Service.query.count() == 1
assert Service.query.count() == 2
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.get_service_by_user_id', user_id="123",
service_id=sample_service.id),
method='GET')
resp = client.get(
url_for('user.get_service_by_user_id', user_id="123", service_id=sample_service.id),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
json_resp = json.loads(resp.get_data(as_text=True))
assert "User not found" in json_resp['message']
def test_get_user_service_service_not_exists(notify_api, notify_db, notify_db_session, sample_service):
def test_get_user_service_service_not_exists(notify_api, notify_db, notify_db_session, sample_service,
sample_admin_service_id):
"""
Tests GET endpoint "/<user_id>/service/<service_id>" 404 is returned for invalid service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
user = User.query.first()
assert Service.query.count() == 1
assert Service.query.count() == 2
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.get_service_by_user_id', user_id=user.id,
service_id="123"),
method='GET')
resp = client.get(
url_for('user.get_service_by_user_id', user_id=user.id, service_id="123"),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
json_resp = json.loads(resp.get_data(as_text=True))
assert "Service not found" in json_resp['message']
def test_delete_user(notify_api, notify_db, notify_db_session, sample_user):
def test_delete_user(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
"""
Tests DELETE endpoint '/<user_id>' delete user.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
user = User.query.first()
assert User.query.count() == 2
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.update_user', user_id=sample_user.id),
method='DELETE')
resp = client.delete(
url_for('user.update_user', user_id=user.id),
headers=[('Content-Type', 'application/json')])
url_for('user.update_user', user_id=sample_user.id),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 202
json_resp = json.loads(resp.get_data(as_text=True))
assert User.query.count() == 0
assert User.query.count() == 1
assert json_resp['data'] == {'id': sample_user.id, 'email_address': sample_user.email_address}
def test_delete_user_not_exists(notify_api, notify_db, notify_db_session, sample_user):
def test_delete_user_not_exists(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
"""
Tests DELETE endpoint '/<user_id>' delete user.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
user = User.query.first()
assert User.query.count() == 2
auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('user.update_user', user_id='123'),
method='DELETE')
resp = client.delete(
url_for('user.update_user', user_id="123"),
headers=[('Content-Type', 'application/json')])
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 404
assert User.query.count() == 1
assert User.query.count() == 2