darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-03 01:41:05 -05:00
Code Issues Packages Projects Releases Wiki Activity

Ensure that querying for notifications does not return CSV data unless:

Browse Source 
- It's explicitly asked for AND
- you are using a LIVE key

Team and Test keys CANNOT retrieve CSV/Job derived notifications
This commit is contained in:
Martyn Inglis
2016-09-15 15:59:02 +01:00
parent c67c819f0a
commit 9f02e94f38
2 changed files with 202 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

201
tests/app/dao/test_notification_dao.py
View File

@@ -16,7 +16,9 @@ from app.models import (
NotificationStatistics,
TemplateStatistics,
NOTIFICATION_STATUS_TYPES,
KEY_TYPE_NORMAL
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST
)
from app.dao.notifications_dao import (
@@ -39,7 +41,8 @@ from app.dao.notifications_dao import (
from notifications_utils.template import get_sms_fragment_count
from tests.app.conftest import (sample_notification, sample_template, sample_email_template, sample_service)
from tests.app.conftest import (sample_notification, sample_template, sample_email_template, sample_service, sample_job,
sample_api_key)
def test_should_have_decorated_notifications_dao_functions():
@@ -259,8 +262,8 @@ def test_should_by_able_to_update_status_by_id(sample_template, sample_job, mmg_
def test_should_not_update_status_by_id_if_not_sending_and_does_not_update_job(notify_db, notify_db_session):
notification = sample_notification(notify_db, notify_db_session, status='delivered')
job = Job.query.get(notification.job_id)
job = sample_job(notify_db, notify_db_session)
notification = sample_notification(notify_db, notify_db_session, status='delivered', job=job)
assert Notification.query.get(notification.id).status == 'delivered'
assert not update_notification_status_by_id(notification.id, 'failed')
assert Notification.query.get(notification.id).status == 'delivered'
@@ -268,8 +271,8 @@ def test_should_not_update_status_by_id_if_not_sending_and_does_not_update_job(n
def test_should_not_update_status_by_reference_if_not_sending_and_does_not_update_job(notify_db, notify_db_session):
notification = sample_notification(notify_db, notify_db_session, status='delivered', reference='reference')
job = Job.query.get(notification.job_id)
job = sample_job(notify_db, notify_db_session)
notification = sample_notification(notify_db, notify_db_session, status='delivered', reference='reference', job=job)
assert Notification.query.get(notification.id).status == 'delivered'
assert not update_notification_status_by_reference('reference', 'failed')
assert Notification.query.get(notification.id).status == 'delivered'
@@ -834,7 +837,7 @@ def _notification_json(sample_template, job_id=None, id=None, status=None):
return data
def test_dao_timeout_notifications(notify_db, notify_db_session,):
def test_dao_timeout_notifications(notify_db, notify_db_session, ):
with freeze_time(datetime.utcnow() - timedelta(minutes=1)):
created = sample_notification(notify_db, notify_db_session)
sending = sample_notification(notify_db, notify_db_session, status='sending')
@@ -874,3 +877,187 @@ def test_dao_timeout_notifications_only_updates_for_older_notifications(notify_d
assert NotificationHistory.query.get(pending.id).status == 'pending'
assert NotificationHistory.query.get(delivered.id).status == 'delivered'
assert updated == 0
def test_should_return_notifications_excluding_jobs_by_default(notify_db, notify_db_session, sample_service):
assert len(Notification.query.all()) == 0
job = sample_job(notify_db, notify_db_session)
with_job = sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), status="delivered", job=job
)
without_job = sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), status="delivered"
)
all_notifications = Notification.query.all()
assert len(all_notifications) == 2
all_notifications = get_notifications_for_service(sample_service.id).items
assert len(all_notifications) == 1
assert all_notifications[0].id == without_job.id
def test_should_return_notifications_including_jobs(notify_db, notify_db_session, sample_service):
assert len(Notification.query.all()) == 0
job = sample_job(notify_db, notify_db_session)
with_job = sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), status="delivered", job=job
)
all_notifications = Notification.query.all()
assert len(all_notifications) == 1
all_notifications = get_notifications_for_service(sample_service.id).items
assert len(all_notifications) == 0
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True).items
assert len(all_notifications) == 1
assert all_notifications[0].id == with_job.id
def test_get_notifications_created_by_api_or_csv_are_returned_correctly(
notify_db,
notify_db_session,
sample_service,
sample_job,
sample_api_key,
sample_team_api_key,
sample_test_api_key
):
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), job=sample_job
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_team_api_key.id,
key_type=sample_team_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_test_api_key.id,
key_type=sample_test_api_key.key_type
)
all_notifications = Notification.query.all()
assert len(all_notifications) == 4
# returns all API derived notifications
all_notifications = get_notifications_for_service(sample_service.id).items
assert len(all_notifications) == 3
# all notifications including jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True).items
assert len(all_notifications) == 4
def test_get_notifications_with_a_live_api_key_type(
notify_db,
notify_db_session,
sample_service,
sample_job,
sample_api_key,
sample_team_api_key,
sample_test_api_key
):
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), job=sample_job
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_team_api_key.id,
key_type=sample_team_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_test_api_key.id,
key_type=sample_test_api_key.key_type
)
all_notifications = Notification.query.all()
assert len(all_notifications) == 4
# only those created with normal API key, no jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, key_type=KEY_TYPE_NORMAL).items
assert len(all_notifications) == 1
# only those created with normal API key, with jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True,
key_type=KEY_TYPE_NORMAL).items
assert len(all_notifications) == 2
def test_get_notifications_with_a_test_api_key_type(
notify_db,
notify_db_session,
sample_service,
sample_job,
sample_api_key,
sample_team_api_key,
sample_test_api_key
):
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), job=sample_job
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_team_api_key.id,
key_type=sample_team_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_test_api_key.id,
key_type=sample_test_api_key.key_type
)
# only those created with test API key, no jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, key_type=KEY_TYPE_TEST).items
assert len(all_notifications) == 1
# only those created with test API key, no jobs, even when requested
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True,
key_type=KEY_TYPE_TEST).items
assert len(all_notifications) == 1
def test_get_notifications_with_a_team_api_key_type(
notify_db,
notify_db_session,
sample_service,
sample_job,
sample_api_key,
sample_team_api_key,
sample_test_api_key
):
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), job=sample_job
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_team_api_key.id,
key_type=sample_team_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key_id=sample_test_api_key.id,
key_type=sample_test_api_key.key_type
)
# only those created with team API key, no jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, key_type=KEY_TYPE_TEAM).items
assert len(all_notifications) == 1
# only those created with team API key, no jobs, even when requested
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True,
key_type=KEY_TYPE_TEAM).items
assert len(all_notifications) == 1