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

Merge pull request #1833 from alphagov/read-redis

Browse Source 
Read template usage stats from new redis keys
This commit is contained in:
Leo Hemsted
2018-04-30 15:15:16 +01:00
committed by GitHub
parent 55236c3173 e35c1bde77
commit 238e85ea9e
11 changed files with 604 additions and 739 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
app/dao/__init__.py
View File

@@ -28,8 +28,3 @@ class DAOClass(object):
db.session.delete(inst)
if _commit:
db.session.commit()
def days_ago(number_of_days):
from datetime import date, timedelta
return date.today() - timedelta(days=number_of_days)

6
app/dao/jobs_dao.py
View File

@@ -4,15 +4,13 @@ from datetime import datetime, timedelta
from flask import current_app
from notifications_utils.statsd_decorators import statsd
from sqlalchemy import (
Date as sql_date,
asc,
cast,
desc,
func,
)
from app import db
from app.dao import days_ago
from app.utils import midnight_n_days_ago
from app.models import (
Job,
JOB_STATUS_PENDING,
@@ -53,7 +51,7 @@ def dao_get_jobs_by_service_id(service_id, limit_days=None, page=1, page_size=50
Job.original_file_name != current_app.config['ONE_OFF_MESSAGE_FILENAME'],
]
if limit_days is not None:
query_filter.append(cast(Job.created_at, sql_date) >= days_ago(limit_days))
query_filter.append(Job.created_at >= midnight_n_days_ago(limit_days))
if statuses is not None and statuses != ['']:
query_filter.append(
Job.job_status.in_(statuses)

49
app/dao/notifications_dao.py
View File

@@ -21,7 +21,7 @@ from sqlalchemy.sql import functions
from notifications_utils.international_billing_rates import INTERNATIONAL_BILLING_RATES
from app import db, create_uuid
from app.dao import days_ago
from app.utils import midnight_n_days_ago
from app.errors import InvalidRequest
from app.models import (
Notification,
@@ -46,51 +46,37 @@ 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):
query_filter = []
table = NotificationHistory
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))
def dao_get_template_usage(service_id, day):
start = get_london_midnight_in_utc(day)
end = get_london_midnight_in_utc(day + timedelta(days=1))
notifications_aggregate_query = db.session.query(
func.count().label('count'),
table.template_id
Notification.template_id
).filter(
*query_filter
Notification.created_at >= start,
Notification.created_at < end,
Notification.service_id == service_id,
Notification.key_type != KEY_TYPE_TEST,
).group_by(
table.template_id
Notification.template_id
).subquery()
query = db.session.query(
Template.id.label('template_id'),
Template.id,
Template.name,
Template.template_type,
Template.is_precompiled_letter,
notifications_aggregate_query.c.count
).join(
func.coalesce(notifications_aggregate_query.c.count, 0).label('count')
).outerjoin(
notifications_aggregate_query,
notifications_aggregate_query.c.template_id == Template.id
).filter(
Template.service_id == service_id
).order_by(Template.name)
return query.all()
@@ -262,8 +248,7 @@ def get_notifications_for_service(
filters = [Notification.service_id == service_id]
if limit_days is not None:
days_ago = date.today() - timedelta(days=limit_days)
filters.append(func.date(Notification.created_at) >= days_ago)
filters.append(Notification.created_at >= midnight_n_days_ago(limit_days))
if older_than is not None:
older_than_created_at = db.session.query(

32
app/dao/templates_dao.py
View File

@@ -2,7 +2,6 @@ from datetime import datetime
import uuid
from sqlalchemy import asc, desc
from sqlalchemy.sql.expression import bindparam
from app import db
from app.models import (
@@ -124,25 +123,16 @@ def dao_get_template_versions(service_id, template_id):
).all()
def dao_get_templates_for_cache(cache):
if not cache or len(cache) == 0:
return []
# First create a subquery that is a union select of the cache values
# Then join templates to the subquery
cache_queries = [
db.session.query(bindparam("template_id" + str(i),
uuid.UUID(template_id.decode())).label('template_id'),
bindparam("count" + str(i), int(count.decode())).label('count'))
for i, (template_id, count) in enumerate(cache)]
cache_subq = cache_queries[0].union(*cache_queries[1:]).subquery()
query = db.session.query(Template.id.label('template_id'),
Template.template_type,
Template.name,
Template.is_precompiled_letter,
cache_subq.c.count.label('count')
).join(cache_subq,
Template.id == cache_subq.c.template_id
).order_by(Template.name)
def dao_get_multiple_template_details(template_ids):
query = db.session.query(
Template.id,
Template.template_type,
Template.name,
Template.is_precompiled_letter
).filter(
Template.id.in_(template_ids)
).order_by(
Template.name
)
return query.all()

115
app/template_statistics/rest.py
View File

@@ -2,7 +2,8 @@ from flask import (
Blueprint,
jsonify,
request,
current_app)
current_app
)
from app import redis_store
from app.dao.notifications_dao import (
@@ -10,15 +11,16 @@ from app.dao.notifications_dao import (
dao_get_last_template_usage
)
from app.dao.templates_dao import (
dao_get_templates_for_cache,
dao_get_multiple_template_details,
dao_get_template_by_id_and_service_id
)
from app.schemas import notification_with_template_schema
from app.utils import cache_key_for_service_template_counter
from app.utils import cache_key_for_service_template_usage_per_day, last_n_days
from app.errors import register_errors, InvalidRequest
from collections import Counter
template_statistics = Blueprint('template-statistics',
template_statistics = Blueprint('template_statistics',
__name__,
url_prefix='/service/<service_id>/template-statistics')
@@ -27,40 +29,22 @@ register_errors(template_statistics)
@template_statistics.route('')
def get_template_statistics_for_service_by_day(service_id):
if request.args.get('limit_days'):
try:
limit_days = int(request.args['limit_days'])
except ValueError as e:
error = '{} is not an integer'.format(request.args['limit_days'])
message = {'limit_days': [error]}
raise InvalidRequest(message, status_code=400)
else:
limit_days = None
try:
limit_days = int(request.args.get('limit_days', ''))
except ValueError:
error = '{} is not an integer'.format(request.args.get('limit_days'))
message = {'limit_days': [error]}
raise InvalidRequest(message, status_code=400)
if limit_days == 7:
stats = get_template_statistics_for_7_days(limit_days, service_id)
else:
stats = dao_get_template_usage(service_id, limit_days=limit_days)
if limit_days < 1 or limit_days > 7:
raise InvalidRequest({'limit_days': ['limit_days must be between 1 and 7']}, status_code=400)
def serialize(data):
return {
'count': data.count,
'template_id': str(data.template_id),
'template_name': data.name,
'template_type': data.template_type,
'is_precompiled_letter': data.is_precompiled_letter
}
return jsonify(data=[serialize(row) for row in stats])
return jsonify(data=_get_template_statistics_for_last_n_days(service_id, limit_days))
@template_statistics.route('/<template_id>')
def get_template_statistics_for_template_id(service_id, template_id):
template = dao_get_template_by_id_and_service_id(template_id, service_id)
if not template:
message = 'No template found for id {}'.format(template_id)
errors = {'template_id': [message]}
raise InvalidRequest(errors, status_code=404)
data = None
notification = dao_get_last_template_usage(template_id, template.template_type)
@@ -70,31 +54,46 @@ def get_template_statistics_for_template_id(service_id, template_id):
return jsonify(data=data)
def get_template_statistics_for_7_days(limit_days, service_id):
cache_key = cache_key_for_service_template_counter(service_id)
template_stats_by_id = redis_store.get_all_from_hash(cache_key)
if not template_stats_by_id:
stats = dao_get_template_usage(service_id, limit_days=limit_days)
cache_values = dict([(x.template_id, x.count) for x in stats])
if cache_values:
redis_store.set_hash_and_expire(cache_key,
cache_values,
current_app.config['EXPIRE_CACHE_TEN_MINUTES'])
else:
stats = dao_get_templates_for_cache(template_stats_by_id.items())
return stats
def _get_template_statistics_for_last_n_days(service_id, limit_days):
template_stats_by_id = Counter()
# TODO: can only switch to this code when redis has been populated (either through time passing or a manual step)
# from collections import Counter
# from notifications_utils.redis_client import RedisException
# template_stats_by_id = Counter()
# for day in last_7_days:
# # "<SERVICE_ID>-template-usage-{YYYY-MM-DD}"
# key = cache_key_for_service_templates_used_per_day(service_id, limit_days)
# try:
# template_stats_by_id += Counter(redis_store.get_all_from_hash(key, raise_exception=True))
# except RedisException:
# # TODO: ????
#
# # TODO: streamline db query and avoid weird unions if possible.
# return dao_get_templates_for_cache(template_stats_by_id.items())
for day in last_n_days(limit_days):
# "{SERVICE_ID}-template-usage-{YYYY-MM-DD}"
key = cache_key_for_service_template_usage_per_day(service_id, day)
stats = redis_store.get_all_from_hash(key)
if stats:
stats = {
k.decode('utf-8'): int(v) for k, v in stats.items()
}
else:
# key didn't exist (or redis was down) - lets populate from DB.
stats = {
str(row.id): row.count for row in dao_get_template_usage(service_id, day=day)
}
# if there is data in db, but not in redis - lets put it in redis so we don't have to do
# this calc again next time. If there isn't any data, we can't put it in redis.
# Zero length hashes aren't a thing in redis. (There'll only be no data if the service has no templates)
# Nothing is stored if redis is down.
if stats:
redis_store.set_hash_and_expire(
key,
stats,
current_app.config['EXPIRE_CACHE_EIGHT_DAYS']
)
template_stats_by_id += Counter(stats)
# attach count from stats to name/type/etc from database
template_details = dao_get_multiple_template_details(template_stats_by_id.keys())
return [
{
'count': template_stats_by_id[str(template.id)],
'template_id': str(template.id),
'template_name': template.name,
'template_type': template.template_type,
'is_precompiled_letter': template.is_precompiled_letter
}
for template in template_details
# we don't want to return templates with no count to the front-end,
# but they're returned from the DB and might be put in redis like that (if there was no data that day)
if template_stats_by_id[str(template.id)] != 0
]

27
app/utils.py
View File

@@ -39,7 +39,7 @@ def get_london_midnight_in_utc(date):
This function converts date to midnight as BST (British Standard Time) to UTC,
the tzinfo is lastly removed from the datetime because the database stores the timestamps without timezone.
:param date: the day to calculate the London midnight in UTC for
:return: the datetime of London midnight in UTC, for example 2016-06-17 = 2016-06-17 23:00:00
:return: the datetime of London midnight in UTC, for example 2016-06-17 = 2016-06-16 23:00:00
"""
return local_timezone.localize(datetime.combine(date, datetime.min.time())).astimezone(
pytz.UTC).replace(
@@ -80,6 +80,9 @@ def cache_key_for_service_template_counter(service_id, limit_days=7):
def cache_key_for_service_template_usage_per_day(service_id, datetime):
"""
You should pass a BST datetime into this function
"""
return "service-{}-template-usage-{}".format(service_id, datetime.date().isoformat())
@@ -90,3 +93,25 @@ def get_public_notify_type_text(notify_type, plural=False):
notify_type_text = 'text message'
return '{}{}'.format(notify_type_text, 's' if plural else '')
def midnight_n_days_ago(number_of_days):
"""
Returns midnight a number of days ago. Takes care of daylight savings etc.
"""
return get_london_midnight_in_utc(datetime.utcnow() - timedelta(days=number_of_days))
def last_n_days(limit_days):
"""
Returns the last n dates, oldest first. Takes care of daylight savings (but returns a date, be careful how you
manipulate it later! Don't directly use the date for comparing to UTC datetimes!). Includes today.
"""
return [
datetime.combine(
(convert_utc_to_bst(datetime.utcnow()) - timedelta(days=x)),
datetime.min.time()
)
# reverse the countdown, -1 from first two args to ensure it stays 0-indexed
for x in range(limit_days - 1, -1, -1)
]

319
tests/app/dao/notification_dao/test_notification_dao.py
View File

@@ -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')

186
tests/app/dao/notification_dao/test_notification_dao_template_usage.py Normal file
View File

@@ -0,0 +1,186 @@
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
@freeze_time('2018-01-01')
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, date.today())
assert results[0].name == sample_email_template.name
assert results[0].template_type == sample_email_template.template_type
assert results[0].count == 2
assert results[1].name == sample_template.name
assert results[1].template_type == sample_template.template_type
assert results[1].count == 3
@freeze_time('2018-01-01')
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, date.today())
assert results[0].name == sample_template.name
assert results[0].template_type == sample_template.template_type
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_3a = create_template(service_3, template_name='a')
template_3b = create_template(service_3, template_name='b') # noqa
# two for service_1, one for service_3
create_notification(template_1)
create_notification(template_1)
create_notification(template_3a)
res1 = dao_get_template_usage(service_1.id, date.today())
res2 = dao_get_template_usage(service_2.id, date.today())
res3 = dao_get_template_usage(service_3.id, date.today())
assert len(res1) == 1
assert res1[0].count == 2
assert len(res2) == 1
assert res2[0].count == 0
assert len(res3) == 2
assert res3[0].count == 1
assert res3[1].count == 0
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, date.today())
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()), date.today())
assert len(results) == 0
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
def test_template_usage_should_by_able_to_get_template_count_for_specific_timezone_boundary(sample_template):
# too early
create_notification(sample_template, created_at=datetime(2018, 3, 24, 23, 59, 0))
# just right
create_notification(sample_template, created_at=datetime(2018, 3, 25, 0, 0, 0))
create_notification(sample_template, created_at=datetime(2018, 3, 25, 0, 0, 0))
create_notification(sample_template, created_at=datetime(2018, 3, 25, 22, 59, 0))
create_notification(sample_template, created_at=datetime(2018, 3, 25, 22, 59, 0))
create_notification(sample_template, created_at=datetime(2018, 3, 25, 22, 59, 0))
# too late
create_notification(sample_template, created_at=datetime(2018, 3, 25, 23, 0, 0))
results = dao_get_template_usage(sample_template.service_id, day=date(2018, 3, 25))
assert len(results) == 1
assert results[0].count == 5

88
tests/app/dao/test_templates_dao.py
View File

@@ -10,14 +10,13 @@ from app.dao.templates_dao import (
dao_get_all_templates_for_service,
dao_update_template,
dao_get_template_versions,
dao_get_templates_for_cache,
dao_get_multiple_template_details,
dao_redact_template, dao_update_template_reply_to
)
from app.models import (
Template,
TemplateHistory,
TemplateRedacted,
PRECOMPILED_TEMPLATE_NAME
TemplateRedacted
)
from tests.app.conftest import sample_template as create_sample_template
@@ -481,77 +480,16 @@ def test_get_template_versions_is_empty_for_hidden_templates(notify_db, notify_d
assert len(versions) == 0
def test_get_templates_by_ids_successful(notify_db, notify_db_session):
template_1 = create_sample_template(
notify_db,
notify_db_session,
template_name='Sample Template 1',
template_type="sms",
content="Template content"
)
template_2 = create_sample_template(
notify_db,
notify_db_session,
template_name='Sample Template 2',
template_type="sms",
content="Template content"
)
create_sample_template(
notify_db,
notify_db_session,
template_name='Sample Template 3',
template_type="email",
content="Template content"
)
sample_cache_dict = {str.encode(str(template_1.id)): str.encode('2'),
str.encode(str(template_2.id)): str.encode('3')}
cache = [[k, v] for k, v in sample_cache_dict.items()]
templates = dao_get_templates_for_cache(cache)
assert len(templates) == 2
assert [(template_1.id, template_1.template_type, template_1.name, False, 2),
(template_2.id, template_2.template_type, template_2.name, False, 3)] == templates
def test_get_multiple_template_details_returns_templates_for_list_of_ids(sample_service):
t1 = create_template(sample_service)
t2 = create_template(sample_service)
create_template(sample_service) # t3
res = dao_get_multiple_template_details([t1.id, t2.id])
def test_get_letter_templates_by_ids_successful(notify_db, notify_db_session):
template_1 = create_sample_template(
notify_db,
notify_db_session,
template_name=PRECOMPILED_TEMPLATE_NAME,
template_type="letter",
content="Template content",
hidden=True
)
template_2 = create_sample_template(
notify_db,
notify_db_session,
template_name='Sample Template 2',
template_type="letter",
content="Template content"
)
sample_cache_dict = {str.encode(str(template_1.id)): str.encode('2'),
str.encode(str(template_2.id)): str.encode('3')}
cache = [[k, v] for k, v in sample_cache_dict.items()]
templates = dao_get_templates_for_cache(cache)
assert len(templates) == 2
assert [(template_1.id, template_1.template_type, template_1.name, True, 2),
(template_2.id, template_2.template_type, template_2.name, False, 3)] == templates
def test_get_templates_by_ids_successful_for_one_cache_item(notify_db, notify_db_session):
template_1 = create_sample_template(
notify_db,
notify_db_session,
template_name='Sample Template 1',
template_type="sms",
content="Template content"
)
sample_cache_dict = {str.encode(str(template_1.id)): str.encode('2')}
cache = [[k, v] for k, v in sample_cache_dict.items()]
templates = dao_get_templates_for_cache(cache)
assert len(templates) == 1
assert [(template_1.id, template_1.template_type, template_1.name, False, 2)] == templates
def test_get_templates_by_ids_returns_empty_list():
assert dao_get_templates_for_cache({}) == []
assert dao_get_templates_for_cache(None) == []
assert {x.id for x in res} == {t1.id, t2.id}
# make sure correct properties are on each row
assert res[0].id
assert res[0].template_type
assert res[0].name
assert not res[0].is_precompiled_letter

469
tests/app/template_statistics/test_rest.py
View File

@@ -1,281 +1,304 @@
from datetime import datetime, timedelta
import json
import uuid
from datetime import datetime
from unittest.mock import Mock, call, ANY
import pytest
from flask import current_app
from freezegun import freeze_time
from app.dao.templates_dao import dao_update_template
from tests import create_authorization_header
from tests.app.conftest import (
sample_template as create_sample_template,
sample_notification,
sample_notification_history,
sample_email_template
from tests.app.db import (
create_notification,
create_template,
)
def test_get_all_template_statistics_with_bad_arg_returns_400(client, sample_service):
auth_header = create_authorization_header()
def set_up_get_all_from_hash(mock_redis, side_effect):
"""
redis returns binary strings for both keys and values - so given a list of side effects (return values),
make sure
"""
assert type(side_effect) == list
side_effects = []
for ret_val in side_effect:
if ret_val is None:
side_effects.append(None)
else:
side_effects += [{str(k).encode('utf-8'): str(v).encode('utf-8') for k, v in ret_val.items()}]
response = client.get(
'/service/{}/template-statistics'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
query_string={'limit_days': 'blurk'}
mock_redis.get_all_from_hash.side_effect = side_effects
# get_template_statistics_for_service_by_day
@pytest.mark.parametrize('query_string', [
{},
{'limit_days': 0},
{'limit_days': 8},
{'limit_days': 3.5},
{'limit_days': 'blurk'},
])
def test_get_template_statistics_for_service_by_day_with_bad_arg_returns_400(admin_request, query_string):
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_service_by_day',
service_id=uuid.uuid4(),
**query_string,
_expected_status=400
)
assert response.status_code == 400
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == {'limit_days': ['blurk is not an integer']}
assert 'limit_days' in json_resp['message']
@freeze_time('2016-08-18')
def test_get_template_statistics_for_service(notify_db, notify_db_session, client, mocker):
email, sms = set_up_notifications(notify_db, notify_db_session)
mocked_redis = mocker.patch('app.redis_store.get_all_from_hash')
auth_header = create_authorization_header()
response = client.get(
'/service/{}/template-statistics'.format(email.service_id),
headers=[('Content-Type', 'application/json'), auth_header]
def test_get_template_statistics_for_service_by_day_returns_template_info(admin_request, mocker, sample_notification):
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_service_by_day',
service_id=sample_notification.service_id,
limit_days=1
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 2
assert len(json_resp['data']) == 1
assert json_resp['data'][0]['count'] == 1
assert json_resp['data'][0]['template_id'] == str(sample_notification.template_id)
assert json_resp['data'][0]['template_name'] == 'Template Name'
assert json_resp['data'][0]['template_type'] == 'sms'
assert json_resp['data'][0]['is_precompiled_letter'] is False
@freeze_time('2018-01-01 12:00:00')
def test_get_template_statistics_for_service_by_day_gets_out_of_redis_if_available(
admin_request,
mocker,
sample_template
):
mock_redis = mocker.patch('app.template_statistics.rest.redis_store')
set_up_get_all_from_hash(mock_redis, [
{sample_template.id: 3}
])
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_service_by_day',
service_id=sample_template.service_id,
limit_days=1
)
assert len(json_resp['data']) == 1
assert json_resp['data'][0]['count'] == 3
assert json_resp['data'][0]['template_id'] == str(email.id)
assert json_resp['data'][0]['template_name'] == email.name
assert json_resp['data'][0]['template_type'] == email.template_type
assert json_resp['data'][1]['count'] == 3
assert json_resp['data'][1]['template_id'] == str(sms.id)
assert json_resp['data'][1]['template_name'] == sms.name
assert json_resp['data'][1]['template_type'] == sms.template_type
mocked_redis.assert_not_called()
@freeze_time('2016-08-18')
def test_get_template_statistics_for_service_limited_1_day(notify_db, notify_db_session, client,
mocker):
email, sms = set_up_notifications(notify_db, notify_db_session)
mock_redis = mocker.patch('app.redis_store.get_all_from_hash')
auth_header = create_authorization_header()
response = client.get(
'/service/{}/template-statistics'.format(email.service_id),
headers=[('Content-Type', 'application/json'), auth_header],
query_string={'limit_days': 1}
assert json_resp['data'][0]['template_id'] == str(sample_template.id)
mock_redis.get_all_from_hash.assert_called_once_with(
'service-{}-template-usage-{}'.format(sample_template.service_id, '2018-01-01')
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))['data']
assert len(json_resp) == 2
assert json_resp[0]['count'] == 1
assert json_resp[0]['template_id'] == str(email.id)
assert json_resp[0]['template_name'] == email.name
assert json_resp[0]['template_type'] == email.template_type
assert json_resp[1]['count'] == 1
assert json_resp[1]['template_id'] == str(sms.id)
assert json_resp[1]['template_name'] == sms.name
assert json_resp[1]['template_type'] == sms.template_type
@freeze_time('2018-01-02 12:00:00')
def test_get_template_statistics_for_service_by_day_goes_to_db_if_not_in_redis(
admin_request,
mocker,
sample_template
):
mock_redis = mocker.patch('app.template_statistics.rest.redis_store')
mock_redis.assert_not_called()
@pytest.mark.parametrize("cache_values", [False, True])
@freeze_time('2016-08-18')
def test_get_template_statistics_for_service_limit_7_days(notify_db, notify_db_session, client,
mocker,
cache_values):
email, sms = set_up_notifications(notify_db, notify_db_session)
mock_cache_values = {str.encode(str(sms.id)): str.encode('3'),
str.encode(str(email.id)): str.encode('3')} if cache_values else None
mocked_redis_get = mocker.patch('app.redis_store.get_all_from_hash', return_value=mock_cache_values)
mocked_redis_set = mocker.patch('app.redis_store.set_hash_and_expire')
auth_header = create_authorization_header()
response_for_a_week = client.get(
'/service/{}/template-statistics'.format(email.service_id),
headers=[('Content-Type', 'application/json'), auth_header],
query_string={'limit_days': 7}
# first time it is called redis returns data, second time returns none
set_up_get_all_from_hash(mock_redis, [
{sample_template.id: 2},
None
])
mock_dao = mocker.patch(
'app.template_statistics.rest.dao_get_template_usage',
return_value=[
Mock(id=sample_template.id, count=3)
]
)
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_service_by_day',
service_id=sample_template.service_id,
limit_days=2
)
assert len(json_resp['data']) == 1
assert json_resp['data'][0]['count'] == 5
assert json_resp['data'][0]['template_id'] == str(sample_template.id)
# first redis call
assert mock_redis.get_all_from_hash.mock_calls == [
call('service-{}-template-usage-{}'.format(sample_template.service_id, '2018-01-01')),
call('service-{}-template-usage-{}'.format(sample_template.service_id, '2018-01-02'))
]
# dao only called for 2nd, since redis returned values for first call
mock_dao.assert_called_once_with(
str(sample_template.service_id), day=datetime(2018, 1, 2)
)
mock_redis.set_hash_and_expire.assert_called_once_with(
'service-{}-template-usage-{}'.format(sample_template.service_id, '2018-01-02'),
# sets the data that the dao returned
{str(sample_template.id): 3},
current_app.config['EXPIRE_CACHE_EIGHT_DAYS']
)
def test_get_template_statistics_for_service_by_day_combines_templates_correctly(
admin_request,
mocker,
sample_service
):
t1 = create_template(sample_service, template_name='1')
t2 = create_template(sample_service, template_name='2')
t3 = create_template(sample_service, template_name='3') # noqa
mock_redis = mocker.patch('app.template_statistics.rest.redis_store')
# first time it is called redis returns data, second time returns none
set_up_get_all_from_hash(mock_redis, [
{t1.id: 2},
None,
{t1.id: 1, t2.id: 4},
])
mock_dao = mocker.patch(
'app.template_statistics.rest.dao_get_template_usage',
return_value=[
Mock(id=t1.id, count=8)
]
)
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_service_by_day',
service_id=sample_service.id,
limit_days=3
)
assert response_for_a_week.status_code == 200
json_resp = json.loads(response_for_a_week.get_data(as_text=True))
assert len(json_resp['data']) == 2
assert json_resp['data'][0]['count'] == 3
assert json_resp['data'][0]['template_name'] == 'New Email Template Name'
assert json_resp['data'][1]['count'] == 3
assert json_resp['data'][1]['template_name'] == 'New SMS Template Name'
assert json_resp['data'][0]['template_id'] == str(t1.id)
assert json_resp['data'][0]['count'] == 11
assert json_resp['data'][1]['template_id'] == str(t2.id)
assert json_resp['data'][1]['count'] == 4
mocked_redis_get.assert_called_once_with("{}-template-counter-limit-7-days".format(email.service_id))
if cache_values:
mocked_redis_set.assert_not_called()
else:
mocked_redis_set.assert_called_once_with("{}-template-counter-limit-7-days".format(email.service_id),
{sms.id: 3, email.id: 3}, 600)
assert mock_redis.get_all_from_hash.call_count == 3
# dao only called for 2nd day
assert mock_dao.call_count == 1
@freeze_time('2016-08-18')
def test_get_template_statistics_for_service_limit_30_days(notify_db, notify_db_session, client,
mocker):
email, sms = set_up_notifications(notify_db, notify_db_session)
mock_redis = mocker.patch('app.redis_store.get_all_from_hash')
@freeze_time('2018-03-28 00:00:00')
def test_get_template_statistics_for_service_by_day_gets_stats_for_correct_days(
admin_request,
mocker,
sample_template
):
mock_redis = mocker.patch('app.template_statistics.rest.redis_store')
auth_header = create_authorization_header()
response_for_a_month = client.get(
'/service/{}/template-statistics'.format(email.service_id),
headers=[('Content-Type', 'application/json'), auth_header],
query_string={'limit_days': 30}
# first time it is called redis returns data, second time returns none
set_up_get_all_from_hash(mock_redis, [
{sample_template.id: 1},
None,
{sample_template.id: 1},
{sample_template.id: 1},
{sample_template.id: 1},
None,
None,
])
mock_dao = mocker.patch(
'app.template_statistics.rest.dao_get_template_usage',
return_value=[
Mock(id=sample_template.id, count=2)
]
)
assert response_for_a_month.status_code == 200
json_resp = json.loads(response_for_a_month.get_data(as_text=True))
assert len(json_resp['data']) == 2
assert json_resp['data'][0]['count'] == 3
assert json_resp['data'][0]['template_name'] == 'New Email Template Name'
assert json_resp['data'][1]['count'] == 3
assert json_resp['data'][1]['template_name'] == 'New SMS Template Name'
mock_redis.assert_not_called()
@freeze_time('2016-08-18')
def test_get_template_statistics_for_service_no_limit(notify_db, notify_db_session, client,
mocker):
email, sms = set_up_notifications(notify_db, notify_db_session)
mock_redis = mocker.patch('app.redis_store.get_all_from_hash')
auth_header = create_authorization_header()
response_for_all = client.get(
'/service/{}/template-statistics'.format(email.service_id),
headers=[('Content-Type', 'application/json'), auth_header]
)
assert response_for_all.status_code == 200
json_resp = json.loads(response_for_all.get_data(as_text=True))
assert len(json_resp['data']) == 2
assert json_resp['data'][0]['count'] == 3
assert json_resp['data'][0]['template_name'] == 'New Email Template Name'
assert json_resp['data'][1]['count'] == 3
assert json_resp['data'][1]['template_name'] == 'New SMS Template Name'
mock_redis.assert_not_called()
def set_up_notifications(notify_db, notify_db_session):
sms = create_sample_template(notify_db, notify_db_session)
email = sample_email_template(notify_db, notify_db_session)
today = datetime.now()
a_week_ago = datetime.now() - timedelta(days=7)
a_month_ago = datetime.now() - timedelta(days=30)
sample_notification(notify_db, notify_db_session, created_at=a_month_ago, template=sms)
sample_notification(notify_db, notify_db_session, created_at=a_month_ago, template=email)
email.name = 'Updated Email Template Name'
dao_update_template(email)
sms.name = 'Updated SMS Template Name'
dao_update_template(sms)
sample_notification(notify_db, notify_db_session, created_at=a_week_ago, template=sms)
sample_notification(notify_db, notify_db_session, created_at=a_week_ago, template=email)
email.name = 'New Email Template Name'
dao_update_template(email)
sms.name = 'New SMS Template Name'
dao_update_template(sms)
sample_notification(notify_db, notify_db_session, created_at=today, template=sms)
sample_notification(notify_db, notify_db_session, created_at=today, template=email)
return email, sms
@freeze_time('2016-08-18')
def test_returns_empty_list_if_no_templates_used(client, sample_service, mocker):
auth_header = create_authorization_header()
mock_redis = mocker.patch('app.redis_store.set_hash_and_expire')
response = client.get(
'/service/{}/template-statistics'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header]
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_service_by_day',
service_id=sample_template.service_id,
limit_days=7
)
assert len(json_resp['data']) == 1
assert json_resp['data'][0]['count'] == 10
assert json_resp['data'][0]['template_id'] == str(sample_template.id)
assert mock_redis.get_all_from_hash.call_count == 7
assert '2018-03-22' in mock_redis.get_all_from_hash.mock_calls[0][1][0]
assert '2018-03-23' in mock_redis.get_all_from_hash.mock_calls[1][1][0]
assert '2018-03-24' in mock_redis.get_all_from_hash.mock_calls[2][1][0]
assert '2018-03-25' in mock_redis.get_all_from_hash.mock_calls[3][1][0]
assert '2018-03-26' in mock_redis.get_all_from_hash.mock_calls[4][1][0]
assert '2018-03-27' in mock_redis.get_all_from_hash.mock_calls[5][1][0]
assert '2018-03-28' in mock_redis.get_all_from_hash.mock_calls[6][1][0]
mock_dao.mock_calls == [
call(ANY, day=datetime(2018, 3, 23)),
call(ANY, day=datetime(2018, 3, 27)),
call(ANY, day=datetime(2018, 3, 28))
]
def test_get_template_statistics_for_service_by_day_returns_empty_list_if_no_templates(
admin_request,
mocker,
sample_service
):
mock_redis = mocker.patch('app.template_statistics.rest.redis_store')
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_service_by_day',
service_id=sample_service.id,
limit_days=7
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 0
mock_redis.assert_not_called()
assert mock_redis.get_all_from_hash.call_count == 7
# make sure we don't try and set any empty hashes in redis
assert mock_redis.set_hash_and_expire.call_count == 0
# get_template_statistics_for_template
def test_get_template_statistics_by_id_returns_last_notification(
notify_db,
notify_db_session,
client):
sample_notification(notify_db, notify_db_session)
sample_notification(notify_db, notify_db_session)
notification_3 = sample_notification(notify_db, notify_db_session)
def test_get_template_statistics_for_template_returns_last_notification(admin_request, sample_template):
create_notification(sample_template)
create_notification(sample_template)
notification_3 = create_notification(sample_template)
auth_header = create_authorization_header()
response = client.get(
'/service/{}/template-statistics/{}'.format(notification_3.service_id, notification_3.template_id),
headers=[('Content-Type', 'application/json'), auth_header],
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_template_id',
service_id=notification_3.service_id,
template_id=notification_3.template_id
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))['data']
assert json_resp['id'] == str(notification_3.id)
assert json_resp['data']['id'] == str(notification_3.id)
def test_get_template_statistics_for_template_returns_empty_if_no_statistics(
client,
admin_request,
sample_template,
):
auth_header = create_authorization_header()
response = client.get(
'/service/{}/template-statistics/{}'.format(sample_template.service_id, sample_template.id),
headers=[('Content-Type', 'application/json'), auth_header],
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_template_id',
service_id=sample_template.service_id,
template_id=sample_template.id
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert not json_resp['data']
def test_get_template_statistics_raises_error_for_nonexistent_template(
client,
def test_get_template_statistics_for_template_raises_error_for_nonexistent_template(
admin_request,
sample_service,
fake_uuid
):
auth_header = create_authorization_header()
response = client.get(
'/service/{}/template-statistics/{}'.format(sample_service.id, fake_uuid),
headers=[('Content-Type', 'application/json'), auth_header],
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_template_id',
service_id=sample_service.id,
template_id=fake_uuid,
_expected_status=404
)
assert response.status_code == 404
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['message'] == 'No result found'
assert json_resp['result'] == 'error'
def test_get_template_statistics_by_id_returns_empty_for_old_notification(
notify_db,
notify_db_session,
client,
sample_template
def test_get_template_statistics_for_template_returns_empty_for_old_notification(
admin_request,
sample_notification_history
):
sample_notification_history(notify_db, notify_db_session, sample_template)
auth_header = create_authorization_header()
response = client.get(
'/service/{}/template-statistics/{}'.format(sample_template.service.id, sample_template.id),
headers=[('Content-Type', 'application/json'), auth_header],
json_resp = admin_request.get(
'template_statistics.get_template_statistics_for_template_id',
service_id=sample_notification_history.service_id,
template_id=sample_notification_history.template_id
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))['data']
assert not json_resp
assert not json_resp['data']

47
tests/app/test_utils.py
View File

@@ -1,11 +1,16 @@
from datetime import datetime
import pytest
from freezegun import freeze_time
from app.utils import (
get_london_midnight_in_utc,
get_midnight_for_day_before,
convert_utc_to_bst,
convert_bst_to_utc)
convert_bst_to_utc,
midnight_n_days_ago,
last_n_days
)
@pytest.mark.parametrize('date, expected_date', [
@@ -44,3 +49,43 @@ def test_convert_bst_to_utc():
bst_datetime = datetime.strptime(bst, "%Y-%m-%d %H:%M")
utc = convert_bst_to_utc(bst_datetime)
assert utc == datetime(2017, 5, 12, 12, 15)
@pytest.mark.parametrize('current_time, arg, expected_datetime', [
# winter
('2018-01-10 23:59', 1, datetime(2018, 1, 9, 0, 0)),
('2018-01-11 00:00', 1, datetime(2018, 1, 10, 0, 0)),
# bst switchover at 1am 25th
('2018-03-25 10:00', 1, datetime(2018, 3, 24, 0, 0)),
('2018-03-26 10:00', 1, datetime(2018, 3, 25, 0, 0)),
('2018-03-27 10:00', 1, datetime(2018, 3, 25, 23, 0)),
# summer
('2018-06-05 10:00', 1, datetime(2018, 6, 3, 23, 0)),
# zero days ago
('2018-01-11 00:00', 0, datetime(2018, 1, 11, 0, 0)),
('2018-06-05 10:00', 0, datetime(2018, 6, 4, 23, 0)),
])
def test_midnight_n_days_ago(current_time, arg, expected_datetime):
with freeze_time(current_time):
assert midnight_n_days_ago(arg) == expected_datetime
def test_last_n_days():
with freeze_time('2018-03-27 12:00'):
res = last_n_days(5)
assert res == [
datetime(2018, 3, 23, 0, 0),
datetime(2018, 3, 24, 0, 0),
datetime(2018, 3, 25, 0, 0),
datetime(2018, 3, 26, 0, 0),
datetime(2018, 3, 27, 0, 0)
]
@pytest.mark.parametrize('arg', [0, -1])
def test_last_n_days_invalid_arg(arg):
assert last_n_days(arg) == []