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

Allow service to set callback url for notifications

Browse Source
This commit is contained in:
chrisw
2017-12-04 15:07:11 +00:00
parent 72dd0b369f
commit 76117c148c
12 changed files with 523 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
app/main/forms.py
View File

@@ -717,15 +717,38 @@ class ServiceInboundNumberForm(Form):
)
class ServiceInboundApiForm(Form):
url = StringField("Callback 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 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",
validators=[DataRequired(message='Can’t be empty'),
Regexp(regex="^https.*", message='Must be a valid https URL')]
)
outbound_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 InternationalSMSForm(Form):

90
app/main/views/api_keys.py
View File

@@ -1,11 +1,13 @@
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, ServiceCallbacksForm
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
@@ -113,3 +115,87 @@ 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', 'POST'])
@login_required
def api_callbacks(service_id):
callback_api, inbound_api = get_apis()
can_receive_inbound = 'inbound_sms' in current_service['permissions']
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,
)
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):
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),
user_id=current_user.id,
callback_api_id=callback_api.get('id')
)
else:
service_api_client.create_service_callback_api(
service_id,
url=form.outbound_url.data,
bearer_token=form.outbound_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 render_template(
'views/api/callbacks.html',
form=form,
can_receive_inbound=can_receive_inbound,
)

61
app/main/views/service_settings.py
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,
)

24
app/notify_client/service_api_client.py
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

27
app/templates/views/service-settings/set-inbound-api.html → app/templates/views/api/callbacks.html
View File

@@ -3,33 +3,44 @@
{% from "components/page-footer.html" import page_footer %}
{% block service_page_title %}
Callback URL
Callbacks
{% 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">Callbacks</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>.
</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.url,
width='2-3',
form.outbound_url,
width='1-1',
hint='Valid https URL'
) }}
{{ textbox(
form.bearer_token,
form.outbound_bearer_token,
width='2-3',
hint='At least 10 characters'
) }}
{{ page_footer(
'Save',
back_link=url_for('.service_settings', service_id=current_service.id),
back_link_text='Back to settings'
back_link=url_for('.api_integration', service_id=current_service.id),
back_link_text='Back to API integration'
) }}
</form>
</div>

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_documentation', service_id=current_service.id) }}">Documentation</a>
<a class="pill-separate-item" href="{{ url_for('.api_callbacks', service_id=current_service.id) }}">Callback Url</a>
</div>
</nav>

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

@@ -31,12 +31,14 @@
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 callback</h2>
<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>
<div class="bottom-gutter-3-2">
{% call mapping_table(
caption='Callback message format',
@@ -60,4 +62,32 @@
{% endcall %}
</div>
<h3 class="heading-small"> Status callback </h3>
<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', 'Notify’s 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'),
('updated_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>
{% endblock %}

8
app/templates/views/service-settings.html
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 %}

2
tests/__init__.py
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,
}

264
tests/app/main/views/test_api_keys.py
View File

@@ -391,3 +391,267 @@ def test_should_validate_whitelist_items(
assert jump_links[1]['href'] == '#phone_numbers'
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,
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://hello2.gov.uk",
'outbound_bearer_token': 'bearer_token_set',
'inbound_url': "https://hello3.gov.uk",
'inbound_bearer_token': 'bearer_token_set',
'user_id': fake_uuid
}
client_request.post(
'main.api_callbacks',
service_id=service_one['id'],
_data=callback_api_data,
_follow_redirects=True,
)
assert mock_update_service_callback_api.called is False
assert mock_update_service_inbound_api.called is False

178
tests/app/main/views/test_service_settings.py
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', [
("", "", "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_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,

62
tests/conftest.py
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)