mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-02-23 20:01:01 -05:00
61fd27c4f6
We changed the schema used by the endpoint that searches for notifications by recipient. So the admin app was looking for the wrong thing in the JSON. This is hard to catch in tests because it relies on our fixtures matching what the API really returns. This commit fixes the code to use the correct key to lookup the template content from the JSON. This also exposed the fact that we weren’t passing in the personalisation any more (perhaps got lost in the re-reverts somehow) so users were only seeing the template in the inbound view, not the full message content.
390 lines
12 KiB
Python
390 lines
12 KiB
Python
import csv
|
|
import pytest
|
|
import uuid
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from flask.testing import FlaskClient
|
|
from flask import url_for
|
|
from flask_login import login_user
|
|
|
|
|
|
class TestClient(FlaskClient):
|
|
def login(self, user, mocker=None, service=None):
|
|
# Skipping authentication here and just log them in
|
|
with self.session_transaction() as session:
|
|
session['user_id'] = user.id
|
|
if mocker:
|
|
mocker.patch('app.user_api_client.get_user', return_value=user)
|
|
mocker.patch('app.events_api_client.create_event')
|
|
if mocker and service:
|
|
with self.session_transaction() as session:
|
|
session['service_id'] = service['id']
|
|
mocker.patch('app.service_api_client.get_service', return_value={'data': service})
|
|
login_user(user)
|
|
|
|
def logout(self, user):
|
|
self.get(url_for("main.logout"))
|
|
|
|
|
|
def sample_uuid():
|
|
return "6ce466d0-fd6a-11e5-82f5-e0accb9d11a6"
|
|
|
|
|
|
def generate_uuid():
|
|
return uuid.uuid4()
|
|
|
|
|
|
def created_by_json(id_, name='', email_address=''):
|
|
return {'id': id_, 'name': name, 'email_address': email_address}
|
|
|
|
|
|
def service_json(
|
|
id_='1234',
|
|
name='Test Service',
|
|
users=None,
|
|
message_limit=1000,
|
|
active=True,
|
|
restricted=True,
|
|
email_from=None,
|
|
reply_to_email_address=None,
|
|
sms_sender='GOVUK',
|
|
research_mode=False,
|
|
organisation=None,
|
|
branding='govuk',
|
|
created_at=None,
|
|
letter_contact_block=None,
|
|
inbound_api=None,
|
|
permissions=['email', 'sms'],
|
|
):
|
|
if users is None:
|
|
users = []
|
|
if permissions is None:
|
|
permissions = []
|
|
if inbound_api is None:
|
|
inbound_api = []
|
|
return {
|
|
'id': id_,
|
|
'name': name,
|
|
'users': users,
|
|
'message_limit': message_limit,
|
|
'active': active,
|
|
'restricted': restricted,
|
|
'email_from': email_from,
|
|
'reply_to_email_address': reply_to_email_address,
|
|
'sms_sender': sms_sender,
|
|
'research_mode': research_mode,
|
|
'organisation': organisation,
|
|
'branding': branding,
|
|
'created_at': created_at or str(datetime.utcnow()),
|
|
'letter_contact_block': letter_contact_block,
|
|
'dvla_organisation': '001',
|
|
'permissions': permissions,
|
|
'inbound_api': inbound_api,
|
|
}
|
|
|
|
|
|
def template_json(service_id,
|
|
id_,
|
|
name="sample template",
|
|
type_="sms",
|
|
content=None,
|
|
subject=None,
|
|
version=1,
|
|
archived=False,
|
|
process_type='normal',
|
|
redact_personalisation=None,
|
|
):
|
|
template = {
|
|
'id': id_,
|
|
'name': name,
|
|
'template_type': type_,
|
|
'content': content,
|
|
'service': service_id,
|
|
'version': version,
|
|
'updated_at': datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f'),
|
|
'archived': archived,
|
|
'process_type': process_type,
|
|
}
|
|
if content is None:
|
|
template['content'] = "template content"
|
|
if subject is None and type_ != 'sms':
|
|
template['subject'] = "template subject"
|
|
if subject is not None:
|
|
template['subject'] = subject
|
|
if redact_personalisation is not None:
|
|
template['redact_personalisation'] = redact_personalisation
|
|
return template
|
|
|
|
|
|
def template_version_json(service_id,
|
|
id_,
|
|
created_by,
|
|
version=1,
|
|
created_at=None,
|
|
**kwargs):
|
|
template = template_json(service_id, id_, **kwargs)
|
|
template['created_by'] = created_by_json(
|
|
created_by.id,
|
|
created_by.name,
|
|
created_by.email_address
|
|
)
|
|
if created_at is None:
|
|
created_at = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
|
|
template['created_at'] = created_at
|
|
template['version'] = version
|
|
return template
|
|
|
|
|
|
def api_key_json(id_, name, expiry_date=None):
|
|
return {'id': id_,
|
|
'name': name,
|
|
'expiry_date': expiry_date
|
|
}
|
|
|
|
|
|
def invite_json(id_, from_user, service_id, email_address, permissions, created_at, status):
|
|
return {'id': id_,
|
|
'from_user': from_user,
|
|
'service': service_id,
|
|
'email_address': email_address,
|
|
'status': status,
|
|
'permissions': permissions,
|
|
'created_at': created_at
|
|
}
|
|
|
|
|
|
TEST_USER_EMAIL = 'test@user.gov.uk'
|
|
|
|
|
|
def create_test_api_user(state, permissions={}):
|
|
from app.notify_client.user_api_client import User
|
|
user_data = {'id': 1,
|
|
'name': 'Test User',
|
|
'password': 'somepassword',
|
|
'email_address': TEST_USER_EMAIL,
|
|
'mobile_number': '+441234123412',
|
|
'state': state,
|
|
'permissions': permissions
|
|
}
|
|
user = User(user_data)
|
|
return user
|
|
|
|
|
|
def job_json(
|
|
service_id,
|
|
created_by,
|
|
job_id=None,
|
|
template_id=None,
|
|
template_version=1,
|
|
created_at=None,
|
|
bucket_name='',
|
|
original_file_name="thisisatest.csv",
|
|
notification_count=1,
|
|
notifications_sent=1,
|
|
notifications_requested=1,
|
|
job_status='finished',
|
|
scheduled_for=''
|
|
):
|
|
if job_id is None:
|
|
job_id = str(generate_uuid())
|
|
if template_id is None:
|
|
template_id = "5d729fbd-239c-44ab-b498-75a985f3198f"
|
|
if created_at is None:
|
|
created_at = str(datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f%z'))
|
|
data = {
|
|
'id': job_id,
|
|
'service': service_id,
|
|
'template': template_id,
|
|
'template_version': template_version,
|
|
'original_file_name': original_file_name,
|
|
'created_at': created_at,
|
|
'notification_count': notification_count,
|
|
'notifications_sent': notifications_sent,
|
|
'notifications_requested': notifications_requested,
|
|
'job_status': job_status,
|
|
'statistics': [],
|
|
'created_by': created_by_json(
|
|
created_by.id,
|
|
created_by.name,
|
|
created_by.email_address
|
|
),
|
|
'scheduled_for': scheduled_for
|
|
}
|
|
return data
|
|
|
|
|
|
def notification_json(
|
|
service_id,
|
|
job=None,
|
|
template=None,
|
|
to='07123456789',
|
|
status=None,
|
|
sent_at=None,
|
|
job_row_number=None,
|
|
created_at=None,
|
|
updated_at=None,
|
|
with_links=False,
|
|
rows=5,
|
|
personalisation=None,
|
|
):
|
|
if template is None:
|
|
template = template_json(service_id, str(generate_uuid()))
|
|
if sent_at is None:
|
|
sent_at = str(datetime.utcnow().time())
|
|
if created_at is None:
|
|
created_at = datetime.now(timezone.utc).isoformat()
|
|
if updated_at is None:
|
|
updated_at = str((datetime.utcnow() + timedelta(minutes=1)).time())
|
|
if status is None:
|
|
status = 'delivered'
|
|
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)
|
|
}
|
|
|
|
job_payload = None
|
|
if job:
|
|
job_payload = {'id': job['id'], 'original_file_name': job['original_file_name']}
|
|
|
|
data = {
|
|
'notifications': [{
|
|
'id': uuid.uuid4(),
|
|
'to': to,
|
|
'template': template,
|
|
'job': job_payload,
|
|
'sent_at': sent_at,
|
|
'status': status,
|
|
'created_at': created_at,
|
|
'created_by': None,
|
|
'updated_at': updated_at,
|
|
'job_row_number': job_row_number,
|
|
'service': service_id,
|
|
'template_version': template['version'],
|
|
'personalisation': personalisation or {},
|
|
} for i in range(rows)],
|
|
'total': rows,
|
|
'page_size': 50,
|
|
'links': links
|
|
}
|
|
return data
|
|
|
|
|
|
def single_notification_json(
|
|
service_id,
|
|
job=None,
|
|
template=None,
|
|
status=None,
|
|
sent_at=None,
|
|
created_at=None,
|
|
updated_at=None
|
|
):
|
|
if template is None:
|
|
template = template_json(service_id, str(generate_uuid()))
|
|
if sent_at is None:
|
|
sent_at = str(datetime.utcnow())
|
|
if created_at is None:
|
|
created_at = str(datetime.utcnow())
|
|
if updated_at is None:
|
|
updated_at = str(datetime.utcnow() + timedelta(minutes=1))
|
|
if status is None:
|
|
status = 'delivered'
|
|
job_payload = None
|
|
if job:
|
|
job_payload = {'id': job['id'], 'original_file_name': job['original_file_name']}
|
|
|
|
data = {
|
|
'sent_at': sent_at,
|
|
'to': '07123456789',
|
|
'billable_units': 1,
|
|
'status': status,
|
|
'created_at': created_at,
|
|
'reference': None,
|
|
'updated_at': updated_at,
|
|
'template_version': 5,
|
|
'service': service_id,
|
|
'id': '29441662-17ce-4ffe-9502-fcaed73b2826',
|
|
'template': template,
|
|
'job_row_number': 0,
|
|
'notification_type': 'sms',
|
|
'api_key': None,
|
|
'job': job_payload,
|
|
'sent_by': 'mmg'
|
|
}
|
|
return data
|
|
|
|
|
|
def validate_route_permission(mocker,
|
|
app_,
|
|
method,
|
|
response_code,
|
|
route,
|
|
permissions,
|
|
usr,
|
|
service):
|
|
usr._permissions[str(service['id'])] = permissions
|
|
mocker.patch(
|
|
'app.user_api_client.check_verify_code',
|
|
return_value=(True, ''))
|
|
mocker.patch(
|
|
'app.service_api_client.get_services',
|
|
return_value={'data': []})
|
|
mocker.patch('app.service_api_client.update_service', return_value=service)
|
|
mocker.patch('app.service_api_client.update_service_with_properties', return_value=service)
|
|
mocker.patch('app.user_api_client.get_user', return_value=usr)
|
|
mocker.patch('app.user_api_client.get_user_by_email', return_value=usr)
|
|
mocker.patch('app.service_api_client.get_service', return_value={'data': service})
|
|
mocker.patch('app.user_api_client.get_users_for_service', return_value=[usr])
|
|
with app_.test_request_context():
|
|
with app_.test_client() as client:
|
|
client.login(usr)
|
|
resp = None
|
|
if method == 'GET':
|
|
resp = client.get(route)
|
|
elif method == 'POST':
|
|
resp = client.post(route)
|
|
else:
|
|
pytest.fail("Invalid method call {}".format(method))
|
|
if resp.status_code != response_code:
|
|
print(resp.status_code)
|
|
pytest.fail("Invalid permissions set for endpoint {}".format(route))
|
|
return resp
|
|
|
|
|
|
def validate_route_permission_with_client(mocker,
|
|
client,
|
|
method,
|
|
response_code,
|
|
route,
|
|
permissions,
|
|
usr,
|
|
service):
|
|
usr._permissions[str(service['id'])] = permissions
|
|
mocker.patch(
|
|
'app.user_api_client.check_verify_code',
|
|
return_value=(True, ''))
|
|
mocker.patch(
|
|
'app.service_api_client.get_services',
|
|
return_value={'data': []})
|
|
mocker.patch('app.service_api_client.update_service', return_value=service)
|
|
mocker.patch('app.service_api_client.update_service_with_properties', return_value=service)
|
|
mocker.patch('app.user_api_client.get_user', return_value=usr)
|
|
mocker.patch('app.user_api_client.get_user_by_email', return_value=usr)
|
|
mocker.patch('app.service_api_client.get_service', return_value={'data': service})
|
|
mocker.patch('app.user_api_client.get_users_for_service', return_value=[usr])
|
|
client.login(usr)
|
|
resp = None
|
|
if method == 'GET':
|
|
resp = client.get(route)
|
|
elif method == 'POST':
|
|
resp = client.post(route)
|
|
else:
|
|
pytest.fail("Invalid method call {}".format(method))
|
|
if resp.status_code != response_code:
|
|
print(resp.status_code)
|
|
pytest.fail("Invalid permissions set for endpoint {}".format(route))
|
|
return resp
|