Merge branch 'master' into hr

This commit is contained in:
Pete Herlihy
2017-12-12 16:38:08 +00:00
committed by GitHub
16 changed files with 649 additions and 300 deletions

View File

@@ -717,15 +717,30 @@ class ServiceInboundNumberForm(Form):
)
class ServiceInboundApiForm(Form):
url = StringField("Callback URL",
validators=[DataRequired(message='Cant be empty'),
Regexp(regex="^https.*",
message='Must be a valid https URL')]
)
bearer_token = PasswordFieldShowHasContent("Bearer token",
validators=[DataRequired(message='Cant be empty'),
Length(min=10, message='Must be at least 10 characters')])
class ServiceReceiveMessagesCallbackForm(Form):
url = StringField(
"URL",
validators=[DataRequired(message='Cant be empty'),
Regexp(regex="^https.*", message='Must be a valid https URL')]
)
bearer_token = PasswordFieldShowHasContent(
"Bearer token",
validators=[DataRequired(message='Cant be empty'),
Length(min=10, message='Must be at least 10 characters')]
)
class ServiceDeliveryStatusCallbackForm(Form):
url = StringField(
"URL",
validators=[DataRequired(message='Cant be empty'),
Regexp(regex="^https.*", message='Must be a valid https URL')]
)
bearer_token = PasswordFieldShowHasContent(
"Bearer token",
validators=[DataRequired(message='Cant be empty'),
Length(min=10, message='Must be at least 10 characters')]
)
class InternationalSMSForm(Form):

View File

@@ -1,18 +1,30 @@
from flask import request, render_template, redirect, url_for, flash, Markup, abort
from flask_login import login_required
from flask_login import login_required, current_user
from app.main import main
from app.main.forms import CreateKeyForm, Whitelist
from app.main.forms import (
CreateKeyForm,
Whitelist,
ServiceReceiveMessagesCallbackForm,
ServiceDeliveryStatusCallbackForm
)
from app import api_key_api_client, service_api_client, notification_api_client, current_service
from app.utils import user_has_permissions, email_safe
from app.notify_client.api_key_api_client import KEY_TYPE_NORMAL, KEY_TYPE_TEST, KEY_TYPE_TEAM
dummy_bearer_token = 'bearer_token_set'
@main.route("/services/<service_id>/api")
@login_required
@user_has_permissions('manage_api_keys', admin_override=True)
def api_integration(service_id):
callbacks_link = (
'.api_callbacks' if 'inbound_sms' in current_service['permissions']
else '.delivery_status_callback'
)
return render_template(
'views/api/index.html',
callbacks_link=callbacks_link,
api_notifications=notification_api_client.get_api_notifications_for_service(service_id)
)
@@ -113,3 +125,137 @@ def revoke_api_key(service_id, key_id):
api_key_api_client.revoke_api_key(service_id=service_id, key_id=key_id)
flash('{} was revoked'.format(key_name), 'default_with_tick')
return redirect(url_for('.api_keys', service_id=service_id))
def get_apis():
callback_api = None
inbound_api = None
if current_service['service_callback_api']:
callback_api = service_api_client.get_service_callback_api(
current_service['id'],
current_service.get('service_callback_api')[0]
)
if current_service['inbound_api']:
inbound_api = service_api_client.get_service_inbound_api(
current_service['id'],
current_service.get('inbound_api')[0]
)
return (callback_api, inbound_api)
def check_token_against_dummy_bearer(token):
if token != dummy_bearer_token:
return token
else:
return ''
@main.route("/services/<service_id>/api/callbacks", methods=['GET'])
@login_required
def api_callbacks(service_id):
if 'inbound_sms' not in current_service['permissions']:
return redirect(url_for('.delivery_status_callback', service_id=service_id))
received_text_messages_callback, delivery_status_callback = get_apis()
return render_template(
'views/api/callbacks.html',
received_text_messages_callback=received_text_messages_callback['url']
if received_text_messages_callback else None,
delivery_status_callback=delivery_status_callback['url'] if delivery_status_callback else None
)
def get_delivery_status_callback_details():
if current_service['service_callback_api']:
return service_api_client.get_service_callback_api(
current_service['id'],
current_service.get('service_callback_api')[0]
)
@main.route("/services/<service_id>/api/callbacks/delivery-status-callback", methods=['GET', 'POST'])
@login_required
def delivery_status_callback(service_id):
delivery_status_callback = get_delivery_status_callback_details()
back_link = (
'.api_callbacks' if 'inbound_sms' in current_service['permissions']
else '.api_integration'
)
form = ServiceDeliveryStatusCallbackForm(
url=delivery_status_callback.get('url') if delivery_status_callback else '',
bearer_token=dummy_bearer_token if delivery_status_callback else ''
)
if form.validate_on_submit():
if delivery_status_callback:
if (delivery_status_callback.get('url') != form.url.data
or form.bearer_token.data != dummy_bearer_token):
service_api_client.update_service_callback_api(
service_id,
url=form.url.data,
bearer_token=check_token_against_dummy_bearer(form.bearer_token.data),
user_id=current_user.id,
callback_api_id=delivery_status_callback.get('id')
)
else:
service_api_client.create_service_callback_api(
service_id,
url=form.url.data,
bearer_token=form.bearer_token.data,
user_id=current_user.id
)
return redirect(url_for(back_link, service_id=service_id))
return render_template(
'views/api/callbacks/delivery-status-callback.html',
back_link=back_link,
form=form,
)
def get_received_text_messages_callback():
if current_service['inbound_api']:
return service_api_client.get_service_inbound_api(
current_service['id'],
current_service.get('inbound_api')[0]
)
@main.route("/services/<service_id>/api/callbacks/received-text-messages-callback", methods=['GET', 'POST'])
@login_required
def received_text_messages_callback(service_id):
if 'inbound_sms' not in current_service['permissions']:
return redirect(url_for('.api_integration', service_id=service_id))
received_text_messages_callback = get_received_text_messages_callback()
form = ServiceReceiveMessagesCallbackForm(
url=received_text_messages_callback.get('url') if received_text_messages_callback else '',
bearer_token=dummy_bearer_token if received_text_messages_callback else ''
)
if form.validate_on_submit():
if received_text_messages_callback:
if (received_text_messages_callback.get('url') != form.url.data
or form.bearer_token.data != dummy_bearer_token):
service_api_client.update_service_inbound_api(
service_id,
url=form.url.data,
bearer_token=check_token_against_dummy_bearer(form.bearer_token.data),
user_id=current_user.id,
inbound_api_id=received_text_messages_callback.get('id')
)
else:
service_api_client.create_service_inbound_api(
service_id,
url=form.url.data,
bearer_token=form.bearer_token.data,
user_id=current_user.id
)
return redirect(url_for('.api_callbacks', service_id=service_id))
return render_template(
'views/api/callbacks/received-text-messages-callback.html',
form=form,
)

View File

@@ -1,5 +1,3 @@
from urllib.parse import urlparse
import requests
from flask import (
render_template,
@@ -33,7 +31,6 @@ from app.main.forms import (
ServiceLetterContactBlockForm,
ServiceBrandingOrg,
LetterBranding,
ServiceInboundApiForm,
InternationalSMSForm,
OrganisationTypeForm,
FreeSMSAllowance,
@@ -44,17 +41,6 @@ from app import user_api_client, current_service, organisations_client, inbound_
from notifications_utils.formatters import formatted_list
dummy_bearer_token = 'bearer_token_set'
def get_inbound_api():
if current_service['inbound_api']:
return service_api_client.get_service_inbound_api(
current_service['id'],
current_service.get('inbound_api')[0]
)
@main.route("/services/<service_id>/service-settings")
@login_required
@user_has_permissions('manage_settings', admin_override=True)
@@ -65,14 +51,6 @@ def service_settings(service_id):
else:
organisation = None
inbound_api = get_inbound_api()
if inbound_api:
parsed_url = urlparse(inbound_api.get('url')) if inbound_api else ''
inbound_api_url = '{uri.scheme}://{uri.netloc}{elide_token}'.format(
uri=parsed_url, elide_token='...' if parsed_url.path else '')
else:
inbound_api_url = ''
inbound_number = inbound_number_client.get_inbound_sms_number_for_service(service_id)
disp_inbound_number = inbound_number['data'].get('number', '')
reply_to_email_addresses = service_api_client.get_reply_to_email_addresses(service_id)
@@ -100,7 +78,6 @@ def service_settings(service_id):
current_service.get('dvla_organisation', '001')
),
can_receive_inbound=('inbound_sms' in current_service['permissions']),
inbound_api_url=inbound_api_url,
inbound_number=disp_inbound_number,
default_reply_to_email_address=default_reply_to_email_address,
reply_to_email_address_count=reply_to_email_address_count,
@@ -807,41 +784,3 @@ def get_branding_as_dict(organisations):
'colour': organisation['colour']
} for organisation in organisations
}
@main.route("/services/<service_id>/service-settings/set-inbound-api", methods=['GET', 'POST'])
@login_required
@user_has_permissions('manage_settings', admin_override=True)
def service_set_inbound_api(service_id):
if 'inbound_sms' not in current_service['permissions']:
abort(403)
inbound_api = get_inbound_api()
form = ServiceInboundApiForm(
url=inbound_api.get('url') if inbound_api else '',
bearer_token=dummy_bearer_token if inbound_api else ''
)
if form.validate_on_submit():
if inbound_api:
if inbound_api.get('url') != form.url.data or form.bearer_token.data != dummy_bearer_token:
service_api_client.update_service_inbound_api(
service_id,
url=form.url.data,
bearer_token=form.bearer_token.data if form.bearer_token.data != dummy_bearer_token else '',
user_id=current_user.id,
inbound_api_id=inbound_api.get('id')
)
else:
service_api_client.create_service_inbound_api(
service_id,
url=form.url.data,
bearer_token=form.bearer_token.data,
user_id=current_user.id
)
return redirect(url_for('.service_settings', service_id=service_id))
return render_template(
'views/service-settings/set-inbound-api.html',
form=form,
)

View File

@@ -387,6 +387,30 @@ class ServiceAPIClient(NotifyAdminAPIClient):
}
)
def get_service_callback_api(self, service_id, callback_api_id):
return self.get(
"/service/{}/delivery-receipt-api/{}".format(
service_id, callback_api_id
)
)['data']
def update_service_callback_api(self, service_id, url, bearer_token, user_id, callback_api_id):
data = {
"url": url,
"updated_by_id": user_id
}
if bearer_token:
data['bearer_token'] = bearer_token
return self.post("/service/{}/delivery-receipt-api/{}".format(service_id, callback_api_id), data)
def create_service_callback_api(self, service_id, url, bearer_token, user_id):
data = {
"url": url,
"bearer_token": bearer_token,
"updated_by_id": user_id
}
return self.post("/service/{}/delivery-receipt-api".format(service_id), data)
class ServicesBrowsableItem(BrowsableItem):
@property

View File

@@ -41,11 +41,12 @@
' form-control-error' if field.errors else ''
)
%}
{{ field(**{
'class': field_class,
'data-module': 'highlight-tags' if highlight_tags else '',
'rows': rows|string
}) }}
{{ field(
class=field_class,
data_module='highlight-tags' if highlight_tags else '',
rows=rows|string,
**kwargs
) }}
{% if suffix %}
<span>{{ suffix }}</span>
{% endif %}

View File

@@ -0,0 +1,33 @@
{% extends "withnav_template.html" %}
{% from "components/textbox.html" import textbox %}
{% from "components/page-footer.html" import page_footer %}
{% from "components/table.html" import mapping_table, row, text_field, edit_field, optional_text_field %}
{% block service_page_title %}
Callbacks
{% endblock %}
{% block maincolumn_content %}
<h1 class="heading-large">Callbacks</h1>
<div class="bottom-gutter-3-2 dashboard-table body-copy-table">
{% call mapping_table(
caption='General',
field_headings=['Label', 'Value', 'Action'],
field_headings_visible=False,
caption_visible=False
) %}
{% call row() %}
{{ text_field('Delivery status callback URL') }}
{{ optional_text_field(received_text_messages_callback) }}
{{ edit_field('Change', url_for('.delivery_status_callback', service_id=current_service.id)) }}
{% endcall %}
{% call row() %}
{{ text_field('Received text messages callback URL') }}
{{ optional_text_field(delivery_status_callback) }}
{{ edit_field('Change', url_for('.received_text_messages_callback', service_id=current_service.id)) }}
{% endcall %}
{% endcall %}
</div>
{% endblock %}

View File

@@ -3,32 +3,33 @@
{% from "components/page-footer.html" import page_footer %}
{% block service_page_title %}
Callback URL
Callback for delivery receipts
{% endblock %}
{% block maincolumn_content %}
<div class="grid-row">
<div class="column-five-sixths">
<h1 class="heading-large">Callback URL for received text messages</h1>
<h1 class="heading-large">Callback for delivery receipts</h1>
<p>
Text messages you receive can be forwarded to your systems with our callback feature.
See our <a href="{{ url_for('.callbacks') }}">documentation on the format of the callback</a>.
When you send an email or text message, we can tell you if Notify was able to deliver it.
Check the <a href="{{ url_for('.callbacks') }}"> callback documentation </a> for more information.
</p>
<form method="post">
{{ textbox(
form.url,
width='2-3',
hint='Valid https URL'
width='1-1',
hint='Must start with https://'
) }}
{{ textbox(
form.bearer_token,
width='2-3',
hint='At least 10 characters'
width='1-1',
hint='At least 10 characters',
autocomplete='new-password'
) }}
{{ page_footer(
'Save',
back_link=url_for('.service_settings', service_id=current_service.id),
back_link=url_for(back_link, service_id=current_service.id),
back_link_text='Back to settings'
) }}
</form>

View File

@@ -0,0 +1,39 @@
{% extends "withnav_template.html" %}
{% from "components/textbox.html" import textbox %}
{% from "components/page-footer.html" import page_footer %}
{% block service_page_title %}
Callback for received text messages
{% endblock %}
{% block maincolumn_content %}
<div class="grid-row">
<div class="column-five-sixths">
<h1 class="heading-large">Callback for received text messages</h1>
<p>
When you receive a text message in Notify, we can forward it to your system.
Check the <a href="{{ url_for('.callbacks') }}"> callback documentation </a> for more information.
</p>
<form method="post">
{{ textbox(
form.url,
width='1-1',
hint='Must start with https://'
) }}
{{ textbox(
form.bearer_token,
width='1-1',
hint='At least 10 characters',
autocomplete='new-password'
) }}
{{ page_footer(
'Save',
back_link=url_for('.api_callbacks', service_id=current_service.id),
back_link_text='Back to settings'
) }}
</form>
</div>
</div>
{% endblock %}

View File

@@ -21,7 +21,7 @@
<a class="pill-separate-item" href="{{ url_for('.whitelist', service_id=current_service.id) }}">Whitelist</a>
</div>
<div class="column-one-third">
<a class="pill-separate-item" href="{{ url_for('.api_documentation', service_id=current_service.id) }}">Documentation</a>
<a class="pill-separate-item" href="{{ url_for(callbacks_link, service_id=current_service.id) }}">Callbacks</a>
</div>
</nav>

View File

@@ -9,34 +9,42 @@
<div class="grid-row">
<div class="column-two-thirds">
<h1 class="heading-large">Callbacks for received text messages</h1>
<p>A callback lets you receive messages from Notify to a URL you choose.</p>
<p>Youll need to provide a bearer token, for security. Well add this to the authorisation header of the callback request.</p>
<p>The callback message is in JSON.</p>
<p>
Text messages you receive can be forwarded to a URL that you specify, using our callback feature.
</p>
<p>
Messages are forwarded as they are received.
</p>
<p>
To protect your service, we require you to provide a bearer token. We put this token in the authorisation header of the callback requests.
</p>
<p>
Once you have receive text messages enabled, you can set up your callback on the settings page of your service.
</p>
<p>
If you dont have receive text messages enabled for your service, <a href="{{ url_for('.support') }}">get in touch</a> and we can turn it on for you.
</p>
<h2 class="heading-medium">Format of the callback</h2>
<p>
The format of the callback message you receive is JSON.
</p>
<h2 class="heading-medium">Email and text message delivery receipts</h2>
<p>When you send an email or text message through Notify, we can send a receipt to your callback URL to tell you if we were able to deliver it or not.</p>
<div class="bottom-gutter-3-2">
{% call mapping_table(
caption='Callback message format',
field_headings=['Key', 'Description', 'Format'],
field_headings_visible=True,
caption_visible=False
) %}
{% for key, description, format in [
('id', 'Notifys id for the status receipts', 'UUID'),
('reference', 'The reference sent by the service', '12345678'),
('to', 'The email address of the recipient', 'hello@gov.uk'),
('status', 'The status of the notification', 'delivered | permanent-failure | temporary-failure | technical-failure'),
('created_at', 'The time the service sent the request', '2017-05-14T12:15:30.000000Z'),
('completed_at', 'The last time the status was updated', '2017-05-14T12:15:30.000000Z'),
('sent_at', 'The time the notification was sent', '2017-05-14T12:15:30.000000Z or nil'),
('notification_type', 'The notification type', 'email | sms | letter')
] %}
{% call row() %}
{% call row_heading() %} {{ key }} {% endcall %}
{{ text_field(description) }}
{{ text_field(format) }}
{% endcall %}
{% endfor %}
{% endcall %}
</div>
<h2 class="heading-medium">Text messages you receive</h2>
<p>If your service receives text messages in Notify, we can forward them to your callback URL as soon as they arrive.</p>
<div class="bottom-gutter-3-2">
{% call mapping_table(
caption='Callback message format',
@@ -59,5 +67,4 @@
{% endfor %}
{% endcall %}
</div>
{% endblock %}

View File

@@ -120,14 +120,6 @@
{{ edit_field('Change', url_for('.service_set_inbound_sms', service_id=current_service.id)) }}
{% endcall %}
{% if can_receive_inbound %}
{% call row() %}
{{ text_field('Callback URL for received text messages') }}
{{ optional_text_field(inbound_api_url) }}
{{ edit_field('Change', url_for('.service_set_inbound_api', service_id=current_service.id)) }}
{% endcall %}
{% endif %}
{% endif %}
{% endcall %}

View File

@@ -120,7 +120,7 @@
<div class="grid-row bottom-gutter">
<div class="column-half">
<h3 class="visually-hidden">Services</h3>
<div class="product-page-big-number">119</div>
<div class="product-page-big-number">120</div>
services
</div>
<div class="column-half">

View File

@@ -53,6 +53,7 @@ def service_json(
created_at=None,
letter_contact_block=None,
inbound_api=None,
service_callback_api=None,
permissions=None,
organisation_type='central',
free_sms_fragment_limit=250000,
@@ -84,6 +85,7 @@ def service_json(
'dvla_organisation': '001',
'permissions': permissions,
'inbound_api': inbound_api,
'service_callback_api': service_callback_api,
'prefix_sms': prefix_sms,
}

View File

@@ -13,6 +13,8 @@ from tests.conftest import (
mock_get_service_with_letters,
normalize_spaces,
SERVICE_ONE_ID,
mock_get_valid_service_callback_api,
mock_get_valid_service_inbound_api,
)
@@ -391,3 +393,267 @@ def test_should_validate_whitelist_items(
assert jump_links[1]['href'] == '#phone_numbers'
mock_update_whitelist.assert_not_called()
@pytest.mark.parametrize('endpoint', [
('main.delivery_status_callback'),
('main.received_text_messages_callback'),
])
@pytest.mark.parametrize('url, bearer_token, expected_errors', [
("", "", "Cant be empty Cant be empty"),
("http://not_https.com", "1234567890", "Must be a valid https URL"),
("https://test.com", "123456789", "Must be at least 10 characters"),
])
def test_callback_forms_validation(
client_request,
service_one,
endpoint,
url,
bearer_token,
expected_errors
):
if endpoint == 'main.received_text_messages_callback':
service_one['permissions'] = ['inbound_sms']
data = {
"url": url,
"bearer_token": bearer_token,
}
response = client_request.post(
endpoint,
service_id=service_one['id'],
_data=data,
_expected_status=200
)
error_msgs = ' '.join(msg.text.strip() for msg in response.select(".error-message"))
assert error_msgs == expected_errors
@pytest.mark.parametrize('has_inbound_sms, expected_link', [
(True, 'main.api_callbacks'),
(False, 'main.delivery_status_callback'),
])
def test_callbacks_button_links_straight_to_delivery_status_if_service_has_no_inbound_sms(
client_request,
service_one,
mocker,
mock_get_notifications,
has_inbound_sms,
expected_link
):
if has_inbound_sms:
service_one['permissions'] = ['inbound_sms']
page = client_request.get(
'main.api_integration',
service_id=service_one['id'],
)
assert page.select('.pill-separate-item')[2]['href'] == url_for(
expected_link, service_id=service_one['id']
)
def test_callbacks_page_redirects_to_delivery_status_if_service_has_no_inbound_sms(
client_request,
service_one,
mocker
):
page = client_request.get(
'main.api_callbacks',
service_id=service_one['id'],
_follow_redirects=True,
)
assert normalize_spaces(page.select_one('h1').text) == "Callback for delivery receipts"
@pytest.mark.parametrize('has_inbound_sms, expected_link', [
(True, 'main.api_callbacks'),
(False, 'main.api_integration'),
])
def test_back_link_directs_to_api_integration_from_delivery_callback_if_no_inbound_sms(
client_request,
service_one,
mocker,
has_inbound_sms,
expected_link
):
if has_inbound_sms:
service_one['permissions'] = ['inbound_sms']
page = client_request.get(
'main.delivery_status_callback',
service_id=service_one['id'],
_follow_redirects=True,
)
assert page.select_one('.page-footer-back-link')['href'] == url_for(
expected_link, service_id=service_one['id']
)
@pytest.mark.parametrize('endpoint', [
('main.delivery_status_callback'),
('main.received_text_messages_callback'),
])
def test_create_delivery_status_and_receive_text_message_callbacks(
client_request,
service_one,
mocker,
mock_get_notifications,
mock_create_service_inbound_api,
mock_create_service_callback_api,
endpoint,
fake_uuid,
):
if endpoint == 'main.received_text_messages_callback':
service_one['permissions'] = ['inbound_sms']
data = {
'url': "https://test.url.com/",
'bearer_token': '1234567890',
'user_id': fake_uuid
}
client_request.post(
endpoint,
service_id=service_one['id'],
_data=data,
)
if endpoint == 'main.received_text_messages_callback':
mock_create_service_inbound_api.assert_called_once_with(
service_one['id'],
url="https://test.url.com/",
bearer_token="1234567890",
user_id=fake_uuid,
)
else:
mock_create_service_callback_api.assert_called_once_with(
service_one['id'],
url="https://test.url.com/",
bearer_token="1234567890",
user_id=fake_uuid,
)
@pytest.mark.parametrize('endpoint, fixture', [
('main.delivery_status_callback', mock_get_valid_service_callback_api),
('main.received_text_messages_callback', mock_get_valid_service_inbound_api),
])
def test_update_delivery_status_and_receive_text_message_callbacks(
client_request,
service_one,
mocker,
mock_get_notifications,
mock_update_service_inbound_api,
mock_update_service_callback_api,
endpoint,
fixture,
fake_uuid,
):
if endpoint == 'main.received_text_messages_callback':
service_one['inbound_api'] = [fake_uuid]
service_one['permissions'] = ['inbound_sms']
else:
service_one['service_callback_api'] = [fake_uuid]
fixture(mocker)
data = {
'url': "https://test.url.com/",
'bearer_token': '1234567890',
'user_id': fake_uuid
}
client_request.post(
endpoint,
service_id=service_one['id'],
_data=data,
)
if endpoint == 'main.received_text_messages_callback':
mock_update_service_inbound_api.assert_called_once_with(
service_one['id'],
url="https://test.url.com/",
bearer_token="1234567890",
user_id=fake_uuid,
inbound_api_id=fake_uuid,
)
else:
mock_update_service_callback_api.assert_called_once_with(
service_one['id'],
url="https://test.url.com/",
bearer_token="1234567890",
user_id=fake_uuid,
callback_api_id=fake_uuid
)
@pytest.mark.parametrize('endpoint, data, fixture', [
(
'main.delivery_status_callback',
{"url": "https://hello2.gov.uk", "bearer_token": "bearer_token_set"},
mock_get_valid_service_callback_api
),
(
'main.received_text_messages_callback',
{"url": "https://hello3.gov.uk", "bearer_token": "bearer_token_set"},
mock_get_valid_service_inbound_api
),
])
def test_update_delivery_status_and_receive_text_message_callbacks_without_changes_do_not_update(
client_request,
service_one,
mocker,
mock_get_notifications,
mock_update_service_callback_api,
mock_update_service_inbound_api,
data,
fixture,
endpoint,
fake_uuid,
):
if endpoint == 'main.received_text_messages_callback':
service_one['inbound_api'] = [fake_uuid]
service_one['permissions'] = ['inbound_sms']
else:
service_one['service_callback_api'] = [fake_uuid]
fixture(mocker)
data['user_id'] = fake_uuid
client_request.post(
endpoint,
service_id=service_one['id'],
_data=data,
)
if endpoint == 'main.received_text_messages_callback':
assert mock_update_service_inbound_api.called is False
else:
assert mock_update_service_callback_api.called is False
def test_callbacks_page_works_when_no_apis_set(
client_request,
service_one,
mocker
):
service_one['permissions'] = ['inbound_sms']
mocker.patch('app.service_api_client.get_service_callback_api', side_effect={})
mocker.patch('app.service_api_client.get_service_inbound_api', side_effect={})
page = client_request.get('main.api_callbacks',
service_id=service_one['id'],
_follow_redirects=True)
expected_rows = ['Delivery status callback URL Not set Change',
'Received text messages callback URL Not set Change']
rows = page.select('tr')
assert len(rows) == 3
for index, row in enumerate(expected_rows):
assert row == " ".join(rows[index + 1].text.split())

View File

@@ -7,7 +7,6 @@ from flask import url_for
from werkzeug.exceptions import InternalServerError
import app
from app.main.views.service_settings import dummy_bearer_token
from app.utils import email_safe
from tests import validate_route_permission, service_json
from tests.conftest import (
@@ -136,7 +135,6 @@ def test_should_show_overview(
'Text messages start with service name On Change',
'International text messages On Change',
'Receive text messages On Change',
'Callback URL for received text messages Not set Change',
'Label Value Action',
'Send letters Off Change',
@@ -186,47 +184,6 @@ def test_should_show_overview_for_service_with_more_things_set(
assert row == " ".join(page.find_all('tr')[index + 1].text.split())
@pytest.mark.parametrize('url, elided_url', [
('https://test.url.com/inbound', 'https://test.url.com...'),
('https://test.url.com/', 'https://test.url.com...'),
('https://test.url.com', 'https://test.url.com'),
])
def test_service_settings_show_elided_api_url_if_needed(
logged_in_platform_admin_client,
service_one,
single_reply_to_email_address,
single_sms_sender,
single_letter_contact_block,
mocker,
fake_uuid,
url,
elided_url,
mock_get_service_settings_page_common,
):
service_one['permissions'] = ['sms', 'email', 'inbound_sms']
service_one['inbound_api'] = [fake_uuid]
mocked_get_fn = mocker.patch(
'app.service_api_client.get',
return_value={'data': {'id': fake_uuid, 'url': url}}
)
response = logged_in_platform_admin_client.get(
url_for(
'main.service_settings',
service_id=service_one['id']
)
)
assert response.status_code == 200
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
non_empty_trs = [tr.find_all('td') for tr in page.find_all('tr') if tr.find_all('td')]
api_url = [api_setting[1].text.strip() for api_setting in non_empty_trs
if api_setting[0].text.strip() == 'Callback URL for received text messages'][0]
assert api_url == elided_url
assert mocked_get_fn.called is True
def test_if_cant_send_letters_then_cant_see_letter_contact_block(
logged_in_client,
service_one,
@@ -1274,34 +1231,6 @@ def test_does_not_show_research_mode_indicator(
assert not element
@pytest.mark.parametrize('url, bearer_token, expected_errors', [
("", "", "Cant be empty Cant be empty"),
("http://not_https.com", "1234567890", "Must be a valid https URL"),
("https://test.com", "123456789", "Must be at least 10 characters"),
])
def test_set_inbound_api_validation(
logged_in_client,
mock_update_service,
service_one,
mock_get_letter_organisations,
url,
bearer_token,
expected_errors,
):
service_one['permissions'] = ['inbound_sms']
response = logged_in_client.post(url_for(
'main.service_set_inbound_api',
service_id=service_one['id']),
data={"url": url, "bearer_token": bearer_token}
)
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
error_msgs = ' '.join(msg.text.strip() for msg in page.select(".error-message"))
assert response.status_code == 200
assert error_msgs == expected_errors
assert not mock_update_service.called
@pytest.mark.parametrize('method', ['get', 'post'])
def test_cant_set_letter_contact_block_if_service_cant_send_letters(
logged_in_client,
@@ -1707,113 +1636,6 @@ def test_switch_service_enable_international_sms(
assert mocked_fn.call_args[0][0] == service_one['id']
def test_set_new_inbound_api_and_valid_bearer_token_calls_create_inbound_api_endpoint(
logged_in_platform_admin_client,
service_one,
mocker,
):
service_one['permissions'] = ['inbound_sms']
service_one['inbound_api'] = []
mocked_post_fn = mocker.patch('app.service_api_client.post', return_value=service_one)
inbound_api_data = {'url': "https://test.url.com/", 'bearer_token': '1234567890'}
response = logged_in_platform_admin_client.post(
url_for(
'main.service_set_inbound_api',
service_id=service_one['id']
),
data=inbound_api_data
)
assert response.status_code == 302
assert response.location == url_for('main.service_settings', service_id=service_one['id'], _external=True)
assert mocked_post_fn.called
inbound_api_data['updated_by_id'] = service_one['users'][0]
assert mocked_post_fn.call_args == call("/service/{}/inbound-api".format(service_one['id']), inbound_api_data)
@pytest.mark.parametrize(
'inbound_api_data', [
{'url': "https://test.url.com/inbound", 'bearer_token': dummy_bearer_token},
{'url': "https://test.url.com/inbound", 'bearer_token': '1234567890'},
{'url': "https://test.url.com/", 'bearer_token': 'new_1234567890'},
]
)
def test_update_inbound_api_and_valid_bearer_token_calls_update_inbound_api_endpoint(
logged_in_platform_admin_client,
service_one,
mocker,
fake_uuid,
inbound_api_data,
):
service_one['permissions'] = ['inbound_sms']
service_one['inbound_api'] = [fake_uuid]
initial_api_data = {'data': {'id': fake_uuid, 'url': "https://test.url.com/"}}
mocked_get_fn = mocker.patch('app.service_api_client.get', return_value=initial_api_data)
mocked_post_fn = mocker.patch('app.service_api_client.post', return_value=service_one)
response = logged_in_platform_admin_client.get(
url_for(
'main.service_set_inbound_api',
service_id=service_one['id']
)
)
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
assert page.find('input', {'id': 'url'}).get('value') == initial_api_data['data']['url']
assert page.find('input', {'id': 'bearer_token'}).get('value') == dummy_bearer_token
response = logged_in_platform_admin_client.post(
url_for(
'main.service_set_inbound_api',
service_id=service_one['id']
),
data=inbound_api_data
)
assert response.status_code == 302
assert response.location == url_for('main.service_settings', service_id=service_one['id'], _external=True)
assert mocked_get_fn.called is True
assert mocked_post_fn.called is True
if inbound_api_data['bearer_token'] == dummy_bearer_token:
del inbound_api_data['bearer_token']
inbound_api_data['updated_by_id'] = service_one['users'][0]
assert mocked_post_fn.call_args == call(
"/service/{}/inbound-api/{}".format(service_one['id'], fake_uuid), inbound_api_data)
def test_save_inbound_api_without_changes_does_not_update_inbound_api(
logged_in_platform_admin_client,
service_one,
mocker,
fake_uuid,
):
service_one['permissions'] = ['inbound_sms']
service_one['inbound_api'] = [fake_uuid]
initial_api_data = {'data': {'id': fake_uuid, 'url': "https://test.url.com/"}}
inbound_api_data = {'url': initial_api_data['data']['url'], 'bearer_token': dummy_bearer_token}
mocked_get_fn = mocker.patch('app.service_api_client.get', return_value=initial_api_data)
mocked_post_fn = mocker.patch('app.service_api_client.post', return_value=service_one)
response = logged_in_platform_admin_client.post(
url_for(
'main.service_set_inbound_api',
service_id=service_one['id']
),
data=inbound_api_data
)
assert response.status_code == 302
assert response.location == url_for('main.service_settings', service_id=service_one['id'], _external=True)
assert mocked_get_fn.called is True
assert mocked_post_fn.called is False
def test_archive_service_after_confirm(
logged_in_platform_admin_client,
service_one,

View File

@@ -2497,3 +2497,65 @@ def valid_token(app_, fake_uuid):
app_.config['SECRET_KEY'],
app_.config['DANGEROUS_SALT']
)
@pytest.fixture(scope='function')
def mock_get_valid_service_inbound_api(mocker):
def _get(service_id, inbound_api_id):
return {
'created_at': '2017-12-04T10:52:55.289026Z',
'updated_by_id': fake_uuid,
'id': inbound_api_id,
'url': 'https://hello3.gov.uk',
'service_id': service_id,
'updated_at': '2017-12-04T11:28:42.575153Z'
}
return mocker.patch('app.service_api_client.get_service_inbound_api', side_effect=_get)
@pytest.fixture(scope='function')
def mock_get_valid_service_callback_api(mocker):
def _get(service_id, callback_api_id):
return {
'created_at': '2017-12-04T10:52:55.289026Z',
'updated_by_id': fake_uuid,
'id': callback_api_id,
'url': 'https://hello2.gov.uk',
'service_id': service_id,
'updated_at': '2017-12-04T11:28:42.575153Z'
}
return mocker.patch('app.service_api_client.get_service_callback_api', side_effect=_get)
@pytest.fixture(scope='function')
def mock_create_service_inbound_api(mocker):
def _create_service_inbound_api(service_id, url, bearer_token, user_id):
return
return mocker.patch('app.service_api_client.create_service_inbound_api', side_effect=_create_service_inbound_api)
@pytest.fixture(scope='function')
def mock_update_service_inbound_api(mocker):
def _update_service_inbound_api(service_id, url, bearer_token, user_id, inbound_api_id):
return
return mocker.patch('app.service_api_client.update_service_inbound_api', side_effect=_update_service_inbound_api)
@pytest.fixture(scope='function')
def mock_create_service_callback_api(mocker):
def _create_service_callback_api(service_id, url, bearer_token, user_id):
return
return mocker.patch('app.service_api_client.create_service_callback_api', side_effect=_create_service_callback_api)
@pytest.fixture(scope='function')
def mock_update_service_callback_api(mocker):
def _update_service_callback_api(service_id, url, bearer_token, user_id, callback_api_id):
return
return mocker.patch('app.service_api_client.update_service_callback_api', side_effect=_update_service_callback_api)