mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-23 08:51:30 -05:00
add day filter to template usage dao, and remove NotificationHistory
we should be very careful with when we get data from NotificationHistory - this should probably only be from scheduled tasks. `dao_get_template_usage` is only called from the template statistics rest endpoint, so shouldn't ever hit template history. also, moved tests out to new file to break up the 2k test file a bit
This commit is contained in:
Leo Hemsted
@@ -46,40 +46,34 @@ from app.models import (
|
||||
)
|
||||
|
||||
from app.dao.dao_utils import transactional
|
||||
from app.utils import convert_utc_to_bst
|
||||
from app.utils import convert_utc_to_bst, get_london_midnight_in_utc
|
||||
|
||||
|
||||
@statsd(namespace="dao")
|
||||
def dao_get_template_usage(service_id, limit_days=None):
|
||||
def dao_get_template_usage(service_id, limit_days=None, day=None):
|
||||
if bool(limit_days) == bool(day):
|
||||
raise ValueError('Must filter on either limit_days or a specific day')
|
||||
|
||||
query_filter = []
|
||||
|
||||
table = NotificationHistory
|
||||
if limit_days:
|
||||
query_filter.append(Notification.created_at >= days_ago(limit_days))
|
||||
else:
|
||||
start = get_london_midnight_in_utc(day)
|
||||
end = get_london_midnight_in_utc(day + timedelta(days=1))
|
||||
query_filter.append(Notification.created_at >= start)
|
||||
query_filter.append(Notification.created_at < end)
|
||||
|
||||
if limit_days is not None and limit_days <= 7:
|
||||
table = Notification
|
||||
|
||||
# only limit days if it's not seven days, as 7 days == the whole of Notifications table.
|
||||
if limit_days != 7:
|
||||
query_filter.append(table.created_at >= days_ago(limit_days))
|
||||
|
||||
elif limit_days is not None:
|
||||
# case where not under 7 days, so using NotificationsHistory so limit allowed
|
||||
query_filter.append(table.created_at >= days_ago(limit_days))
|
||||
|
||||
query_filter.append(table.service_id == service_id)
|
||||
query_filter.append(table.key_type != KEY_TYPE_TEST)
|
||||
|
||||
# only limit days if it's not seven days, as 7 days == the whole of Notifications table.
|
||||
if limit_days is not None and limit_days != 7:
|
||||
query_filter.append(table.created_at >= days_ago(limit_days))
|
||||
query_filter.append(Notification.service_id == service_id)
|
||||
query_filter.append(Notification.key_type != KEY_TYPE_TEST)
|
||||
|
||||
notifications_aggregate_query = db.session.query(
|
||||
func.count().label('count'),
|
||||
table.template_id
|
||||
Notification.template_id
|
||||
).filter(
|
||||
*query_filter
|
||||
).group_by(
|
||||
table.template_id
|
||||
Notification.template_id
|
||||
).subquery()
|
||||
|
||||
query = db.session.query(
|
||||
|
||||
@@ -35,7 +35,7 @@ def get_template_statistics_for_service_by_day(service_id):
|
||||
message = {'limit_days': [error]}
|
||||
raise InvalidRequest(message, status_code=400)
|
||||
else:
|
||||
limit_days = None
|
||||
limit_days = 7
|
||||
|
||||
if limit_days == 7:
|
||||
stats = get_template_statistics_for_7_days(limit_days, service_id)
|
||||
|
||||
@@ -83,325 +83,6 @@ def test_should_have_decorated_notifications_dao_functions():
|
||||
assert dao_delete_notifications_and_history_by_id.__wrapped__.__name__ == 'dao_delete_notifications_and_history_by_id' # noqa
|
||||
|
||||
|
||||
def test_should_be_able_to_get_template_usage_history(notify_db, notify_db_session, sample_service):
|
||||
with freeze_time('2000-01-01 12:00:00'):
|
||||
sms = create_sample_template(notify_db, notify_db_session)
|
||||
notification = sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
|
||||
results = dao_get_last_template_usage(sms.id, 'sms')
|
||||
assert results.template.name == 'Template Name'
|
||||
assert results.template.template_type == 'sms'
|
||||
assert results.created_at == datetime(year=2000, month=1, day=1, hour=12, minute=0, second=0)
|
||||
assert results.template_id == sms.id
|
||||
assert results.id == notification.id
|
||||
|
||||
|
||||
@pytest.mark.parametrize("notification_type",
|
||||
['sms', 'email', 'letter'])
|
||||
def test_should_be_able_to_get_all_template_usage_history_order_by_notification_created_at(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_service,
|
||||
notification_type
|
||||
):
|
||||
template = create_sample_template(notify_db, notify_db_session, template_type=notification_type)
|
||||
|
||||
sample_notification(notify_db, notify_db_session, service=sample_service, template=template)
|
||||
sample_notification(notify_db, notify_db_session, service=sample_service, template=template)
|
||||
sample_notification(notify_db, notify_db_session, service=sample_service, template=template)
|
||||
most_recent = sample_notification(notify_db, notify_db_session, service=sample_service, template=template)
|
||||
|
||||
results = dao_get_last_template_usage(template.id, notification_type)
|
||||
assert results.id == most_recent.id
|
||||
|
||||
|
||||
def test_template_usage_should_ignore_test_keys(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_team_api_key,
|
||||
sample_test_api_key
|
||||
):
|
||||
sms = create_sample_template(notify_db, notify_db_session)
|
||||
|
||||
one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
|
||||
two_minutes_ago = datetime.utcnow() - timedelta(minutes=2)
|
||||
|
||||
team_key = sample_notification(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
created_at=two_minutes_ago,
|
||||
template=sms,
|
||||
api_key=sample_team_api_key,
|
||||
key_type=KEY_TYPE_TEAM)
|
||||
sample_notification(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
created_at=one_minute_ago,
|
||||
template=sms,
|
||||
api_key=sample_test_api_key,
|
||||
key_type=KEY_TYPE_TEST)
|
||||
|
||||
results = dao_get_last_template_usage(sms.id, 'sms')
|
||||
assert results.id == team_key.id
|
||||
|
||||
|
||||
def test_should_be_able_to_get_no_template_usage_history_if_no_notifications_using_template(
|
||||
notify_db,
|
||||
notify_db_session):
|
||||
sms = create_sample_template(notify_db, notify_db_session)
|
||||
|
||||
results = dao_get_last_template_usage(sms.id, 'sms')
|
||||
assert not results
|
||||
|
||||
|
||||
def test_should_by_able_to_get_template_count(notify_db, notify_db_session, sample_service):
|
||||
sms = create_sample_template(notify_db, notify_db_session)
|
||||
email = sample_email_template(notify_db, notify_db_session)
|
||||
sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
|
||||
sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
|
||||
sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
|
||||
sample_notification(notify_db, notify_db_session, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, service=sample_service, template=email)
|
||||
|
||||
results = dao_get_template_usage(sample_service.id)
|
||||
assert results[0].name == 'Email Template Name'
|
||||
assert results[0].template_type == 'email'
|
||||
assert results[0].count == 2
|
||||
|
||||
assert results[1].name == 'Template Name'
|
||||
assert results[1].template_type == 'sms'
|
||||
assert results[1].count == 3
|
||||
|
||||
|
||||
def test_template_history_should_ignore_test_keys(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_team_api_key,
|
||||
sample_test_api_key,
|
||||
sample_api_key
|
||||
):
|
||||
sms = create_sample_template(notify_db, notify_db_session)
|
||||
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, template=sms, api_key=sample_api_key, key_type=KEY_TYPE_NORMAL)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, template=sms, api_key=sample_team_api_key, key_type=KEY_TYPE_TEAM)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, template=sms, api_key=sample_test_api_key, key_type=KEY_TYPE_TEST)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, template=sms)
|
||||
|
||||
results = dao_get_template_usage(sms.service_id)
|
||||
assert results[0].name == 'Template Name'
|
||||
assert results[0].template_type == 'sms'
|
||||
assert results[0].count == 3
|
||||
|
||||
|
||||
def test_should_by_able_to_get_template_count_limited_for_service(
|
||||
notify_db,
|
||||
notify_db_session):
|
||||
service_1 = sample_service(notify_db, notify_db_session, service_name="test1", email_from="test1")
|
||||
service_2 = sample_service(notify_db, notify_db_session, service_name="test2", email_from="test2")
|
||||
service_3 = sample_service(notify_db, notify_db_session, service_name="test3", email_from="test3")
|
||||
|
||||
sms = create_sample_template(notify_db, notify_db_session)
|
||||
|
||||
sample_notification(notify_db, notify_db_session, service=service_1, template=sms)
|
||||
sample_notification(notify_db, notify_db_session, service=service_1, template=sms)
|
||||
sample_notification(notify_db, notify_db_session, service=service_2, template=sms)
|
||||
|
||||
assert dao_get_template_usage(service_1.id)[0].count == 2
|
||||
assert dao_get_template_usage(service_2.id)[0].count == 1
|
||||
assert len(dao_get_template_usage(service_3.id)) == 0
|
||||
|
||||
|
||||
def test_should_by_able_to_get_zero_count_from_notifications_history_if_no_rows(sample_service):
|
||||
results = dao_get_template_usage(sample_service.id)
|
||||
assert len(results) == 0
|
||||
|
||||
|
||||
def test_should_by_able_to_get_zero_count_from_notifications_history_if_no_service():
|
||||
results = dao_get_template_usage(str(uuid.uuid4()))
|
||||
assert len(results) == 0
|
||||
|
||||
|
||||
def test_should_by_able_to_get_template_count_across_days(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_service):
|
||||
sms = create_sample_template(notify_db, notify_db_session)
|
||||
email = sample_email_template(notify_db, notify_db_session)
|
||||
|
||||
today = datetime.now()
|
||||
yesterday = datetime.now() - timedelta(days=1)
|
||||
one_month_ago = datetime.now() - timedelta(days=30)
|
||||
|
||||
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=sms)
|
||||
|
||||
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=sms)
|
||||
|
||||
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
|
||||
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
|
||||
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
|
||||
|
||||
results = dao_get_template_usage(sample_service.id)
|
||||
|
||||
assert len(results) == 2
|
||||
|
||||
assert [(row.name, row.template_type, row.count) for row in results] == [
|
||||
('Email Template Name', 'email', 5),
|
||||
('Template Name', 'sms', 5)
|
||||
]
|
||||
|
||||
|
||||
def test_should_by_able_to_get_template_count_for_under_seven_days(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_service,
|
||||
sample_template):
|
||||
|
||||
yesterday = datetime.now() - timedelta(days=1)
|
||||
six_days_ago = datetime.now() - timedelta(days=6)
|
||||
seven_days_ago = datetime.now() - timedelta(days=7)
|
||||
eight_days_ago = datetime.now() - timedelta(days=8)
|
||||
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=sample_template
|
||||
)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=six_days_ago, service=sample_service, template=sample_template
|
||||
)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=seven_days_ago, service=sample_service, template=sample_template
|
||||
)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=eight_days_ago, service=sample_service, template=sample_template
|
||||
)
|
||||
|
||||
results = dao_get_template_usage(sample_service.id, limit_days=6)
|
||||
assert len(results) == 1
|
||||
assert [(row.name, row.template_type, row.count) for row in results] == [
|
||||
('Template Name', 'sms', 2)
|
||||
]
|
||||
|
||||
|
||||
def test_should_by_able_to_get_template_count_for_whole_of_notifications_table_if_seven_days_exactly(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_service,
|
||||
sample_template):
|
||||
yesterday = datetime.now() - timedelta(days=1)
|
||||
six_days_ago = datetime.now() - timedelta(days=6)
|
||||
seven_days_ago = datetime.now() - timedelta(days=7)
|
||||
eight_days_ago = datetime.now() - timedelta(days=8)
|
||||
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=sample_template
|
||||
)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=six_days_ago, service=sample_service, template=sample_template
|
||||
)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=seven_days_ago, service=sample_service, template=sample_template
|
||||
)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=eight_days_ago, service=sample_service, template=sample_template
|
||||
)
|
||||
|
||||
results = dao_get_template_usage(sample_service.id, limit_days=7)
|
||||
assert len(results) == 1
|
||||
# note as we haven't run the delete task they'll ALL be in the notifications table.
|
||||
assert [(row.name, row.template_type, row.count) for row in results] == [
|
||||
('Template Name', 'sms', 4)
|
||||
]
|
||||
|
||||
|
||||
def test_should_by_able_to_get_all_template_count_for_more_than_seven_days(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_service,
|
||||
sample_template):
|
||||
yesterday = datetime.now() - timedelta(days=1)
|
||||
six_days_ago = datetime.now() - timedelta(days=6)
|
||||
seven_days_ago = datetime.now() - timedelta(days=7)
|
||||
eight_days_ago = datetime.now() - timedelta(days=8)
|
||||
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=sample_template
|
||||
)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=six_days_ago, service=sample_service, template=sample_template
|
||||
)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=seven_days_ago, service=sample_service, template=sample_template
|
||||
)
|
||||
sample_notification(
|
||||
notify_db, notify_db_session, created_at=eight_days_ago, service=sample_service, template=sample_template
|
||||
)
|
||||
|
||||
Notification.query.delete()
|
||||
# gets all from history table
|
||||
results = dao_get_template_usage(sample_service.id, limit_days=10)
|
||||
assert len(results) == 1
|
||||
assert [(row.name, row.template_type, row.count) for row in results] == [
|
||||
('Template Name', 'sms', 4)
|
||||
]
|
||||
|
||||
|
||||
def test_should_by_able_to_get_template_count_from_notifications_history_with_day_limit(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_service):
|
||||
sms = create_sample_template(notify_db, notify_db_session)
|
||||
|
||||
email = sample_email_template(notify_db, notify_db_session)
|
||||
|
||||
today = datetime.now()
|
||||
yesterday = datetime.now() - timedelta(days=1)
|
||||
one_month_ago = datetime.now() - timedelta(days=30)
|
||||
|
||||
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=sms)
|
||||
|
||||
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
|
||||
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=sms)
|
||||
|
||||
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
|
||||
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
|
||||
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
|
||||
|
||||
results_day_one = dao_get_template_usage(sample_service.id, limit_days=0)
|
||||
assert len(results_day_one) == 2
|
||||
|
||||
results_day_two = dao_get_template_usage(sample_service.id, limit_days=1)
|
||||
assert len(results_day_two) == 2
|
||||
|
||||
results_day_30 = dao_get_template_usage(sample_service.id, limit_days=31)
|
||||
assert len(results_day_30) == 2
|
||||
|
||||
assert [(row.name, row.template_type, row.count) for row in results_day_one] == [
|
||||
('Email Template Name', 'email', 2),
|
||||
('Template Name', 'sms', 1)
|
||||
]
|
||||
|
||||
assert [(row.name, row.template_type, row.count) for row in results_day_two] == [
|
||||
('Email Template Name', 'email', 5),
|
||||
('Template Name', 'sms', 2),
|
||||
]
|
||||
|
||||
assert [(row.name, row.template_type, row.count) for row in results_day_30] == [
|
||||
('Email Template Name', 'email', 5),
|
||||
('Template Name', 'sms', 5),
|
||||
]
|
||||
|
||||
|
||||
def test_should_by_able_to_update_status_by_reference(sample_email_template, ses_provider):
|
||||
data = _notification_json(sample_email_template, status='sending')
|
||||
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
import uuid
|
||||
from datetime import datetime, timedelta, date
|
||||
|
||||
import pytest
|
||||
from freezegun import freeze_time
|
||||
|
||||
from app.dao.notifications_dao import (
|
||||
dao_get_last_template_usage,
|
||||
dao_get_template_usage
|
||||
)
|
||||
from app.models import (
|
||||
KEY_TYPE_NORMAL,
|
||||
KEY_TYPE_TEST,
|
||||
KEY_TYPE_TEAM
|
||||
)
|
||||
from tests.app.db import (
|
||||
create_notification,
|
||||
create_service,
|
||||
create_template
|
||||
)
|
||||
|
||||
|
||||
def test_last_template_usage_should_get_right_data(sample_notification):
|
||||
results = dao_get_last_template_usage(sample_notification.template_id, 'sms')
|
||||
assert results.template.name == 'Template Name'
|
||||
assert results.template.template_type == 'sms'
|
||||
assert results.created_at == sample_notification.created_at
|
||||
assert results.template_id == sample_notification.template_id
|
||||
assert results.id == sample_notification.id
|
||||
|
||||
|
||||
@pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
|
||||
def test_last_template_usage_should_be_able_to_get_all_template_usage_history_order_by_notification_created_at(
|
||||
sample_service,
|
||||
notification_type
|
||||
):
|
||||
template = create_template(sample_service, template_type=notification_type)
|
||||
|
||||
create_notification(template)
|
||||
create_notification(template)
|
||||
create_notification(template)
|
||||
most_recent = create_notification(template)
|
||||
|
||||
results = dao_get_last_template_usage(template.id, notification_type)
|
||||
assert results.id == most_recent.id
|
||||
|
||||
|
||||
def test_last_template_usage_should_ignore_test_keys(
|
||||
sample_template,
|
||||
sample_team_api_key,
|
||||
sample_test_api_key
|
||||
):
|
||||
one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
|
||||
two_minutes_ago = datetime.utcnow() - timedelta(minutes=2)
|
||||
|
||||
team_key = create_notification(
|
||||
template=sample_template,
|
||||
created_at=two_minutes_ago,
|
||||
api_key=sample_team_api_key)
|
||||
create_notification(
|
||||
template=sample_template,
|
||||
created_at=one_minute_ago,
|
||||
api_key=sample_test_api_key)
|
||||
|
||||
results = dao_get_last_template_usage(sample_template.id, 'sms')
|
||||
assert results.id == team_key.id
|
||||
|
||||
|
||||
def test_last_template_usage_should_be_able_to_get_no_template_usage_history_if_no_notifications_using_template(
|
||||
sample_template):
|
||||
results = dao_get_last_template_usage(sample_template.id, 'sms')
|
||||
assert not results
|
||||
|
||||
|
||||
def test_should_by_able_to_get_template_count(sample_template, sample_email_template):
|
||||
create_notification(sample_template)
|
||||
create_notification(sample_template)
|
||||
create_notification(sample_template)
|
||||
create_notification(sample_email_template)
|
||||
create_notification(sample_email_template)
|
||||
|
||||
results = dao_get_template_usage(sample_template.service_id, limit_days=1)
|
||||
assert results[0].name == 'Email Template Name'
|
||||
assert results[0].template_type == 'email'
|
||||
assert results[0].count == 2
|
||||
|
||||
assert results[1].name == 'Template Name'
|
||||
assert results[1].template_type == 'sms'
|
||||
assert results[1].count == 3
|
||||
|
||||
|
||||
def test_template_usage_should_ignore_test_keys(
|
||||
sample_team_api_key,
|
||||
sample_test_api_key,
|
||||
sample_api_key,
|
||||
sample_template
|
||||
):
|
||||
|
||||
create_notification(sample_template, api_key=sample_api_key, key_type=KEY_TYPE_NORMAL)
|
||||
create_notification(sample_template, api_key=sample_team_api_key, key_type=KEY_TYPE_TEAM)
|
||||
create_notification(sample_template, api_key=sample_test_api_key, key_type=KEY_TYPE_TEST)
|
||||
create_notification(sample_template)
|
||||
|
||||
results = dao_get_template_usage(sample_template.service_id, limit_days=1)
|
||||
assert results[0].name == 'Template Name'
|
||||
assert results[0].template_type == 'sms'
|
||||
assert results[0].count == 3
|
||||
|
||||
|
||||
def test_template_usage_should_filter_by_service(notify_db_session):
|
||||
service_1 = create_service(service_name='test1')
|
||||
service_2 = create_service(service_name='test2')
|
||||
service_3 = create_service(service_name='test3')
|
||||
|
||||
template_1 = create_template(service_1)
|
||||
template_2 = create_template(service_2) # noqa
|
||||
template_3 = create_template(service_3)
|
||||
|
||||
# two for service_1, one for service_3
|
||||
create_notification(template_1)
|
||||
create_notification(template_1)
|
||||
|
||||
create_notification(template_3)
|
||||
|
||||
res1 = dao_get_template_usage(service_1.id, limit_days=1)
|
||||
res2 = dao_get_template_usage(service_2.id, limit_days=1)
|
||||
res3 = dao_get_template_usage(service_3.id, limit_days=1)
|
||||
|
||||
assert len(res1) == 1
|
||||
assert len(res2) == 0
|
||||
assert len(res3) == 1
|
||||
assert res1[0].count == 2
|
||||
assert res3[0].count == 1
|
||||
|
||||
|
||||
def test_template_usage_should_by_able_to_get_zero_count_from_notifications_history_if_no_rows(sample_service):
|
||||
results = dao_get_template_usage(sample_service.id, limit_days=1)
|
||||
assert len(results) == 0
|
||||
|
||||
|
||||
def test_template_usage_should_by_able_to_get_zero_count_from_notifications_history_if_no_service():
|
||||
results = dao_get_template_usage(str(uuid.uuid4()), limit_days=1)
|
||||
assert len(results) == 0
|
||||
|
||||
|
||||
@freeze_time('2017-06-10T12:00:00')
|
||||
def test_template_usage_should_by_able_to_get_template_count_with_limit_days(sample_template):
|
||||
# too early
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 7, 22, 59, 0))
|
||||
# just right
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 7, 23, 0, 0))
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 7, 23, 0, 0))
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 8, 22, 59, 0))
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 8, 22, 59, 0))
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 8, 22, 59, 0))
|
||||
# next day, still included
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 8, 23, 0, 0))
|
||||
|
||||
results = dao_get_template_usage(sample_template.service_id, limit_days=2)
|
||||
|
||||
assert len(results) == 1
|
||||
assert results[0].count == 6
|
||||
|
||||
|
||||
@freeze_time('2017-06-10T12:00:00')
|
||||
def test_template_usage_should_by_able_to_get_template_count_for_specific_day(sample_template):
|
||||
# too early
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 7, 22, 59, 0))
|
||||
# just right
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 7, 23, 0, 0))
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 7, 23, 0, 0))
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 8, 22, 59, 0))
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 8, 22, 59, 0))
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 8, 22, 59, 0))
|
||||
# too late
|
||||
create_notification(sample_template, created_at=datetime(2017, 6, 8, 23, 0, 0))
|
||||
|
||||
results = dao_get_template_usage(sample_template.service_id, day=date(2017, 6, 8))
|
||||
|
||||
assert len(results) == 1
|
||||
assert results[0].count == 5
|
||||
|
||||
|
||||
@pytest.mark.parametrize('kwargs', [
|
||||
{},
|
||||
{'limit_days': 0},
|
||||
{'limit_days': 1, 'day': date(2017, 1, 1)}
|
||||
])
|
||||
def test_template_usage_rejects_invalid_params(kwargs):
|
||||
with pytest.raises(ValueError):
|
||||
dao_get_template_usage(uuid.uuid4(), **kwargs)
|
||||
Reference in New Issue
Block a user