Merge pull request #1904 from alphagov/merge-all-the-things

Merge all the things

app/config.py

@@ -69,39 +69,6 @@ class Config(object):
     STATSD_HOST = "statsd.hostedgraphite.com"
     STATSD_PORT = 8125
     NOTIFY_ENVIRONMENT = 'development'
-
-    EMAIL_DOMAIN_REGEXES = [
-        r"gov\.uk",
-        r"mod\.uk",
-        r"mil\.uk",
-        r"ddc-mod\.org",
-        r"slc\.co\.uk",
-        r"gov\.scot",
-        r"parliament\.uk",
-        r"nhs\.uk",
-        r"nhs\.net",
-        r"police\.uk",
-        r"dclgdatamart\.co\.uk",
-        r"ucds\.email",
-        r"naturalengland\.org\.uk",
-        r"hmcts\.net",
-        r"scotent\.co\.uk",
-        r"assembly\.wales",
-        r"cjsm\.net",
-        r"cqc\.org\.uk",
-        r"bl\.uk",
-        r"stfc\.ac\.uk",
-        r"wmfs\.net",
-        r"bbsrc\.ac\.uk",
-        r"acas\.org\.uk",
-        r"gov\.wales",
-        r"biglotteryfund\.org\.uk",
-        r"marinemanagement\.org\.uk",
-        r"britishmuseum\.org",
-        r"derrystrabane\.com",
-        r"highwaysengland\.co\.uk",
-    ]
-
     LOGO_UPLOAD_BUCKET_NAME = 'public-logos-local'
     ROUTE_SECRET_KEY_1 = os.environ.get('ROUTE_SECRET_KEY_1', '')
     ROUTE_SECRET_KEY_2 = os.environ.get('ROUTE_SECRET_KEY_2', '')

app/email_domains.yml

@@ -0,0 +1,30 @@
+---
+- gov.uk
+- mod.uk
+- mil.uk
+- ddc-mod.org
+- slc.co.uk
+- gov.scot
+- parliament.uk
+- nhs.uk
+- nhs.net
+- police.uk
+- dclgdatamart.co.uk
+- ucds.email
+- naturalengland.org.uk
+- hmcts.net
+- scotent.co.uk
+- assembly.wales
+- cjsm.net
+- cqc.org.uk
+- bl.uk
+- stfc.ac.uk
+- wmfs.net
+- bbsrc.ac.uk
+- acas.org.uk
+- gov.wales
+- biglotteryfund.org.uk
+- marinemanagement.org.uk
+- britishmuseum.org
+- derrystrabane.com
+- highwaysengland.co.uk

app/notify_client/models.py

@@ -1,6 +1,6 @@
 from itertools import chain
 from flask_login import UserMixin, AnonymousUserMixin
-from flask import session
+from flask import request, session
 
 
 roles = {
@@ -13,6 +13,10 @@ roles = {
 all_permissions = set(chain.from_iterable(roles.values())) | {'view_activity'}
 
 
+def _get_service_id_from_view_args():
+    return request.view_args.get('service_id', None)
+
+
 class User(UserMixin):
     def __init__(self, fields, max_failed_login_count=3):
         self._id = fields.get('id')
@@ -117,9 +121,8 @@ class User(UserMixin):
         if admin_override and not permissions:
             return False
 
-        from flask import request
         # Service id is always set on the request for service specific views.
-        service_id = request.view_args.get('service_id', None)
+        service_id = _get_service_id_from_view_args()
         if service_id in self._permissions:
             if any_:
                 return any([x in self._permissions[service_id] for x in permissions])

app/notify_client/service_api_client.py

@@ -226,6 +226,16 @@ class ServiceAPIClient(NotifyAdminAPIClient):
             service_id=service_id)
         return self.get(endpoint, *params)
 
+    def count_service_templates(self, service_id, template_type=None):
+        return len([
+            template for template in
+            self.get_service_templates(service_id)['data']
+            if (
+                not template_type
+                or template['template_type'] == template_type
+            )
+        ])
+
     def delete_service_template(self, service_id, template_id):
         """
         Set a service template's archived flag to True

app/notify_client/user_api_client.py

@@ -131,6 +131,12 @@ class UserApiClient(NotifyAdminAPIClient):
         resp = self.get(endpoint)
         return [User(data) for data in resp['data']]
 
+    def get_count_of_users_with_permission(self, service_id, permission):
+        return len([
+            user for user in self.get_users_for_service(service_id)
+            if user.has_permissions(permission, any_=True)
+        ])
+
     def add_user_to_service(self, service_id, user_id, permissions):
         endpoint = '/service/{}/users/{}'.format(service_id, user_id)
         data = [{'permission': x} for x in permissions]

app/templates/views/service-settings.html

@@ -227,9 +227,14 @@
       </ul>
 
       <p>
-        To remove these restrictions
-        <a href="{{ url_for('.request_to_go_live', service_id=current_service.id) }}">request to go live</a>.
-      </p>
+        {% if current_user.has_permissions('manage_settings') %}
+          To remove these restrictions
+          <a href="{{ url_for('.request_to_go_live', service_id=current_service.id) }}">request to go live</a>.
+        {% else %}
+          Your service manager can ask to have these restrictions removed.
+        {% endif %}
+        </p>
+
     {% else %}
       <h2 class="heading-medium">Your service is live</h2>
 
@@ -328,7 +333,7 @@
             <a href="{{ url_for('.service_switch_can_send_precompiled_letter', service_id=current_service.id) }}" class="button">
               {{ 'Stop sending precompiled letters' if 'precompiled_letter' in current_service.permissions else 'Allow to send precompiled letters' }}
             </a>
-          </li>  
+          </li>
         {% endif %}
         <li class="bottom-gutter">
           <a href="{{ url_for('.service_switch_email_auth', service_id=current_service.id) }}" class="button">

app/templates/views/signedout.html

@@ -120,12 +120,12 @@
       <div class="grid-row bottom-gutter">
         <div class="column-half">
           <h3 class="visually-hidden">Services</h3>
-          <div class="product-page-big-number">160</div>
+          <div class="product-page-big-number">166</div>
           services
         </div>
         <div class="column-half">
           <h3 class="visually-hidden">Organisations</h3>
-          <div class="product-page-big-number">65</div>
+          <div class="product-page-big-number">66</div>
           organisations
         </div>
       </div>

app/utils.py

@@ -1,5 +1,8 @@
+import os
 import re
 import csv
+import yaml
+
 from itertools import chain
 
 import pytz
@@ -280,9 +283,11 @@ def get_help_argument():
 
 
 def is_gov_user(email_address):
-    valid_domains = current_app.config['EMAIL_DOMAIN_REGEXES']
-    email_regex = (r"[\.|@]({})$".format("|".join(valid_domains)))
-    return bool(re.search(email_regex, email_address.lower()))
+    try:
+        GovernmentEmailDomain(email_address)
+        return True
+    except NotGovernmentEmailDomain:
+        return False
 
 
 def get_template(
@@ -428,3 +433,76 @@ def set_status_filters(filter_args):
         SENDING_STATUSES if 'sending' in status_filters else [],
         FAILURE_STATUSES if 'failed' in status_filters else []
     )))
+
+
+_dir_path = os.path.dirname(os.path.realpath(__file__))
+
+
+class GovernmentDomain:
+
+    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.owner,
+            self.crown_status,
+            self.agreement_signed
+        ) = self._get_details_of_domain()
+
+    @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_details_of_domain(self):
+
+        details = self.domains.get(self._match) or {}
+
+        if isinstance(details, str):
+            return GovernmentDomain(details)._get_details_of_domain()
+
+        elif isinstance(details, dict):
+            return(
+                details.get("owner"),
+                details.get("crown"),
+                details.get("agreement_signed"),
+            )
+
+
+class NotGovernmentEmailDomain(Exception):
+    pass
+
+
+class GovernmentEmailDomain(GovernmentDomain):
+
+    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()

tests/app/main/views/test_service_settings.py

@@ -11,6 +11,7 @@ from tests import validate_route_permission, service_json
 from tests.conftest import (
     active_user_with_permissions,
     active_user_no_api_key_permission,
+    active_user_no_settings_permission,
     platform_admin_user,
     normalize_spaces,
     multiple_reply_to_email_addresses,
@@ -281,20 +282,49 @@ def test_should_redirect_after_change_service_name(
     assert mock_service_name_is_unique.called
 
 
+@pytest.mark.parametrize('user, expected_text, expected_link', [
+    (
+        active_user_with_permissions,
+        'To remove these restrictions request to go live.',
+        True,
+    ),
+    (
+        active_user_no_settings_permission,
+        'Your service manager can ask to have these restrictions removed.',
+        False,
+    ),
+])
 def test_show_restricted_service(
-        logged_in_client,
-        service_one,
-        single_reply_to_email_address,
-        single_letter_contact_block,
-        mock_get_service_organisation,
-        single_sms_sender,
-        mock_get_service_settings_page_common,
+    client,
+    mocker,
+    fake_uuid,
+    service_one,
+    single_reply_to_email_address,
+    single_letter_contact_block,
+    mock_get_service_organisation,
+    single_sms_sender,
+    mock_get_service_settings_page_common,
+    user,
+    expected_text,
+    expected_link,
 ):
-    response = logged_in_client.get(url_for('main.service_settings', service_id=service_one['id']))
+    client.login(user(fake_uuid), mocker, service_one)
+    response = client.get(url_for('main.service_settings', service_id=service_one['id']))
     page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
     assert page.find('h1').text == 'Settings'
     assert page.find_all('h2')[0].text == 'Your service is in trial mode'
 
+    request_to_live = page.select_one('main p')
+    request_to_live_link = request_to_live.select_one('a')
+
+    assert normalize_spaces(request_to_live.text) == expected_text
+
+    if expected_link:
+        assert request_to_live_link.text.strip() == 'request to go live'
+        assert request_to_live_link['href'] == url_for('main.request_to_go_live', service_id=service_one['id'])
+    else:
+        assert not request_to_live_link
+
 
 def test_switch_service_to_live(
         logged_in_platform_admin_client,

tests/app/notify_client/test_service_api_client.py

@@ -1,7 +1,8 @@
 import pytest
 
+from app import service_api_client
 from app.notify_client.service_api_client import ServiceAPIClient
-from tests.conftest import fake_uuid
+from tests.conftest import fake_uuid, SERVICE_ONE_ID
 
 
 def test_client_posts_archived_true_when_deleting_template(mocker):
@@ -78,3 +79,57 @@ def test_client_creates_service_with_correct_data(
             email_from='test@example.com',
         ),
     )
+
+
+@pytest.mark.parametrize('template_data, extra_args, expected_count', (
+    (
+        [],
+        {},
+        0,
+    ),
+    (
+        [],
+        {'template_type': 'email'},
+        0,
+    ),
+    (
+        [
+            {'template_type': 'email'},
+            {'template_type': 'sms'},
+        ],
+        {},
+        2,
+    ),
+    (
+        [
+            {'template_type': 'email'},
+            {'template_type': 'sms'},
+        ],
+        {'template_type': 'email'},
+        1,
+    ),
+    (
+        [
+            {'template_type': 'email'},
+            {'template_type': 'sms'},
+        ],
+        {'template_type': 'letter'},
+        0,
+    ),
+))
+def test_client_returns_count_of_service_templates(
+    app_,
+    mocker,
+    template_data,
+    extra_args,
+    expected_count,
+):
+
+    mocker.patch(
+        'app.service_api_client.get_service_templates',
+        return_value={'data': template_data}
+    )
+
+    assert service_api_client.count_service_templates(
+        SERVICE_ONE_ID, **extra_args
+    ) == expected_count

tests/app/notify_client/test_user_client.py

@@ -1,6 +1,69 @@
 import pytest
-
+from unittest.mock import call
 from app import user_api_client
+from app.notify_client.models import User
+from tests.conftest import SERVICE_ONE_ID
+
+
+def test_client_gets_all_users_for_service(
+    mocker,
+    fake_uuid,
+):
+
+    user_api_client.max_failed_login_count = 99  # doesn't matter for this test
+    mock_get = mocker.patch(
+        'app.notify_client.user_api_client.UserApiClient.get',
+        return_value={'data': [
+            {'id': fake_uuid},
+        ]}
+    )
+
+    users = user_api_client.get_users_for_service(SERVICE_ONE_ID)
+
+    mock_get.assert_called_once_with('/service/{}/users'.format(SERVICE_ONE_ID))
+    assert len(users) == 1
+    assert users[0].id == fake_uuid
+
+
+def test_client_returns_count_of_users_with_manage_service(
+    app_,
+    client,
+    mocker,
+    fake_uuid,
+):
+
+    def _service_one_user_with_permissions(*permissions):
+        return User({'permissions': {SERVICE_ONE_ID: list(permissions)}})
+
+    mock_get_users = mocker.patch(
+        'app.notify_client.user_api_client.UserApiClient.get_users_for_service',
+        return_value=[
+            _service_one_user_with_permissions('manage_settings', 'view_activity'),
+            _service_one_user_with_permissions('manage_settings'),
+            _service_one_user_with_permissions('view_activity'),
+            _service_one_user_with_permissions('manage_templates'),
+        ]
+    )
+
+    mocker.patch(
+        'app.notify_client.models._get_service_id_from_view_args',
+        return_value=SERVICE_ONE_ID,
+    )
+
+    assert user_api_client.get_count_of_users_with_permission(
+        SERVICE_ONE_ID,
+        'manage_settings'
+    ) == 2
+
+    assert user_api_client.get_count_of_users_with_permission(
+        SERVICE_ONE_ID,
+        'manage_templates'
+    ) == 1
+
+    assert mock_get_users.call_args_list == [
+        call(SERVICE_ONE_ID),
+        call(SERVICE_ONE_ID)
+    ]
 
 
 def test_client_uses_correct_find_by_email(mocker, api_user_active):

tests/app/test_utils.py

@@ -13,7 +13,8 @@ from app.utils import (
     generate_next_dict,
     Spreadsheet,
     get_letter_timings,
-    get_cdn_domain
+    get_cdn_domain,
+    GovernmentDomain,
 )
 from tests.conftest import fake_uuid
 
@@ -307,3 +308,64 @@ def test_get_cdn_domain_on_non_localhost(client, mocker):
     mocker.patch.dict('app.current_app.config', values={'ADMIN_BASE_URL': 'https://some.admintest.com'})
     domain = get_cdn_domain()
     assert domain == 'static-logos.admintest.com'
+
+
+@pytest.mark.parametrize("domain_or_email_address", (
+    "test@dclgdatamart.co.uk", "test@communities.gsi.gov.uk", "test@communities.gov.uk",
+))
+def test_get_valid_government_domain_known_details(domain_or_email_address):
+    government_domain = GovernmentDomain(domain_or_email_address)
+    assert government_domain.crown_status is None
+    assert government_domain.owner == "Ministry of Housing, Communities & Local Government"
+    assert government_domain.agreement_signed is True
+
+
+@pytest.mark.parametrize("domain_or_email_address", (
+    "test@police.gov.uk", "police.gov.uk",
+))
+def test_get_valid_government_domain_unknown_details(domain_or_email_address):
+    government_domain = GovernmentDomain(domain_or_email_address)
+    assert government_domain.crown_status is None
+    assert government_domain.owner is None
+    assert government_domain.agreement_signed is None
+
+
+def test_get_valid_government_domain_some_known_details():
+    government_domain = GovernmentDomain("marinemanagement.org.uk")
+    assert government_domain.crown_status is None
+    assert government_domain.owner == "Marine Management Organisation"
+    assert government_domain.agreement_signed is True
+
+
+def test_get_valid_government_domain_gets_most_specific_first():
+
+    generic = GovernmentDomain("gov.uk")
+    assert generic.crown_status is None
+    assert generic.owner is None
+    assert generic.agreement_signed is None
+
+    specific = GovernmentDomain("dacorum.gov.uk")
+    assert specific.crown_status is False
+    assert specific.owner == 'Dacorum Borough Council'
+    assert specific.agreement_signed is True
+
+
+def test_validate_government_domain_data():
+
+    for domain in GovernmentDomain.domains.keys():
+
+        government_domain = GovernmentDomain(domain)
+
+        assert government_domain.crown_status in {
+            True, False, None
+        }
+
+        assert (
+            government_domain.owner is None
+        ) or (
+            isinstance(government_domain.owner, str)
+        )
+
+        assert government_domain.agreement_signed in {
+            True, False, None
+        }

tests/conftest.py

@@ -1261,6 +1261,31 @@ def active_user_no_api_key_permission(fake_uuid):
     return user
 
 
+@pytest.fixture
+def active_user_no_settings_permission(fake_uuid):
+    from app.notify_client.user_api_client import User
+
+    user_data = {
+        'id': fake_uuid,
+        'name': 'Test User With Permissions',
+        'password': 'somepassword',
+        'password_changed_at': str(datetime.utcnow()),
+        'email_address': 'test@user.gov.uk',
+        'mobile_number': '07700 900762',
+        'state': 'active',
+        'failed_login_count': 0,
+        'permissions': {SERVICE_ONE_ID: [
+            'manage_templates',
+            'manage_api_keys',
+            'view_activity',
+        ]},
+        'platform_admin': False,
+        'auth_type': 'sms_auth'
+    }
+    user = User(user_data)
+    return user
+
+
 @pytest.fixture(scope='function')
 def api_user_locked(fake_uuid):
     from app.notify_client.user_api_client import User