Enable update of url / bearer_token inbound api

This commit is contained in:
Ken Tsang
2017-06-21 17:34:22 +01:00
parent ba8ee3e60a
commit c5e1de9b33
4 changed files with 78 additions and 18 deletions

View File

@@ -9,6 +9,7 @@ from notifications_utils.recipients import (
)
from notifications_utils.columns import Columns
from wtforms import (
widgets,
validators,
StringField,
PasswordField,
@@ -651,15 +652,19 @@ class PlaceholderForm(Form):
pass
class PasswordFieldShowHasContent(StringField):
widget = widgets.PasswordInput(hide_value=False)
class ServiceInboundApiForm(Form):
url = StringField("Inbound sms url",
validators=[DataRequired(message='Cant be empty'),
Regexp(regex="^https.*",
message='Must be a valid https url')]
)
bearer_token = PasswordField("Bearer token",
validators=[DataRequired(message='Cant be empty'),
Length(min=10, message='Must be at least 10 characters')])
bearer_token = PasswordFieldShowHasContent("Bearer token",
validators=[DataRequired(message='Cant be empty'),
Length(min=10, message='Must be at least 10 characters')])
def get_placeholder_form_instance(

View File

@@ -36,6 +36,9 @@ from app.main.forms import (
from app import user_api_client, current_service, organisations_client
dummy_bearer_token = 'bearer_token_set'
def get_inbound_api():
if current_service['inbound_api']:
return service_api_client.get_service_inbound_api(
@@ -441,17 +444,21 @@ def service_set_inbound_api(service_id):
abort(403)
inbound_api = get_inbound_api()
form = ServiceInboundApiForm(url=inbound_api.get('url') if inbound_api else '')
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:
service_api_client.update_service_inbound_api(
service_id,
url=form.url.data,
bearer_token=form.bearer_token.data,
user_id=current_user.id,
inbound_api_id=inbound_api.get('id')
)
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,

View File

@@ -277,9 +277,10 @@ class ServiceAPIClient(NotifyAdminAPIClient):
def update_service_inbound_api(self, service_id, url, bearer_token, user_id, inbound_api_id):
data = {
"url": url,
"bearer_token": bearer_token,
"updated_by_id": user_id
}
if bearer_token:
data['bearer_token'] = bearer_token
return self.post("/service/{}/inbound-api/{}".format(service_id, inbound_api_id), data)
def get_service_inbound_api(self, service_id, inbound_sms_api_id):

View File

@@ -7,6 +7,7 @@ from bs4 import BeautifulSoup
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.app.test_utils import normalize_spaces
@@ -1112,21 +1113,39 @@ def test_set_new_inbound_api_and_valid_bearer_token_calls_create_inbound_api_end
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
fake_uuid,
inbound_api_data
):
service_one['permissions'] = ['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': "https://test.url.com/"}})
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)
inbound_api_data = {'url': "https://test.url.com/", 'bearer_token': '1234567890', 'inbound_api_id': fake_uuid}
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',
@@ -1138,13 +1157,41 @@ def test_update_inbound_api_and_valid_bearer_token_calls_update_inbound_api_endp
assert response.location == url_for('main.service_settings', service_id=service_one['id'], _external=True)
assert mocked_post_fn.called
del inbound_api_data['inbound_api_id']
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_post_fn.called is False
def test_archive_service_after_confirm(
logged_in_platform_admin_client,
service_one,