Merge 2419f0d into f6b566f

app/notify_client/__init__.py

@@ -1,5 +1,5 @@
 from flask_login import current_user
-from flask import has_request_context, request
+from flask import has_request_context, request, abort
 from notifications_python_client.base import BaseAPIClient
 from notifications_python_client.version import __version__
 
@@ -26,3 +26,25 @@ def _add_request_id_header(headers):
             return headers
         headers['NotifyRequestID'] = request.request_id
         return headers
+
+    def check_inactive_service(self):
+        # this file is imported in app/__init__.py before current_service is initialised, so need to import later
+        # to prevent cyclical imports
+        from app import current_service
+
+        # if the current service is inactive and the user isn't a platform admin, we should block them from making any
+        # stateful modifications to that service
+        if current_service and not current_service['active'] and not current_user.platform_admin:
+            abort(403)
+
+    def post(self, *args, **kwargs):
+        self.check_inactive_service()
+        return super().post(*args, **kwargs)
+
+    def put(self, *args, **kwargs):
+        self.check_inactive_service()
+        return super().put(*args, **kwargs)
+
+    def delete(self, *args, **kwargs):
+        self.check_inactive_service()
+        return super().delete(*args, **kwargs)

tests/app/notify_client/test_notify_admin_api_client.py

@@ -0,0 +1,96 @@
+import uuid
+import pytest
+from unittest.mock import patch
+
+import werkzeug
+
+from tests import service_json
+from tests.conftest import api_user_active, platform_admin_user
+from app.notify_client import NotifyAdminAPIClient
+
+
+@pytest.mark.parametrize('method', [
+    'put',
+    'post',
+    'delete'
+])
+@pytest.mark.parametrize('user', [
+    api_user_active(str(uuid.uuid4())),
+    platform_admin_user(str(uuid.uuid4()))
+], ids=['api_user', 'platform_admin'])
+@pytest.mark.parametrize('service', [
+    service_json(active=True),
+    None
+], ids=['active_service', 'no_service'])
+def test_active_service_can_be_modified(app_, method, user, service):
+    api_client = NotifyAdminAPIClient('api_key', 'base_url', 'service_id')
+
+    with app_.test_request_context() as request_context, app_.test_client() as client:
+        client.login(user)
+        request_context.service = service
+
+        with patch.object(api_client, 'request') as request:
+            ret = getattr(api_client, method)('url', 'data')
+
+    assert request.called
+    assert ret == request.return_value
+
+
+@pytest.mark.parametrize('method', [
+    'put',
+    'post',
+    'delete'
+])
+def test_inactive_service_cannot_be_modified_by_normal_user(app_, api_user_active, method):
+    api_client = NotifyAdminAPIClient('api_key', 'base_url', 'service_id')
+
+    with app_.test_request_context() as request_context, app_.test_client() as client:
+        client.login(api_user_active)
+        request_context.service = service_json(active=False)
+
+        with patch.object(api_client, 'request') as request:
+            with pytest.raises(werkzeug.exceptions.Forbidden):
+                getattr(api_client, method)('url', 'data')
+
+    assert not request.called
+
+
+@pytest.mark.parametrize('method', [
+    'put',
+    'post',
+    'delete'
+])
+def test_inactive_service_can_be_modified_by_platform_admin(app_, platform_admin_user, method):
+    api_client = NotifyAdminAPIClient('api_key', 'base_url', 'service_id')
+
+    with app_.test_request_context() as request_context, app_.test_client() as client:
+        client.login(platform_admin_user)
+        request_context.service = service_json(active=False)
+
+        with patch.object(api_client, 'request') as request:
+            ret = getattr(api_client, method)('url', 'data')
+
+    assert request.called
+    assert ret == request.return_value
+
+
+def test_generate_headers_sets_standard_headers():
+    api_client = NotifyAdminAPIClient('api_key', 'base_url', 'service_id')
+
+    # with patch('app.notify_client.has_request_context', return_value=False):
+    headers = api_client.generate_headers('api_token')
+
+    assert set(headers.keys()) == {'Authorization', 'Content-type', 'User-agent'}
+    assert headers['Authorization'] == 'Bearer api_token'
+    assert headers['Content-type'] == 'application/json'
+    assert headers['User-agent'].startswith('NOTIFY-API-PYTHON-CLIENT')
+
+
+def test_generate_headers_sets_request_id_if_in_request_context(app_):
+    api_client = NotifyAdminAPIClient('api_key', 'base_url', 'service_id')
+
+    with app_.test_request_context() as request_context:
+        headers = api_client.generate_headers('api_token')
+
+    assert set(headers.keys()) == {'Authorization', 'Content-type', 'User-agent', 'NotifyRequestID'}
+    assert headers['NotifyRequestID'] == request_context.request.request_id