darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-18 21:48:49 -04:00
Code Issues Packages Projects Releases Wiki Activity

Refactor ApiKeys.secret and ServiceInboundApi.bearer_token to use the same encryption method and get rid of the duplicate code.

Browse Source
This commit is contained in:
Rebecca Law
2017-06-19 14:32:22 +01:00
parent 6202da7dea
commit 3a66027d6a
15 changed files with 55 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
tests/app/authentication/test_utils.py
View File

@@ -1,8 +0,0 @@
from app.authentication.utils import generate_secret, get_secret
def test_secret_is_signed_and_can_be_read_again(notify_api):
with notify_api.test_request_context():
signed_secret = generate_secret('some_uuid')
assert signed_secret != 'some_uuid'
assert 'some_uuid' == get_secret(signed_secret)

10
tests/app/dao/test_api_key_dao.py
View File

@@ -4,7 +4,7 @@ import pytest
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import NoResultFound
from app.authentication.utils import get_secret
from app import encryption
from app.dao.api_key_dao import (save_model_api_key,
get_model_api_keys,
get_unsigned_secrets,
@@ -63,14 +63,14 @@ def test_should_return_api_key_for_service(notify_api, notify_db, notify_db_sess
def test_should_return_unsigned_api_keys_for_service_id(sample_api_key):
unsigned_api_key = get_unsigned_secrets(sample_api_key.service_id)
assert len(unsigned_api_key) == 1
assert sample_api_key.secret != unsigned_api_key[0]
assert unsigned_api_key[0] == get_secret(sample_api_key.secret)
assert sample_api_key._secret != unsigned_api_key[0]
assert unsigned_api_key[0] == sample_api_key.secret
def test_get_unsigned_secret_returns_key(sample_api_key):
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)
assert sample_api_key._secret != unsigned_api_key
assert unsigned_api_key == sample_api_key.secret
def test_should_not_allow_duplicate_key_names_per_service(sample_api_key, fake_uuid):

18
tests/app/dao/test_service_inbound_api_dao.py
View File

@@ -3,7 +3,7 @@ import uuid
import pytest
from sqlalchemy.exc import SQLAlchemyError
from app.authentication.utils import get_secret
from app import encryption
from app.dao.service_inbound_api_dao import (
save_service_inbound_api,
reset_service_inbound_api,
@@ -29,8 +29,8 @@ def test_save_service_inbound_api(sample_service):
assert inbound_api.service_id == sample_service.id
assert inbound_api.updated_by_id == sample_service.users[0].id
assert inbound_api.url == "https://some_service/inbound_messages"
assert inbound_api.unsigned_bearer_token == "some_unique_string"
assert inbound_api.bearer_token != "some_unique_string"
assert inbound_api.bearer_token == "some_unique_string"
assert inbound_api._bearer_token != "some_unique_string"
assert inbound_api.updated_at is None
versioned = ServiceInboundApi.get_history_model().query.filter_by(id=inbound_api.id).one()
@@ -38,7 +38,7 @@ def test_save_service_inbound_api(sample_service):
assert versioned.service_id == sample_service.id
assert versioned.updated_by_id == sample_service.users[0].id
assert versioned.url == "https://some_service/inbound_messages"
assert versioned.bearer_token != "some_unique_string"
assert encryption.decrypt(versioned._bearer_token) == "some_unique_string"
assert versioned.updated_at is None
assert versioned.version == 1
@@ -77,8 +77,8 @@ def test_update_service_inbound_api(sample_service):
assert updated.service_id == sample_service.id
assert updated.updated_by_id == sample_service.users[0].id
assert updated.url == "https://some_service/changed_url"
assert updated.unsigned_bearer_token == "some_unique_string"
assert updated.bearer_token != "some_unique_string"
assert updated.bearer_token == "some_unique_string"
assert updated._bearer_token != "some_unique_string"
assert updated.updated_at is not None
versioned_results = ServiceInboundApi.get_history_model().query.filter_by(id=saved_inbound_api.id).all()
@@ -95,7 +95,7 @@ def test_update_service_inbound_api(sample_service):
assert x.id is not None
assert x.service_id == sample_service.id
assert x.updated_by_id == sample_service.users[0].id
assert get_secret(x.bearer_token) == "some_unique_string"
assert encryption.decrypt(x._bearer_token) == "some_unique_string"
def test_get_service_inbound_api(sample_service):
@@ -112,6 +112,6 @@ def test_get_service_inbound_api(sample_service):
assert inbound_api.service_id == sample_service.id
assert inbound_api.updated_by_id == sample_service.users[0].id
assert inbound_api.url == "https://some_service/inbound_messages"
assert inbound_api.unsigned_bearer_token == "some_unique_string"
assert inbound_api.bearer_token != "some_unique_string"
assert inbound_api.bearer_token == "some_unique_string"
assert inbound_api._bearer_token != "some_unique_string"
assert inbound_api.updated_at is None

24
tests/app/notifications/rest/test_send_notification.py
View File

@@ -321,7 +321,6 @@ def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocked = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
data = {
'to': '07700 900 855',
@@ -374,7 +373,6 @@ def test_should_allow_valid_email_notification(notify_api, sample_email_template
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
data = {
'to': 'ok@ok.com',
@@ -411,7 +409,6 @@ def test_should_block_api_call_if_over_day_limit_for_live_service(
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
service = create_sample_service(notify_db, notify_db_session, limit=1, restricted=False)
email_template = create_sample_email_template(notify_db, notify_db_session, service=service)
@@ -443,7 +440,6 @@ def test_should_block_api_call_if_over_day_limit_for_restricted_service(
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
service = create_sample_service(notify_db, notify_db_session, limit=1, restricted=True)
email_template = create_sample_email_template(notify_db, notify_db_session, service=service)
@@ -479,7 +475,6 @@ def test_should_allow_api_call_if_under_day_limit_regardless_of_type(
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
service = create_sample_service(notify_db, notify_db_session, limit=2, restricted=restricted)
email_template = create_sample_email_template(notify_db, notify_db_session, service=service)
@@ -586,7 +581,7 @@ def test_should_send_email_if_team_api_key_and_a_service_user(notify_api, sample
created_by=sample_email_template.created_by,
key_type=KEY_TYPE_TEAM)
save_model_api_key(api_key)
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
auth_header = create_jwt_token(secret=api_key.secret, client_id=str(api_key.service_id))
response = client.post(
path='/notifications/email',
@@ -619,7 +614,7 @@ def test_should_send_sms_to_anyone_with_test_key(
key_type=KEY_TYPE_TEST
)
save_model_api_key(api_key)
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
auth_header = create_jwt_token(secret=api_key.secret, client_id=str(api_key.service_id))
response = client.post(
path='/notifications/sms',
@@ -654,7 +649,7 @@ def test_should_send_email_to_anyone_with_test_key(
key_type=KEY_TYPE_TEST
)
save_model_api_key(api_key)
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
auth_header = create_jwt_token(secret=api_key.secret, client_id=str(api_key.service_id))
response = client.post(
path='/notifications/email',
@@ -682,7 +677,7 @@ def test_should_send_sms_if_team_api_key_and_a_service_user(notify_api, sample_t
created_by=sample_template.created_by,
key_type=KEY_TYPE_TEAM)
save_model_api_key(api_key)
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
auth_header = create_jwt_token(secret=api_key.secret, client_id=str(api_key.service_id))
response = client.post(
path='/notifications/sms',
@@ -715,7 +710,7 @@ def test_should_persist_notification(notify_api, sample_template,
created_by=template.created_by,
key_type=KEY_TYPE_TEAM)
save_model_api_key(api_key)
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
auth_header = create_jwt_token(secret=api_key.secret, client_id=str(api_key.service_id))
response = client.post(
path='/notifications/{}'.format(template_type),
@@ -758,7 +753,7 @@ def test_should_delete_notification_and_return_error_if_sqs_fails(
created_by=template.created_by,
key_type=KEY_TYPE_TEAM)
save_model_api_key(api_key)
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
auth_header = create_jwt_token(secret=api_key.secret, client_id=str(api_key.service_id))
response = client.post(
path='/notifications/{}'.format(template_type),
@@ -862,7 +857,7 @@ def test_should_not_send_notification_to_non_whitelist_recipient_in_trial_mode(
}
api_key = create_sample_api_key(notify_db, notify_db_session, service, key_type=key_type)
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
auth_header = create_jwt_token(secret=api_key.secret, client_id=str(api_key.service_id))
response = client.post(
path='/notifications/{}'.format(notification_type),
@@ -923,7 +918,7 @@ def test_should_send_notification_to_whitelist_recipient(
}
sample_key = create_sample_api_key(notify_db, notify_db_session, service, key_type=key_type)
auth_header = create_jwt_token(secret=sample_key.unsigned_secret, client_id=str(sample_key.service_id))
auth_header = create_jwt_token(secret=sample_key.secret, client_id=str(sample_key.service_id))
response = client.post(
path='/notifications/{}'.format(notification_type),
@@ -1101,7 +1096,6 @@ def test_returns_a_429_limit_exceeded_if_rate_limit_exceeded(
def test_should_allow_store_original_number_on_sms_notification(client, sample_template, mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
data = {
'to': '+(44) 7700-900 855',
@@ -1128,7 +1122,6 @@ def test_should_allow_store_original_number_on_sms_notification(client, sample_t
def test_should_not_allow_international_number_on_sms_notification(client, sample_template, mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
data = {
'to': '20-12-1234-1234',
@@ -1151,7 +1144,6 @@ def test_should_not_allow_international_number_on_sms_notification(client, sampl
def test_should_allow_international_number_on_sms_notification(client, notify_db, notify_db_session, mocker):
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
service = sample_service(notify_db, notify_db_session, can_send_international_sms=True)
template = create_sample_template(notify_db, notify_db_session, service=service)

2
tests/app/notifications/test_rest.py
View File

@@ -741,5 +741,5 @@ def test_get_notification_selects_correct_template_for_personalisation(client,
def _create_auth_header_from_key(api_key):
token = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
token = create_jwt_token(secret=api_key.secret, client_id=str(api_key.service_id))
return [('Authorization', 'Bearer {}'.format(token))]

2
tests/app/public_contracts/test_POST_notification.py
View File

@@ -21,7 +21,6 @@ def _post_notification(client, template, url, to):
def test_post_sms_contract(client, mocker, sample_template):
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
response_json = return_json_from_response(_post_notification(
client, sample_template, url='/notifications/sms', to='07700 900 855'
@@ -31,7 +30,6 @@ def test_post_sms_contract(client, mocker, sample_template):
def test_post_email_contract(client, mocker, sample_email_template):
mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
response_json = return_json_from_response(_post_notification(
client, sample_email_template, url='/notifications/email', to='foo@bar.com'

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

@@ -8,7 +8,7 @@ import pytest
from flask import url_for, current_app
from freezegun import freeze_time
from app.authentication.utils import get_secret
from app import encryption
from app.dao.users_dao import save_model_user
from app.dao.services_dao import dao_remove_user_from_service
from app.models import (
@@ -2168,7 +2168,6 @@ def test_create_service_inbound_api(client, sample_service):
assert resp_json["id"]
assert resp_json["service_id"] == str(sample_service.id)
assert resp_json["url"] == "https://some_service/inbound-sms"
assert resp_json["bearer_token"] != "some-unique-string" # returned encrypted
assert resp_json["updated_by_id"] == str(sample_service.users[0].id)
assert resp_json["created_at"]
assert not resp_json["updated_at"]
@@ -2217,9 +2216,7 @@ def test_update_service_inbound_api_updates_bearer_token(client, sample_service)
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()])
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))["data"]
assert get_secret(resp_json["bearer_token"]) == "different_token"
assert service_inbound_api.unsigned_bearer_token == "different_token"
assert service_inbound_api.bearer_token == "different_token"
def test_fetch_service_inbound_api(client, sample_service):