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

Fix conflict

Browse Source
This commit is contained in:
Imdad Ahad
2016-09-27 16:54:00 +01:00
parent af2cbaa9c5 b1654c8808
commit f9b539d71e
34 changed files with 1540 additions and 839 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
tests/app/dao/test_jobs_dao.py
View File

@@ -1,5 +1,6 @@
from datetime import (datetime, timedelta)
from datetime import datetime, timedelta
import uuid
from freezegun import freeze_time
from app.dao.jobs_dao import (
@@ -10,10 +11,15 @@ from app.dao.jobs_dao import (
dao_get_scheduled_jobs,
dao_get_future_scheduled_job_by_id_and_service_id,
dao_get_notification_outcomes_for_job,
dao_get_jobs_older_than)
dao_get_jobs_older_than
)
from app.models import Job
from tests.app.conftest import sample_notification, sample_job, sample_service
from tests.app.conftest import sample_notification as create_notification
from tests.app.conftest import sample_job as create_job
from tests.app.conftest import sample_service as create_service
from tests.app.conftest import sample_template as create_template
from tests.app.conftest import sample_user as create_user
def test_should_have_decorated_notifications_dao_functions():
@@ -26,14 +32,14 @@ def test_should_get_all_statuses_for_notifications_associated_with_job(
sample_service,
sample_job):
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='created')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='sending')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='delivered')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='pending')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='failed')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='technical-failure') # noqa
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='temporary-failure') # noqa
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='permanent-failure') # noqa
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='created')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='sending')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='delivered')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='pending')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='failed')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='technical-failure') # noqa
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='temporary-failure') # noqa
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='permanent-failure') # noqa
results = dao_get_notification_outcomes_for_job(sample_service.id, sample_job.id)
assert [(row.count, row.status) for row in results] == [
@@ -53,14 +59,14 @@ def test_should_count_of_statuses_for_notifications_associated_with_job(
notify_db_session,
sample_service,
sample_job):
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='created')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='created')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='sending')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='sending')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='sending')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='sending')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='delivered')
sample_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='delivered')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='created')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='created')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='sending')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='sending')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='sending')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='sending')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='delivered')
create_notification(notify_db, notify_db_session, service=sample_service, job=sample_job, status='delivered')
results = dao_get_notification_outcomes_for_job(sample_service.id, sample_job.id)
assert [(row.count, row.status) for row in results] == [
@@ -75,11 +81,11 @@ def test_should_return_zero_length_array_if_no_notifications_for_job(sample_serv
def test_should_return_notifications_only_for_this_job(notify_db, notify_db_session, sample_service):
job_1 = sample_job(notify_db, notify_db_session, service=sample_service)
job_2 = sample_job(notify_db, notify_db_session, service=sample_service)
job_1 = create_job(notify_db, notify_db_session, service=sample_service)
job_2 = create_job(notify_db, notify_db_session, service=sample_service)
sample_notification(notify_db, notify_db_session, service=sample_service, job=job_1, status='created')
sample_notification(notify_db, notify_db_session, service=sample_service, job=job_2, status='created')
create_notification(notify_db, notify_db_session, service=sample_service, job=job_1, status='created')
create_notification(notify_db, notify_db_session, service=sample_service, job=job_2, status='created')
results = dao_get_notification_outcomes_for_job(sample_service.id, job_1.id)
assert [(row.count, row.status) for row in results] == [
@@ -88,14 +94,14 @@ def test_should_return_notifications_only_for_this_job(notify_db, notify_db_sess
def test_should_return_notifications_only_for_this_service(notify_db, notify_db_session):
service_1 = sample_service(notify_db, notify_db_session, service_name="one", email_from="one")
service_2 = sample_service(notify_db, notify_db_session, service_name="two", email_from="two")
service_1 = create_service(notify_db, notify_db_session, service_name="one", email_from="one")
service_2 = create_service(notify_db, notify_db_session, service_name="two", email_from="two")
job_1 = sample_job(notify_db, notify_db_session, service=service_1)
job_2 = sample_job(notify_db, notify_db_session, service=service_2)
job_1 = create_job(notify_db, notify_db_session, service=service_1)
job_2 = create_job(notify_db, notify_db_session, service=service_2)
sample_notification(notify_db, notify_db_session, service=service_1, job=job_1, status='created')
sample_notification(notify_db, notify_db_session, service=service_2, job=job_2, status='created')
create_notification(notify_db, notify_db_session, service=service_1, job=job_1, status='created')
create_notification(notify_db, notify_db_session, service=service_2, job=job_2, status='created')
assert len(dao_get_notification_outcomes_for_job(service_1.id, job_2.id)) == 0
@@ -130,11 +136,6 @@ def test_get_job_by_id(sample_job):
def test_get_jobs_for_service(notify_db, notify_db_session, sample_template):
from tests.app.conftest import sample_job as create_job
from tests.app.conftest import sample_service as create_service
from tests.app.conftest import sample_template as create_template
from tests.app.conftest import sample_user as create_user
one_job = create_job(notify_db, notify_db_session, sample_template.service, sample_template)
other_user = create_user(notify_db, notify_db_session, email="test@digital.cabinet-office.gov.uk")
@@ -143,8 +144,8 @@ def test_get_jobs_for_service(notify_db, notify_db_session, sample_template):
other_template = create_template(notify_db, notify_db_session, service=other_service)
other_job = create_job(notify_db, notify_db_session, service=other_service, template=other_template)
one_job_from_db = dao_get_jobs_by_service_id(one_job.service_id)
other_job_from_db = dao_get_jobs_by_service_id(other_job.service_id)
one_job_from_db = dao_get_jobs_by_service_id(one_job.service_id).items
other_job_from_db = dao_get_jobs_by_service_id(other_job.service_id).items
assert len(one_job_from_db) == 1
assert one_job == one_job_from_db[0]
@@ -156,27 +157,23 @@ def test_get_jobs_for_service(notify_db, notify_db_session, sample_template):
def test_get_jobs_for_service_with_limit_days_param(notify_db, notify_db_session, sample_template):
from tests.app.conftest import sample_job as create_job
one_job = create_job(notify_db, notify_db_session, sample_template.service, sample_template)
old_job = create_job(notify_db, notify_db_session, sample_template.service, sample_template,
created_at=datetime.now() - timedelta(days=8))
jobs = dao_get_jobs_by_service_id(one_job.service_id)
jobs = dao_get_jobs_by_service_id(one_job.service_id).items
assert len(jobs) == 2
assert one_job in jobs
assert old_job in jobs
jobs_limit_days = dao_get_jobs_by_service_id(one_job.service_id, limit_days=7)
jobs_limit_days = dao_get_jobs_by_service_id(one_job.service_id, limit_days=7).items
assert len(jobs_limit_days) == 1
assert one_job in jobs_limit_days
assert old_job not in jobs_limit_days
def test_get_jobs_for_service_with_limit_days_edge_case(notify_db, notify_db_session, sample_template):
from tests.app.conftest import sample_job as create_job
one_job = create_job(notify_db, notify_db_session, sample_template.service, sample_template)
job_two = create_job(notify_db, notify_db_session, sample_template.service, sample_template,
created_at=(datetime.now() - timedelta(days=7)).date())
@@ -187,7 +184,7 @@ def test_get_jobs_for_service_with_limit_days_edge_case(notify_db, notify_db_ses
job_eight_days_old = create_job(notify_db, notify_db_session, sample_template.service, sample_template,
created_at=datetime.now() - timedelta(days=8))
jobs_limit_days = dao_get_jobs_by_service_id(one_job.service_id, limit_days=7)
jobs_limit_days = dao_get_jobs_by_service_id(one_job.service_id, limit_days=7).items
assert len(jobs_limit_days) == 3
assert one_job in jobs_limit_days
assert job_two in jobs_limit_days
@@ -196,8 +193,6 @@ def test_get_jobs_for_service_with_limit_days_edge_case(notify_db, notify_db_ses
def test_get_jobs_for_service_in_created_at_order(notify_db, notify_db_session, sample_template):
from tests.app.conftest import sample_job as create_job
job_1 = create_job(
notify_db, notify_db_session, sample_template.service, sample_template, created_at=datetime.utcnow())
job_2 = create_job(
@@ -207,7 +202,7 @@ def test_get_jobs_for_service_in_created_at_order(notify_db, notify_db_session,
job_4 = create_job(
notify_db, notify_db_session, sample_template.service, sample_template, created_at=datetime.utcnow())
jobs = dao_get_jobs_by_service_id(sample_template.service.id)
jobs = dao_get_jobs_by_service_id(sample_template.service.id).items
assert len(jobs) == 4
assert jobs[0].id == job_4.id
@@ -231,8 +226,8 @@ def test_update_job(sample_job):
def test_get_scheduled_jobs_gets_all_jobs_in_scheduled_state_scheduled_before_now(notify_db, notify_db_session):
one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
one_hour_ago = datetime.utcnow() - timedelta(minutes=60)
job_new = sample_job(notify_db, notify_db_session, scheduled_for=one_minute_ago, job_status='scheduled')
job_old = sample_job(notify_db, notify_db_session, scheduled_for=one_hour_ago, job_status='scheduled')
job_new = create_job(notify_db, notify_db_session, scheduled_for=one_minute_ago, job_status='scheduled')
job_old = create_job(notify_db, notify_db_session, scheduled_for=one_hour_ago, job_status='scheduled')
jobs = dao_get_scheduled_jobs()
assert len(jobs) == 2
assert jobs[0].id == job_old.id
@@ -241,8 +236,8 @@ def test_get_scheduled_jobs_gets_all_jobs_in_scheduled_state_scheduled_before_no
def test_get_scheduled_jobs_gets_ignores_jobs_not_scheduled(notify_db, notify_db_session):
one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
sample_job(notify_db, notify_db_session)
job_scheduled = sample_job(notify_db, notify_db_session, scheduled_for=one_minute_ago, job_status='scheduled')
create_job(notify_db, notify_db_session)
job_scheduled = create_job(notify_db, notify_db_session, scheduled_for=one_minute_ago, job_status='scheduled')
jobs = dao_get_scheduled_jobs()
assert len(jobs) == 1
assert jobs[0].id == job_scheduled.id
@@ -263,11 +258,32 @@ def test_should_get_jobs_older_than_seven_days(notify_db, notify_db_session):
midnight = datetime(2016, 10, 10, 0, 0, 0, 0)
one_millisecond_past_midnight = datetime(2016, 10, 10, 0, 0, 0, 1)
job_1 = sample_job(notify_db, notify_db_session, created_at=one_millisecond_before_midnight)
sample_job(notify_db, notify_db_session, created_at=midnight)
sample_job(notify_db, notify_db_session, created_at=one_millisecond_past_midnight)
job_1 = create_job(notify_db, notify_db_session, created_at=one_millisecond_before_midnight)
create_job(notify_db, notify_db_session, created_at=midnight)
create_job(notify_db, notify_db_session, created_at=one_millisecond_past_midnight)
with freeze_time('2016-10-17T00:00:00'):
jobs = dao_get_jobs_older_than(7)
assert len(jobs) == 1
assert jobs[0].id == job_1.id
def test_get_jobs_for_service_is_paginated(notify_db, notify_db_session, sample_service, sample_template):
with freeze_time('2015-01-01T00:00:00') as the_time:
for _ in range(10):
the_time.tick(timedelta(hours=1))
create_job(notify_db, notify_db_session, sample_service, sample_template)
res = dao_get_jobs_by_service_id(sample_service.id, page=1, page_size=2)
assert res.per_page == 2
assert res.total == 10
assert len(res.items) == 2
assert res.items[0].created_at == datetime(2015, 1, 1, 10)
assert res.items[1].created_at == datetime(2015, 1, 1, 9)
res = dao_get_jobs_by_service_id(sample_service.id, page=2, page_size=2)
assert len(res.items) == 2
assert res.items[0].created_at == datetime(2015, 1, 1, 8)
assert res.items[1].created_at == datetime(2015, 1, 1, 7)

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

@@ -1000,11 +1000,15 @@ def test_get_notifications_created_by_api_or_csv_are_returned_correctly_excludin
all_notifications = Notification.query.all()
assert len(all_notifications) == 4
# returns all API derived notifications
# returns all real API derived notifications
all_notifications = get_notifications_for_service(sample_service.id).items
assert len(all_notifications) == 2
# all notifications including jobs
# returns all API derived notifications, including those created with test key
all_notifications = get_notifications_for_service(sample_service.id, include_from_test_key=True).items
assert len(all_notifications) == 3
# all real notifications including jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True).items
assert len(all_notifications) == 3