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

carving out precompile, almost completely

Browse Source
This commit is contained in:
stvnrlly
2023-02-15 16:45:40 -05:00
parent 4da08adff2
commit 518e2264a6
14 changed files with 22 additions and 610 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/dao/fact_notification_status_dao.py
+2 -7
View File
@@ -156,7 +156,7 @@ def fetch_notification_status_for_service_for_today_and_7_previous_days(service_
query = db.session.query(
*([
Template.name.label("template_name"),
Template.is_precompiled_letter,
False, # TODO: this is related to is_precompiled_letter
all_stats_table.c.template_id
] if by_template else []),
all_stats_table.c.notification_type,
@@ -168,7 +168,7 @@ def fetch_notification_status_for_service_for_today_and_7_previous_days(service_
query = query.filter(all_stats_table.c.template_id == Template.id)
return query.group_by(
*([Template.name, Template.is_precompiled_letter, all_stats_table.c.template_id] if by_template else []),
*([Template.name, all_stats_table.c.template_id] if by_template else []),
all_stats_table.c.notification_type,
all_stats_table.c.status,
).all()
@@ -333,7 +333,6 @@ def fetch_monthly_template_usage_for_service(start_date, end_date, service_id):
FactNotificationStatus.template_id.label('template_id'),
Template.name.label('name'),
Template.template_type.label('template_type'),
Template.is_precompiled_letter.label('is_precompiled_letter'),
extract('month', FactNotificationStatus.local_date).label('month'),
extract('year', FactNotificationStatus.local_date).label('year'),
func.sum(FactNotificationStatus.notification_count).label('count')
@@ -349,7 +348,6 @@ def fetch_monthly_template_usage_for_service(start_date, end_date, service_id):
FactNotificationStatus.template_id,
Template.name,
Template.template_type,
Template.is_precompiled_letter,
extract('month', FactNotificationStatus.local_date).label('month'),
extract('year', FactNotificationStatus.local_date).label('year'),
).order_by(
@@ -366,7 +364,6 @@ def fetch_monthly_template_usage_for_service(start_date, end_date, service_id):
Notification.template_id.label('template_id'),
Template.name.label('name'),
Template.template_type.label('template_type'),
Template.is_precompiled_letter.label('is_precompiled_letter'),
extract('month', month).label('month'),
extract('year', month).label('year'),
func.count().label('count')
@@ -389,7 +386,6 @@ def fetch_monthly_template_usage_for_service(start_date, end_date, service_id):
query = db.session.query(
all_stats_table.c.template_id,
all_stats_table.c.name,
all_stats_table.c.is_precompiled_letter,
all_stats_table.c.template_type,
func.cast(all_stats_table.c.month, Integer).label('month'),
func.cast(all_stats_table.c.year, Integer).label('year'),
@@ -397,7 +393,6 @@ def fetch_monthly_template_usage_for_service(start_date, end_date, service_id):
).group_by(
all_stats_table.c.template_id,
all_stats_table.c.name,
all_stats_table.c.is_precompiled_letter,
all_stats_table.c.template_type,
all_stats_table.c.month,
all_stats_table.c.year,
app/models.py
-13
View File
@@ -30,7 +30,6 @@ from sqlalchemy import (
from sqlalchemy.dialects.postgresql import JSON, JSONB, UUID
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import validates
from sqlalchemy.orm.collections import attribute_mapped_collection
@@ -296,7 +295,6 @@ INTERNATIONAL_SMS_TYPE = 'international_sms'
INBOUND_SMS_TYPE = 'inbound_sms'
SCHEDULE_NOTIFICATIONS = 'schedule_notifications'
EMAIL_AUTH = 'email_auth'
PRECOMPILED_LETTER = 'precompiled_letter'
UPLOAD_DOCUMENT = 'upload_document'
EDIT_FOLDER_PERMISSIONS = 'edit_folder_permissions'
UPLOAD_LETTERS = 'upload_letters'
@@ -880,9 +878,6 @@ template_folder_map = db.Table(
)
PRECOMPILED_TEMPLATE_NAME = 'Pre-compiled PDF'
class TemplateBase(db.Model):
__abstract__ = True
@@ -960,14 +955,6 @@ class TemplateBase(db.Model):
else:
return None
@hybrid_property
def is_precompiled_letter(self):
return self.hidden and self.name == PRECOMPILED_TEMPLATE_NAME and self.template_type == LETTER_TYPE
@is_precompiled_letter.setter
def is_precompiled_letter(self, value):
pass
def _as_utils_template(self):
if self.template_type == EMAIL_TYPE:
return PlainTextEmailTemplate(self.__dict__)
app/schemas.py
+1 -1
View File
@@ -389,7 +389,7 @@ class TemplateSchemaNested(TemplateSchema):
is_precompiled_letter = fields.Method('get_is_precompiled_letter')
def get_is_precompiled_letter(self, template):
return template.is_precompiled_letter
return False
class TemplateSchemaNoDetail(TemplateSchema):
app/service/rest.py
+1 -2
View File
@@ -706,7 +706,6 @@ def get_monthly_template_usage(service_id):
'month': i.month,
'year': i.year,
'count': i.count,
'is_precompiled_letter': i.is_precompiled_letter
}
)
@@ -1057,7 +1056,7 @@ def get_returned_letters(service_id):
'client_reference': x.client_reference if x.api_key_id else None,
'reported_at': x.reported_at.strftime(DATE_FORMAT),
'created_at': x.created_at.strftime(DATETIME_FORMAT_NO_TIMEZONE),
# it doesn't make sense to show hidden/precompiled templates
# it doesn't make sense to show hidden templates
'template_name': x.template_name if not x.hidden else None,
'template_id': x.template_id if not x.hidden else None,
'template_version': x.template_version if not x.hidden else None,
app/template_statistics/rest.py
-1
View File
@@ -37,7 +37,6 @@ def get_template_statistics_for_service_by_day(service_id):
'template_id': str(row.template_id),
'template_name': row.template_name,
'template_type': row.notification_type,
'is_precompiled_letter': row.is_precompiled_letter,
'status': row.status
}
for row in data
app/utils.py
+1 -3
View File
@@ -86,14 +86,12 @@ def get_local_month_from_utc_column(column):
def get_public_notify_type_text(notify_type, plural=False):
from app.models import PRECOMPILED_LETTER, SMS_TYPE, UPLOAD_DOCUMENT
from app.models import SMS_TYPE, UPLOAD_DOCUMENT
notify_type_text = notify_type
if notify_type == SMS_TYPE:
notify_type_text = 'text message'
elif notify_type == UPLOAD_DOCUMENT:
notify_type_text = 'document'
elif notify_type == PRECOMPILED_LETTER:
notify_type_text = 'precompiled letter'
return '{}{}'.format(notify_type_text, 's' if plural else '')
app/v2/notifications/notification_schemas.py
-13
View File
@@ -231,19 +231,6 @@ post_letter_request = {
"additionalProperties": False
}
post_precompiled_letter_request = {
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "POST precompiled letter notification schema",
"type": "object",
"title": "POST v2/notifications/letter",
"properties": {
"reference": {"type": "string"},
"content": {"type": "string"},
},
"required": ["reference", "content"],
"additionalProperties": False
}
letter_content = {
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "Letter content for POST letter notification",
app/v2/notifications/post_notifications.py
-52
View File
@@ -19,18 +19,11 @@ from app.config import QueueNames
from app.models import (
EMAIL_TYPE,
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
NOTIFICATION_CREATED,
NOTIFICATION_DELIVERED,
NOTIFICATION_SENDING,
PRIORITY,
SMS_TYPE,
Notification,
)
from app.notifications.process_letter_notifications import (
create_letter_notification,
)
from app.notifications.process_notifications import (
persist_notification,
send_notification_to_queue_detached,
@@ -280,51 +273,6 @@ def process_document_uploads(personalisation_data, service, simulated=False):
return personalisation_data, len(file_keys)
# TODO: remove precompiled var
def process_letter_notification(
*, letter_data, api_key, service, template, template_with_content, reply_to_text, precompiled=False
):
if api_key.key_type == KEY_TYPE_TEAM:
raise BadRequestError(message='Cannot send letters with a team api key', status_code=403)
if not service.research_mode and service.restricted and api_key.key_type != KEY_TYPE_TEST:
raise BadRequestError(message='Cannot send letters when service is in trial mode', status_code=403)
test_key = api_key.key_type == KEY_TYPE_TEST
status = NOTIFICATION_CREATED
updated_at = None
if test_key:
# if we don't want to actually send the letter, then start it off in SENDING so we don't pick it up
if current_app.config['NOTIFY_ENVIRONMENT'] in ['preview', 'development']:
status = NOTIFICATION_SENDING
# mark test letter as delivered and do not create a fake response later
else:
status = NOTIFICATION_DELIVERED
updated_at = datetime.utcnow()
notification = create_letter_notification(letter_data=letter_data,
service=service,
template=template,
api_key=api_key,
status=status,
reply_to_text=reply_to_text,
updated_at=updated_at
)
resp = create_response_for_post_notification(
notification_id=notification.id,
client_reference=notification.client_reference,
template_id=notification.template_id,
template_version=notification.template_version,
notification_type=notification.notification_type,
reply_to=reply_to_text,
service_id=notification.service_id,
template_with_content=template_with_content
)
return resp
def get_reply_to_text(notification_type, form, template):
reply_to = None
if notification_type == EMAIL_TYPE:
tests/app/dao/test_fact_notification_status_dao.py
-4
View File
@@ -402,14 +402,12 @@ def test_fetch_monthly_template_usage_for_service(sample_service):
assert results[0].template_id == template_one.id
assert results[0].name == template_one.name
assert results[0].is_precompiled_letter is False
assert results[0].template_type == template_one.template_type
assert results[0].month == 12
assert results[0].year == 2017
assert results[0].count == 6
assert results[1].template_id == template_two.id
assert results[1].name == template_two.name
assert results[1].is_precompiled_letter is False
assert results[1].template_type == template_two.template_type
assert results[1].month == 12
assert results[1].year == 2017
@@ -417,7 +415,6 @@ def test_fetch_monthly_template_usage_for_service(sample_service):
assert results[2].template_id == template_one.id
assert results[2].name == template_one.name
assert results[2].is_precompiled_letter is False
assert results[2].template_type == template_one.template_type
assert results[2].month == 1
assert results[2].year == 2018
@@ -425,7 +422,6 @@ def test_fetch_monthly_template_usage_for_service(sample_service):
assert results[3].template_id == template_three.id
assert results[3].name == template_three.name
assert results[3].is_precompiled_letter is False
assert results[3].template_type == template_three.template_type
assert results[3].month == 3
assert results[3].year == 2018
tests/app/service/test_rest.py
+13 -250
View File
@@ -27,7 +27,6 @@ from app.models import (
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
LETTER_TYPE,
NOTIFICATION_RETURNED_LETTER,
SMS_TYPE,
UPLOAD_LETTERS,
AnnualBilling,
@@ -45,19 +44,15 @@ from app.models import (
from tests import create_admin_authorization_header
from tests.app.db import (
create_annual_billing,
create_api_key,
create_domain,
create_email_branding,
create_ft_billing,
create_ft_notification_status,
create_inbound_number,
create_job,
create_letter_contact,
create_notification,
create_notification_history,
create_organisation,
create_reply_to_email,
create_returned_letter,
create_service,
create_service_sms_sender,
create_service_with_defined_sms_sender,
@@ -2465,31 +2460,23 @@ def test_get_all_notifications_for_service_includes_template_redacted(admin_requ
assert resp['notifications'][1]['template']['redact_personalisation'] is True
def test_get_all_notifications_for_service_includes_template_hidden(admin_request, sample_service):
letter_template = create_template(sample_service, template_type=LETTER_TYPE)
precompiled_template = create_template(
sample_service,
template_type=LETTER_TYPE,
template_name='Pre-compiled PDF',
subject='Pre-compiled PDF',
hidden=True
)
# TODO: check whether all hidden templates are also precompiled letters
# def test_get_all_notifications_for_service_includes_template_hidden(admin_request, sample_service):
# letter_template = create_template(sample_service, template_type=LETTER_TYPE)
with freeze_time('2000-01-01'):
letter_noti = create_notification(letter_template)
with freeze_time('2000-01-02'):
precompiled_noti = create_notification(precompiled_template)
# with freeze_time('2000-01-01'):
# letter_noti = create_notification(letter_template)
resp = admin_request.get(
'service.get_all_notifications_for_service',
service_id=sample_service.id
)
# resp = admin_request.get(
# 'service.get_all_notifications_for_service',
# service_id=sample_service.id
# )
assert resp['notifications'][0]['id'] == str(precompiled_noti.id)
assert resp['notifications'][0]['template']['is_precompiled_letter'] is True
# assert resp['notifications'][0]['id'] == str(precompiled_noti.id)
# assert resp['notifications'][0]['template']['is_precompiled_letter'] is True
assert resp['notifications'][1]['id'] == str(letter_noti.id)
assert resp['notifications'][1]['template']['is_precompiled_letter'] is False
# assert resp['notifications'][1]['id'] == str(letter_noti.id)
# assert resp['notifications'][1]['template']['is_precompiled_letter'] is False
def test_search_for_notification_by_to_field_returns_personlisation(
@@ -3279,47 +3266,6 @@ def test_cancel_notification_for_service_raises_invalid_request_when_notificatio
assert response['result'] == 'error'
@pytest.mark.skip(reason="Needs updating for TTS: Remove letters")
@pytest.mark.parametrize('notification_status', ['created', 'pending-virus-check'])
@freeze_time('2018-07-07 16:00:00')
def test_cancel_notification_for_service_updates_letter_if_letter_is_in_cancellable_state(
admin_request,
sample_letter_notification,
notification_status,
):
sample_letter_notification.status = notification_status
sample_letter_notification.created_at = datetime.now()
response = admin_request.post(
'service.cancel_notification_for_service',
service_id=sample_letter_notification.service_id,
notification_id=sample_letter_notification.id,
)
assert response['status'] == 'cancelled'
@pytest.mark.skip(reason="Needs updating for TTS: Remove letters")
@pytest.mark.parametrize('created_at', [
datetime(2018, 7, 6, 22, 30), # yesterday evening
datetime(2018, 7, 6, 23, 30), # this morning early hours (in bst)
datetime(2018, 7, 7, 10, 0), # this morning normal hours
])
@freeze_time('2018-7-7 16:00:00')
def test_cancel_notification_for_service_updates_letter_if_still_time_to_cancel(
admin_request,
sample_letter_notification,
created_at,
):
sample_letter_notification.created_at = created_at
response = admin_request.post(
'service.cancel_notification_for_service',
service_id=sample_letter_notification.service_id,
notification_id=sample_letter_notification.id,
)
assert response['status'] == 'cancelled'
def test_get_monthly_notification_data_by_service(sample_service, admin_request):
create_ft_notification_status(date(2019, 4, 17), notification_type='letter', service=sample_service,
notification_status='delivered')
@@ -3335,186 +3281,3 @@ def test_get_monthly_notification_data_by_service(sample_service, admin_request)
['2019-03-01', str(sample_service.id), 'Sample service', 'email', 4, 0, 0, 0, 0, 0],
['2019-04-01', str(sample_service.id), 'Sample service', 'letter', 0, 1, 0, 0, 0, 0],
]
@freeze_time('2019-12-11 13:30')
def test_get_returned_letter_statistics(admin_request, sample_service):
create_returned_letter(sample_service, reported_at=datetime.utcnow() - timedelta(days=3))
create_returned_letter(sample_service, reported_at=datetime.utcnow() - timedelta(days=2))
create_returned_letter(sample_service, reported_at=datetime.utcnow() - timedelta(days=1))
response = admin_request.get('service.returned_letter_statistics', service_id=sample_service.id)
assert response == {
'returned_letter_count': 3,
'most_recent_report': '2019-12-10 00:00:00.000000'
}
@freeze_time('2019-12-11 13:30')
def test_get_returned_letter_statistics_with_old_returned_letters(
mocker,
admin_request,
sample_service,
):
create_returned_letter(sample_service, reported_at=datetime.utcnow() - timedelta(days=8))
create_returned_letter(sample_service, reported_at=datetime.utcnow() - timedelta(days=800))
count_mock = mocker.patch(
'app.service.rest.fetch_recent_returned_letter_count',
)
assert admin_request.get(
'service.returned_letter_statistics',
service_id=sample_service.id,
) == {
'returned_letter_count': 0,
'most_recent_report': '2019-12-03 00:00:00.000000',
}
assert count_mock.called is False
def test_get_returned_letter_statistics_with_no_returned_letters(
mocker,
admin_request,
sample_service,
):
count_mock = mocker.patch(
'app.service.rest.fetch_recent_returned_letter_count',
)
assert admin_request.get(
'service.returned_letter_statistics',
service_id=sample_service.id,
) == {
'returned_letter_count': 0,
'most_recent_report': None,
}
assert count_mock.called is False
@freeze_time('2019-12-11 13:30')
def test_get_returned_letter_summary(admin_request, sample_service):
create_returned_letter(sample_service, reported_at=datetime.utcnow() - timedelta(days=3))
create_returned_letter(sample_service, reported_at=datetime.utcnow())
create_returned_letter(sample_service, reported_at=datetime.utcnow())
response = admin_request.get('service.returned_letter_summary', service_id=sample_service.id)
assert len(response) == 2
assert response[0] == {'returned_letter_count': 2, 'reported_at': '2019-12-11'}
assert response[1] == {'returned_letter_count': 1, 'reported_at': '2019-12-08'}
@freeze_time('2019-12-11 13:30')
def test_get_returned_letter(admin_request, sample_letter_template):
job = create_job(template=sample_letter_template)
letter_from_job = create_notification(template=sample_letter_template, client_reference='letter_from_job',
status=NOTIFICATION_RETURNED_LETTER,
job=job, job_row_number=2,
created_at=datetime.utcnow() - timedelta(days=1),
created_by_id=sample_letter_template.service.users[0].id)
create_returned_letter(service=sample_letter_template.service, reported_at=datetime.utcnow(),
notification_id=letter_from_job.id)
one_off_letter = create_notification(template=sample_letter_template,
status=NOTIFICATION_RETURNED_LETTER,
created_at=datetime.utcnow() - timedelta(days=2),
created_by_id=sample_letter_template.service.users[0].id)
create_returned_letter(service=sample_letter_template.service, reported_at=datetime.utcnow(),
notification_id=one_off_letter.id)
api_key = create_api_key(service=sample_letter_template.service)
api_letter = create_notification(template=sample_letter_template, client_reference='api_letter',
status=NOTIFICATION_RETURNED_LETTER,
created_at=datetime.utcnow() - timedelta(days=3),
api_key=api_key)
create_returned_letter(service=sample_letter_template.service, reported_at=datetime.utcnow(),
notification_id=api_letter.id)
precompiled_template = create_template(service=sample_letter_template.service, template_type='letter', hidden=True,
template_name='hidden template')
precompiled_letter = create_notification_history(template=precompiled_template, api_key=api_key,
client_reference='precompiled letter',
created_at=datetime.utcnow() - timedelta(days=4))
create_returned_letter(service=sample_letter_template.service, reported_at=datetime.utcnow(),
notification_id=precompiled_letter.id)
uploaded_letter = create_notification_history(template=precompiled_template, client_reference='filename.pdf',
created_at=datetime.utcnow() - timedelta(days=5),
created_by_id=sample_letter_template.service.users[0].id)
create_returned_letter(service=sample_letter_template.service, reported_at=datetime.utcnow(),
notification_id=uploaded_letter.id)
not_included_in_results_template = create_template(service=create_service(service_name='not included in results'),
template_type='letter')
letter_4 = create_notification_history(template=not_included_in_results_template,
status=NOTIFICATION_RETURNED_LETTER)
create_returned_letter(service=not_included_in_results_template.service, reported_at=datetime.utcnow(),
notification_id=letter_4.id)
response = admin_request.get('service.get_returned_letters', service_id=sample_letter_template.service_id,
reported_at='2019-12-11')
assert len(response) == 5
assert response[0]['notification_id'] == str(letter_from_job.id)
assert not response[0]['client_reference']
assert response[0]['reported_at'] == '2019-12-11'
assert response[0]['created_at'] == '2019-12-10 13:30:00.000000'
assert response[0]['template_name'] == sample_letter_template.name
assert response[0]['template_id'] == str(sample_letter_template.id)
assert response[0]['template_version'] == sample_letter_template.version
assert response[0]['user_name'] == sample_letter_template.service.users[0].name
assert response[0]['original_file_name'] == job.original_file_name
assert response[0]['job_row_number'] == 3
assert not response[0]['uploaded_letter_file_name']
assert response[1]['notification_id'] == str(one_off_letter.id)
assert not response[1]['client_reference']
assert response[1]['reported_at'] == '2019-12-11'
assert response[1]['created_at'] == '2019-12-09 13:30:00.000000'
assert response[1]['template_name'] == sample_letter_template.name
assert response[1]['template_id'] == str(sample_letter_template.id)
assert response[1]['template_version'] == sample_letter_template.version
assert response[1]['user_name'] == sample_letter_template.service.users[0].name
assert not response[1]['original_file_name']
assert not response[1]['job_row_number']
assert not response[1]['uploaded_letter_file_name']
assert response[2]['notification_id'] == str(api_letter.id)
assert response[2]['client_reference'] == 'api_letter'
assert response[2]['reported_at'] == '2019-12-11'
assert response[2]['created_at'] == '2019-12-08 13:30:00.000000'
assert response[2]['template_name'] == sample_letter_template.name
assert response[2]['template_id'] == str(sample_letter_template.id)
assert response[2]['template_version'] == sample_letter_template.version
assert response[2]['user_name'] == 'API'
assert not response[2]['original_file_name']
assert not response[2]['job_row_number']
assert not response[2]['uploaded_letter_file_name']
assert response[3]['notification_id'] == str(precompiled_letter.id)
assert response[3]['client_reference'] == 'precompiled letter'
assert response[3]['reported_at'] == '2019-12-11'
assert response[3]['created_at'] == '2019-12-07 13:30:00.000000'
assert not response[3]['template_name']
assert not response[3]['template_id']
assert not response[3]['template_version']
assert response[3]['user_name'] == 'API'
assert not response[3]['original_file_name']
assert not response[3]['job_row_number']
assert not response[3]['uploaded_letter_file_name']
assert response[4]['notification_id'] == str(uploaded_letter.id)
assert not response[4]['client_reference']
assert response[4]['reported_at'] == '2019-12-11'
assert response[4]['created_at'] == '2019-12-06 13:30:00.000000'
assert not response[4]['template_name']
assert not response[4]['template_id']
assert not response[4]['template_version']
assert response[4]['user_name'] == sample_letter_template.service.users[0].name
assert response[4]['email_address'] == sample_letter_template.service.users[0].email_address
assert not response[4]['original_file_name']
assert not response[4]['job_row_number']
assert response[4]['uploaded_letter_file_name'] == 'filename.pdf'
tests/app/service/test_statistics_rest.py
+2 -6
View File
@@ -10,7 +10,6 @@ from app.models import (
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
LETTER_TYPE,
PRECOMPILED_TEMPLATE_NAME,
SMS_TYPE,
)
from tests.app.db import (
@@ -57,8 +56,8 @@ def test_get_template_usage_by_month_returns_correct_data(
def test_get_template_usage_by_month_returns_two_templates(admin_request, sample_template, sample_service):
template_one = create_template(
sample_service,
template_type=LETTER_TYPE,
template_name=PRECOMPILED_TEMPLATE_NAME,
template_type=SMS_TYPE,
template_name="TEST TEMPLATE",
hidden=True
)
create_ft_notification_status(local_date=datetime(2017, 4, 2), template=template_one, count=1)
@@ -80,7 +79,6 @@ def test_get_template_usage_by_month_returns_two_templates(admin_request, sample
assert resp_json[0]["month"] == 4
assert resp_json[0]["year"] == 2017
assert resp_json[0]["count"] == 1
assert resp_json[0]["is_precompiled_letter"] is True
assert resp_json[1]["template_id"] == str(sample_template.id)
assert resp_json[1]["name"] == sample_template.name
@@ -88,7 +86,6 @@ def test_get_template_usage_by_month_returns_two_templates(admin_request, sample
assert resp_json[1]["month"] == 4
assert resp_json[1]["year"] == 2017
assert resp_json[1]["count"] == 3
assert resp_json[1]["is_precompiled_letter"] is False
assert resp_json[2]["template_id"] == str(sample_template.id)
assert resp_json[2]["name"] == sample_template.name
@@ -96,7 +93,6 @@ def test_get_template_usage_by_month_returns_two_templates(admin_request, sample
assert resp_json[2]["month"] == 11
assert resp_json[2]["year"] == 2017
assert resp_json[2]["count"] == 1
assert resp_json[2]["is_precompiled_letter"] is False
@pytest.mark.parametrize('today_only, stats', [
tests/app/template_statistics/test_rest.py
+2 -5
View File
@@ -42,7 +42,6 @@ def test_get_template_statistics_for_service_by_day_returns_template_info(admin_
assert json_resp['data'][0]['template_id'] == str(sample_notification.template_id)
assert json_resp['data'][0]['template_name'] == 'sms Template Name'
assert json_resp['data'][0]['template_type'] == 'sms'
assert json_resp['data'][0]['is_precompiled_letter'] is False
@pytest.mark.parametrize('var_name', ['limit_days', 'whole_days'])
@@ -77,8 +76,7 @@ def test_get_template_statistics_for_service_by_day_goes_to_db(
count=3,
template_name=sample_template.name,
notification_type=sample_template.template_type,
status='created',
is_precompiled_letter=False
status='created'
)
]
)
@@ -93,8 +91,7 @@ def test_get_template_statistics_for_service_by_day_goes_to_db(
"count": 3,
"template_name": sample_template.name,
"template_type": sample_template.template_type,
"status": "created",
"is_precompiled_letter": False
"status": "created"
}]
# dao only called for 2nd, since redis returned values for first call
tests/app/upload/__init__.py
View File
tests/app/upload/test_rest.py
-253
View File
@@ -1,253 +0,0 @@
from datetime import datetime, timedelta
from freezegun import freeze_time
from app.models import JOB_STATUS_FINISHED, JOB_STATUS_PENDING, LETTER_TYPE
from tests.app.db import (
create_ft_notification_status,
create_job,
create_notification,
create_template,
)
from tests.conftest import set_config
def create_uploaded_letter(letter_template, service, status='created', created_at=None):
return create_notification(
template=letter_template,
to_field="742 Evergreen Terrace",
status=status,
reference="dvla-reference",
client_reference="file-name",
one_off=True,
created_by_id=service.users[0].id,
created_at=created_at
)
def create_precompiled_template(service):
return create_template(
service,
template_type=LETTER_TYPE,
template_name='Pre-compiled PDF',
subject='Pre-compiled PDF',
content="",
hidden=True,
)
@freeze_time('2020-02-02 14:00')
def test_get_uploads(admin_request, sample_template):
letter_template = create_precompiled_template(sample_template.service)
create_uploaded_letter(letter_template, sample_template.service, status='delivered',
created_at=datetime.utcnow() - timedelta(minutes=4))
upload_2 = create_job(template=sample_template,
processing_started=datetime.utcnow() - timedelta(minutes=3),
job_status=JOB_STATUS_FINISHED)
create_uploaded_letter(letter_template, sample_template.service, status='delivered',
created_at=datetime.utcnow() - timedelta(minutes=2))
upload_4 = create_job(template=sample_template,
processing_started=datetime.utcnow() - timedelta(minutes=1),
job_status=JOB_STATUS_FINISHED)
upload_5 = create_job(template=sample_template, processing_started=None,
job_status=JOB_STATUS_PENDING, notification_count=10)
service_id = sample_template.service.id
resp_json = admin_request.get('upload.get_uploads_by_service', service_id=service_id)
data = resp_json['data']
assert len(data) == 4
assert data[0] == {'id': str(upload_5.id),
'original_file_name': 'some.csv',
'recipient': None,
'notification_count': 10,
'template_type': 'sms',
'created_at': upload_5.created_at.strftime("%Y-%m-%d %H:%M:%S"),
'statistics': [],
'upload_type': 'job'}
assert data[1] == {'id': None,
'original_file_name': 'Uploaded letters',
'recipient': None,
'notification_count': 2,
'template_type': 'letter',
'created_at': upload_4.created_at.replace(hour=22, minute=30).strftime(
"%Y-%m-%d %H:%M:%S"),
'statistics': [],
'upload_type': 'letter_day'}
assert data[3] == {'id': str(upload_2.id),
'original_file_name': "some.csv",
'recipient': None,
'notification_count': 1,
'template_type': 'sms',
'created_at': upload_2.created_at.strftime(
"%Y-%m-%d %H:%M:%S"),
'statistics': [],
'upload_type': 'job'}
def test_get_uploads_should_return_statistics(admin_request, sample_template):
now = datetime.utcnow()
earlier = datetime.utcnow() - timedelta(days=1)
job_1 = create_job(template=sample_template, job_status='pending')
job_2 = create_job(sample_template, processing_started=earlier)
for _ in range(3):
create_notification(template=sample_template, job=job_2, status='created')
job_3 = create_job(sample_template, processing_started=now)
for _ in range(4):
create_notification(template=sample_template, job=job_3, status='sending')
letter_template = create_precompiled_template(sample_template.service)
create_uploaded_letter(letter_template, sample_template.service, status='delivered',
created_at=datetime.utcnow() - timedelta(days=3))
resp_json = admin_request.get('upload.get_uploads_by_service', service_id=sample_template.service_id)['data']
assert len(resp_json) == 4
assert resp_json[0]['id'] == str(job_1.id)
assert resp_json[0]['statistics'] == []
assert resp_json[1]['id'] == str(job_3.id)
assert resp_json[1]['statistics'] == [{'status': 'sending', 'count': 4}]
assert resp_json[2]['id'] == str(job_2.id)
assert resp_json[2]['statistics'] == [{'status': 'created', 'count': 3}]
assert resp_json[3]['id'] is None
assert resp_json[3]['statistics'] == []
def test_get_uploads_should_paginate(admin_request, sample_template):
for _ in range(10):
create_job(sample_template)
with set_config(admin_request.app, 'PAGE_SIZE', 2):
resp_json = admin_request.get('upload.get_uploads_by_service', service_id=sample_template.service_id)
assert len(resp_json['data']) == 2
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_uploads_accepts_page_parameter(admin_request, sample_template):
for _ in range(10):
create_job(sample_template)
with set_config(admin_request.app, 'PAGE_SIZE', 2):
resp_json = admin_request.get('upload.get_uploads_by_service', service_id=sample_template.service_id, page=2)
assert len(resp_json['data']) == 2
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'}
@freeze_time('2017-06-10 12:00')
def test_get_uploads_should_retrieve_from_ft_notification_status_for_old_jobs(admin_request, sample_template):
# it's the 10th today, so 3 days should include all of 7th, 8th, 9th, and some of 10th.
just_three_days_ago = datetime(2017, 6, 7, 3, 59, 59)
not_quite_three_days_ago = just_three_days_ago + timedelta(seconds=1)
job_1 = create_job(sample_template, created_at=just_three_days_ago, processing_started=just_three_days_ago)
job_2 = create_job(sample_template, created_at=just_three_days_ago, processing_started=not_quite_three_days_ago)
# is old but hasn't started yet (probably a scheduled job). We don't have any stats for this job yet.
job_3 = create_job(sample_template, created_at=just_three_days_ago, processing_started=None)
# some notifications created more than three days ago, some created after the midnight cutoff
create_ft_notification_status(datetime(2017, 6, 6, 12), job=job_1, notification_status='delivered', count=2)
create_ft_notification_status(datetime(2017, 6, 7, 12), job=job_1, notification_status='delivered', count=4)
# job2's new enough
create_notification(job=job_2, status='created', created_at=not_quite_three_days_ago)
# this isn't picked up because the job is too new
create_ft_notification_status(datetime(2017, 6, 7, 12), job=job_2, notification_status='delivered', count=8)
# this isn't picked up - while the job is old, it started in last 3 days so we look at notification table instead
create_ft_notification_status(datetime(2017, 6, 7, 12), job=job_3, notification_status='delivered', count=16)
# this isn't picked up because we're using the ft status table for job_1 as it's old
create_notification(job=job_1, status='created', created_at=not_quite_three_days_ago)
resp_json = admin_request.get('upload.get_uploads_by_service', service_id=sample_template.service_id)['data']
assert resp_json[0]['id'] == str(job_3.id)
assert resp_json[0]['statistics'] == []
assert resp_json[1]['id'] == str(job_2.id)
assert resp_json[1]['statistics'] == [{'status': 'created', 'count': 1}]
assert resp_json[2]['id'] == str(job_1.id)
assert resp_json[2]['statistics'] == [{'status': 'delivered', 'count': 6}]
@freeze_time('2020-02-02 14:00')
def test_get_uploaded_letters_by_print_date(admin_request, sample_template):
letter_template = create_precompiled_template(sample_template.service)
letter_1 = create_uploaded_letter(
letter_template, sample_template.service, status='delivered',
created_at=datetime.utcnow() - timedelta(minutes=1)
)
letter_2 = create_uploaded_letter(
letter_template, sample_template.service, status='delivered',
created_at=datetime.utcnow() - timedelta(minutes=2)
)
service_id = sample_template.service.id
resp_json = admin_request.get(
'upload.get_uploaded_letter_by_service_and_print_day',
service_id=service_id,
letter_print_date='2020-02-02',
)
assert resp_json['total'] == 2
assert resp_json['page_size'] == 50
assert len(resp_json['notifications']) == 2
assert resp_json['notifications'][0]['id'] == str(letter_1.id)
assert resp_json['notifications'][0]['created_at'] == letter_1.created_at.strftime(
'%Y-%m-%dT%H:%M:%S+00:00'
)
assert resp_json['notifications'][1]['id'] == str(letter_2.id)
assert resp_json['notifications'][1]['created_at'] == letter_2.created_at.strftime(
'%Y-%m-%dT%H:%M:%S+00:00'
)
@freeze_time('2020-02-02 14:00')
def test_get_uploaded_letters_by_print_date_paginates(admin_request, sample_template):
letter_template = create_precompiled_template(sample_template.service)
for _ in range(101):
create_uploaded_letter(
letter_template, sample_template.service, status='delivered',
created_at=datetime.utcnow() - timedelta(minutes=1)
)
service_id = sample_template.service.id
resp_json = admin_request.get(
'upload.get_uploaded_letter_by_service_and_print_day',
service_id=service_id,
letter_print_date='2020-02-02',
page=2,
)
assert resp_json['total'] == 101
assert resp_json['page_size'] == 50
assert len(resp_json['notifications']) == 50
assert resp_json['links']['prev'] == (
f'/service/{service_id}/upload/uploaded-letters/2020-02-02?page=1'
)
assert resp_json['links']['next'] == (
f'/service/{service_id}/upload/uploaded-letters/2020-02-02?page=3'
)
def test_get_uploaded_letters_by_print_date_404s_for_bad_date(
admin_request,
sample_service,
):
admin_request.get(
'upload.get_uploaded_letter_by_service_and_print_day',
service_id=sample_service.id,
letter_print_date='foo',
_expected_status=400,
)