diff --git a/app/authentication/auth.py b/app/authentication/auth.py index 2762d5052..7b9eccb96 100644 --- a/app/authentication/auth.py +++ b/app/authentication/auth.py @@ -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) } diff --git a/app/service/rest.py b/app/service/rest.py index 7b8c52b8b..64a3c1cc8 100644 --- a/app/service/rest.py +++ b/app/service/rest.py @@ -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 diff --git a/tests/__init__.py b/tests/__init__.py index e69de29bb..e7796406e 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -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) diff --git a/tests/app/authentication/test_authentication.py b/tests/app/authentication/test_authentication.py index bbcdf07c0..3b1c54652 100644 --- a/tests/app/authentication/test_authentication.py +++ b/tests/app/authentication/test_authentication.py @@ -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 + ) diff --git a/tests/app/conftest.py b/tests/app/conftest.py index 99930cd4e..b279505d0 100644 --- a/tests/app/conftest.py +++ b/tests/app/conftest.py @@ -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 diff --git a/tests/app/main/__init__.py b/tests/app/main/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/app/main/views/__init__.py b/tests/app/main/views/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/app/service/views/test_rest.py b/tests/app/service/views/test_rest.py index 3b4f4f018..d5f5f7a58 100644 --- a/tests/app/service/views/test_rest.py +++ b/tests/app/service/views/test_rest.py @@ -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 '/' 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 '/' 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 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 '/' 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 '/' 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 '/' 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 '/' 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 '//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 '//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 '//template/' 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 '//template/' 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 '//template/' 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 '//template/' 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)) diff --git a/tests/app/template/test_rest.py b/tests/app/template/test_rest.py index 312072d6a..7f3330b95 100644 --- a/tests/app/template/test_rest.py +++ b/tests/app/template/test_rest.py @@ -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 diff --git a/tests/app/user/views/test_rest.py b/tests/app/user/views/test_rest.py index 43cf08949..7a681af08 100644 --- a/tests/app/user/views/test_rest.py +++ b/tests/app/user/views/test_rest.py @@ -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 '/' 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 "//service/" 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 "//service/" 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 "//service/" 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 "//service/" 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 '/' 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 '/' 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