darkhelm/notifications-admin
1
0
Fork 0
mirror of https://github.com/GSA/notifications-admin.git synced 2026-08-11 17:38:50 -04:00
Code Issues Packages Projects Releases Wiki Activity

refactor of api callbacks admin work

Browse Source
This commit is contained in:
chrisw
2017-12-08 10:52:38 +00:00
parent cd3d8edccd
commit 9c0d2c2a39
8 changed files with 469 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
app/main/forms.py
View File

@@ -717,38 +717,30 @@ class ServiceInboundNumberForm(Form):
)
class ServiceCallbacksForm(Form):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.can_receive_inbound = kwargs['can_receive_inbound']
inbound_url = StringField("Inbound URL")
inbound_bearer_token = PasswordFieldShowHasContent("Bearer token")
outbound_url = StringField(
"Outbound URL",
class ServiceReceiveMessagesCallbackForm(Form):
url = StringField(
"URL",
validators=[DataRequired(message='Can’t be empty'),
Regexp(regex="^https.*", message='Must be a valid https URL')]
)
outbound_bearer_token = PasswordFieldShowHasContent(
bearer_token = PasswordFieldShowHasContent(
"Bearer token",
validators=[DataRequired(message='Can’t be empty'),
Length(min=10, message='Must be at least 10 characters')]
)
def validate_inbound_url(self, field):
pattern = re.compile("^https.*")
if self.can_receive_inbound:
if not field.data:
raise ValidationError('Can’t be empty')
elif not pattern.match(field.data):
raise ValidationError('Must be a valid https URL')
def validate_inbound_bearer_token(self, field):
if self.can_receive_inbound:
if not field.data:
raise ValidationError('Can’t be empty')
elif len(field.data) < 10:
raise ValidationError('Must be at least 10 characters')
class ServiceDeliveryStatusCallbackForm(Form):
url = StringField(
"URL",
validators=[DataRequired(message='Can’t be empty'),
Regexp(regex="^https.*", message='Must be a valid https URL')]
)
bearer_token = PasswordFieldShowHasContent(
"Bearer token",
validators=[DataRequired(message='Can’t be empty'),
Length(min=10, message='Must be at least 10 characters')]
)
class InternationalSMSForm(Form):

137
app/main/views/api_integration.py
View File

@@ -1,7 +1,12 @@
from flask import request, render_template, redirect, url_for, flash, Markup, abort
from flask_login import login_required, current_user
from app.main import main
from app.main.forms import CreateKeyForm, Whitelist, ServiceCallbacksForm
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
@@ -13,8 +18,13 @@ dummy_bearer_token = 'bearer_token_set'
@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)
)
@@ -144,58 +154,107 @@ def check_token_against_dummy_bearer(token):
@main.route("/services/<service_id>/api/callbacks", methods=['GET', 'POST'])
@login_required
def api_callbacks(service_id):
callback_api, inbound_api = get_apis()
can_receive_inbound = 'inbound_sms' in current_service['permissions']
if 'inbound_sms' not in current_service['permissions']:
return redirect(url_for('.delivery_status_callback', service_id=service_id))
form = ServiceCallbacksForm(
inbound_url=inbound_api.get('url') if inbound_api else '',
inbound_bearer_token=dummy_bearer_token if inbound_api else '',
outbound_url=callback_api.get('url') if callback_api else '',
outbound_bearer_token=dummy_bearer_token if callback_api else '',
can_receive_inbound=can_receive_inbound,
received_text_messages_callback, delivery_status_callback = get_apis()
return render_template(
'views/api/callbacks.html',
received_text_messages_callback=received_text_messages_callback.get('url', "Not Set"),
delivery_status_callback=delivery_status_callback.get('url', "Not Set")
)
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 callback_api:
if (callback_api.get('url') != form.outbound_url.data
or form.outbound_bearer_token.data != dummy_bearer_token):
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.outbound_url.data,
bearer_token=check_token_against_dummy_bearer(form.outbound_bearer_token.data),
url=form.url.data,
bearer_token=check_token_against_dummy_bearer(form.bearer_token.data),
user_id=current_user.id,
callback_api_id=callback_api.get('id')
callback_api_id=delivery_status_callback.get('id')
)
else:
service_api_client.create_service_callback_api(
service_id,
url=form.outbound_url.data,
bearer_token=form.outbound_bearer_token.data,
url=form.url.data,
bearer_token=form.bearer_token.data,
user_id=current_user.id
)
if can_receive_inbound:
if inbound_api:
if (inbound_api.get('url') != form.inbound_url.data
or form.inbound_bearer_token.data != dummy_bearer_token):
service_api_client.update_service_inbound_api(
service_id,
url=form.inbound_url.data,
bearer_token=check_token_against_dummy_bearer(form.inbound_bearer_token.data),
user_id=current_user.id,
inbound_api_id=inbound_api.get('id')
)
else:
service_api_client.create_service_inbound_api(
service_id,
url=form.inbound_url.data,
bearer_token=form.inbound_bearer_token.data,
user_id=current_user.id
)
return redirect(url_for('.api_integration', service_id=service_id))
return redirect(url_for(back_link, service_id=service_id))
return render_template(
'views/api/callbacks.html',
'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,
can_receive_inbound=can_receive_inbound,
)

60
app/templates/views/api/callbacks.html
View File

@@ -1,49 +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 %}
{% block service_page_title %}
Callbacks
{% endblock %}
{% block maincolumn_content %}
<div class="grid-row">
<div class="column-five-sixths">
<h1 class="heading-large">Callbacks</h1>
<p>
See our <a href="{{ url_for('.callbacks') }}">documentation on the format of the callback</a>.
</p>
<form method="post">
{% if can_receive_inbound %}
{{ textbox(
form.inbound_url,
width='1-1',
hint='Valid https URL'
) }}
{{ textbox(
form.inbound_bearer_token,
width='2-3',
hint='At least 10 characters'
) }}
{% endif %}
{{ textbox(
form.outbound_url,
width='1-1',
hint='Valid https URL'
) }}
{{ textbox(
form.outbound_bearer_token,
width='2-3',
hint='At least 10 characters'
) }}
{{ page_footer(
'Save',
back_link=url_for('.api_integration', service_id=current_service.id),
back_link_text='Back to API integration'
) }}
</form>
</div>
</div>
<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') }}
{{ 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') }}
{{ text_field(delivery_status_callback) }}
{{ edit_field('Change', url_for('.received_text_messages_callback', service_id=current_service.id)) }}
{% endcall %}
{% endcall %}
</div>
{% endblock %}

39
app/templates/views/api/callbacks/delivery-status-callback.html Normal file
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 delivery receipts
{% endblock %}
{% block maincolumn_content %}
<div class="grid-row">
<div class="column-five-sixths">
<h1 class="heading-large">Callback for delivery receipts</h1>
<p>
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'
) }}
{{ textbox(
form.bearer_token,
width='2-3',
hint='At least 10 characters',
autocomplete='new-password'
) }}
{{ page_footer(
'Save',
back_link=url_for(back_link, service_id=current_service.id),
back_link_text='Back to settings'
) }}
</form>
</div>
</div>
{% endblock %}

39
app/templates/views/api/callbacks/received-text-messages-callback.html Normal file
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='2-3',
hint='Valid https URL'
) }}
{{ textbox(
form.bearer_token,
width='2-3',
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 %}

2
app/templates/views/api/index.html
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_callbacks', service_id=current_service.id) }}">Callback Url</a>
<a class="pill-separate-item" href="{{ url_for(callbacks_link, service_id=current_service.id) }}">Callbacks</a>
</div>
</nav>

40
app/templates/views/callbacks.html
View File

@@ -9,36 +9,14 @@
<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>You’ll need to provide a bearer token, for security. We’ll 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 don’t 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 callbacks</h2>
<p>
The format of the callback message you receive is JSON.
</p>
<h3 class="heading-small"> Received text messages callback </h3>
<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',
@@ -62,8 +40,8 @@
{% endcall %}
</div>
<h3 class="heading-small"> Status callback </h3>
<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',
@@ -77,7 +55,7 @@
('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'),
('updated_at', 'The last time the status was updated', '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')
] %}

476
tests/app/main/views/test_api_integration.py
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,
)
@@ -393,236 +395,6 @@ def test_should_validate_whitelist_items(
mock_update_whitelist.assert_not_called()
@pytest.mark.parametrize('url, bearer_token, expected_errors', [
("", "", "Can’t be empty Can’t 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_outbound_api_validation(
client_request,
service_one,
url,
bearer_token,
expected_errors,
mock_create_service_callback_api,
mock_update_service_callback_api
):
response = client_request.post(
'main.api_callbacks',
service_id=service_one['id'],
_data={"outbound_url": url, "outbound_bearer_token": bearer_token},
_expected_status=200
)
error_msgs = ' '.join(msg.text.strip() for msg in response.select(".error-message"))
assert error_msgs == expected_errors
mock_create_service_callback_api.assert_not_called()
mock_update_service_callback_api.assert_not_called()
@pytest.mark.parametrize('url, bearer_token, expected_errors', [
("", "", "Can’t be empty Can’t be empty Can’t be empty Can’t be empty"),
("http://not_https.com", "1234567890", "Must be a valid https URL Must be a valid https URL"),
("https://test.com", "123456789", "Must be at least 10 characters Must be at least 10 characters"),
])
def test_set_inbound_api_validation(
client_request,
service_one,
url,
bearer_token,
expected_errors,
fake_uuid
):
service_one['permissions'] = ['inbound_sms']
data = {
"inbound_url": url,
"inbound_bearer_token": bearer_token,
"outbound_url": url,
"outbound_bearer_token": bearer_token
}
response = client_request.post(
'main.api_callbacks',
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
def test_create_new_callback_api_without_inbound_set(
client_request,
service_one,
mock_create_service_callback_api,
mock_get_notifications,
fake_uuid,
mocker,
):
service_one['service_callback_api'] = []
callback_api_data = {
'outbound_url': "https://test.url.com/",
'outbound_bearer_token': '1234567890',
'user_id': fake_uuid
}
client_request.post(
'main.api_callbacks',
service_id=service_one['id'],
_data=callback_api_data,
_follow_redirects=True,
)
callback_api_data['updated_by_id'] = service_one['users'][0]
mock_create_service_callback_api.assert_called_once_with(
service_one['id'],
url="https://test.url.com/",
bearer_token="1234567890",
user_id=fake_uuid
)
def test_update_new_callback_api_without_inbound_set(
client_request,
service_one,
mock_get_valid_service_callback_api,
mock_update_service_callback_api,
mock_get_notifications,
fake_uuid,
mocker,
):
service_one['service_callback_api'] = [fake_uuid]
callback_api_data = {
'outbound_url': "https://test.url.com/",
'outbound_bearer_token': '1234567890',
'user_id': fake_uuid
}
client_request.post(
'main.api_callbacks',
service_id=service_one['id'],
_data=callback_api_data,
_follow_redirects=True,
)
callback_api_data['updated_by_id'] = service_one['users'][0]
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
)
def test_inbound_fields_do_not_show_if_inbound_is_disabled(
client_request,
service_one,
mocker
):
response = client_request.get(
'main.api_callbacks',
service_id=service_one['id'],
_expected_status=200
)
assert response.select_one('label[for="inbound_url"]') is None
assert response.select_one('label[for="inbound_bearer_token"]') is None
def test_create_inbound_and_outbound_apis(
client_request,
service_one,
mocker,
mock_create_service_callback_api,
mock_create_service_inbound_api,
mock_get_notifications,
fake_uuid,
):
service_one['permissions'] = ['inbound_sms']
callback_api_data = {
'outbound_url': "https://test.url.com/",
'outbound_bearer_token': '1234567890',
'inbound_url': "https://test2.url.com/",
'inbound_bearer_token': '5678901234',
'user_id': fake_uuid
}
client_request.post(
'main.api_callbacks',
service_id=service_one['id'],
_data=callback_api_data,
_follow_redirects=True,
)
mock_create_service_callback_api.assert_called_once_with(
service_one['id'],
url="https://test.url.com/",
bearer_token="1234567890",
user_id=fake_uuid
)
mock_create_service_inbound_api.assert_called_once_with(
service_one['id'],
url="https://test2.url.com/",
bearer_token="5678901234",
user_id=fake_uuid
)
def test_update_inbound_and_outbound_apis(
client_request,
service_one,
mocker,
mock_get_valid_service_callback_api,
mock_get_valid_service_inbound_api,
mock_update_service_callback_api,
mock_update_service_inbound_api,
mock_get_notifications,
fake_uuid,
):
service_one['service_callback_api'] = [fake_uuid]
service_one['inbound_api'] = [fake_uuid]
service_one['permissions'] = ['inbound_sms']
callback_api_data = {
'outbound_url': "https://test.url.com/",
'outbound_bearer_token': '1234567890',
'inbound_url': "https://test2.url.com/",
'inbound_bearer_token': '5678901234',
'user_id': fake_uuid
}
client_request.post(
'main.api_callbacks',
service_id=service_one['id'],
_data=callback_api_data,
_follow_redirects=True,
)
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
)
mock_update_service_inbound_api.assert_called_once_with(
service_one['id'],
url="https://test2.url.com/",
bearer_token="5678901234",
user_id=fake_uuid,
inbound_api_id=fake_uuid
)
def test_save_callback_apis_without_changes_does_not_update_callback_apis(
client_request,
service_one,
@@ -655,3 +427,247 @@ def test_save_callback_apis_without_changes_does_not_update_callback_apis(
assert mock_update_service_callback_api.called is False
assert mock_update_service_inbound_api.called is False
@pytest.mark.parametrize('endpoint', [
('main.delivery_status_callback'),
('main.received_text_messages_callback'),
])
@pytest.mark.parametrize('url, bearer_token, expected_errors', [
("", "", "Can’t be empty Can’t 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