Make a service model and use for permissions

Having the service floating about as JSON is a bit flakey. Could easily
introduce a mistake where you mistype the name of a key and silently
get `None`.

Also means doing awkward things like `if 'permission' in
current_service['permissions']`, whereas for users we can do the
much cleaner `user.has_permission()`.

So this commit:
- introduces a model
- adds a `.has_permission` method similar to the one we have for users
This commit is contained in:
Chris Hill-Scott
2018-07-20 08:43:02 +01:00
parent 3056731cbb
commit 036923c382
12 changed files with 47 additions and 31 deletions

View File

@@ -40,6 +40,7 @@ from werkzeug.local import LocalProxy
from app import proxy_fix from app import proxy_fix
from app.config import configs from app.config import configs
from app.asset_fingerprinter import AssetFingerprinter from app.asset_fingerprinter import AssetFingerprinter
from app.notify_client.models import Service
from app.navigation import ( from app.navigation import (
CaseworkNavigation, CaseworkNavigation,
HeaderNavigation, HeaderNavigation,
@@ -94,8 +95,13 @@ billing_api_client = BillingAPIClient()
complaint_api_client = ComplaintApiClient() complaint_api_client = ComplaintApiClient()
platform_stats_api_client = PlatformStatsAPIClient() platform_stats_api_client = PlatformStatsAPIClient()
# The current service attached to the request stack. # The current service attached to the request stack.
current_service = LocalProxy(partial(_lookup_req_object, 'service')) def _get_current_service():
return Service(_lookup_req_object('service'))
current_service = LocalProxy(_get_current_service)
# The current organisation attached to the request stack. # The current organisation attached to the request stack.
current_organisation = LocalProxy(partial(_lookup_req_object, 'organisation')) current_organisation = LocalProxy(partial(_lookup_req_object, 'organisation'))

View File

@@ -37,7 +37,7 @@ dummy_bearer_token = 'bearer_token_set'
@user_has_permissions('manage_api_keys') @user_has_permissions('manage_api_keys')
def api_integration(service_id): def api_integration(service_id):
callbacks_link = ( callbacks_link = (
'.api_callbacks' if 'inbound_sms' in current_service['permissions'] '.api_callbacks' if current_service.has_permission('inbound_sms')
else '.delivery_status_callback' else '.delivery_status_callback'
) )
return render_template( return render_template(
@@ -104,7 +104,7 @@ def create_api_key(service_id):
'Not available because your service is in ' 'Not available because your service is in '
'<a href="{}#trial-mode">trial mode</a>'.format(url_for(".using_notify")) '<a href="{}#trial-mode">trial mode</a>'.format(url_for(".using_notify"))
) )
if 'letter' in current_service['permissions']: if current_service.has_permission('letter'):
option_hints[KEY_TYPE_TEAM] = 'Cant be used to send letters' option_hints[KEY_TYPE_TEAM] = 'Cant be used to send letters'
if form.validate_on_submit(): if form.validate_on_submit():
if form.key_type.data in disabled_options: if form.key_type.data in disabled_options:
@@ -198,7 +198,7 @@ def get_delivery_status_callback_details():
def delivery_status_callback(service_id): def delivery_status_callback(service_id):
delivery_status_callback = get_delivery_status_callback_details() delivery_status_callback = get_delivery_status_callback_details()
back_link = ( back_link = (
'.api_callbacks' if 'inbound_sms' in current_service['permissions'] '.api_callbacks' if current_service.has_permission('inbound_sms')
else '.api_integration' else '.api_integration'
) )

View File

@@ -152,7 +152,7 @@ def usage(service_id):
yearly_usage = billing_api_client.get_service_usage_ft(service_id, year) yearly_usage = billing_api_client.get_service_usage_ft(service_id, year)
usage_template = 'views/usage.html' usage_template = 'views/usage.html'
if 'letter' in current_service['permissions']: if current_service.has_permission('letter'):
usage_template = 'views/usage-with-letters.html' usage_template = 'views/usage-with-letters.html'
return render_template( return render_template(
usage_template, usage_template,
@@ -293,7 +293,7 @@ def get_dashboard_partials(service_id):
stats = service_api_client.get_service_statistics(service_id, today_only=False) stats = service_api_client.get_service_statistics(service_id, today_only=False)
column_width, max_notifiction_count = get_column_properties( column_width, max_notifiction_count = get_column_properties(
number_of_columns=( number_of_columns=(
3 if 'letter' in current_service['permissions'] else 2 3 if current_service.has_permission('letter') else 2
) )
) )
dashboard_totals = get_dashboard_totals(stats), dashboard_totals = get_dashboard_totals(stats),
@@ -313,7 +313,7 @@ def get_dashboard_partials(service_id):
'views/dashboard/_inbox.html', 'views/dashboard/_inbox.html',
inbound_sms_summary=( inbound_sms_summary=(
service_api_client.get_inbound_sms_summary(service_id) service_api_client.get_inbound_sms_summary(service_id)
if 'inbound_sms' in current_service['permissions'] else None if current_service.has_permission('inbound_sms') else None
), ),
), ),
'totals': render_template( 'totals': render_template(

View File

@@ -48,14 +48,14 @@ def manage_users(service_id):
@user_has_permissions('manage_service') @user_has_permissions('manage_service')
def invite_user(service_id): def invite_user(service_id):
if 'caseworking' in current_service['permissions']: if current_service.has_permission('caseworking'):
form = CaseworkingInviteUserForm form = CaseworkingInviteUserForm
else: else:
form = AdminInviteUserForm form = AdminInviteUserForm
form = form(invalid_email_address=current_user.email_address) form = form(invalid_email_address=current_user.email_address)
service_has_email_auth = 'email_auth' in current_service['permissions'] service_has_email_auth = current_service.has_permission('email_auth')
if not service_has_email_auth: if not service_has_email_auth:
form.login_authentication.data = 'sms_auth' form.login_authentication.data = 'sms_auth'
@@ -83,13 +83,13 @@ def invite_user(service_id):
@login_required @login_required
@user_has_permissions('manage_service') @user_has_permissions('manage_service')
def edit_user_permissions(service_id, user_id): def edit_user_permissions(service_id, user_id):
service_has_email_auth = 'email_auth' in current_service['permissions'] service_has_email_auth = current_service.has_permission('email_auth')
# TODO we should probably using the service id here in the get user # TODO we should probably using the service id here in the get user
# call as well. eg. /user/<user_id>?&service=service_id # call as well. eg. /user/<user_id>?&service=service_id
user = user_api_client.get_user(user_id) user = user_api_client.get_user(user_id)
user_has_no_mobile_number = user.mobile_number is None user_has_no_mobile_number = user.mobile_number is None
if 'caseworking' in current_service['permissions']: if current_service.has_permission('caseworking'):
form = partial( form = partial(
CaseworkingPermissionsForm, CaseworkingPermissionsForm,
user_type='admin' if user.has_permission_for_service(service_id, 'view_activity') else 'caseworker', user_type='admin' if user.has_permission_for_service(service_id, 'view_activity') else 'caseworker',

View File

@@ -91,7 +91,7 @@ def view_notification(service_id, notification_id):
help=get_help_argument(), help=get_help_argument(),
estimated_letter_delivery_date=get_letter_timings(notification['created_at']).earliest_delivery, estimated_letter_delivery_date=get_letter_timings(notification['created_at']).earliest_delivery,
notification_id=notification['id'], notification_id=notification['id'],
can_receive_inbound=('inbound_sms' in current_service['permissions']), can_receive_inbound=(current_service.has_permission('inbound_sms')),
is_precompiled_letter=notification['template']['is_precompiled_letter'] is_precompiled_letter=notification['template']['is_precompiled_letter']
) )

View File

@@ -401,7 +401,7 @@ def send_test_step(service_id, template_id, step_index):
dict_to_populate_from=get_normalised_placeholders_from_session(), dict_to_populate_from=get_normalised_placeholders_from_session(),
template_type=template.template_type, template_type=template.template_type,
optional_placeholder=optional_placeholder, optional_placeholder=optional_placeholder,
allow_international_phone_numbers='international_sms' in current_service['permissions'], allow_international_phone_numbers=current_service.has_permission('international_sms'),
) )
if form.validate_on_submit(): if form.validate_on_submit():
@@ -551,7 +551,7 @@ def _check_messages(service_id, template_id, upload_id, preview_row, letters_as_
[user.name, user.mobile_number, user.email_address] for user in users [user.name, user.mobile_number, user.email_address] for user in users
) if current_service['restricted'] else None, ) if current_service['restricted'] else None,
remaining_messages=remaining_messages, remaining_messages=remaining_messages,
international_sms='international_sms' in current_service['permissions'], international_sms=current_service.has_permission('international_sms'),
) )
if request.args.get('from_test'): if request.args.get('from_test'):

View File

@@ -93,7 +93,7 @@ def service_settings(service_id):
letter_branding=letter_branding_organisations.get( letter_branding=letter_branding_organisations.get(
current_service.get('dvla_organisation', '001') current_service.get('dvla_organisation', '001')
), ),
can_receive_inbound=('inbound_sms' in current_service['permissions']), can_receive_inbound=(current_service.has_permission('inbound_sms')),
inbound_number=disp_inbound_number, inbound_number=disp_inbound_number,
default_reply_to_email_address=default_reply_to_email_address, default_reply_to_email_address=default_reply_to_email_address,
reply_to_email_address_count=reply_to_email_address_count, reply_to_email_address_count=reply_to_email_address_count,
@@ -289,9 +289,7 @@ def force_service_permission(service_id, permission, on=False, sms_sender=None):
def update_service_permissions(service_id, permissions, sms_sender=None): def update_service_permissions(service_id, permissions, sms_sender=None):
current_service['permissions'] = list(permissions) data = {'permissions': list(permissions)}
data = {'permissions': current_service['permissions']}
if sms_sender: if sms_sender:
data['sms_sender'] = sms_sender data['sms_sender'] = sms_sender
@@ -339,7 +337,7 @@ def service_switch_can_upload_document(service_id):
# If turning the permission off, or turning it on and the service already has a contact_link, # If turning the permission off, or turning it on and the service already has a contact_link,
# don't show the form to add the link # don't show the form to add the link
if 'upload_document' in current_service['permissions'] or current_service.get('contact_link'): if current_service.has_permission('upload_document') or current_service.get('contact_link'):
switch_service_permissions(service_id, 'upload_document') switch_service_permissions(service_id, 'upload_document')
return redirect(url_for('.service_settings', service_id=service_id)) return redirect(url_for('.service_settings', service_id=service_id))
@@ -569,7 +567,7 @@ def service_set_sms_prefix(service_id):
@user_has_permissions('manage_service') @user_has_permissions('manage_service')
def service_set_international_sms(service_id): def service_set_international_sms(service_id):
form = InternationalSMSForm( form = InternationalSMSForm(
enabled='on' if 'international_sms' in current_service['permissions'] else 'off' enabled='on' if current_service.has_permission('international_sms') else 'off'
) )
if form.validate_on_submit(): if form.validate_on_submit():
force_service_permission( force_service_permission(
@@ -602,7 +600,7 @@ def service_set_inbound_sms(service_id):
@user_has_permissions('manage_service') @user_has_permissions('manage_service')
def service_set_letters(service_id): def service_set_letters(service_id):
form = ServiceSwitchLettersForm( form = ServiceSwitchLettersForm(
enabled='on' if 'letter' in current_service['permissions'] else 'off' enabled='on' if current_service.has_permission('letter') else 'off'
) )
if form.validate_on_submit(): if form.validate_on_submit():
force_service_permission( force_service_permission(
@@ -640,7 +638,7 @@ def service_set_basic_view(service_id):
abort(403) abort(403)
form = ServiceBasicViewForm( form = ServiceBasicViewForm(
enabled='caseworking' in current_service['permissions'] enabled=current_service.has_permission('caseworking')
) )
if form.validate_on_submit(): if form.validate_on_submit():
force_service_permission( force_service_permission(

View File

@@ -108,7 +108,7 @@ def choose_template(service_id, template_type='all'):
templates = service_api_client.get_service_templates(service_id)['data'] templates = service_api_client.get_service_templates(service_id)['data']
letters_available = ( letters_available = (
'letter' in current_service['permissions'] and current_service.has_permission('letter') and
current_user.has_permissions('view_activity') current_user.has_permissions('view_activity')
) )
@@ -207,9 +207,9 @@ def view_template_version_preview(service_id, template_id, version, filetype):
def add_template_by_type(service_id): def add_template_by_type(service_id):
form = ChooseTemplateType( form = ChooseTemplateType(
include_letters='letter' in current_service['permissions'], include_letters=current_service.has_permission('letter'),
include_copy=any(( include_copy=any((
service_api_client.count_service_templates(service_id), service_api_client.count_service_templates(service_id) > 0,
len(user_api_client.get_service_ids_for_user(current_user)) > 1, len(user_api_client.get_service_ids_for_user(current_user)) > 1,
)), )),
) )
@@ -672,5 +672,5 @@ def get_human_readable_delta(from_time, until_time):
def should_show_template(template_type): def should_show_template(template_type):
return ( return (
template_type != 'letter' or template_type != 'letter' or
'letter' in current_service['permissions'] current_service.has_permission('letter')
) )

View File

@@ -261,3 +261,13 @@ class AnonymousUser(AnonymousUserMixin):
# set the anonymous user so that if a new browser hits us we don't error http://stackoverflow.com/a/19275188 # set the anonymous user so that if a new browser hits us we don't error http://stackoverflow.com/a/19275188
def logged_in_elsewhere(self): def logged_in_elsewhere(self):
return False return False
class Service(dict):
def __init__(self, _dict):
# in the case of a bad request current service may be `None`
super().__init__(_dict or {})
def has_permission(self, permission):
return permission in self['permissions']

View File

@@ -27,7 +27,7 @@
smaller=smaller_font_size smaller=smaller_font_size
) }} ) }}
</div> </div>
{% if 'letter' in current_service['permissions'] %} {% if current_service.has_permission('letter') %}
<div id="total-letters" class="{{column_width}}"> <div id="total-letters" class="{{column_width}}">
{{ big_number_with_status( {{ big_number_with_status(
statistics['letter']['requested'], statistics['letter']['requested'],

View File

@@ -92,7 +92,7 @@
'Basic view' 'Basic view'
) }} ) }}
{% endif %} {% endif %}
{% if 'email_auth' in current_service['permissions'] %} {% if current_service.has_permission('email_auth') %}
<div class="tick-cross-list-hint"> <div class="tick-cross-list-hint">
{% if user.auth_type == 'sms_auth' %} {% if user.auth_type == 'sms_auth' %}
Signs in with a text message code Signs in with a text message code

View File

@@ -2039,8 +2039,10 @@ def test_service_switch_can_upload_document_changes_the_permission_if_not_adding
follow_redirects=True follow_redirects=True
) )
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser') page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
mock_update_service.assert_called_once_with(
assert service_one['permissions'] == end_permissions SERVICE_ONE_ID,
permissions=end_permissions,
)
assert page.h1.text.strip() == 'Settings' assert page.h1.text.strip() == 'Settings'
@@ -2080,7 +2082,7 @@ def test_service_switch_can_upload_document_lets_contact_link_be_added_and_switc
) )
page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser') page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
assert 'upload_document' in service_one['permissions'] assert 'upload_document' in mock_update_service.call_args[1]['permissions']
assert page.h1.text.strip() == 'Settings' assert page.h1.text.strip() == 'Settings'