Add a maintenance mode for running application

conf/development.ini

@@ -38,6 +38,11 @@ datameta.tfa.enabled =
 datameta.tfa.encrypt_key =
 datameta.tfa.otp_issuer =
 
+# Uncomment and set to configure and enable support of maintenance mode
+# If the file at the path not exists, the application will return a 503
+datameta.maintenance_mode.path = /tmp/maintenance_mode
+datameta.maintenance_mode.exclude_request_paths = []
+
 pyramid.reload_templates = true
 pyramid.debug_authorization = false
 pyramid.debug_notfound = false

datameta/__init__.py

@@ -38,6 +38,9 @@ def main(global_config, **settings):
         session_factory = session_factory_from_settings(settings)
         config.set_session_factory(session_factory)
 
+        # Tweens
+        config.add_tween('datameta.tweens.maintenance_mode_tween_factory')
+
         config.include("pyramid_openapi3")
         config.pyramid_openapi3_spec(
             os.path.join(os.path.dirname(__file__), "api", "openapi.yaml")

datameta/tweens.py

@@ -0,0 +1,43 @@
+# Copyright 2021 Universität Tübingen, DKFZ and EMBL for the German Human Genome-Phenome Archive (GHGA)
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from pyramid.settings import aslist
+from pyramid.httpexceptions import HTTPServiceUnavailable
+
+"""Module defines tween factories for the application"""
+
+
+def maintenance_mode_tween_factory(handler, registry):
+    """Returns a tween that checks if the application is in maintenance mode."""
+    def maintenance_mode(request):
+        enabled = request.registry.settings.get('datameta.maintenance_mode.path', None)
+        exclude_request_paths = aslist(request.registry.settings.get('datameta.maintenance_mode.exclude_request_paths', []))
+        if enabled:
+            # If request path is excluded, return response
+            if request.path in exclude_request_paths:
+                response = handler(request)
+                return response
+            # If file exists, return response
+            elif os.path.isfile(enabled):
+                response = handler(request)
+                return response
+            # If file does not exist, return 503
+            else:
+                return HTTPServiceUnavailable('Maintenance Mode')
+        else:
+            response = handler(request)
+            return response
+
+    return maintenance_mode

tests/integration/__init__.py

@@ -3,6 +3,7 @@
 import unittest
 from webtest import TestApp
 import tempfile
+import os
 
 import transaction
 from sqlalchemy_utils import create_database, drop_database, database_exists
@@ -57,6 +58,7 @@ def setUp(self):
         self.storage_path_obj = tempfile.TemporaryDirectory()
         self.storage_path = self.storage_path_obj.name
         self.settings["datameta.storage_path"] = self.storage_path
+        self.settings["datameta.maintenance_mode.path"] = os.path.join(self.storage_path, "maintenance_mode")
 
         # initialize DB
         self.initDb()

tests/integration/test_maintenance_mode.py

@@ -0,0 +1,56 @@
+# Copyright 2021 Universität Tübingen, DKFZ and EMBL for the German Human Genome-Phenome Archive (GHGA)
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from parameterized import parameterized
+
+from . import BaseIntegrationTest
+from datameta.api import base_url
+
+
+class TestMaintenanceMode(BaseIntegrationTest):
+    """Test if application reactes correctly to existance of maintenance mode file"""
+
+    def setUp(self):
+        super().setUp()
+        self.settings["datameta.maintenance_mode.exclude_request_paths"] = ["/login", base_url + "/metadata"]
+        self.maintenance_mode_file = self.settings["datameta.maintenance_mode.path"]
+
+    def ensure_exists_maintenance_mode_file(self):
+        with open(self.maintenance_mode_file, 'a'):
+            os.utime(self.maintenance_mode_file, None)
+
+    def ensure_exists_not_maintenance_mode_file(self):
+        if os.path.exists(self.maintenance_mode_file):
+            os.remove(self.maintenance_mode_file)
+
+    @parameterized.expand([
+        # test_name route expected_response_exists expected_response_not_exists
+        ("Normal view", "/", 302, 503),
+        ("Excluded view", "/login", 200, 200),
+        ("Normal view", "/register", 200, 503),
+        ("Normal api", "/api", 302, 503),
+        ("Normal api", base_url + "/server", 500, 503),
+        ("Normal api", base_url + "/rpc/whoami", 401, 503),
+        ("Excluded api", base_url + "/metadata", 401, 401),
+        ("Normal api", base_url + "/files", 404, 503),
+        ])
+    def test_maintenance_mode(self, _, route: str, expected_response: int, expected_response_not_exists: int):
+        """Test that the maintenance mode is working as expected"""
+        self.ensure_exists_maintenance_mode_file()
+        self.testapp.get(route, status=expected_response)
+        self.ensure_exists_not_maintenance_mode_file()
+        self.testapp.get(route, status=expected_response_not_exists)
+        self.ensure_exists_maintenance_mode_file()
+        self.testapp.get(route, status=expected_response)