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

Merge pull request #485 from alphagov/api_user-cleanup

Browse Source 
Api user cleanup
This commit is contained in:
Leo Hemsted
2016-07-01 10:58:18 +01:00
committed by GitHub
parent d5382120ad 3838715077
commit 11093530f9
16 changed files with 341 additions and 357 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,9 +1,12 @@
import uuid
from datetime import datetime, timedelta
from notifications_python_client.authentication import create_jwt_token
from datetime import datetime
import pytest
from flask import json, current_app
from notifications_python_client.authentication import create_jwt_token
from app import api_user
from app.dao.api_key_dao import get_unsigned_secrets, save_model_api_key, get_unsigned_secret, expire_api_key
from app.models import ApiKey, KEY_TYPE_NORMAL
from app.models import ApiKey, KEY_TYPE_NORMAL, KEY_TYPE_TEAM
def test_should_not_allow_request_with_no_token(notify_api):
@@ -90,13 +93,6 @@ def test_should_allow_valid_token_when_service_has_multiple_keys(notify_api, sam
assert response.status_code == 200
JSON_BODY = json.dumps({
"key1": "value1",
"key2": "value2",
"key3": "value3"
})
def test_authentication_passes_admin_client_token(notify_api,
notify_db,
notify_db_session,
@@ -172,12 +168,12 @@ def test_authentication_returns_token_expired_when_service_uses_expired_key_and_
headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 403
data = json.loads(response.get_data())
assert data['message'] == {"token": ['Invalid token: signature']}
assert data['message'] == {"token": ['Invalid token: revoked']}
def test_authentication_returns_error_when_api_client_has_no_secrets(notify_api,
notify_db,
notify_db_session):
def test_authentication_returns_error_when_admin_client_has_no_secrets(notify_api,
notify_db,
notify_db_session):
with notify_api.test_request_context():
with notify_api.test_client() as client:
api_secret = notify_api.config.get('ADMIN_CLIENT_SECRET')
@@ -228,3 +224,17 @@ def __create_post_token(service_id, request_body):
secret=get_unsigned_secrets(service_id)[0],
client_id=str(service_id)
)
def test_should_attach_the_current_api_key_to_current_app(notify_api, sample_service, sample_api_key):
with notify_api.test_request_context() as context, notify_api.test_client() as client:
with pytest.raises(AttributeError):
print(api_user)
token = __create_get_token(sample_api_key.service_id)
response = client.get(
'/service/{}'.format(str(sample_api_key.service_id)),
headers={'Authorization': 'Bearer {}'.format(token)}
)
assert response.status_code == 200
assert api_user == sample_api_key

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

@@ -0,0 +1,8 @@
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)

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

@@ -4,23 +4,15 @@ import pytest
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import NoResultFound
from app.authentication.utils import get_secret
from app.dao.api_key_dao import (save_model_api_key,
get_model_api_keys,
get_unsigned_secrets,
get_unsigned_secret,
_generate_secret,
_get_secret, expire_api_key)
expire_api_key)
from app.models import ApiKey, KEY_TYPE_NORMAL
def test_secret_is_signed_and_can_be_read_again(notify_api, mocker):
with notify_api.test_request_context():
mocker.patch("uuid.uuid4", return_value='some_uuid')
signed_secret = _generate_secret()
assert 'some_uuid' == _get_secret(signed_secret)
assert signed_secret != 'some_uuid'
def test_save_api_key_should_create_new_api_key_and_history(sample_service):
api_key = ApiKey(**{'service': sample_service,
'name': sample_service.name,
@@ -72,13 +64,13 @@ 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 unsigned_api_key[0] == get_secret(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 unsigned_api_key == get_secret(sample_api_key.secret)
def test_should_not_allow_duplicate_key_names_per_service(sample_api_key, fake_uuid):

113
tests/app/job/test_rest.py
View File

@@ -2,11 +2,16 @@ import json
import uuid
from datetime import datetime, timedelta
import pytest
import app.celery.tasks
from tests import create_authorization_header
from tests.app.conftest import sample_job as create_job
from tests.app.conftest import (
sample_job as create_job,
sample_notification as create_sample_notification)
from app.dao.templates_dao import dao_update_template
from app.models import NOTIFICATION_STATUS_TYPES
def test_get_jobs(notify_api, notify_db, notify_db_session, sample_template):
@@ -239,3 +244,109 @@ def _setup_jobs(notify_db, notify_db_session, template, number_of_jobs=5):
notify_db_session,
service=template.service,
template=template)
def test_get_all_notifications_for_job_in_order_of_job_number(notify_api,
notify_db,
notify_db_session,
sample_service):
with notify_api.test_request_context(), notify_api.test_client() as client:
main_job = create_job(notify_db, notify_db_session, service=sample_service)
another_job = create_job(notify_db, notify_db_session, service=sample_service)
notification_1 = create_sample_notification(
notify_db,
notify_db_session,
job=main_job,
to_field="1",
created_at=datetime.utcnow(),
job_row_number=1
)
notification_2 = create_sample_notification(
notify_db,
notify_db_session,
job=main_job,
to_field="2",
created_at=datetime.utcnow(),
job_row_number=2
)
notification_3 = create_sample_notification(
notify_db,
notify_db_session,
job=main_job,
to_field="3",
created_at=datetime.utcnow(),
job_row_number=3
)
create_sample_notification(notify_db, notify_db_session, job=another_job)
auth_header = create_authorization_header()
response = client.get(
path='/service/{}/job/{}/notifications'.format(sample_service.id, main_job.id),
headers=[auth_header])
resp = json.loads(response.get_data(as_text=True))
assert len(resp['notifications']) == 3
assert resp['notifications'][0]['to'] == notification_1.to
assert resp['notifications'][0]['job_row_number'] == notification_1.job_row_number
assert resp['notifications'][1]['to'] == notification_2.to
assert resp['notifications'][1]['job_row_number'] == notification_2.job_row_number
assert resp['notifications'][2]['to'] == notification_3.to
assert resp['notifications'][2]['job_row_number'] == notification_3.job_row_number
assert response.status_code == 200
@pytest.mark.parametrize(
"expected_notification_count, status_args",
[
(
1,
'?status={}'.format(NOTIFICATION_STATUS_TYPES[0])
),
(
0,
'?status={}'.format(NOTIFICATION_STATUS_TYPES[1])
),
(
1,
'?status={}&status={}&status={}'.format(
*NOTIFICATION_STATUS_TYPES[0:3]
)
),
(
0,
'?status={}&status={}&status={}'.format(
*NOTIFICATION_STATUS_TYPES[3:6]
)
),
]
)
def test_get_all_notifications_for_job_filtered_by_status(
notify_api,
notify_db,
notify_db_session,
sample_service,
expected_notification_count,
status_args
):
with notify_api.test_request_context(), notify_api.test_client() as client:
job = create_job(notify_db, notify_db_session, service=sample_service)
create_sample_notification(
notify_db,
notify_db_session,
job=job,
to_field="1",
created_at=datetime.utcnow(),
status=NOTIFICATION_STATUS_TYPES[0],
job_row_number=1
)
response = client.get(
path='/service/{}/job/{}/notifications{}'.format(sample_service.id, job.id, status_args),
headers=[create_authorization_header()]
)
resp = json.loads(response.get_data(as_text=True))
assert len(resp['notifications']) == expected_notification_count
assert response.status_code == 200

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

@@ -1,15 +1,12 @@
from datetime import datetime, timedelta
import uuid
import pytest
from flask import json
import app.celery.tasks
from app.models import NOTIFICATION_STATUS_TYPES
from app.dao.notifications_dao import get_notification_by_id, dao_get_notification_statistics_for_service
from tests import create_authorization_header
from tests.app.conftest import sample_notification as create_sample_notification
from tests.app.conftest import sample_job as create_sample_job
from tests.app.conftest import sample_service as create_sample_service
@@ -137,170 +134,6 @@ def test_get_all_notifications_newest_first(notify_api, notify_db, notify_db_ses
assert response.status_code == 200
def test_get_all_notifications_for_service_in_order(notify_api, notify_db, notify_db_session):
with notify_api.test_request_context():
with notify_api.test_client() as client:
service_1 = create_sample_service(notify_db, notify_db_session, service_name="1", email_from='1')
service_2 = create_sample_service(notify_db, notify_db_session, service_name="2", email_from='2')
create_sample_notification(notify_db, notify_db_session, service=service_2)
notification_1 = create_sample_notification(notify_db, notify_db_session, service=service_1)
notification_2 = create_sample_notification(notify_db, notify_db_session, service=service_1)
notification_3 = create_sample_notification(notify_db, notify_db_session, service=service_1)
auth_header = create_authorization_header()
response = client.get(
path='/service/{}/notifications'.format(service_1.id),
headers=[auth_header])
resp = json.loads(response.get_data(as_text=True))
assert len(resp['notifications']) == 3
assert resp['notifications'][0]['to'] == notification_3.to
assert resp['notifications'][1]['to'] == notification_2.to
assert resp['notifications'][2]['to'] == notification_1.to
assert response.status_code == 200
def test_get_all_notifications_for_job_in_order_of_job_number(notify_api,
notify_db,
notify_db_session,
sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
main_job = create_sample_job(notify_db, notify_db_session, service=sample_service)
another_job = create_sample_job(notify_db, notify_db_session, service=sample_service)
notification_1 = create_sample_notification(
notify_db,
notify_db_session,
job=main_job,
to_field="1",
created_at=datetime.utcnow(),
job_row_number=1
)
notification_2 = create_sample_notification(
notify_db,
notify_db_session,
job=main_job,
to_field="2",
created_at=datetime.utcnow(),
job_row_number=2
)
notification_3 = create_sample_notification(
notify_db,
notify_db_session,
job=main_job,
to_field="3",
created_at=datetime.utcnow(),
job_row_number=3
)
create_sample_notification(notify_db, notify_db_session, job=another_job)
auth_header = create_authorization_header()
response = client.get(
path='/service/{}/job/{}/notifications'.format(sample_service.id, main_job.id),
headers=[auth_header])
resp = json.loads(response.get_data(as_text=True))
assert len(resp['notifications']) == 3
assert resp['notifications'][0]['to'] == notification_1.to
assert resp['notifications'][0]['job_row_number'] == notification_1.job_row_number
assert resp['notifications'][1]['to'] == notification_2.to
assert resp['notifications'][1]['job_row_number'] == notification_2.job_row_number
assert resp['notifications'][2]['to'] == notification_3.to
assert resp['notifications'][2]['job_row_number'] == notification_3.job_row_number
assert response.status_code == 200
@pytest.mark.parametrize(
"expected_notification_count, status_args",
[
(
1,
'?status={}'.format(NOTIFICATION_STATUS_TYPES[0])
),
(
0,
'?status={}'.format(NOTIFICATION_STATUS_TYPES[1])
),
(
1,
'?status={}&status={}&status={}'.format(
*NOTIFICATION_STATUS_TYPES[0:3]
)
),
(
0,
'?status={}&status={}&status={}'.format(
*NOTIFICATION_STATUS_TYPES[3:6]
)
),
]
)
def test_get_all_notifications_for_job_filtered_by_status(
notify_api,
notify_db,
notify_db_session,
sample_service,
expected_notification_count,
status_args
):
with notify_api.test_request_context():
with notify_api.test_client() as client:
job = create_sample_job(notify_db, notify_db_session, service=sample_service)
create_sample_notification(
notify_db,
notify_db_session,
job=job,
to_field="1",
created_at=datetime.utcnow(),
status=NOTIFICATION_STATUS_TYPES[0],
job_row_number=1
)
response = client.get(
path='/service/{}/job/{}/notifications{}'.format(sample_service.id, job.id, status_args),
headers=[create_authorization_header()]
)
resp = json.loads(response.get_data(as_text=True))
assert len(resp['notifications']) == expected_notification_count
assert response.status_code == 200
def test_should_not_get_notifications_by_service_with_client_credentials(notify_api, sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(service_id=sample_api_key.service.id)
response = client.get(
'/service/{}/notifications'.format(sample_api_key.service.id),
headers=[auth_header])
resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 403
assert resp['result'] == 'error'
assert resp['message'] == 'Forbidden, invalid authentication token provided'
def test_should_not_get_notifications_by_job_and_service_with_client_credentials(notify_api, sample_job):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(service_id=sample_job.service.id)
response = client.get(
'/service/{}/job/{}/notifications'.format(sample_job.service.id, sample_job.id),
headers=[auth_header])
resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 403
assert resp['result'] == 'error'
assert resp['message'] == 'Forbidden, invalid authentication token provided'
def test_should_reject_invalid_page_param(notify_api, sample_email_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:

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

@@ -6,9 +6,12 @@ from app.dao.users_dao import save_model_user
from app.dao.services_dao import dao_remove_user_from_service
from app.models import User
from tests import create_authorization_header
from tests.app.conftest import sample_service as create_sample_service
from tests.app.conftest import sample_service_permission as create_sample_service_permission
from tests.app.conftest import sample_user as create_sample_user
from tests.app.conftest import (
sample_service as create_sample_service,
sample_service_permission as create_sample_service_permission,
sample_user as create_sample_user,
sample_notification as create_sample_notification
)
def test_get_service_list(notify_api, service_factory):
@@ -1001,3 +1004,28 @@ def test_set_reply_to_email_for_service(notify_api, sample_service):
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['reply_to_email_address'] == 'reply_test@service.gov.uk'
def test_get_all_notifications_for_service_in_order(notify_api, notify_db, notify_db_session):
with notify_api.test_request_context(), notify_api.test_client() as client:
service_1 = create_sample_service(notify_db, notify_db_session, service_name="1", email_from='1')
service_2 = create_sample_service(notify_db, notify_db_session, service_name="2", email_from='2')
create_sample_notification(notify_db, notify_db_session, service=service_2)
notification_1 = create_sample_notification(notify_db, notify_db_session, service=service_1)
notification_2 = create_sample_notification(notify_db, notify_db_session, service=service_1)
notification_3 = create_sample_notification(notify_db, notify_db_session, service=service_1)
auth_header = create_authorization_header()
response = client.get(
path='/service/{}/notifications'.format(service_1.id),
headers=[auth_header])
resp = json.loads(response.get_data(as_text=True))
assert len(resp['notifications']) == 3
assert resp['notifications'][0]['to'] == notification_3.to
assert resp['notifications'][1]['to'] == notification_2.to
assert resp['notifications'][2]['to'] == notification_1.to
assert response.status_code == 200