Read only AdminFlag

tests/common/db/utils.py → tests/common/db/admin.py

@@ -12,7 +12,7 @@
 
 import factory.fuzzy
 
-from warehouse.utils.admin_flags import AdminFlag
+from warehouse.admin.flags import AdminFlag
 
 from .base import WarehouseFactory
 

tests/conftest.py

@@ -27,6 +27,7 @@
 
 from warehouse.config import configure
 from warehouse.accounts import services
+from warehouse.admin.flags import Flags
 
 from .common.db import Session
 
@@ -215,6 +216,7 @@ def query_recorder(app_config):
 def db_request(pyramid_request, db_session, datadog):
     pyramid_request.registry.datadog = datadog
     pyramid_request.db = db_session
+    pyramid_request.flags = Flags(pyramid_request)
     return pyramid_request
 
 

tests/unit/accounts/test_views.py

@@ -25,7 +25,7 @@
     IUserService, ITokenService, TokenExpired, TokenInvalid, TokenMissing,
     TooManyFailedLogins
 )
-from warehouse.utils.admin_flags import AdminFlag
+from warehouse.admin.flags import AdminFlag
 
 from ...common.db.accounts import EmailFactory, UserFactory
 
@@ -394,11 +394,13 @@ def test_register_redirect(self, db_request, monkeypatch):
         assert send_email.calls == [pretend.call(db_request, email)]
 
     def test_register_fails_with_admin_flag_set(self, db_request):
-        admin_flag = (db_request.db.query(AdminFlag)
-                      .filter(
-                          AdminFlag.id == 'disallow-new-user-registration')
-                      .first())
-        admin_flag.enabled = True
+        # This flag was already set via migration, just need to enable it
+        flag = (
+            db_request.db.query(AdminFlag)
+            .get('disallow-new-user-registration')
+        )
+        flag.enabled = True
+
         db_request.method = "POST"
 
         db_request.POST.update({

tests/unit/admin/test_core.py

@@ -34,7 +34,10 @@ def test_includeme():
     assert config.add_static_view.calls == [
         pretend.call("admin/static", "static", cache_max_age=0),
     ]
-    assert config.include.calls == [pretend.call(".routes")]
+    assert config.include.calls == [
+        pretend.call(".routes"),
+        pretend.call(".flags"),
+    ]
     assert config.add_view.calls == [
         pretend.call(
             accounts_views.login,

tests/unit/utils/test_admin_flags.py → tests/unit/admin/test_flags.py

@@ -10,16 +10,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from warehouse.utils.admin_flags import AdminFlag
-
-from ...common.db.utils import AdminFlagFactory as DBAdminFlagFactory
+from ...common.db.admin import AdminFlagFactory
 
 
 class TestAdminFlag:
 
-    def test_default(self, db_session):
-        assert not AdminFlag.is_enabled(db_session, 'not-a-real-flag')
+    def test_default(self, db_request):
+        assert not db_request.flags.enabled('not-a-real-flag')
+
+    def test_enabled(self, db_request):
+        AdminFlagFactory(id='this-flag-is-enabled')
 
-    def test_enabled(self, db_session):
-        DBAdminFlagFactory.create(id='this-flag-is-enabled', enabled=True)
-        assert AdminFlag.is_enabled(db_session, 'this-flag-is-enabled')
+        assert db_request.flags.enabled('this-flag-is-enabled')

tests/unit/admin/test_routes.py

@@ -116,4 +116,14 @@ def test_includeme():
             "/admin/blacklist/remove/",
             domain=warehouse,
         ),
+        pretend.call(
+            "admin.flags",
+            "/admin/flags/",
+            domain=warehouse,
+        ),
+        pretend.call(
+            "admin.flags.edit",
+            "/admin/flags/edit/",
+            domain=warehouse,
+        ),
     ]

tests/unit/admin/views/test_flags.py

@@ -0,0 +1,100 @@
+# 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
+#
+# http://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 pytest
+
+from warehouse.admin.flags import AdminFlag
+from warehouse.admin.views import flags as views
+
+from ....common.db.admin import AdminFlagFactory
+
+
+class TestGetFlags:
+
+    def test_get_classifiers(self, db_request):
+        # Clear out any existing flags added from migrations
+        db_request.db.query(AdminFlag).delete()
+
+        flag_a = AdminFlagFactory(id='flag-a')
+        flag_b = AdminFlagFactory(id='flag-b')
+
+        assert views.get_flags(db_request) == {
+            'flags': [flag_a, flag_b],
+        }
+
+
+class TestEditFlag:
+
+    @pytest.mark.parametrize(
+        "description, enabled, post, expected_description, expected_enabled",
+        [
+            (
+                # Nothing changed when enabled
+                'old', True,
+                {'id': 'foo-bar', 'description': 'old', 'enabled': 'on'},
+                'old', True,
+            ),
+            (
+                # Nothing changed when disabled
+                'old', False,
+                {'id': 'foo-bar', 'description': 'old'},
+                'old', False,
+            ),
+            (
+                # Enable flag
+                'old', False,
+                {'id': 'foo-bar', 'description': 'old', 'enabled': 'on'},
+                'old', True,
+            ),
+            (
+                # Disable flag
+                'old', True,
+                {'id': 'foo-bar', 'description': 'old'},
+                'old', False,
+            ),
+            (
+                # Change description when enabled
+                'old', True,
+                {'id': 'foo-bar', 'description': 'new', 'enabled': 'on'},
+                'new', True,
+            ),
+            (
+                # Change description when disabled
+                'old', False,
+                {'id': 'foo-bar', 'description': 'new'},
+                'new', False,
+            ),
+        ]
+    )
+    def test_edit_flag(
+            self, db_request, description, enabled, post, expected_description,
+            expected_enabled):
+
+        # Clear out any existing flags added from migrations
+        db_request.db.query(AdminFlag).delete()
+
+        flag = AdminFlagFactory(
+            id='foo-bar',
+            description=description,
+            enabled=enabled,
+        )
+
+        db_request.POST = post
+        db_request.route_path = lambda *a: '/the/redirect'
+        db_request.flash = lambda *a: None
+
+        views.edit_flag(db_request)
+
+        db_request.db.flush()
+
+        assert flag.enabled == expected_enabled
+        assert flag.description == expected_description

tests/unit/forklift/test_legacy.py

@@ -36,7 +36,7 @@
     File, Filename, Dependency, DependencyKind, Release, Project, Role,
     JournalEntry,
 )
-from warehouse.utils.admin_flags import AdminFlag
+from warehouse.admin.flags import AdminFlag
 
 from ...common.db.accounts import UserFactory, EmailFactory
 from ...common.db.packaging import (
@@ -738,6 +738,7 @@ def test_fails_invalid_version(self, pyramid_config, pyramid_request,
                                    version):
         pyramid_config.testing_securitypolicy(userid=1)
         pyramid_request.POST["protocol_version"] = version
+        pyramid_request.flags = pretend.stub(enabled=lambda *a: False)
 
         with pytest.raises(HTTPBadRequest) as excinfo:
             legacy.file_upload(pyramid_request)
@@ -2797,7 +2798,21 @@ def test_upload_purges_legacy(self, pyramid_config, db_request,
             ),
         ]
 
+    def test_fails_in_read_only_mode(self, pyramid_request):
+        pyramid_request.flags = pretend.stub(enabled=lambda *a: True)
+
+        with pytest.raises(HTTPForbidden) as excinfo:
+            legacy.file_upload(pyramid_request)
+
+        resp = excinfo.value
+
+        assert resp.status_code == 403
+        assert resp.status == (
+            '403 Read Only Mode: Uploads are temporarily disabled'
+        )
+
     def test_fails_without_user(self, pyramid_config, pyramid_request):
+        pyramid_request.flags = pretend.stub(enabled=lambda *a: False)
         pyramid_config.testing_securitypolicy(userid=None)
 
         with pytest.raises(HTTPForbidden) as excinfo:

tests/unit/test_db.py

@@ -147,7 +147,10 @@ def test_creates_engine(monkeypatch):
     ],
 )
 def test_create_session(monkeypatch, read_only, tx_status):
-    session_obj = pretend.stub(close=pretend.call_recorder(lambda: None))
+    session_obj = pretend.stub(
+        close=pretend.call_recorder(lambda: None),
+        query=lambda *a: pretend.stub(get=lambda *a: None),
+    )
     session_cls = pretend.call_recorder(lambda bind: session_obj)
     monkeypatch.setattr(db, "Session", session_cls)
 
@@ -200,6 +203,57 @@ def test_create_session(monkeypatch, read_only, tx_status):
         connection.connection.rollback.calls == [pretend.call()]
 
 
+@pytest.mark.parametrize(
+    "admin_flag, is_superuser, doom_calls",
+    [
+        (None, True, []),
+        (None, False, []),
+        (pretend.stub(enabled=False), True, []),
+        (pretend.stub(enabled=False), False, []),
+        (pretend.stub(enabled=True, description='flag description'), True, []),
+        (
+            pretend.stub(enabled=True, description='flag description'),
+            False,
+            [pretend.call()],
+        ),
+    ],
+)
+def test_create_session_read_only_mode(
+        admin_flag, is_superuser, doom_calls, monkeypatch):
+    get = pretend.call_recorder(lambda *a: admin_flag)
+    session_obj = pretend.stub(
+        close=lambda: None,
+        query=lambda *a: pretend.stub(get=get),
+    )
+    session_cls = pretend.call_recorder(lambda bind: session_obj)
+    monkeypatch.setattr(db, "Session", session_cls)
+
+    register = pretend.call_recorder(lambda session, transaction_manager: None)
+    monkeypatch.setattr(zope.sqlalchemy, "register", register)
+
+    connection = pretend.stub(
+        connection=pretend.stub(
+            get_transaction_status=lambda: pretend.stub(),
+            set_session=lambda **kw: None,
+            rollback=lambda: None,
+        ),
+        info={},
+        close=lambda: None,
+    )
+    engine = pretend.stub(connect=pretend.call_recorder(lambda: connection))
+    request = pretend.stub(
+        registry={"sqlalchemy.engine": engine},
+        tm=pretend.stub(doom=pretend.call_recorder(lambda: None)),
+        read_only=False,
+        add_finished_callback=lambda callback: None,
+        user=pretend.stub(is_superuser=is_superuser),
+    )
+
+    assert _create_session(request) is session_obj
+    assert get.calls == [pretend.call('read-only')]
+    assert request.tm.doom.calls == doom_calls
+
+
 @pytest.mark.parametrize(
     ("predicates", "expected"),
     [

warehouse/accounts/views.py

@@ -35,7 +35,6 @@
     send_password_reset_email, send_email_verification_email,
 )
 from warehouse.packaging.models import Project, Release
-from warehouse.utils.admin_flags import AdminFlag
 from warehouse.utils.http import is_safe_url
 
 
@@ -233,7 +232,7 @@ def register(request, _form_class=RegistrationForm):
     if request.method == "POST" and request.POST.get('confirm_form'):
         return HTTPSeeOther(request.route_path("index"))
 
-    if AdminFlag.is_enabled(request.db, 'disallow-new-user-registration'):
+    if request.flags.enabled('disallow-new-user-registration'):
         request.session.flash(
             ("New User Registration Temporarily Disabled "
              "See https://pypi.org/help#admin-intervention for details"),

warehouse/admin/__init__.py

@@ -10,10 +10,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from warehouse.accounts.views import login, logout
-
 
 def includeme(config):
+    from warehouse.accounts.views import login, logout
+
     # Setup Jinja2 Rendering for the Admin application
     config.add_jinja2_search_path("templates", name=".html")
 
@@ -23,6 +23,9 @@ def includeme(config):
     # Add our routes
     config.include(".routes")
 
+    # Add our flags
+    config.include(".flags")
+
     config.add_view(
         login,
         route_name="admin.login",