mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-08-26 09:13:58 -04:00
Merge pull request #2896 from alphagov/remove-domains-yml
Use organisations from database rather than YAML file
This commit is contained in:
4429
app/domains.yml
4429
app/domains.yml
File diff suppressed because it is too large
Load Diff
@@ -36,10 +36,8 @@ from wtforms.widgets import CheckboxInput, ListWidget
|
||||
|
||||
from app.main.validators import (
|
||||
Blacklist,
|
||||
CanonicalGovernmentDomain,
|
||||
CsvFileValidator,
|
||||
DoesNotStartWithDoubleZero,
|
||||
KnownGovernmentDomain,
|
||||
LettersNumbersAndFullStopsOnly,
|
||||
NoCommasInPlaceHolders,
|
||||
OnlyGSMCharacters,
|
||||
@@ -47,7 +45,7 @@ from app.main.validators import (
|
||||
ValidGovEmail,
|
||||
)
|
||||
from app.models.user import permissions, roles
|
||||
from app.utils import AgreementInfo, guess_name_from_email_address
|
||||
from app.utils import guess_name_from_email_address
|
||||
|
||||
|
||||
def get_time_value_and_label(future_time):
|
||||
@@ -990,23 +988,9 @@ class PreviewBranding(StripWhitespaceForm):
|
||||
branding_style = HiddenFieldWithNoneOption('branding_style')
|
||||
|
||||
|
||||
class GovernmentDomainField(StringField):
|
||||
validators = [
|
||||
KnownGovernmentDomain(),
|
||||
CanonicalGovernmentDomain(),
|
||||
]
|
||||
|
||||
def post_validate(self, form, validation_stopped):
|
||||
if self.data == '':
|
||||
self.data = None
|
||||
if self.data and not self.errors:
|
||||
self.data = AgreementInfo(self.data).canonical_domain
|
||||
|
||||
|
||||
class ServiceUpdateEmailBranding(StripWhitespaceForm):
|
||||
name = StringField('Name of brand')
|
||||
text = StringField('Text')
|
||||
domain = GovernmentDomainField('Domain')
|
||||
colour = StringField(
|
||||
'Colour',
|
||||
validators=[
|
||||
@@ -1041,7 +1025,6 @@ class SVGFileUpload(StripWhitespaceForm):
|
||||
|
||||
class ServiceLetterBrandingDetails(StripWhitespaceForm):
|
||||
name = StringField('Name of brand', validators=[DataRequired()])
|
||||
domain = GovernmentDomainField('Domain')
|
||||
|
||||
|
||||
class PDFUploadForm(StripWhitespaceForm):
|
||||
|
||||
@@ -11,7 +11,7 @@ from wtforms.validators import Email
|
||||
|
||||
from app import formatted_list
|
||||
from app.main._blacklisted_passwords import blacklisted_passwords
|
||||
from app.utils import AgreementInfo, Spreadsheet, is_gov_user
|
||||
from app.utils import Spreadsheet, is_gov_user
|
||||
|
||||
|
||||
class Blacklist:
|
||||
@@ -111,34 +111,3 @@ class DoesNotStartWithDoubleZero:
|
||||
def __call__(self, form, field):
|
||||
if field.data and field.data.startswith("00"):
|
||||
raise ValidationError(self.message)
|
||||
|
||||
|
||||
class KnownGovernmentDomain:
|
||||
|
||||
message = 'Not a known government domain (you might need to update domains.yml)'
|
||||
|
||||
def __call__(self, form, field):
|
||||
if field.data and AgreementInfo(field.data).owner is None:
|
||||
raise ValidationError(self.message)
|
||||
|
||||
|
||||
class CanonicalGovernmentDomain:
|
||||
|
||||
message = 'Not {} domain (use {} if appropriate)'
|
||||
|
||||
def __call__(self, form, field):
|
||||
|
||||
if not field.data:
|
||||
return
|
||||
|
||||
domain = AgreementInfo(field.data)
|
||||
|
||||
if not domain.is_canonical:
|
||||
raise ValidationError(
|
||||
self.message.format('a canonical', domain.canonical_domain)
|
||||
)
|
||||
|
||||
if field.data != domain.canonical_domain:
|
||||
raise ValidationError(
|
||||
self.message.format('an organisation-level', domain.canonical_domain)
|
||||
)
|
||||
|
||||
@@ -2,17 +2,15 @@ from flask import current_app, redirect, render_template, session, url_for
|
||||
from flask_login import login_required
|
||||
from notifications_python_client.errors import HTTPError
|
||||
|
||||
from app import billing_api_client, email_branding_client, service_api_client
|
||||
from app import billing_api_client, service_api_client
|
||||
from app.main import main
|
||||
from app.main.forms import CreateServiceForm
|
||||
from app.utils import AgreementInfo, email_safe, user_is_gov_user
|
||||
from app.utils import email_safe, user_is_gov_user
|
||||
|
||||
|
||||
def _create_service(service_name, organisation_type, email_from, form):
|
||||
free_sms_fragment_limit = current_app.config['DEFAULT_FREE_SMS_FRAGMENT_LIMITS'].get(organisation_type)
|
||||
|
||||
domain = 'nhs.uk' if organisation_type == 'nhs' else AgreementInfo.from_current_user().canonical_domain
|
||||
email_branding = email_branding_client.get_email_branding_id_for_domain(domain)
|
||||
try:
|
||||
service_id = service_api_client.create_service(
|
||||
service_name=service_name,
|
||||
@@ -21,13 +19,9 @@ def _create_service(service_name, organisation_type, email_from, form):
|
||||
restricted=True,
|
||||
user_id=session['user_id'],
|
||||
email_from=email_from,
|
||||
service_domain=domain
|
||||
)
|
||||
session['service_id'] = service_id
|
||||
|
||||
if email_branding:
|
||||
service_api_client.update_service(service_id, email_branding=email_branding)
|
||||
|
||||
billing_api_client.create_or_update_free_sms_fragment_limit(service_id, free_sms_fragment_limit)
|
||||
|
||||
return service_id, None
|
||||
|
||||
@@ -1,21 +1,17 @@
|
||||
from flask import abort, render_template, request, send_file, url_for
|
||||
from flask_login import login_required
|
||||
from flask_login import current_user, login_required
|
||||
|
||||
from app.main import main
|
||||
from app.main.views.sub_navigation_dictionaries import features_nav
|
||||
from app.s3_client.s3_mou_client import get_mou
|
||||
from app.utils import AgreementInfo
|
||||
|
||||
|
||||
@main.route('/agreement')
|
||||
@login_required
|
||||
def agreement():
|
||||
|
||||
agreement_info = AgreementInfo.from_current_user()
|
||||
|
||||
return render_template(
|
||||
'views/{}.html'.format(agreement_info.as_jinja_template),
|
||||
owner=agreement_info.owner,
|
||||
'views/{}.html'.format(current_user.default_organisation.as_jinja_template),
|
||||
owner=current_user.default_organisation.name,
|
||||
navigation_links=features_nav(),
|
||||
)
|
||||
|
||||
@@ -24,7 +20,7 @@ def agreement():
|
||||
@login_required
|
||||
def download_agreement():
|
||||
return send_file(**get_mou(
|
||||
AgreementInfo.from_current_user().crown_status_or_404
|
||||
current_user.default_organisation.crown_status_or_404
|
||||
))
|
||||
|
||||
|
||||
@@ -40,10 +36,8 @@ def public_agreement(variant):
|
||||
organisation_is_crown=(variant == 'crown')
|
||||
))
|
||||
|
||||
agreement_info = AgreementInfo.from_current_user()
|
||||
|
||||
return render_template(
|
||||
'views/agreement-public.html',
|
||||
owner=agreement_info.owner,
|
||||
owner=current_user.default_organisation.name,
|
||||
download_link=url_for('.public_download_agreement', variant=variant),
|
||||
)
|
||||
|
||||
@@ -39,7 +39,6 @@ def update_email_branding(branding_id, logo=None):
|
||||
name=email_branding['name'],
|
||||
text=email_branding['text'],
|
||||
colour=email_branding['colour'],
|
||||
domain=email_branding['domain'],
|
||||
brand_type=email_branding['brand_type']
|
||||
)
|
||||
|
||||
@@ -67,7 +66,6 @@ def update_email_branding(branding_id, logo=None):
|
||||
name=form.name.data,
|
||||
text=form.text.data,
|
||||
colour=form.colour.data,
|
||||
domain=form.domain.data,
|
||||
brand_type=form.brand_type.data,
|
||||
)
|
||||
|
||||
@@ -115,7 +113,6 @@ def create_email_branding(logo=None):
|
||||
name=form.name.data,
|
||||
text=form.text.data,
|
||||
colour=form.colour.data,
|
||||
domain=form.domain.data,
|
||||
brand_type=form.brand_type.data,
|
||||
)
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@ from app import email_branding_client, letter_branding_client, status_api_client
|
||||
from app.main import main
|
||||
from app.main.forms import FieldWithNoneOption, SearchByNameForm
|
||||
from app.main.views.sub_navigation_dictionaries import features_nav
|
||||
from app.utils import AgreementInfo, get_logo_cdn_domain
|
||||
from app.utils import get_logo_cdn_domain
|
||||
|
||||
|
||||
@main.route('/')
|
||||
@@ -80,7 +80,6 @@ def pricing():
|
||||
for cc, country in INTERNATIONAL_BILLING_RATES.items()
|
||||
], key=lambda x: x[0]),
|
||||
search_form=SearchByNameForm(),
|
||||
agreement_info=AgreementInfo.from_current_user(),
|
||||
)
|
||||
|
||||
|
||||
@@ -254,7 +253,6 @@ def terms():
|
||||
return render_template(
|
||||
'views/terms-of-use.html',
|
||||
navigation_links=features_nav(),
|
||||
agreement_info=AgreementInfo.from_current_user(),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -56,7 +56,6 @@ def update_letter_branding(branding_id, logo=None):
|
||||
file_upload_form = SVGFileUpload()
|
||||
letter_branding_details_form = ServiceLetterBrandingDetails(
|
||||
name=letter_branding['name'],
|
||||
domain=letter_branding['domain']
|
||||
)
|
||||
|
||||
file_upload_form_submitted = file_upload_form.file.data
|
||||
@@ -87,7 +86,6 @@ def update_letter_branding(branding_id, logo=None):
|
||||
branding_id=branding_id,
|
||||
filename=db_filename,
|
||||
name=letter_branding_details_form.name.data,
|
||||
domain=letter_branding_details_form.domain.data
|
||||
)
|
||||
|
||||
return redirect(url_for('main.letter_branding'))
|
||||
@@ -98,7 +96,6 @@ def update_letter_branding(branding_id, logo=None):
|
||||
branding_id=branding_id,
|
||||
filename=db_filename,
|
||||
name=letter_branding_details_form.name.data,
|
||||
domain=letter_branding_details_form.domain.data
|
||||
)
|
||||
|
||||
upload_letter_logos(logo, db_filename, png_file, session['user_id'])
|
||||
@@ -106,9 +103,7 @@ def update_letter_branding(branding_id, logo=None):
|
||||
return redirect(url_for('main.letter_branding'))
|
||||
|
||||
except HTTPError as e:
|
||||
if 'domain' in e.message:
|
||||
letter_branding_details_form.domain.errors.append(e.message['domain'][0])
|
||||
elif 'name' in e.message:
|
||||
if 'name' in e.message:
|
||||
letter_branding_details_form.name.errors.append(e.message['name'][0])
|
||||
else:
|
||||
raise e
|
||||
@@ -118,7 +113,6 @@ def update_letter_branding(branding_id, logo=None):
|
||||
branding_id=branding_id,
|
||||
filename=letter_branding['filename'],
|
||||
name=letter_branding['name'],
|
||||
domain=letter_branding['domain']
|
||||
)
|
||||
file_upload_form.file.errors = ['Error saving uploaded file - try uploading again']
|
||||
|
||||
@@ -165,7 +159,6 @@ def create_letter_branding(logo=None):
|
||||
letter_branding_client.create_letter_branding(
|
||||
filename=db_filename,
|
||||
name=letter_branding_details_form.name.data,
|
||||
domain=letter_branding_details_form.domain.data,
|
||||
)
|
||||
|
||||
upload_letter_logos(logo, db_filename, png_file, session['user_id'])
|
||||
@@ -173,9 +166,7 @@ def create_letter_branding(logo=None):
|
||||
return redirect(url_for('main.letter_branding'))
|
||||
|
||||
except HTTPError as e:
|
||||
if 'domain' in e.message:
|
||||
letter_branding_details_form.domain.errors.append(e.message['domain'][0])
|
||||
elif 'name' in e.message:
|
||||
if 'name' in e.message:
|
||||
letter_branding_details_form.name.errors.append(e.message['name'][0])
|
||||
else:
|
||||
raise e
|
||||
|
||||
@@ -54,7 +54,6 @@ from app.main.forms import (
|
||||
branding_options_dict,
|
||||
)
|
||||
from app.utils import (
|
||||
AgreementInfo,
|
||||
email_safe,
|
||||
user_has_permissions,
|
||||
user_is_gov_user,
|
||||
@@ -180,7 +179,7 @@ def estimate_usage(service_id):
|
||||
@user_has_permissions('manage_service')
|
||||
def request_to_go_live(service_id):
|
||||
|
||||
agreement_signed = AgreementInfo.from_current_user().agreement_signed
|
||||
agreement_signed = current_service.organisation.agreement_signed
|
||||
|
||||
return render_template(
|
||||
'views/service-settings/request-to-go-live.html',
|
||||
@@ -226,7 +225,7 @@ def submit_request_to_go_live(service_id):
|
||||
service_name=current_service.name,
|
||||
service_dashboard=url_for('main.service_dashboard', service_id=current_service.id, _external=True),
|
||||
organisation_type=str(current_service.organisation_type).title(),
|
||||
agreement=AgreementInfo.from_current_user().as_human_readable,
|
||||
agreement=current_service.organisation.as_human_readable(current_user.email_domain),
|
||||
checklist=current_service.go_live_checklist_completed_as_yes_no,
|
||||
volume_email=print_if_number(current_service.volume_email),
|
||||
volume_email_formatted=format_if_number(current_service.volume_email),
|
||||
@@ -237,7 +236,7 @@ def submit_request_to_go_live(service_id):
|
||||
research_consent='Yes' if current_service.consent_to_research else 'No',
|
||||
existing_live='Yes' if user_api_client.user_has_live_services(current_user) else 'No',
|
||||
service_id=current_service.id,
|
||||
organisation=AgreementInfo.from_current_user().owner,
|
||||
organisation=current_service.organisation.name,
|
||||
user_name=current_user.name,
|
||||
user_email=current_user.email_address,
|
||||
date=datetime.now(tz=pytz.timezone('Europe/London')).strftime('%d/%m/%Y'),
|
||||
@@ -245,7 +244,7 @@ def submit_request_to_go_live(service_id):
|
||||
ticket_type=zendesk_client.TYPE_QUESTION,
|
||||
user_email=current_user.email_address,
|
||||
user_name=current_user.name,
|
||||
tags=get_request_to_go_live_tags(current_service, current_user),
|
||||
tags=current_service.request_to_go_live_tags,
|
||||
)
|
||||
|
||||
flash('Thanks for your request to go live. We’ll get back to you within one working day.', 'default')
|
||||
@@ -980,7 +979,7 @@ def branding_request(service_id):
|
||||
'\nCurrent branding: {current_branding}'
|
||||
'\nBranding requested: {branding_requested}'
|
||||
).format(
|
||||
organisation=AgreementInfo.from_current_user().as_info_for_branding_request,
|
||||
organisation=current_service.organisation.as_info_for_branding_request(current_user.email_domain),
|
||||
service_name=current_service.name,
|
||||
dashboard_url=url_for('main.service_dashboard', service_id=current_service.id, _external=True),
|
||||
current_branding=current_service.email_branding_name,
|
||||
@@ -1068,37 +1067,6 @@ def check_contact_details_type(contact_details):
|
||||
return 'phone_number'
|
||||
|
||||
|
||||
def get_request_to_go_live_tags(service, user):
|
||||
return list(_get_request_to_go_live_tags(
|
||||
service,
|
||||
AgreementInfo.from_user(user).agreement_signed,
|
||||
))
|
||||
|
||||
|
||||
def _get_request_to_go_live_tags(service, agreement_signed):
|
||||
|
||||
BASE = 'notify_request_to_go_live'
|
||||
|
||||
yield BASE
|
||||
|
||||
if service.go_live_checklist_completed and agreement_signed:
|
||||
yield BASE + '_complete'
|
||||
return
|
||||
|
||||
for test, tag in (
|
||||
(True, ''),
|
||||
(not service.volumes, '_volumes'),
|
||||
(not service.go_live_checklist_completed, '_checklist'),
|
||||
(not agreement_signed, '_mou'),
|
||||
(service.needs_to_add_email_reply_to_address, '_email_reply_to'),
|
||||
(not service.has_team_members, '_team_member'),
|
||||
(not service.has_templates, '_template_content'),
|
||||
(service.needs_to_change_sms_sender, '_sms_sender'),
|
||||
):
|
||||
if test:
|
||||
yield BASE + '_incomplete' + tag
|
||||
|
||||
|
||||
def print_if_number(value):
|
||||
return value if isinstance(value, int) else ''
|
||||
|
||||
|
||||
27
app/models/__init__.py
Normal file
27
app/models/__init__.py
Normal file
@@ -0,0 +1,27 @@
|
||||
from flask import abort
|
||||
|
||||
|
||||
class JSONModel():
|
||||
|
||||
ALLOWED_PROPERTIES = set()
|
||||
|
||||
def __init__(self, _dict):
|
||||
# in the case of a bad request _dict may be `None`
|
||||
self._dict = _dict or {}
|
||||
|
||||
def __bool__(self):
|
||||
return self._dict != {}
|
||||
|
||||
def __getattr__(self, attr):
|
||||
if attr in self.ALLOWED_PROPERTIES:
|
||||
return self._dict[attr]
|
||||
raise AttributeError('`{}` is not a {} attribute'.format(
|
||||
attr,
|
||||
self.__class__.__name__.lower(),
|
||||
))
|
||||
|
||||
def _get_by_id(self, things, id):
|
||||
try:
|
||||
return next(thing for thing in things if thing['id'] == str(id))
|
||||
except StopIteration:
|
||||
abort(404)
|
||||
134
app/models/organisation.py
Normal file
134
app/models/organisation.py
Normal file
@@ -0,0 +1,134 @@
|
||||
from flask import Markup, abort
|
||||
|
||||
from app.models import JSONModel
|
||||
|
||||
|
||||
class Organisation(JSONModel):
|
||||
|
||||
ALLOWED_PROPERTIES = {
|
||||
'id',
|
||||
'name',
|
||||
'active',
|
||||
'crown',
|
||||
'organisation_type',
|
||||
'letter_branding_id',
|
||||
'email_branding_id',
|
||||
'agreement_signed',
|
||||
'agreement_signed_at',
|
||||
'agreement_signed_by_id',
|
||||
'agreement_signed_version',
|
||||
'domains',
|
||||
}
|
||||
|
||||
def __init__(self, _dict):
|
||||
|
||||
super().__init__(_dict)
|
||||
|
||||
if self._dict == {}:
|
||||
self.name, self.crown, self.agreement_signed = None, None, None
|
||||
|
||||
def as_human_readable(self, fallback_domain):
|
||||
if 'dwp.' in ''.join(self.domains):
|
||||
return 'DWP - Requires OED approval'
|
||||
if self.agreement_signed:
|
||||
return 'Yes, on behalf of {}'.format(self.name)
|
||||
elif self.name:
|
||||
return '{} (organisation is {}, {})'.format(
|
||||
{
|
||||
False: 'No',
|
||||
None: 'Can’t tell',
|
||||
}.get(self.agreement_signed),
|
||||
self.name,
|
||||
{
|
||||
True: 'a crown body',
|
||||
False: 'a non-crown body',
|
||||
None: 'crown status unknown',
|
||||
}.get(self.crown),
|
||||
)
|
||||
else:
|
||||
return 'Can’t tell (domain is {})'.format(fallback_domain)
|
||||
|
||||
def as_info_for_branding_request(self, fallback_domain):
|
||||
return self.name or 'Can’t tell (domain is {})'.format(fallback_domain)
|
||||
|
||||
@property
|
||||
def as_jinja_template(self):
|
||||
if self.crown is None:
|
||||
return 'agreement-choose'
|
||||
if self.agreement_signed:
|
||||
return 'agreement-signed'
|
||||
return 'agreement'
|
||||
|
||||
def as_terms_of_use_paragraph(self, **kwargs):
|
||||
return Markup(self._as_terms_of_use_paragraph(**kwargs))
|
||||
|
||||
def _as_terms_of_use_paragraph(self, terms_link, download_link, support_link, signed_in):
|
||||
|
||||
if not signed_in:
|
||||
return ((
|
||||
'{} <a href="{}">Sign in</a> to download a copy '
|
||||
'or find out if one is already in place.'
|
||||
).format(self._acceptance_required, terms_link))
|
||||
|
||||
if self.agreement_signed is None:
|
||||
return ((
|
||||
'{} <a href="{}">Download the agreement</a> or '
|
||||
'<a href="{}">contact us</a> to find out if we already '
|
||||
'have one in place with your organisation.'
|
||||
).format(self._acceptance_required, download_link, support_link))
|
||||
|
||||
if self.agreement_signed is False:
|
||||
return ((
|
||||
'{} <a href="{}">Download a copy</a>.'
|
||||
).format(self._acceptance_required, download_link))
|
||||
|
||||
return (
|
||||
'Your organisation ({}) has already accepted the '
|
||||
'GOV.UK Notify data sharing and financial '
|
||||
'agreement.'.format(self.name)
|
||||
)
|
||||
|
||||
def as_pricing_paragraph(self, **kwargs):
|
||||
return Markup(self._as_pricing_paragraph(**kwargs))
|
||||
|
||||
def _as_pricing_paragraph(self, pricing_link, download_link, support_link, signed_in):
|
||||
|
||||
if not signed_in:
|
||||
return ((
|
||||
'<a href="{}">Sign in</a> to download a copy or find '
|
||||
'out if one is already in place with your organisation.'
|
||||
).format(pricing_link))
|
||||
|
||||
if self.agreement_signed is None:
|
||||
return ((
|
||||
'<a href="{}">Download the agreement</a> or '
|
||||
'<a href="{}">contact us</a> to find out if we already '
|
||||
'have one in place with your organisation.'
|
||||
).format(download_link, support_link))
|
||||
|
||||
return (
|
||||
'<a href="{}">Download the agreement</a> '
|
||||
'({} {}).'.format(
|
||||
download_link,
|
||||
self.name,
|
||||
{
|
||||
True: 'has already accepted it',
|
||||
False: 'hasn’t accepted it yet'
|
||||
}.get(self.agreement_signed)
|
||||
)
|
||||
)
|
||||
|
||||
@property
|
||||
def _acceptance_required(self):
|
||||
return (
|
||||
'Your organisation {} must also accept our data sharing '
|
||||
'and financial agreement.'.format(
|
||||
'({})'.format(self.name) if self.name else '',
|
||||
)
|
||||
)
|
||||
|
||||
@property
|
||||
def crown_status_or_404(self):
|
||||
if self.crown is None:
|
||||
abort(404)
|
||||
return self.crown
|
||||
@@ -2,6 +2,8 @@ from flask import abort, current_app
|
||||
from notifications_utils.field import Field
|
||||
from werkzeug.utils import cached_property
|
||||
|
||||
from app.models import JSONModel
|
||||
from app.models.organisation import Organisation
|
||||
from app.notify_client.api_key_api_client import api_key_api_client
|
||||
from app.notify_client.billing_api_client import billing_api_client
|
||||
from app.notify_client.email_branding_client import email_branding_client
|
||||
@@ -18,7 +20,7 @@ from app.notify_client.user_api_client import user_api_client
|
||||
from app.utils import get_default_sms_sender
|
||||
|
||||
|
||||
class Service():
|
||||
class Service(JSONModel):
|
||||
|
||||
ALLOWED_PROPERTIES = {
|
||||
'active',
|
||||
@@ -50,25 +52,12 @@ class Service():
|
||||
)
|
||||
|
||||
def __init__(self, _dict):
|
||||
# in the case of a bad request current service may be `None`
|
||||
self._dict = _dict or {}
|
||||
|
||||
super().__init__(_dict)
|
||||
|
||||
if 'permissions' not in self._dict:
|
||||
self.permissions = {'email', 'sms', 'letter'}
|
||||
|
||||
def __bool__(self):
|
||||
return self._dict != {}
|
||||
|
||||
def __getattr__(self, attr):
|
||||
if attr in self.ALLOWED_PROPERTIES:
|
||||
return self._dict[attr]
|
||||
raise AttributeError('`{}` is not a service attribute'.format(attr))
|
||||
|
||||
def _get_by_id(self, things, id):
|
||||
try:
|
||||
return next(thing for thing in things if thing['id'] == str(id))
|
||||
except StopIteration:
|
||||
abort(404)
|
||||
|
||||
def update(self, **kwargs):
|
||||
return service_api_client.update_service(self.id, **kwargs)
|
||||
|
||||
@@ -405,8 +394,10 @@ class Service():
|
||||
return None
|
||||
|
||||
@cached_property
|
||||
def organisation_name(self):
|
||||
return organisations_client.get_service_organisation(self.id).get('name', None)
|
||||
def organisation(self):
|
||||
return Organisation(
|
||||
organisations_client.get_service_organisation(self.id)
|
||||
)
|
||||
|
||||
@cached_property
|
||||
def inbound_number(self):
|
||||
@@ -555,3 +546,30 @@ class Service():
|
||||
|
||||
def get_api_key(self, id):
|
||||
return self._get_by_id(self.api_keys, id)
|
||||
|
||||
@property
|
||||
def request_to_go_live_tags(self):
|
||||
return list(self._get_request_to_go_live_tags())
|
||||
|
||||
def _get_request_to_go_live_tags(self):
|
||||
|
||||
BASE = 'notify_request_to_go_live'
|
||||
|
||||
yield BASE
|
||||
|
||||
if self.go_live_checklist_completed and self.organisation.agreement_signed:
|
||||
yield BASE + '_complete'
|
||||
return
|
||||
|
||||
for test, tag in (
|
||||
(True, ''),
|
||||
(not self.volumes, '_volumes'),
|
||||
(not self.go_live_checklist_completed, '_checklist'),
|
||||
(not self.organisation.agreement_signed, '_mou'),
|
||||
(self.needs_to_add_email_reply_to_address, '_email_reply_to'),
|
||||
(not self.has_team_members, '_team_member'),
|
||||
(not self.has_templates, '_template_content'),
|
||||
(self.needs_to_change_sms_sender, '_sms_sender'),
|
||||
):
|
||||
if test:
|
||||
yield BASE + '_incomplete' + tag
|
||||
|
||||
@@ -2,7 +2,10 @@ from itertools import chain
|
||||
|
||||
from flask import abort, request, session
|
||||
from flask_login import AnonymousUserMixin, UserMixin
|
||||
from werkzeug.utils import cached_property
|
||||
|
||||
from app.models.organisation import Organisation
|
||||
from app.notify_client.organisations_api_client import organisations_client
|
||||
from app.utils import is_gov_user
|
||||
|
||||
roles = {
|
||||
@@ -192,6 +195,16 @@ class User(UserMixin):
|
||||
def is_locked(self):
|
||||
return self.failed_login_count >= self.max_failed_login_count
|
||||
|
||||
@property
|
||||
def email_domain(self):
|
||||
return self.email_address.split('@')[-1]
|
||||
|
||||
@cached_property
|
||||
def default_organisation(self):
|
||||
return Organisation(
|
||||
organisations_client.get_organisation_by_domain(self.email_domain)
|
||||
)
|
||||
|
||||
def serialize(self):
|
||||
dct = {
|
||||
"id": self.id,
|
||||
@@ -322,3 +335,7 @@ class AnonymousUser(AnonymousUserMixin):
|
||||
# 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):
|
||||
return False
|
||||
|
||||
@property
|
||||
def default_organisation(self):
|
||||
return Organisation(None)
|
||||
|
||||
@@ -14,33 +14,25 @@ class EmailBrandingClient(NotifyAdminAPIClient):
|
||||
brandings.sort(key=lambda branding: branding[sort_key].lower())
|
||||
return brandings
|
||||
|
||||
def get_email_branding_id_for_domain(self, domain):
|
||||
for branding in self.get_all_email_branding():
|
||||
if domain and branding.get('domain') == domain:
|
||||
return branding['id']
|
||||
return None
|
||||
|
||||
@cache.delete('email_branding')
|
||||
def create_email_branding(self, logo, name, text, colour, domain, brand_type):
|
||||
def create_email_branding(self, logo, name, text, colour, brand_type):
|
||||
data = {
|
||||
"logo": logo,
|
||||
"name": name,
|
||||
"text": text,
|
||||
"colour": colour,
|
||||
"domain": domain,
|
||||
"brand_type": brand_type
|
||||
}
|
||||
return self.post(url="/email-branding", data=data)
|
||||
|
||||
@cache.delete('email_branding')
|
||||
@cache.delete('email_branding-{branding_id}')
|
||||
def update_email_branding(self, branding_id, logo, name, text, colour, domain, brand_type):
|
||||
def update_email_branding(self, branding_id, logo, name, text, colour, brand_type):
|
||||
data = {
|
||||
"logo": logo,
|
||||
"name": name,
|
||||
"text": text,
|
||||
"colour": colour,
|
||||
"domain": domain,
|
||||
"brand_type": brand_type
|
||||
}
|
||||
return self.post(url="/email-branding/{}".format(branding_id), data=data)
|
||||
|
||||
@@ -12,22 +12,19 @@ class LetterBrandingClient(NotifyAdminAPIClient):
|
||||
return self.get(url='/letter-branding')
|
||||
|
||||
@cache.delete('letter_branding')
|
||||
def create_letter_branding(self, filename, name, domain):
|
||||
def create_letter_branding(self, filename, name):
|
||||
data = {
|
||||
"filename": filename,
|
||||
"name": name,
|
||||
"domain": domain,
|
||||
}
|
||||
return self.post(url="/letter-branding", data=data)
|
||||
|
||||
@cache.delete('letter_branding')
|
||||
@cache.delete('letter_branding-{branding_id}')
|
||||
def update_letter_branding(self, branding_id, filename, name, domain):
|
||||
def update_letter_branding(self, branding_id, filename, name):
|
||||
data = {
|
||||
"filename": filename,
|
||||
"name": name,
|
||||
"domain": domain,
|
||||
|
||||
}
|
||||
return self.post(url="/letter-branding/{}".format(branding_id), data=data)
|
||||
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from notifications_python_client.errors import HTTPError
|
||||
|
||||
from app.notify_client import NotifyAdminAPIClient, _attach_current_user, cache
|
||||
|
||||
|
||||
@@ -9,6 +11,16 @@ class OrganisationsClient(NotifyAdminAPIClient):
|
||||
def get_organisation(self, org_id):
|
||||
return self.get(url='/organisations/{}'.format(org_id))
|
||||
|
||||
def get_organisation_by_domain(self, domain):
|
||||
try:
|
||||
return self.get(
|
||||
url='/organisations/by-domain?domain={}'.format(domain),
|
||||
)
|
||||
except HTTPError as error:
|
||||
if error.status_code == 404:
|
||||
return None
|
||||
raise error
|
||||
|
||||
def create_organisation(self, name):
|
||||
data = {
|
||||
"name": name
|
||||
|
||||
@@ -28,7 +28,6 @@
|
||||
<div style='margin-top:15px;'>{{textbox(form.name)}}</div>
|
||||
<div style='margin-top:15px;'>{{textbox(form.text)}}</div>
|
||||
{{ textbox(form.colour, width='1-4', colour_preview=True) }}
|
||||
<div style='margin-top:15px;'>{{textbox(form.domain)}}</div>
|
||||
{{ radios(form.brand_type) }}
|
||||
{{ page_footer(
|
||||
'Save',
|
||||
|
||||
@@ -24,13 +24,6 @@
|
||||
{{ brand.name or 'Unnamed' }}
|
||||
</a>
|
||||
</div>
|
||||
<p class="message-type">
|
||||
{% if brand.domain %}
|
||||
Default for {{ brand.domain }}
|
||||
{% else %}
|
||||
–
|
||||
{% endif %}
|
||||
</p>
|
||||
</div>
|
||||
{% endfor %}
|
||||
</nav>
|
||||
|
||||
@@ -25,7 +25,6 @@
|
||||
{% call form_wrapper() %}
|
||||
<div class="form-group">
|
||||
<div style='margin-top:15px;'>{{textbox(letter_branding_details_form.name)}}</div>
|
||||
<div style='margin-top:15px;'>{{textbox(letter_branding_details_form.domain)}}</div>
|
||||
{{ page_footer(
|
||||
'Save',
|
||||
button_name='operation',
|
||||
|
||||
@@ -24,13 +24,6 @@
|
||||
{{ brand.name }}
|
||||
</a>
|
||||
</div>
|
||||
<p class="message-type">
|
||||
{% if brand.domain %}
|
||||
Default for {{ brand.domain }}
|
||||
{% else %}
|
||||
–
|
||||
{% endif %}
|
||||
</p>
|
||||
</div>
|
||||
{% endfor %}
|
||||
</nav>
|
||||
|
||||
@@ -123,7 +123,7 @@
|
||||
<h2 class="heading-medium" id="paying">How to pay</h2>
|
||||
<p>You can find details of how to pay for Notify in our data sharing and financial agreement.</p>
|
||||
<p>
|
||||
{{ agreement_info.as_pricing_paragraph(
|
||||
{{ current_user.default_organisation.as_pricing_paragraph(
|
||||
pricing_link=url_for('main.sign_in', next=url_for('main.pricing', _anchor='paying')),
|
||||
download_link=url_for('main.agreement'),
|
||||
support_link=url_for('.feedback', ticket_type='ask-question-give-feedback', body='agreement'),
|
||||
|
||||
@@ -300,7 +300,7 @@
|
||||
|
||||
{% call row() %}
|
||||
{{ text_field('Organisation')}}
|
||||
{{ optional_text_field(current_service.organisation_name) }}
|
||||
{{ optional_text_field(current_service.organisation.name) }}
|
||||
{{ edit_field('Change', url_for('.link_service_to_organisation', service_id=current_service.id)) }}
|
||||
{% endcall %}
|
||||
{% call row() %}
|
||||
|
||||
@@ -21,7 +21,7 @@ Terms of use
|
||||
</p>
|
||||
|
||||
<p>
|
||||
{{ agreement_info.as_terms_of_use_paragraph(
|
||||
{{ current_user.default_organisation.as_terms_of_use_paragraph(
|
||||
terms_link=url_for('main.sign_in', next=url_for('main.terms')),
|
||||
download_link=url_for('.agreement'),
|
||||
support_link=url_for('.feedback', ticket_type='ask-question-give-feedback', body='agreement'),
|
||||
|
||||
225
app/utils.py
225
app/utils.py
@@ -13,15 +13,7 @@ import ago
|
||||
import dateutil
|
||||
import pyexcel
|
||||
import yaml
|
||||
from flask import (
|
||||
Markup,
|
||||
abort,
|
||||
current_app,
|
||||
redirect,
|
||||
request,
|
||||
session,
|
||||
url_for,
|
||||
)
|
||||
from flask import abort, current_app, redirect, request, session, url_for
|
||||
from flask_login import current_user
|
||||
from notifications_utils.field import Field
|
||||
from notifications_utils.formatters import make_quotes_smart
|
||||
@@ -44,6 +36,11 @@ FAILURE_STATUSES = ['failed', 'temporary-failure', 'permanent-failure',
|
||||
'technical-failure', 'virus-scan-failed', 'validation-failed']
|
||||
REQUESTED_STATUSES = SENDING_STATUSES + DELIVERED_STATUSES + FAILURE_STATUSES
|
||||
|
||||
with open('{}/email_domains.yml'.format(
|
||||
os.path.dirname(os.path.realpath(__file__))
|
||||
)) as email_domains:
|
||||
GOVERNMENT_EMAIL_DOMAIN_NAMES = yaml.safe_load(email_domains)
|
||||
|
||||
|
||||
def user_has_permissions(*permissions, **permission_kwargs):
|
||||
def wrap(func):
|
||||
@@ -288,11 +285,13 @@ def get_help_argument():
|
||||
|
||||
|
||||
def is_gov_user(email_address):
|
||||
try:
|
||||
GovernmentEmailDomain(email_address)
|
||||
return True
|
||||
except NotGovernmentEmailDomain:
|
||||
return False
|
||||
return any(
|
||||
email_address.lower().endswith((
|
||||
"@{}".format(known),
|
||||
".{}".format(known),
|
||||
))
|
||||
for known in GOVERNMENT_EMAIL_DOMAIN_NAMES
|
||||
)
|
||||
|
||||
|
||||
def get_template(
|
||||
@@ -406,204 +405,6 @@ def set_status_filters(filter_args):
|
||||
)))
|
||||
|
||||
|
||||
_dir_path = os.path.dirname(os.path.realpath(__file__))
|
||||
|
||||
|
||||
class AgreementInfo:
|
||||
|
||||
with open('{}/domains.yml'.format(_dir_path)) as domains:
|
||||
domains = yaml.safe_load(domains)
|
||||
domain_names = sorted(domains.keys(), key=len, reverse=True)
|
||||
|
||||
def __init__(self, email_address_or_domain):
|
||||
|
||||
self._match = next(filter(
|
||||
self.get_matching_function(email_address_or_domain),
|
||||
self.domain_names,
|
||||
), None)
|
||||
|
||||
self._domain = email_address_or_domain.split('@')[-1]
|
||||
|
||||
(
|
||||
self.owner,
|
||||
self.crown_status,
|
||||
self.agreement_signed,
|
||||
self.canonical_domain,
|
||||
) = self._get_info()
|
||||
|
||||
@classmethod
|
||||
def from_user(cls, user):
|
||||
return cls(user.email_address if user.is_authenticated else '')
|
||||
|
||||
@classmethod
|
||||
def from_current_user(cls):
|
||||
return cls.from_user(current_user)
|
||||
|
||||
@property
|
||||
def as_human_readable(self):
|
||||
if self.canonical_domain and 'dwp' in self.canonical_domain:
|
||||
return 'DWP - Requires OED approval'
|
||||
if self.agreement_signed:
|
||||
return 'Yes, on behalf of {}'.format(self.owner)
|
||||
elif self.owner:
|
||||
return '{} (organisation is {}, {})'.format(
|
||||
{
|
||||
False: 'No',
|
||||
None: 'Can’t tell',
|
||||
}.get(self.agreement_signed),
|
||||
self.owner,
|
||||
{
|
||||
True: 'a crown body',
|
||||
False: 'a non-crown body',
|
||||
None: 'crown status unknown',
|
||||
}.get(self.crown_status),
|
||||
)
|
||||
else:
|
||||
return 'Can’t tell (domain is {})'.format(self._domain)
|
||||
|
||||
@property
|
||||
def as_info_for_branding_request(self):
|
||||
return self.owner or 'Can’t tell (domain is {})'.format(self._domain)
|
||||
|
||||
@property
|
||||
def as_jinja_template(self):
|
||||
if self.crown_status is None:
|
||||
return 'agreement-choose'
|
||||
if self.agreement_signed:
|
||||
return 'agreement-signed'
|
||||
return 'agreement'
|
||||
|
||||
def as_terms_of_use_paragraph(self, **kwargs):
|
||||
return Markup(self._as_terms_of_use_paragraph(**kwargs))
|
||||
|
||||
def _as_terms_of_use_paragraph(self, terms_link, download_link, support_link, signed_in):
|
||||
|
||||
if not signed_in:
|
||||
return ((
|
||||
'{} <a href="{}">Sign in</a> to download a copy '
|
||||
'or find out if one is already in place.'
|
||||
).format(self._acceptance_required, terms_link))
|
||||
|
||||
if self.agreement_signed is None:
|
||||
return ((
|
||||
'{} <a href="{}">Download the agreement</a> or '
|
||||
'<a href="{}">contact us</a> to find out if we already '
|
||||
'have one in place with your organisation.'
|
||||
).format(self._acceptance_required, download_link, support_link))
|
||||
|
||||
if self.agreement_signed is False:
|
||||
return ((
|
||||
'{} <a href="{}">Download a copy</a>.'
|
||||
).format(self._acceptance_required, download_link))
|
||||
|
||||
return (
|
||||
'Your organisation ({}) has already accepted the '
|
||||
'GOV.UK Notify data sharing and financial '
|
||||
'agreement.'.format(self.owner)
|
||||
)
|
||||
|
||||
def as_pricing_paragraph(self, **kwargs):
|
||||
return Markup(self._as_pricing_paragraph(**kwargs))
|
||||
|
||||
def _as_pricing_paragraph(self, pricing_link, download_link, support_link, signed_in):
|
||||
|
||||
if not signed_in:
|
||||
return ((
|
||||
'<a href="{}">Sign in</a> to download a copy or find '
|
||||
'out if one is already in place with your organisation.'
|
||||
).format(pricing_link))
|
||||
|
||||
if self.agreement_signed is None:
|
||||
return ((
|
||||
'<a href="{}">Download the agreement</a> or '
|
||||
'<a href="{}">contact us</a> to find out if we already '
|
||||
'have one in place with your organisation.'
|
||||
).format(download_link, support_link))
|
||||
|
||||
return (
|
||||
'<a href="{}">Download the agreement</a> '
|
||||
'({} {}).'.format(
|
||||
download_link,
|
||||
self.owner,
|
||||
{
|
||||
True: 'has already accepted it',
|
||||
False: 'hasn’t accepted it yet'
|
||||
}.get(self.agreement_signed)
|
||||
)
|
||||
)
|
||||
|
||||
@property
|
||||
def _acceptance_required(self):
|
||||
return (
|
||||
'Your organisation {} must also accept our data sharing '
|
||||
'and financial agreement.'.format(
|
||||
'({})'.format(self.owner) if self.owner else '',
|
||||
)
|
||||
)
|
||||
|
||||
@property
|
||||
def crown_status_or_404(self):
|
||||
if self.crown_status is None:
|
||||
abort(404)
|
||||
return self.crown_status
|
||||
|
||||
@staticmethod
|
||||
def get_matching_function(email_address_or_domain):
|
||||
|
||||
email_address_or_domain = email_address_or_domain.lower()
|
||||
|
||||
def fn(domain):
|
||||
|
||||
return (
|
||||
email_address_or_domain == domain
|
||||
) or (
|
||||
email_address_or_domain.endswith("@{}".format(domain))
|
||||
) or (
|
||||
email_address_or_domain.endswith(".{}".format(domain))
|
||||
)
|
||||
|
||||
return fn
|
||||
|
||||
def _get_info(self):
|
||||
|
||||
details = self.domains.get(self._match, {})
|
||||
|
||||
if details is None:
|
||||
raise TypeError('Domain must have details ({})'.format(self._domain))
|
||||
|
||||
if isinstance(details, str):
|
||||
self.is_canonical = False
|
||||
return AgreementInfo(details)._get_info()
|
||||
|
||||
elif isinstance(details, dict):
|
||||
self.is_canonical = bool(details)
|
||||
return(
|
||||
details.get("owner"),
|
||||
details.get("crown"),
|
||||
details.get("agreement_signed"),
|
||||
self._match,
|
||||
)
|
||||
|
||||
|
||||
class NotGovernmentEmailDomain(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class GovernmentEmailDomain(AgreementInfo):
|
||||
|
||||
with open('{}/email_domains.yml'.format(_dir_path)) as email_domains:
|
||||
domain_names = yaml.safe_load(email_domains)
|
||||
|
||||
def __init__(self, email_address_or_domain):
|
||||
try:
|
||||
self._match = next(filter(
|
||||
self.get_matching_function(email_address_or_domain),
|
||||
self.domain_names,
|
||||
))
|
||||
except StopIteration:
|
||||
raise NotGovernmentEmailDomain()
|
||||
|
||||
|
||||
def unicode_truncate(s, length):
|
||||
encoded = s.encode('utf-8')[:length]
|
||||
return encoded.decode('utf-8', 'ignore')
|
||||
|
||||
Reference in New Issue
Block a user