darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-07-09 10:53:55 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #683 from alphagov/paginate-dashboard

Browse Source 
Paginate `GET /service/{}/job` query
This commit is contained in:
Leo Hemsted
2016-09-23 09:54:47 +01:00
committed by GitHub
parent e757c6463c 00b905bef0
commit 281323b435
7 changed files with 260 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
app/dao/jobs_dao.py
View File

@@ -1,10 +1,11 @@
from datetime import date, timedelta, datetime
from sqlalchemy import desc, asc, cast, Date as sql_date
from datetime import datetime
from sqlalchemy import func, desc, asc, cast, Date as sql_date
from app import db
from app.dao import days_ago
from app.models import Job, NotificationHistory, JOB_STATUS_SCHEDULED
from app.statsd_decorators import statsd
from sqlalchemy import func, asc
@statsd(namespace="dao")
@@ -26,11 +27,14 @@ def dao_get_job_by_service_id_and_job_id(service_id, job_id):
return Job.query.filter_by(service_id=service_id, id=job_id).one()
def dao_get_jobs_by_service_id(service_id, limit_days=None):
def dao_get_jobs_by_service_id(service_id, limit_days=None, page=1, page_size=50):
query_filter = [Job.service_id == service_id]
if limit_days is not None:
query_filter.append(cast(Job.created_at, sql_date) >= days_ago(limit_days))
return Job.query.filter(*query_filter).order_by(desc(Job.created_at)).all()
return Job.query \
.filter(*query_filter) \
.order_by(desc(Job.created_at)) \
.paginate(page=page, per_page=page_size)
def dao_get_job_by_id(job_id):

37
app/job/rest.py
View File

@@ -96,20 +96,14 @@ def get_jobs_by_service(service_id):
if request.args.get('limit_days'):
try:
limit_days = int(request.args['limit_days'])
except ValueError as e:
except ValueError:
errors = {'limit_days': ['{} is not an integer'.format(request.args['limit_days'])]}
raise InvalidRequest(errors, status_code=400)
else:
limit_days = None
jobs = dao_get_jobs_by_service_id(service_id, limit_days)
data = job_schema.dump(jobs, many=True).data
for job_data in data:
statistics = dao_get_notification_outcomes_for_job(service_id, job_data['id'])
job_data['statistics'] = [{'status': statistic[1], 'count': statistic[0]} for statistic in statistics]
return jsonify(data=data)
page = int(request.args.get('page', 1))
return jsonify(**get_paginated_jobs(service_id, limit_days, page))
@job.route('', methods=['POST'])
@@ -144,3 +138,28 @@ def create_job(service_id):
job_json['statistics'] = []
return jsonify(data=job_json), 201
def get_paginated_jobs(service_id, limit_days, page):
pagination = dao_get_jobs_by_service_id(
service_id,
limit_days=limit_days,
page=page,
page_size=current_app.config['PAGE_SIZE']
)
data = job_schema.dump(pagination.items, many=True).data
for job_data in data:
statistics = dao_get_notification_outcomes_for_job(service_id, job_data['id'])
job_data['statistics'] = [{'status': statistic[1], 'count': statistic[0]} for statistic in statistics]
return {
'data': data,
'page_size': pagination.per_page,
'total': pagination.total,
'links': pagination_links(
pagination,
'.get_jobs_by_service',
service_id=service_id
)
}

2
app/utils.py
View File

@@ -4,7 +4,7 @@ from flask import url_for
def pagination_links(pagination, endpoint, **kwargs):
if 'page' in kwargs:
kwargs.pop('page', None)
links = dict()
links = {}
if pagination.has_prev:
links['prev'] = url_for(endpoint, page=pagination.prev_num, **kwargs)
if pagination.has_next:

4
tests/app/conftest.py
View File

@@ -257,7 +257,7 @@ def sample_job(notify_db,
service=None,
template=None,
notification_count=1,
created_at=datetime.utcnow(),
created_at=None,
job_status='pending',
scheduled_for=None):
if service is None:
@@ -273,7 +273,7 @@ def sample_job(notify_db,
'template_version': template.version,
'original_file_name': 'some.csv',
'notification_count': notification_count,
'created_at': created_at,
'created_at': created_at or datetime.utcnow(),
'created_by': service.created_by,
'job_status': job_status,
'scheduled_for': scheduled_for

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)

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

@@ -1,61 +1,22 @@
import json
import uuid
from datetime import datetime, timedelta
from freezegun import freeze_time
from freezegun import freeze_time
import pytest
import pytz
import app.celery.tasks
from tests import create_authorization_header
from tests.conftest import set_config
from tests.app.conftest import (
sample_job as create_job,
sample_notification as create_sample_notification, sample_notification, sample_job)
sample_notification as create_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):
_setup_jobs(notify_db, notify_db_session, sample_template)
service_id = sample_template.service.id
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = '/service/{}/job'.format(service_id)
auth_header = create_authorization_header(service_id=service_id)
response = client.get(path, headers=[auth_header])
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert len(resp_json['data']) == 5
def test_get_jobs_with_limit_days(notify_api, notify_db, notify_db_session, sample_template):
create_job(
notify_db,
notify_db_session,
service=sample_template.service,
template=sample_template,
)
create_job(
notify_db,
notify_db_session,
service=sample_template.service,
template=sample_template,
created_at=datetime.now() - timedelta(days=7))
service_id = sample_template.service.id
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = '/service/{}/job'.format(service_id)
auth_header = create_authorization_header(service_id=service_id)
response = client.get(path, headers=[auth_header], query_string={'limit_days': 5})
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert len(resp_json['data']) == 1
def test_get_job_with_invalid_service_id_returns404(notify_api, sample_api_key, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
@@ -391,7 +352,7 @@ def test_get_all_notifications_for_job_in_order_of_job_number(notify_api,
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(
notification_1 = create_notification(
notify_db,
notify_db_session,
job=main_job,
@@ -399,7 +360,7 @@ def test_get_all_notifications_for_job_in_order_of_job_number(notify_api,
created_at=datetime.utcnow(),
job_row_number=1
)
notification_2 = create_sample_notification(
notification_2 = create_notification(
notify_db,
notify_db_session,
job=main_job,
@@ -407,7 +368,7 @@ def test_get_all_notifications_for_job_in_order_of_job_number(notify_api,
created_at=datetime.utcnow(),
job_row_number=2
)
notification_3 = create_sample_notification(
notification_3 = create_notification(
notify_db,
notify_db_session,
job=main_job,
@@ -415,7 +376,7 @@ def test_get_all_notifications_for_job_in_order_of_job_number(notify_api,
created_at=datetime.utcnow(),
job_row_number=3
)
create_sample_notification(notify_db, notify_db_session, job=another_job)
create_notification(notify_db, notify_db_session, job=another_job)
auth_header = create_authorization_header()
@@ -454,7 +415,7 @@ def test_get_all_notifications_for_job_filtered_by_status(
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(
create_notification(
notify_db,
notify_db_session,
job=job,
@@ -492,14 +453,14 @@ def test_get_job_by_id_should_return_statistics(notify_db, notify_db_session, no
job_id = str(sample_job.id)
service_id = sample_job.service.id
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='created')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='sending')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='delivered')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='pending')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='failed')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='technical-failure') # noqa
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='temporary-failure') # noqa
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='permanent-failure') # noqa
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='created')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='sending')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='delivered')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='pending')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='failed')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='technical-failure') # noqa
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='temporary-failure') # noqa
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='permanent-failure') # noqa
with notify_api.test_request_context():
with notify_api.test_client() as client:
@@ -524,16 +485,16 @@ def test_get_job_by_id_should_return_summed_statistics(notify_db, notify_db_sess
job_id = str(sample_job.id)
service_id = sample_job.service.id
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='created')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='created')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='created')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='sending')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='failed')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='failed')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='failed')
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='technical-failure') # noqa
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='temporary-failure') # noqa
sample_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='temporary-failure') # noqa
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='created')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='created')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='created')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='sending')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='failed')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='failed')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='failed')
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='technical-failure') # noqa
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='temporary-failure') # noqa
create_notification(notify_db, notify_db_session, service=sample_job.service, job=sample_job, status='temporary-failure') # noqa
with notify_api.test_request_context():
with notify_api.test_client() as client:
@@ -551,22 +512,63 @@ def test_get_job_by_id_should_return_summed_statistics(notify_db, notify_db_sess
assert resp_json['data']['created_by']['name'] == 'Test User'
def test_get_jobs_for_service_should_return_statistics(notify_db, notify_db_session, notify_api, sample_service):
now = datetime.utcnow()
earlier = datetime.utcnow() - timedelta(days=1)
job_1 = sample_job(notify_db, notify_db_session, service=sample_service, created_at=earlier)
job_2 = sample_job(notify_db, notify_db_session, service=sample_service, created_at=now)
def test_get_jobs(notify_api, notify_db, notify_db_session, sample_template):
_setup_jobs(notify_db, notify_db_session, sample_template)
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_1, status='created')
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='sending')
sample_notification(notify_db, notify_db_session, service=sample_service, job=job_2, status='sending')
sample_notification(notify_db, notify_db_session, service=sample_service, job=job_2, status='sending')
service_id = sample_template.service.id
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = '/service/{}/job'.format(str(sample_service.id))
path = '/service/{}/job'.format(service_id)
auth_header = create_authorization_header(service_id=service_id)
response = client.get(path, headers=[auth_header])
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert len(resp_json['data']) == 5
def test_get_jobs_with_limit_days(notify_api, notify_db, notify_db_session, sample_template):
create_job(
notify_db,
notify_db_session,
service=sample_template.service,
template=sample_template,
)
create_job(
notify_db,
notify_db_session,
service=sample_template.service,
template=sample_template,
created_at=datetime.now() - timedelta(days=7))
service_id = sample_template.service.id
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = '/service/{}/job'.format(service_id)
auth_header = create_authorization_header(service_id=service_id)
response = client.get(path, headers=[auth_header], query_string={'limit_days': 5})
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert len(resp_json['data']) == 1
def test_get_jobs_should_return_statistics(notify_db, notify_db_session, notify_api, sample_service):
now = datetime.utcnow()
earlier = datetime.utcnow() - timedelta(days=1)
job_1 = create_job(notify_db, notify_db_session, service=sample_service, created_at=earlier)
job_2 = create_job(notify_db, notify_db_session, service=sample_service, created_at=now)
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_1, 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='sending')
create_notification(notify_db, notify_db_session, service=sample_service, job=job_2, status='sending')
create_notification(notify_db, notify_db_session, service=sample_service, job=job_2, status='sending')
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = '/service/{}/job'.format(sample_service.id)
auth_header = create_authorization_header(service_id=str(sample_service.id))
response = client.get(path, headers=[auth_header])
assert response.status_code == 200
@@ -578,7 +580,7 @@ def test_get_jobs_for_service_should_return_statistics(notify_db, notify_db_sess
assert {'status': 'created', 'count': 3} in resp_json['data'][1]['statistics']
def test_get_jobs_for_service_should_return_no_stats_if_no_rows_in_notifications(
def test_get_jobs_should_return_no_stats_if_no_rows_in_notifications(
notify_db,
notify_db_session,
notify_api,
@@ -586,12 +588,12 @@ def test_get_jobs_for_service_should_return_no_stats_if_no_rows_in_notifications
now = datetime.utcnow()
earlier = datetime.utcnow() - timedelta(days=1)
job_1 = sample_job(notify_db, notify_db_session, service=sample_service, created_at=earlier)
job_2 = sample_job(notify_db, notify_db_session, service=sample_service, created_at=now)
job_1 = create_job(notify_db, notify_db_session, service=sample_service, created_at=earlier)
job_2 = create_job(notify_db, notify_db_session, service=sample_service, created_at=now)
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = '/service/{}/job'.format(str(sample_service.id))
path = '/service/{}/job'.format(sample_service.id)
auth_header = create_authorization_header(service_id=str(sample_service.id))
response = client.get(path, headers=[auth_header])
assert response.status_code == 200
@@ -601,3 +603,60 @@ def test_get_jobs_for_service_should_return_no_stats_if_no_rows_in_notifications
assert resp_json['data'][0]['statistics'] == []
assert resp_json['data'][1]['id'] == str(job_1.id)
assert resp_json['data'][1]['statistics'] == []
def test_get_jobs_should_paginate(
notify_db,
notify_db_session,
client,
sample_template
):
create_10_jobs(notify_db, notify_db_session, sample_template.service, sample_template)
path = '/service/{}/job'.format(sample_template.service_id)
auth_header = create_authorization_header(service_id=str(sample_template.service_id))
with set_config(client.application, 'PAGE_SIZE', 2):
response = client.get(path, headers=[auth_header])
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert len(resp_json['data']) == 2
assert resp_json['data'][0]['created_at'] == '2015-01-01T10:00:00+00:00'
assert resp_json['data'][1]['created_at'] == '2015-01-01T09:00:00+00:00'
assert resp_json['page_size'] == 2
assert resp_json['total'] == 10
assert 'links' in resp_json
assert set(resp_json['links'].keys()) == {'next', 'last'}
def test_get_jobs_accepts_page_parameter(
notify_db,
notify_db_session,
client,
sample_template
):
create_10_jobs(notify_db, notify_db_session, sample_template.service, sample_template)
path = '/service/{}/job'.format(sample_template.service_id)
auth_header = create_authorization_header(service_id=str(sample_template.service_id))
with set_config(client.application, 'PAGE_SIZE', 2):
response = client.get(path, headers=[auth_header], query_string={'page': 2})
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert len(resp_json['data']) == 2
assert resp_json['data'][0]['created_at'] == '2015-01-01T08:00:00+00:00'
assert resp_json['data'][1]['created_at'] == '2015-01-01T07:00:00+00:00'
assert resp_json['page_size'] == 2
assert resp_json['total'] == 10
assert 'links' in resp_json
assert set(resp_json['links'].keys()) == {'prev', 'next', 'last'}
def create_10_jobs(db, session, service, template):
with freeze_time('2015-01-01T00:00:00') as the_time:
for _ in range(10):
the_time.tick(timedelta(hours=1))
create_job(db, session, service, template)

9
tests/conftest.py
View File

@@ -1,3 +1,4 @@
from contextlib import contextmanager
import os
import boto3
@@ -83,3 +84,11 @@ def pytest_generate_tests(metafunc):
argnames, testdata = idparametrize.args
ids, argvalues = zip(*sorted(testdata.items()))
metafunc.parametrize(argnames, argvalues, ids=ids)
@contextmanager
def set_config(app, name, value):
old_val = app.config.get(name)
app.config[name] = value
yield
app.config[name] = old_val