mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-23 00:41:35 -05:00
Fix all other dependant classes that are affected by the DAO / Service / Model changes
This commit is contained in:
@@ -9,7 +9,7 @@ from app.models import ApiKey, Service
|
||||
def test_should_not_allow_request_with_no_token(notify_api):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
response = client.get(url_for('service.get_service'))
|
||||
response = client.get('/service')
|
||||
assert response.status_code == 401
|
||||
data = json.loads(response.get_data())
|
||||
assert data['error'] == 'Unauthorized, authentication token must be provided'
|
||||
@@ -18,89 +18,103 @@ def test_should_not_allow_request_with_no_token(notify_api):
|
||||
def test_should_not_allow_request_with_incorrect_header(notify_api):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': 'Basic 1234'})
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': 'Basic 1234'})
|
||||
assert response.status_code == 401
|
||||
data = json.loads(response.get_data())
|
||||
assert data['error'] == 'Unauthorized, authentication bearer scheme must be used'
|
||||
|
||||
|
||||
def test_should_not_allow_request_with_incorrect_token(notify_api, notify_db, notify_db_session, sample_api_key):
|
||||
def test_should_not_allow_request_with_incorrect_token(notify_api, sample_user):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': 'Bearer 1234'})
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': 'Bearer 1234'})
|
||||
assert response.status_code == 403
|
||||
data = json.loads(response.get_data())
|
||||
assert data['error'] == 'Invalid token: signature'
|
||||
|
||||
|
||||
def test_should_not_allow_incorrect_path(notify_api, notify_db, notify_db_session, sample_api_key):
|
||||
def test_should_not_allow_incorrect_path(notify_api, sample_api_key):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
token = create_jwt_token(request_method="GET",
|
||||
request_path="/bad",
|
||||
secret=get_unsigned_secrets(sample_api_key.service_id)[0],
|
||||
client_id=str(sample_api_key.service_id))
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': "Bearer {}".format(token)})
|
||||
token = create_jwt_token(
|
||||
request_method="GET",
|
||||
request_path="/bad",
|
||||
secret=get_unsigned_secrets(sample_api_key.service_id)[0],
|
||||
client_id=str(sample_api_key.service_id)
|
||||
)
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': "Bearer {}".format(token)})
|
||||
assert response.status_code == 403
|
||||
data = json.loads(response.get_data())
|
||||
assert data['error'] == 'Invalid token: request'
|
||||
|
||||
|
||||
def test_should_not_allow_incorrect_method(notify_api, notify_db, notify_db_session, sample_api_key):
|
||||
def test_should_not_allow_incorrect_method(notify_api, sample_api_key):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
token = __create_post_token(sample_api_key.service_id, {})
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': "Bearer {}".format(token)})
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': "Bearer {}".format(token)}
|
||||
)
|
||||
assert response.status_code == 403
|
||||
data = json.loads(response.get_data())
|
||||
assert data['error'] == 'Invalid token: request'
|
||||
|
||||
|
||||
def test_should_not_allow_invalid_secret(notify_api, notify_db, notify_db_session, sample_api_key):
|
||||
def test_should_not_allow_invalid_secret(notify_api, sample_api_key):
|
||||
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=str(sample_api_key.service_id))
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': "Bearer {}".format(token)})
|
||||
token = create_jwt_token(
|
||||
request_method="POST",
|
||||
request_path="/service",
|
||||
secret="not-so-secret",
|
||||
client_id=str(sample_api_key.service_id))
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': "Bearer {}".format(token)}
|
||||
)
|
||||
assert response.status_code == 403
|
||||
data = json.loads(response.get_data())
|
||||
assert data['error'] == 'Invalid token: signature'
|
||||
|
||||
|
||||
def test_should_allow_valid_token(notify_api, notify_db, notify_db_session, sample_api_key):
|
||||
def test_should_allow_valid_token(notify_api, 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=str(sample_api_key.service_id)),
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
response = client.get(
|
||||
'/service/{}'.format(str(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):
|
||||
def test_should_allow_valid_token_for_request_with_path_params(notify_api, 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=str(sample_api_key.service_id)),
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
response = client.get(
|
||||
'/service/{}'.format(str(sample_api_key.service_id)),
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_should_allow_valid_token_when_service_has_multiple_keys(notify_api, notify_db, notify_db_session,
|
||||
sample_api_key):
|
||||
def test_should_allow_valid_token_when_service_has_multiple_keys(notify_api, sample_api_key):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
data = {'service_id': sample_api_key.service_id, 'name': 'some key name'}
|
||||
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', service_id=str(sample_api_key.service_id)),
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
response = client.get(
|
||||
'/service/{}'.format(str(sample_api_key.service_id)),
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@@ -111,26 +125,28 @@ JSON_BODY = json.dumps({
|
||||
})
|
||||
|
||||
|
||||
def test_should_allow_valid_token_with_post_body(notify_api, notify_db, notify_db_session, sample_api_key):
|
||||
def test_should_allow_valid_token_with_post_body(notify_api, sample_api_key):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
service = Service.query.get(sample_api_key.service_id)
|
||||
data = {'name': 'new name',
|
||||
'users': [service.users[0].id],
|
||||
'limit': 1000,
|
||||
'restricted': False,
|
||||
'active': False}
|
||||
data = {
|
||||
'email_from': 'new name',
|
||||
'name': 'new name',
|
||||
'users': [service.users[0].id],
|
||||
'limit': 1000,
|
||||
'restricted': False,
|
||||
'active': False}
|
||||
|
||||
token = create_jwt_token(
|
||||
request_method="PUT",
|
||||
request_path=url_for('service.update_service', service_id=str(sample_api_key.service_id)),
|
||||
request_method="POST",
|
||||
request_path='/service/{}'.format(str(sample_api_key.service_id)),
|
||||
secret=get_unsigned_secret(sample_api_key.id),
|
||||
client_id=str(sample_api_key.service_id),
|
||||
request_body=json.dumps(data)
|
||||
)
|
||||
headers = [('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(token))]
|
||||
response = client.put(
|
||||
url_for('service.update_service', service_id=service.id),
|
||||
response = client.post(
|
||||
'/service/{}'.format(service.id),
|
||||
data=json.dumps(data),
|
||||
headers=headers)
|
||||
assert response.status_code == 200
|
||||
@@ -140,9 +156,10 @@ def test_should_not_allow_valid_token_with_invalid_post_body(notify_api, notify_
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
token = __create_post_token(str(sample_api_key.service_id), JSON_BODY)
|
||||
response = client.post(url_for('service.create_service'),
|
||||
data="spurious",
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
response = client.post(
|
||||
'/service',
|
||||
data="spurious",
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 403
|
||||
data = json.loads(response.get_data())
|
||||
assert data['error'] == 'Invalid token: payload'
|
||||
@@ -154,19 +171,22 @@ def test_authentication_passes_admin_client_token(notify_api,
|
||||
sample_api_key):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
token = create_jwt_token(request_method="GET",
|
||||
request_path=url_for('service.get_service'),
|
||||
secret=current_app.config.get('ADMIN_CLIENT_SECRET'),
|
||||
client_id=current_app.config.get('ADMIN_CLIENT_USER_NAME'))
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
token = create_jwt_token(
|
||||
request_method="GET",
|
||||
request_path='/service',
|
||||
secret=current_app.config.get('ADMIN_CLIENT_SECRET'),
|
||||
client_id=current_app.config.get('ADMIN_CLIENT_USER_NAME'))
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_authentication_passes_when_service_has_multiple_keys_some_expired(notify_api,
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_api_key):
|
||||
def test_authentication_passes_when_service_has_multiple_keys_some_expired(
|
||||
notify_api,
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_api_key):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
exprired_key = {'service_id': sample_api_key.service_id, 'name': 'expired_key',
|
||||
@@ -176,12 +196,14 @@ def test_authentication_passes_when_service_has_multiple_keys_some_expired(notif
|
||||
another_key = {'service_id': sample_api_key.service_id, 'name': 'another_key'}
|
||||
api_key = ApiKey(**another_key)
|
||||
save_model_api_key(api_key)
|
||||
token = create_jwt_token(request_method="GET",
|
||||
request_path=url_for('service.get_service'),
|
||||
secret=get_unsigned_secret(api_key.id),
|
||||
client_id=str(sample_api_key.service_id))
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
token = create_jwt_token(
|
||||
request_method="GET",
|
||||
request_path='/service',
|
||||
secret=get_unsigned_secret(api_key.id),
|
||||
client_id=str(sample_api_key.service_id))
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@@ -197,18 +219,20 @@ def test_authentication_returns_token_expired_when_service_uses_expired_key_and_
|
||||
another_key = {'service_id': sample_api_key.service_id, 'name': 'another_key'}
|
||||
api_key = ApiKey(**another_key)
|
||||
save_model_api_key(api_key)
|
||||
token = create_jwt_token(request_method="GET",
|
||||
request_path=url_for('service.get_service'),
|
||||
secret=get_unsigned_secret(expired_api_key.id),
|
||||
client_id=str(sample_api_key.service_id))
|
||||
token = create_jwt_token(
|
||||
request_method="GET",
|
||||
request_path='/service',
|
||||
secret=get_unsigned_secret(expired_api_key.id),
|
||||
client_id=str(sample_api_key.service_id))
|
||||
# expire the key
|
||||
expire_the_key = {'id': expired_api_key.id,
|
||||
'service_id': str(sample_api_key.service_id),
|
||||
'name': 'expired_key',
|
||||
'expiry_date': datetime.now() + timedelta(hours=-2)}
|
||||
save_model_api_key(expired_api_key, expire_the_key)
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 403
|
||||
data = json.loads(response.get_data())
|
||||
assert data['error'] == 'Invalid token: signature'
|
||||
@@ -220,14 +244,16 @@ def test_authentication_returns_error_when_api_client_has_no_secrets(notify_api,
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
api_secret = notify_api.config.get('ADMIN_CLIENT_SECRET')
|
||||
token = create_jwt_token(request_method="GET",
|
||||
request_path=url_for('service.get_service'),
|
||||
secret=api_secret,
|
||||
client_id=notify_api.config.get('ADMIN_CLIENT_USER_NAME')
|
||||
)
|
||||
token = create_jwt_token(
|
||||
request_method="GET",
|
||||
request_path='/service',
|
||||
secret=api_secret,
|
||||
client_id=notify_api.config.get('ADMIN_CLIENT_USER_NAME')
|
||||
)
|
||||
notify_api.config['ADMIN_CLIENT_SECRET'] = ''
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 403
|
||||
error_message = json.loads(response.get_data())
|
||||
assert error_message['error'] == 'Invalid token: signature'
|
||||
@@ -237,12 +263,12 @@ def test_authentication_returns_error_when_api_client_has_no_secrets(notify_api,
|
||||
def __create_get_token(service_id):
|
||||
if service_id:
|
||||
return create_jwt_token(request_method="GET",
|
||||
request_path=url_for('service.get_service', service_id=service_id),
|
||||
request_path='/service/{}'.format(service_id),
|
||||
secret=get_unsigned_secrets(service_id)[0],
|
||||
client_id=str(service_id))
|
||||
else:
|
||||
return create_jwt_token(request_method="GET",
|
||||
request_path=url_for('service.get_service'),
|
||||
request_path='/service',
|
||||
secret=get_unsigned_secrets(service_id)[0],
|
||||
client_id=service_id)
|
||||
|
||||
@@ -250,7 +276,7 @@ def __create_get_token(service_id):
|
||||
def __create_post_token(service_id, request_body):
|
||||
return create_jwt_token(
|
||||
request_method="POST",
|
||||
request_path=url_for('service.create_service'),
|
||||
request_path='/service',
|
||||
secret=get_unsigned_secrets(service_id)[0],
|
||||
client_id=str(service_id),
|
||||
request_body=request_body
|
||||
|
||||
Reference in New Issue
Block a user