[ENG-10768] Have a display of some form when the OSF is down for planned maintenance

admin/maintenance/views.py

@@ -1,7 +1,7 @@
 import pytz
 import datetime
 
-from osf.models import MaintenanceState
+from osf.models import MaintenanceState, MaintenanceMode
 import website.maintenance as maintenance
 from admin.maintenance.forms import MaintenanceForm
 
@@ -36,15 +36,17 @@ def get_context_data(self, **kwargs):
         maintenance = MaintenanceState.objects.first()
         kwargs['form'] = MaintenanceForm()
         kwargs['current_alert'] = model_to_dict(maintenance) if maintenance else None
+        kwargs['maintenance_mode'] = MaintenanceMode.is_under_maintenance()
         return super().get_context_data(**kwargs)
 
     def post(self, request, *args, **kwargs):
         data = request.POST
-
-        start = convert_eastern_to_utc(data['start']).isoformat() if data.get('start') else None
-        end = convert_eastern_to_utc(data['end']).isoformat() if data.get('end') else None
-
-        maintenance.set_maintenance(data.get('message', ''), data['level'], start, end)
+        if maintenance_mode := data.get('maintenance_mode'):
+            MaintenanceMode(maintenance_mode=False if maintenance_mode == 'True' else True).save()
+        else:
+            start = convert_eastern_to_utc(data['start']).isoformat() if data.get('start') else None
+            end = convert_eastern_to_utc(data['end']).isoformat() if data.get('end') else None
+            maintenance.set_maintenance(data.get('message', ''), data['level'], start, end)
         return redirect('maintenance:display')
 
 

admin/templates/maintenance/display.html

@@ -66,6 +66,21 @@ <h4>Put up an alert:</h4>
                 </form>
             </div>
         </div>
+
+
+    <div class="row">
+         <div class="col-md-9" style="margin-top: 15px;">
+         <form action="" method="post">
+             {% csrf_token %}
+             <input type="hidden" name="maintenance_mode" value={{maintenance_mode}}>
+             {% if maintenance_mode %}
+             <input class="btn btn-success" type="submit" value="Turn off Maintenance Mode" />
+             {% else %}
+             <input class="btn btn-danger" type="submit" value="Turn on Maintenance Mode" />
+             {% endif %}
+         </form>
+         </div>
+    </div>
     {% endif %}
 
 {% endblock content %}

admin_tests/maintenance/test_views.py

@@ -8,7 +8,7 @@
 from django.core.exceptions import PermissionDenied
 
 import website.maintenance as maintenance
-from osf.models import MaintenanceState
+from osf.models import MaintenanceState, MaintenanceMode
 from osf_tests.factories import AuthUserFactory
 
 from admin_tests.utilities import setup_view
@@ -105,3 +105,59 @@ def test_correct_view_permissions(self, req, user, plain_view):
 
         res = plain_view.as_view()(req)
         assert res.status_code == 200
+
+
+@pytest.mark.urls('admin.base.urls')
+class TestMaintenanceMode:
+
+    @pytest.fixture()
+    def user(self):
+        user = AuthUserFactory()
+        view_permission = Permission.objects.get(codename='change_maintenancestate')
+        user.user_permissions.add(view_permission)
+        user.save()
+        return user
+
+    @pytest.fixture()
+    def plain_view(self):
+        return views.MaintenanceDisplay
+
+    @pytest.fixture()
+    def view(self, user, plain_view):
+        req = RequestFactory().get('/fake_path')
+        req.user = user
+        view = plain_view()
+        setup_view(view, req)
+        return view
+
+    def test_get_context_data_includes_maintenance_mode(self, view):
+        MaintenanceMode(maintenance_mode=True).save()
+        context = view.get_context_data()
+        assert context['maintenance_mode'] is True
+        MaintenanceMode(maintenance_mode=False).save()
+        context = view.get_context_data()
+        assert context['maintenance_mode'] is False
+
+    def test_post_toggles_maintenance_mode_on(self, user, plain_view):
+        MaintenanceMode(maintenance_mode=False).save()
+        req = RequestFactory().post('/fake_path', data={'maintenance_mode': 'False'})
+        req.user = user
+        view = plain_view()
+        setup_view(view, req)
+        response = view.post(req)
+        # It should redirect back to the display page
+        assert response.status_code == 302
+        # The database state should now be True
+        assert MaintenanceMode.is_under_maintenance() is True
+
+    def test_post_toggles_maintenance_mode_off(self, user, plain_view):
+        MaintenanceMode(maintenance_mode=True).save()
+        req = RequestFactory().post('/fake_path', data={'maintenance_mode': 'True'})
+        req.user = user
+        view = plain_view()
+        setup_view(view, req)
+        response = view.post(req)
+        # It should redirect back to the display page
+        assert response.status_code == 302
+        # The database state should now be False
+        assert MaintenanceMode.is_under_maintenance() is False

api/base/middleware.py

@@ -5,6 +5,7 @@
 from importlib import import_module
 
 from django.conf import settings
+from django.http import JsonResponse
 from django.contrib.sessions.middleware import SessionMiddleware
 from django.utils.deprecation import MiddlewareMixin
 from sentry_sdk import init
@@ -24,6 +25,7 @@
 from .api_globals import api_globals
 from api.base import settings as api_settings
 from api.base.authentication.drf import drf_get_session_from_cookie
+from osf.models import MaintenanceMode
 
 SessionStore = import_module(settings.SESSION_ENGINE).SessionStore
 
@@ -132,3 +134,20 @@ def process_request(self, request):
             request.session = drf_get_session_from_cookie(cookie)
         else:
             request.session = SessionStore()
+
+
+class MaintenanceModeMiddleware:
+    def __init__(self, get_response):
+        self.get_response = get_response
+
+    def __call__(self, request):
+        if MaintenanceMode.is_under_maintenance():
+            return JsonResponse(
+                {
+                    'meta': {
+                        'maintenance_mode': True,
+                        'status_page': 'https://status.cos.io',
+                    },
+                }, status=503,
+            )
+        return self.get_response(request)

api/base/settings/defaults.py

@@ -232,6 +232,7 @@
     'django.middleware.common.CommonMiddleware',
     'django.middleware.csrf.CsrfViewMiddleware',
     'api.base.middleware.UnsignCookieSessionMiddleware',
+    'api.base.middleware.MaintenanceModeMiddleware',
     'django.contrib.auth.middleware.AuthenticationMiddleware',
     'django.contrib.messages.middleware.MessageMiddleware',
     'django.middleware.clickjacking.XFrameOptionsMiddleware',

osf/migrations/0039_maintenancemode.py

@@ -0,0 +1,38 @@
+# Generated by Django 4.2.26 on 2026-04-23 14:25
+
+from django.db import migrations, models
+
+
+def create_initial_record(apps, schema_editor):
+    MaintenanceMode = apps.get_model('osf', 'MaintenanceMode')
+    MaintenanceMode.objects.get_or_create(
+        pk=1,
+        defaults={'maintenance_mode': False}
+    )
+
+
+def reverse_initial_record(apps, schema_editor):
+    # the reverse 'reverse_initial_record' does nothing
+    # because the table will be removed
+    pass
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('osf', '0038_abstractnode_date_last_indexed_and_more'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='MaintenanceMode',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('maintenance_mode', models.BooleanField(default=False)),
+            ],
+        ),
+        migrations.RunPython(
+            create_initial_record,
+            reverse_code=reverse_initial_record
+        ),
+    ]

osf/models/__init__.py

@@ -52,7 +52,7 @@
 from .institution_affiliation import InstitutionAffiliation
 from .institution_storage_region import InstitutionStorageRegion
 from .licenses import NodeLicense, NodeLicenseRecord
-from .maintenance_state import MaintenanceState
+from .maintenance_state import MaintenanceState, MaintenanceMode
 from .metadata import GuidMetadataRecord
 from .metaschema import (
     FileMetadataSchema,

osf/models/maintenance_state.py

@@ -14,3 +14,16 @@ class MaintenanceState(models.Model):
     start = NonNaiveDateTimeField()
     end = NonNaiveDateTimeField()
     message = models.TextField(blank=True)
+
+
+class MaintenanceMode(models.Model):
+    maintenance_mode = models.BooleanField(default=False)
+
+    def save(self, *args, **kwargs):
+        self.pk = 1
+        super().save(*args, **kwargs)
+
+    @classmethod
+    def is_under_maintenance(cls):
+        obj, _ = cls.objects.get_or_create(pk=1)
+        return obj.maintenance_mode

osf_tests/test_middleware.py

@@ -0,0 +1,105 @@
+import pytest
+from unittest import mock
+from tests.base import ApiTestCase
+from osf.utils import permissions
+from osf_tests.factories import (
+    AuthUserFactory,
+    CollectionProviderFactory,
+    ProjectFactory,
+)
+
+
+@pytest.fixture()
+def provider():
+    provider = CollectionProviderFactory()
+    provider.update_group_permissions()
+    return provider
+
+
+@pytest.fixture()
+def admin(provider):
+    user = AuthUserFactory()
+    provider.get_group(permissions.ADMIN).user_set.add(user)
+    return user
+
+
+@pytest.fixture()
+def node(admin):
+    return ProjectFactory(creator=admin)
+
+
+class TestMaintenanceModeMiddlewareIntegration(ApiTestCase):
+    MAINTENANCE_MOCK_PATH = 'api.base.middleware.MaintenanceMode.is_under_maintenance'
+
+    def setUp(self):
+        super().setUp()
+        self.provider = CollectionProviderFactory()
+        self.provider.update_group_permissions()
+        self.admin = AuthUserFactory()
+        self.provider.get_group(permissions.ADMIN).user_set.add(self.admin)
+        self.node = ProjectFactory(creator=self.admin)
+
+    @mock.patch(MAINTENANCE_MOCK_PATH, return_value=True)
+    def test_middleware_blocks_post_if_maintenance_mode_on(self, mock_maintenance):
+        url = f'/v2/nodes/{self.node._id}/'
+        response = self.app.post_json(url, {}, expect_errors=True)
+        assert response.status_code == 503
+        assert response.json['meta']['maintenance_mode'] is True
+        assert response.json['meta']['status_page'] == 'https://status.cos.io'
+
+    @mock.patch(MAINTENANCE_MOCK_PATH, return_value=True)
+    def test_middleware_blocks_patch_if_maintenance_mode_on(self, mock_maintenance):
+        url = f'/v2/nodes/{self.node._id}/'
+        original_title = self.node.title
+        payload = {
+            'data': {
+                'id': self.node._id,
+                'type': 'nodes',
+                'attributes': {'title': 'Updated Title'}
+            }
+        }
+        response = self.app.patch_json(url, payload, expect_errors=True)
+        assert response.status_code == 503
+        assert response.json['meta']['maintenance_mode'] is True
+        self.node.reload()
+        assert self.node.title == original_title
+
+    @mock.patch(MAINTENANCE_MOCK_PATH, return_value=True)
+    def test_middleware_blocks_delete_if_maintenance_mode_on(self, mock_maintenance):
+        url = f'/v2/nodes/{self.node._id}/'
+        response = self.app.delete(url, expect_errors=True)
+        assert response.status_code == 503
+        assert response.json['meta']['maintenance_mode'] is True
+        self.node.reload()
+        assert self.node.is_deleted is False
+
+    @mock.patch(MAINTENANCE_MOCK_PATH, return_value=False)
+    def test_go_to_post_view_if_maintenance_mode_off(self, mock_maintenance):
+        url = '/v2/nodes/'
+        payload = {
+            'data': {
+                'type': 'nodes',
+                'attributes': {'title': 'New Node', 'category': 'project'}
+            }
+        }
+        response = self.app.post_json(url, payload, auth=self.admin.auth)
+        assert response.status_code == 201
+
+    @mock.patch(MAINTENANCE_MOCK_PATH, return_value=False)
+    def test_go_to_patch_view_if_maintenance_mode_off(self, mock_maintenance):
+        url = f'/v2/nodes/{self.node._id}/'
+        payload = {
+            'data': {
+                'id': self.node._id,
+                'type': 'nodes',
+                'attributes': {'title': 'Updated Title'}
+            }
+        }
+        response = self.app.patch_json(url, payload, auth=self.admin.auth)
+        assert response.status_code == 200
+
+    @mock.patch(MAINTENANCE_MOCK_PATH, return_value=False)
+    def test_go_to_delete_view_if_maintenance_mode_off(self, mock_maintenance):
+        url = f'/v2/nodes/{self.node._id}/'
+        response = self.app.delete(url, auth=self.admin.auth)
+        assert response.status_code == 204