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

clean up test imports

Browse Source
This commit is contained in:
Leo Hemsted
2016-09-21 15:45:26 +01:00
parent 29abd4c382
commit 4bfcfcd2fb
1 changed files with 38 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -15,7 +15,11 @@ from app.dao.jobs_dao import (
) )
from app.models import Job 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(): def test_should_have_decorated_notifications_dao_functions():
@@ -28,14 +32,14 @@ def test_should_get_all_statuses_for_notifications_associated_with_job(
sample_service, sample_service,
sample_job): sample_job):
sample_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')
sample_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')
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='delivered')
sample_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='pending')
sample_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='failed')
sample_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='technical-failure') # noqa
sample_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='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='permanent-failure') # noqa
results = dao_get_notification_outcomes_for_job(sample_service.id, sample_job.id) results = dao_get_notification_outcomes_for_job(sample_service.id, sample_job.id)
assert [(row.count, row.status) for row in results] == [ assert [(row.count, row.status) for row in results] == [
@@ -55,14 +59,14 @@ def test_should_count_of_statuses_for_notifications_associated_with_job(
notify_db_session, notify_db_session,
sample_service, sample_service,
sample_job): sample_job):
sample_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')
sample_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')
sample_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')
sample_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')
sample_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')
sample_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')
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='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='delivered')
results = dao_get_notification_outcomes_for_job(sample_service.id, sample_job.id) results = dao_get_notification_outcomes_for_job(sample_service.id, sample_job.id)
assert [(row.count, row.status) for row in results] == [ assert [(row.count, row.status) for row in results] == [
@@ -77,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): 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_1 = create_job(notify_db, notify_db_session, service=sample_service)
job_2 = sample_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') create_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_2, status='created')
results = dao_get_notification_outcomes_for_job(sample_service.id, job_1.id) results = dao_get_notification_outcomes_for_job(sample_service.id, job_1.id)
assert [(row.count, row.status) for row in results] == [ assert [(row.count, row.status) for row in results] == [
@@ -90,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): 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_1 = create_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_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_1 = create_job(notify_db, notify_db_session, service=service_1)
job_2 = sample_job(notify_db, notify_db_session, service=service_2) 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') create_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_2, job=job_2, status='created')
assert len(dao_get_notification_outcomes_for_job(service_1.id, job_2.id)) == 0 assert len(dao_get_notification_outcomes_for_job(service_1.id, job_2.id)) == 0
@@ -132,11 +136,6 @@ def test_get_job_by_id(sample_job):
def test_get_jobs_for_service(notify_db, notify_db_session, sample_template): 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) 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") other_user = create_user(notify_db, notify_db_session, email="test@digital.cabinet-office.gov.uk")
@@ -158,8 +157,6 @@ 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): 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) 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, old_job = create_job(notify_db, notify_db_session, sample_template.service, sample_template,
created_at=datetime.now() - timedelta(days=8)) created_at=datetime.now() - timedelta(days=8))
@@ -177,8 +174,6 @@ def test_get_jobs_for_service_with_limit_days_param(notify_db, notify_db_session
def test_get_jobs_for_service_with_limit_days_edge_case(notify_db, notify_db_session, sample_template): 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) 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, job_two = create_job(notify_db, notify_db_session, sample_template.service, sample_template,
created_at=(datetime.now() - timedelta(days=7)).date()) created_at=(datetime.now() - timedelta(days=7)).date())
@@ -198,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): 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( job_1 = create_job(
notify_db, notify_db_session, sample_template.service, sample_template, created_at=datetime.utcnow()) notify_db, notify_db_session, sample_template.service, sample_template, created_at=datetime.utcnow())
job_2 = create_job( job_2 = create_job(
@@ -233,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): 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_minute_ago = datetime.utcnow() - timedelta(minutes=1)
one_hour_ago = datetime.utcnow() - timedelta(minutes=60) 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_new = create_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_old = create_job(notify_db, notify_db_session, scheduled_for=one_hour_ago, job_status='scheduled')
jobs = dao_get_scheduled_jobs() jobs = dao_get_scheduled_jobs()
assert len(jobs) == 2 assert len(jobs) == 2
assert jobs[0].id == job_old.id assert jobs[0].id == job_old.id
@@ -243,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): def test_get_scheduled_jobs_gets_ignores_jobs_not_scheduled(notify_db, notify_db_session):
one_minute_ago = datetime.utcnow() - timedelta(minutes=1) one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
sample_job(notify_db, notify_db_session) create_job(notify_db, notify_db_session)
job_scheduled = sample_job(notify_db, notify_db_session, scheduled_for=one_minute_ago, job_status='scheduled') job_scheduled = create_job(notify_db, notify_db_session, scheduled_for=one_minute_ago, job_status='scheduled')
jobs = dao_get_scheduled_jobs() jobs = dao_get_scheduled_jobs()
assert len(jobs) == 1 assert len(jobs) == 1
assert jobs[0].id == job_scheduled.id assert jobs[0].id == job_scheduled.id
@@ -265,9 +258,9 @@ def test_should_get_jobs_older_than_seven_days(notify_db, notify_db_session):
midnight = datetime(2016, 10, 10, 0, 0, 0, 0) midnight = datetime(2016, 10, 10, 0, 0, 0, 0)
one_millisecond_past_midnight = datetime(2016, 10, 10, 0, 0, 0, 1) 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) job_1 = create_job(notify_db, notify_db_session, created_at=one_millisecond_before_midnight)
sample_job(notify_db, notify_db_session, created_at=midnight) create_job(notify_db, notify_db_session, created_at=midnight)
sample_job(notify_db, notify_db_session, created_at=one_millisecond_past_midnight) create_job(notify_db, notify_db_session, created_at=one_millisecond_past_midnight)
with freeze_time('2016-10-17T00:00:00'): with freeze_time('2016-10-17T00:00:00'):
jobs = dao_get_jobs_older_than(7) jobs = dao_get_jobs_older_than(7)
@@ -276,9 +269,6 @@ def test_should_get_jobs_older_than_seven_days(notify_db, notify_db_session):
def test_get_jobs_for_service_is_paginated(notify_db, notify_db_session, sample_service, sample_template): def test_get_jobs_for_service_is_paginated(notify_db, notify_db_session, sample_service, sample_template):
from tests.app.conftest import sample_job as create_job
with freeze_time('2015-01-01T00:00:00') as the_time: with freeze_time('2015-01-01T00:00:00') as the_time:
for _ in range(10): for _ in range(10):
the_time.tick(timedelta(hours=1)) the_time.tick(timedelta(hours=1))