mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-02-21 10:54:49 -05:00
0934843815
When downloading a report of a which messages from a job have been delivered and which have failed we currently only include the Notify data. This makes it hard to reconcile or do analysis on these reports, because often the thing that people want to reconcile on is in the data they’ve uploaded (eg a reference number). Here’s an example of a user talking about this problem: > It would also be helpful if the format of the delivery and failure > reports could include the fields from the recipient's file. While I > can, of course, cross-reference one report with the other it would be > easier if I did not have to. We send emails to individuals within > organisations and it is not always easy to establish the organisation > from a recipient's email address. This is particularly important when > emails fail to be delivered as we need to contact the organisation to > establish a new contact. – ticket 677 We’ve also seen it when doing research with a local council. This commit takes the original file, the data from the API, and munges them together.
433 lines
12 KiB
Python
433 lines
12 KiB
Python
from collections import OrderedDict
|
|
from csv import DictReader
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from freezegun import freeze_time
|
|
from tests.conftest import fake_uuid
|
|
|
|
from app.utils import (
|
|
GovernmentDomain,
|
|
Spreadsheet,
|
|
email_safe,
|
|
generate_next_dict,
|
|
generate_notifications_csv,
|
|
generate_previous_dict,
|
|
get_cdn_domain,
|
|
get_letter_timings,
|
|
)
|
|
|
|
|
|
def _get_notifications_csv(
|
|
service_id,
|
|
page=1,
|
|
row_number=1,
|
|
recipient='foo@bar.com',
|
|
template_name='foo',
|
|
template_type='sms',
|
|
job_name='bar.csv',
|
|
status='Delivered',
|
|
created_at='Thursday 19 April at 12:00',
|
|
rows=1,
|
|
with_links=False,
|
|
job_id=fake_uuid
|
|
):
|
|
links = {}
|
|
if with_links:
|
|
links = {
|
|
'prev': '/service/{}/notifications?page=0'.format(service_id),
|
|
'next': '/service/{}/notifications?page=1'.format(service_id),
|
|
'last': '/service/{}/notifications?page=2'.format(service_id)
|
|
}
|
|
|
|
data = {
|
|
'notifications': [{
|
|
"row_number": row_number + i,
|
|
"to": recipient,
|
|
"recipient": recipient,
|
|
"template_name": template_name,
|
|
"template_type": template_type,
|
|
"template": {"name": template_name, "template_type": template_type},
|
|
"job_name": job_name,
|
|
"status": status,
|
|
"created_at": created_at,
|
|
"updated_at": None
|
|
} for i in range(rows)],
|
|
'total': rows,
|
|
'page_size': 50,
|
|
'links': links
|
|
}
|
|
return data
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def _get_notifications_csv_mock(
|
|
mocker,
|
|
api_user_active,
|
|
job_id=fake_uuid
|
|
):
|
|
return mocker.patch(
|
|
'app.notification_api_client.get_notifications_for_service',
|
|
side_effect=_get_notifications_csv
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize('service_name, safe_email', [
|
|
('name with spaces', 'name.with.spaces'),
|
|
('singleword', 'singleword'),
|
|
('UPPER CASE', 'upper.case'),
|
|
('Service - with dash', 'service.with.dash'),
|
|
('lots of spaces', 'lots.of.spaces'),
|
|
('name.with.dots', 'name.with.dots'),
|
|
('name-with-other-delimiters', 'namewithotherdelimiters'),
|
|
('.leading', 'leading'),
|
|
('trailing.', 'trailing'),
|
|
('üńïçödë wördś', 'unicode.words'),
|
|
])
|
|
def test_email_safe_return_dot_separated_email_domain(service_name, safe_email):
|
|
assert email_safe(service_name) == safe_email
|
|
|
|
|
|
def test_generate_previous_dict(client):
|
|
ret = generate_previous_dict('main.view_jobs', 'foo', 2, {})
|
|
assert 'page=1' in ret['url']
|
|
assert ret['title'] == 'Previous page'
|
|
assert ret['label'] == 'page 1'
|
|
|
|
|
|
def test_generate_next_dict(client):
|
|
ret = generate_next_dict('main.view_jobs', 'foo', 2, {})
|
|
assert 'page=3' in ret['url']
|
|
assert ret['title'] == 'Next page'
|
|
assert ret['label'] == 'page 3'
|
|
|
|
|
|
def test_generate_previous_next_dict_adds_other_url_args(client):
|
|
ret = generate_next_dict('main.view_notifications', 'foo', 2, {'message_type': 'blah'})
|
|
assert 'notifications/blah' in ret['url']
|
|
|
|
|
|
def test_can_create_spreadsheet_from_large_excel_file():
|
|
with open(str(Path.cwd() / 'tests' / 'spreadsheet_files' / 'excel 2007.xlsx'), 'rb') as xl:
|
|
ret = Spreadsheet.from_file(xl, filename='xl.xlsx')
|
|
assert ret.as_csv_data
|
|
|
|
|
|
def test_can_create_spreadsheet_from_dict():
|
|
assert Spreadsheet.from_dict(OrderedDict(
|
|
foo='bar',
|
|
name='Jane',
|
|
)).as_csv_data == (
|
|
"foo,name\r\n"
|
|
"bar,Jane\r\n"
|
|
)
|
|
|
|
|
|
def test_can_create_spreadsheet_from_dict_with_filename():
|
|
assert Spreadsheet.from_dict({}, filename='empty.csv').as_dict['file_name'] == "empty.csv"
|
|
|
|
|
|
@pytest.mark.parametrize('original_file_contents, expected_column_headers, expected_1st_row', [
|
|
(
|
|
"""
|
|
phone_number
|
|
07700900123
|
|
""",
|
|
['Row number', 'phone_number', 'Template', 'Type', 'Job', 'Status', 'Time'],
|
|
['1', '07700900123', 'foo', 'sms', 'bar.csv', 'Delivered', 'Thursday 19 April at 12:00'],
|
|
),
|
|
(
|
|
"""
|
|
phone_number, a, b, c
|
|
07700900123, 🐜,🐝,🦀
|
|
""",
|
|
['Row number', 'phone_number', 'a', 'b', 'c', 'Template', 'Type', 'Job', 'Status', 'Time'],
|
|
['1', '07700900123', '🐜', '🐝', '🦀', 'foo', 'sms', 'bar.csv', 'Delivered', 'Thursday 19 April at 12:00'],
|
|
),
|
|
])
|
|
def test_generate_notifications_csv_returns_correct_csv_file(
|
|
app_,
|
|
mocker,
|
|
_get_notifications_csv_mock,
|
|
original_file_contents,
|
|
expected_column_headers,
|
|
expected_1st_row,
|
|
):
|
|
mocker.patch(
|
|
'app.main.s3_client.s3download',
|
|
return_value=original_file_contents,
|
|
)
|
|
csv_content = generate_notifications_csv(service_id='1234', job_id=fake_uuid, template_type='sms')
|
|
csv_file = DictReader(StringIO('\n'.join(csv_content)))
|
|
assert csv_file.fieldnames == expected_column_headers
|
|
assert next(csv_file) == dict(zip(expected_column_headers, expected_1st_row))
|
|
|
|
|
|
def test_generate_notifications_csv_only_calls_once_if_no_next_link(
|
|
app_,
|
|
_get_notifications_csv_mock,
|
|
):
|
|
list(generate_notifications_csv(service_id='1234'))
|
|
|
|
assert _get_notifications_csv_mock.call_count == 1
|
|
|
|
|
|
@pytest.mark.parametrize("job_id", ["some", None])
|
|
def test_generate_notifications_csv_calls_twice_if_next_link(
|
|
app_,
|
|
mocker,
|
|
job_id,
|
|
):
|
|
|
|
mocker.patch(
|
|
'app.main.s3_client.s3download',
|
|
return_value="""
|
|
phone_number
|
|
07700900000
|
|
07700900001
|
|
07700900002
|
|
07700900003
|
|
07700900004
|
|
07700900005
|
|
07700900006
|
|
07700900007
|
|
07700900008
|
|
07700900009
|
|
"""
|
|
)
|
|
|
|
service_id = '1234'
|
|
response_with_links = _get_notifications_csv(service_id, rows=7, with_links=True)
|
|
response_with_no_links = _get_notifications_csv(service_id, rows=3, row_number=8, with_links=False)
|
|
|
|
mock_get_notifications = mocker.patch(
|
|
'app.notification_api_client.get_notifications_for_service',
|
|
side_effect=[
|
|
response_with_links,
|
|
response_with_no_links,
|
|
]
|
|
)
|
|
|
|
csv_content = generate_notifications_csv(
|
|
service_id=service_id,
|
|
job_id=job_id or fake_uuid,
|
|
template_type='sms',
|
|
)
|
|
csv = list(DictReader(StringIO('\n'.join(csv_content))))
|
|
|
|
assert len(csv) == 10
|
|
assert csv[0]['phone_number'] == '07700900000'
|
|
assert csv[9]['phone_number'] == '07700900009'
|
|
assert mock_get_notifications.call_count == 2
|
|
# mock_calls[0][2] is the kwargs from first call
|
|
assert mock_get_notifications.mock_calls[0][2]['page'] == 1
|
|
assert mock_get_notifications.mock_calls[1][2]['page'] == 2
|
|
|
|
|
|
@freeze_time('2017-07-14 14:59:59') # Friday, before print deadline
|
|
@pytest.mark.parametrize('upload_time, expected_print_time, is_printed, expected_earliest, expected_latest', [
|
|
|
|
# BST
|
|
# ==================================================================
|
|
# First thing Monday
|
|
(
|
|
'2017-07-10 00:00:01',
|
|
'Tuesday 15:00',
|
|
True,
|
|
'Thursday 2017-07-13',
|
|
'Friday 2017-07-14'
|
|
),
|
|
# Monday at 16:59 BST
|
|
(
|
|
'2017-07-10 15:59:59',
|
|
'Tuesday 15:00',
|
|
True,
|
|
'Thursday 2017-07-13',
|
|
'Friday 2017-07-14'
|
|
),
|
|
# Monday at 17:00 BST
|
|
(
|
|
'2017-07-10 16:00:01',
|
|
'Wednesday 15:00',
|
|
True,
|
|
'Friday 2017-07-14',
|
|
'Saturday 2017-07-15'
|
|
),
|
|
# Tuesday before 17:00 BST
|
|
(
|
|
'2017-07-11 12:00:00',
|
|
'Wednesday 15:00',
|
|
True,
|
|
'Friday 2017-07-14',
|
|
'Saturday 2017-07-15'
|
|
),
|
|
# Wednesday before 17:00 BST
|
|
(
|
|
'2017-07-12 12:00:00',
|
|
'Thursday 15:00',
|
|
True,
|
|
'Saturday 2017-07-15',
|
|
'Monday 2017-07-17'
|
|
),
|
|
# Thursday before 17:00 BST
|
|
(
|
|
'2017-07-13 12:00:00',
|
|
'Friday 15:00',
|
|
True, # WRONG
|
|
'Monday 2017-07-17',
|
|
'Tuesday 2017-07-18'
|
|
),
|
|
# Friday anytime
|
|
(
|
|
'2017-07-14 00:00:00',
|
|
'Monday 15:00',
|
|
False,
|
|
'Wednesday 2017-07-19',
|
|
'Thursday 2017-07-20'
|
|
),
|
|
(
|
|
'2017-07-14 12:00:00',
|
|
'Monday 15:00',
|
|
False,
|
|
'Wednesday 2017-07-19',
|
|
'Thursday 2017-07-20'
|
|
),
|
|
(
|
|
'2017-07-14 22:00:00',
|
|
'Monday 15:00',
|
|
False,
|
|
'Wednesday 2017-07-19',
|
|
'Thursday 2017-07-20'
|
|
),
|
|
# Saturday anytime
|
|
(
|
|
'2017-07-14 12:00:00',
|
|
'Monday 15:00',
|
|
False,
|
|
'Wednesday 2017-07-19',
|
|
'Thursday 2017-07-20'
|
|
),
|
|
# Sunday before 1700 BST
|
|
(
|
|
'2017-07-15 15:59:59',
|
|
'Monday 15:00',
|
|
False,
|
|
'Wednesday 2017-07-19',
|
|
'Thursday 2017-07-20'
|
|
),
|
|
# Sunday after 17:00 BST
|
|
(
|
|
'2017-07-16 16:00:01',
|
|
'Tuesday 15:00',
|
|
False,
|
|
'Thursday 2017-07-20',
|
|
'Friday 2017-07-21'
|
|
),
|
|
|
|
# GMT
|
|
# ==================================================================
|
|
# Monday at 16:59 GMT
|
|
(
|
|
'2017-01-02 16:59:59',
|
|
'Tuesday 15:00',
|
|
True,
|
|
'Thursday 2017-01-05',
|
|
'Friday 2017-01-06',
|
|
),
|
|
# Monday at 17:00 GMT
|
|
(
|
|
'2017-01-02 17:00:01',
|
|
'Wednesday 15:00',
|
|
True,
|
|
'Friday 2017-01-06',
|
|
'Saturday 2017-01-07',
|
|
),
|
|
|
|
])
|
|
def test_get_estimated_delivery_date_for_letter(
|
|
upload_time,
|
|
expected_print_time,
|
|
is_printed,
|
|
expected_earliest,
|
|
expected_latest,
|
|
):
|
|
timings = get_letter_timings(upload_time)
|
|
assert timings.printed_by.strftime('%A %H:%M') == expected_print_time
|
|
assert timings.is_printed == is_printed
|
|
assert timings.earliest_delivery.strftime('%A %Y-%m-%d') == expected_earliest
|
|
assert timings.latest_delivery.strftime('%A %Y-%m-%d') == expected_latest
|
|
|
|
|
|
def test_get_cdn_domain_on_localhost(client, mocker):
|
|
mocker.patch.dict('app.current_app.config', values={'ADMIN_BASE_URL': 'http://localhost:6012'})
|
|
domain = get_cdn_domain()
|
|
assert domain == 'static-logos.notify.tools'
|
|
|
|
|
|
def test_get_cdn_domain_on_non_localhost(client, mocker):
|
|
mocker.patch.dict('app.current_app.config', values={'ADMIN_BASE_URL': 'https://some.admintest.com'})
|
|
domain = get_cdn_domain()
|
|
assert domain == 'static-logos.admintest.com'
|
|
|
|
|
|
@pytest.mark.parametrize("domain_or_email_address", (
|
|
"test@dclgdatamart.co.uk", "test@communities.gsi.gov.uk", "test@communities.gov.uk",
|
|
))
|
|
def test_get_valid_government_domain_known_details(domain_or_email_address):
|
|
government_domain = GovernmentDomain(domain_or_email_address)
|
|
assert government_domain.crown_status is None
|
|
assert government_domain.owner == "Ministry of Housing, Communities & Local Government"
|
|
assert government_domain.agreement_signed is True
|
|
|
|
|
|
@pytest.mark.parametrize("domain_or_email_address", (
|
|
"test@police.gov.uk", "police.gov.uk",
|
|
))
|
|
def test_get_valid_government_domain_unknown_details(domain_or_email_address):
|
|
government_domain = GovernmentDomain(domain_or_email_address)
|
|
assert government_domain.crown_status is None
|
|
assert government_domain.owner is None
|
|
assert government_domain.agreement_signed is None
|
|
|
|
|
|
def test_get_valid_government_domain_some_known_details():
|
|
government_domain = GovernmentDomain("marinemanagement.org.uk")
|
|
assert government_domain.crown_status is None
|
|
assert government_domain.owner == "Marine Management Organisation"
|
|
assert government_domain.agreement_signed is True
|
|
|
|
|
|
def test_get_valid_government_domain_gets_most_specific_first():
|
|
|
|
generic = GovernmentDomain("gov.uk")
|
|
assert generic.crown_status is None
|
|
assert generic.owner is None
|
|
assert generic.agreement_signed is None
|
|
|
|
specific = GovernmentDomain("dacorum.gov.uk")
|
|
assert specific.crown_status is False
|
|
assert specific.owner == 'Dacorum Borough Council'
|
|
assert specific.agreement_signed is True
|
|
|
|
|
|
def test_validate_government_domain_data():
|
|
|
|
for domain in GovernmentDomain.domains.keys():
|
|
|
|
government_domain = GovernmentDomain(domain)
|
|
|
|
assert government_domain.crown_status in {
|
|
True, False, None
|
|
}
|
|
|
|
assert (
|
|
government_domain.owner is None
|
|
) or (
|
|
isinstance(government_domain.owner, str)
|
|
)
|
|
|
|
assert government_domain.agreement_signed in {
|
|
True, False, None
|
|
}
|