Merged with master.

This commit is contained in:
Nicholas Staples
2016-01-22 14:46:03 +00:00
9 changed files with 177 additions and 12 deletions

View File

@@ -76,7 +76,17 @@ def test_should_allow_valid_token(notify_api, notify_db, notify_db_session, samp
with notify_api.test_request_context():
with notify_api.test_client() as client:
token = __create_get_token(sample_api_key.service_id)
response = client.get(url_for('service.get_service'),
response = client.get(url_for('service.get_service', service_id=sample_api_key.service_id),
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
def test_should_allow_valid_token_for_request_with_path_params(notify_api, notify_db, notify_db_session,
sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
token = __create_get_token(sample_api_key.service_id)
response = client.get(url_for('service.get_service', service_id=sample_api_key.service_id),
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
@@ -89,7 +99,7 @@ def test_should_allow_valid_token_when_service_has_multiple_keys(notify_api, not
api_key = ApiKey(**data)
save_model_api_key(api_key)
token = __create_get_token(sample_api_key.service_id)
response = client.get(url_for('service.get_service'),
response = client.get(url_for('service.get_service', service_id=sample_api_key.service_id),
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200
@@ -205,10 +215,16 @@ def test_authentication_returns_token_expired_when_service_uses_expired_key_and_
def __create_get_token(service_id):
return create_jwt_token(request_method="GET",
request_path=url_for('service.get_service'),
secret=get_unsigned_secrets(service_id)[0],
client_id=service_id)
if service_id:
return create_jwt_token(request_method="GET",
request_path=url_for('service.get_service', service_id=service_id),
secret=get_unsigned_secrets(service_id)[0],
client_id=service_id)
else:
return create_jwt_token(request_method="GET",
request_path=url_for('service.get_service'),
secret=get_unsigned_secrets(service_id)[0],
client_id=service_id)
def __create_post_token(service_id, request_body):

View File

@@ -73,3 +73,16 @@ def test_get_unsigned_secret_returns_key(notify_api,
unsigned_api_key = get_unsigned_secret(sample_api_key.id)
assert sample_api_key.secret != unsigned_api_key
assert unsigned_api_key == _get_secret(sample_api_key.secret)
def test_should_not_allow_duplicate_key_names_per_service(notify_api,
notify_db,
notify_db_session,
sample_api_key):
api_key = ApiKey(
**{'id': sample_api_key.id + 1, 'service_id': sample_api_key.service_id, 'name': sample_api_key.name})
try:
save_model_api_key(api_key)
fail("should throw IntegrityError")
except:
pass

View File

@@ -2,6 +2,7 @@ import pytest
from app.dao.services_dao import (
save_model_service, get_model_services, DAOException, delete_model_service)
from tests.app.conftest import sample_service as create_sample_service
from tests.app.conftest import sample_user as create_sample_user
from app.models import Service
@@ -47,6 +48,23 @@ def test_get_user_service(notify_api, notify_db, notify_db_session, sample_user)
assert Service.query.count() == 1
def test_get_services_for_user(notify_api, notify_db, notify_db_session, sample_service):
assert Service.query.count() == 1
service_name = "Random service"
second_user = create_sample_user(notify_db, notify_db_session, 'an@other.gov.uk')
create_sample_service(notify_db, notify_db_session, service_name='another service', user=second_user)
sample_service = create_sample_service(notify_db,
notify_db_session,
service_name=service_name,
user=sample_service.users[0])
assert Service.query.count() == 3
services = get_model_services(user_id=sample_service.users[0].id)
assert len(services) == 2
assert service_name in [x.name for x in services]
assert 'Sample service' in [x.name for x in services]
def test_missing_user_attribute(notify_api, notify_db, notify_db_session):
assert Service.query.count() == 0
try:

View File

@@ -6,6 +6,7 @@ from app.dao.services_dao import save_model_service
from app.models import (Service, ApiKey, Template)
from tests import create_authorization_header
from tests.app.conftest import sample_user as create_sample_user
from tests.app.conftest import sample_service as create_sample_service
def test_get_service_list(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):
@@ -45,6 +46,26 @@ def test_get_service(notify_api, notify_db, notify_db_session, sample_service, s
assert json_resp['data']['id'] == sample_service.id
def test_get_service_for_user(notify_api, notify_db, notify_db_session, sample_service):
second_user = create_sample_user(notify_db, notify_db_session, 'an@other.gov.uk')
create_sample_service(notify_db, notify_db_session, service_name='Second Service', user=second_user)
create_sample_service(notify_db, notify_db_session, service_name='Another Service', user=sample_service.users[0])
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(
path='/service',
method='GET')
resp = client.get('/service?user_id={}'.format(sample_service.users[0].id),
headers=[auth_header])
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
assert len(json_resp['data']) == 2
print(x for x in json_resp['data'])
assert 'Another Service' in [x.get('name') for x in json_resp['data']]
assert 'Sample service' in [x.get('name') for x in json_resp['data']]
assert 'Second Service' not in [x.get('name') for x in json_resp['data']]
def test_post_service(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
"""
Tests POST endpoint '/' to create a service.
@@ -547,3 +568,51 @@ def test_create_template_unicode_content(notify_api, notify_db, notify_db_sessio
assert json_resp['data']['name'] == template_name
assert json_resp['data']['template_type'] == template_type
assert json_resp['data']['content'] == template_content
def test_get_template_list(notify_api, notify_db, notify_db_session, sample_template):
"""
Tests GET endpoint '/' to retrieve entire template list.
"""
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(
'service.get_service_template',
service_id=sample_template.service_id),
method='GET')
response = client.get(
url_for(
'service.get_service_template',
service_id=sample_template.service_id),
headers=[auth_header])
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 1
assert json_resp['data'][0]['name'] == sample_template.name
assert json_resp['data'][0]['id'] == sample_template.id
def test_get_template(notify_api, notify_db, notify_db_session, sample_template):
"""
Tests GET endpoint '/<template_id>' to retrieve a single 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(
'service.get_service_template',
template_id=sample_template.id,
service_id=sample_template.service_id),
method='GET')
resp = client.get(url_for(
'service.get_service_template',
template_id=sample_template.id,
service_id=sample_template.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_template.name
assert json_resp['data']['id'] == sample_template.id