Add disallow deletion AdminFlag

See pypi#3218

tests/unit/manage/test_views.py

@@ -27,6 +27,7 @@
 import warehouse.utils.otp as otp
 
 from warehouse.accounts.interfaces import IPasswordBreachedService, IUserService
+from warehouse.admin.flags import AdminFlagValue
 from warehouse.macaroons.interfaces import IMacaroonService
 from warehouse.manage import views
 from warehouse.packaging.models import (
@@ -2014,6 +2015,7 @@ def test_delete_project_no_confirm(self):
         project = pretend.stub(normalized_name="foo")
         request = pretend.stub(
             POST={},
+            flags=pretend.stub(enabled=pretend.call_recorder(lambda *a: False)),
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
             route_path=lambda *a, **kw: "/foo/bar/",
         )
@@ -2023,6 +2025,9 @@ def test_delete_project_no_confirm(self):
             assert exc.value.status_code == 303
             assert exc.value.headers["Location"] == "/foo/bar/"
 
+        assert request.flags.enabled.calls == [
+            pretend.call(AdminFlagValue.DISALLOW_DELETION)
+        ]
         assert request.session.flash.calls == [
             pretend.call("Confirm the request", queue="error")
         ]
@@ -2031,6 +2036,7 @@ def test_delete_project_wrong_confirm(self):
         project = pretend.stub(normalized_name="foo")
         request = pretend.stub(
             POST={"confirm_project_name": "bar"},
+            flags=pretend.stub(enabled=pretend.call_recorder(lambda *a: False)),
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
             route_path=lambda *a, **kw: "/foo/bar/",
         )
@@ -2040,13 +2046,46 @@ def test_delete_project_wrong_confirm(self):
             assert exc.value.status_code == 303
             assert exc.value.headers["Location"] == "/foo/bar/"
 
+        assert request.flags.enabled.calls == [
+            pretend.call(AdminFlagValue.DISALLOW_DELETION)
+        ]
         assert request.session.flash.calls == [
             pretend.call(
                 "Could not delete project - 'bar' is not the same as 'foo'",
                 queue="error",
             )
         ]
 
+    def test_delete_project_disallow_deletion(self):
+        project = pretend.stub(name="foo", normalized_name="foo")
+        request = pretend.stub(
+            flags=pretend.stub(enabled=pretend.call_recorder(lambda *a: True)),
+            route_path=pretend.call_recorder(lambda *a, **kw: "/the-redirect"),
+            session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
+        )
+
+        result = views.delete_project(project, request)
+        assert isinstance(result, HTTPSeeOther)
+        assert result.headers["Location"] == "/the-redirect"
+
+        assert request.flags.enabled.calls == [
+            pretend.call(AdminFlagValue.DISALLOW_DELETION)
+        ]
+
+        assert request.session.flash.calls == [
+            pretend.call(
+                (
+                    "Project deletion temporarily disabled. "
+                    "See https://pypi.org/help#admin-intervention for details."
+                ),
+                queue="error",
+            )
+        ]
+
+        assert request.route_path.calls == [
+            pretend.call("manage.project.settings", project_name="foo")
+        ]
+
     def test_delete_project(self, db_request):
         project = ProjectFactory.create(name="foo")
 
@@ -2159,6 +2198,7 @@ def test_manage_project_releases(self, db_request):
             filename=f"foobar-{release.version}.tar.gz",
             packagetype="sdist",
         )
+        db_request.flags = pretend.stub(enabled=pretend.call_recorder(lambda *a: False))
 
         assert views.manage_project_releases(project, db_request) == {
             "project": project,
@@ -2182,6 +2222,48 @@ def test_manage_project_release(self):
             "files": files,
         }
 
+    def test_delete_project_release_disallow_deletion(self, monkeypatch):
+        release = pretend.stub(
+            version="1.2.3",
+            canonical_version="1.2.3",
+            project=pretend.stub(
+                name="foobar", record_event=pretend.call_recorder(lambda *a, **kw: None)
+            ),
+        )
+        request = pretend.stub(
+            flags=pretend.stub(enabled=pretend.call_recorder(lambda *a: True)),
+            method="POST",
+            route_path=pretend.call_recorder(lambda *a, **kw: "/the-redirect"),
+            session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
+        )
+        view = views.ManageProjectRelease(release, request)
+
+        result = view.delete_project_release()
+        assert isinstance(result, HTTPSeeOther)
+        assert result.headers["Location"] == "/the-redirect"
+
+        assert request.flags.enabled.calls == [
+            pretend.call(AdminFlagValue.DISALLOW_DELETION)
+        ]
+
+        assert request.session.flash.calls == [
+            pretend.call(
+                (
+                    "Project deletion temporarily disabled. "
+                    "See https://pypi.org/help#admin-intervention for details."
+                ),
+                queue="error",
+            )
+        ]
+
+        assert request.route_path.calls == [
+            pretend.call(
+                "manage.project.release",
+                project_name=release.project.name,
+                version=release.version,
+            )
+        ]
+
     def test_delete_project_release(self, monkeypatch):
         release = pretend.stub(
             version="1.2.3",
@@ -2197,6 +2279,7 @@ def test_delete_project_release(self, monkeypatch):
                 delete=pretend.call_recorder(lambda a: None),
                 add=pretend.call_recorder(lambda a: None),
             ),
+            flags=pretend.stub(enabled=pretend.call_recorder(lambda *a: False)),
             route_path=pretend.call_recorder(lambda *a, **kw: "/the-redirect"),
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
             user=pretend.stub(username=pretend.stub()),
@@ -2215,6 +2298,9 @@ def test_delete_project_release(self, monkeypatch):
 
         assert request.db.delete.calls == [pretend.call(release)]
         assert request.db.add.calls == [pretend.call(journal_obj)]
+        assert request.flags.enabled.calls == [
+            pretend.call(AdminFlagValue.DISALLOW_DELETION)
+        ]
         assert journal_cls.calls == [
             pretend.call(
                 name=release.project.name,
@@ -2247,6 +2333,7 @@ def test_delete_project_release_no_confirm(self):
             POST={"confirm_version": ""},
             method="POST",
             db=pretend.stub(delete=pretend.call_recorder(lambda a: None)),
+            flags=pretend.stub(enabled=pretend.call_recorder(lambda *a: False)),
             route_path=pretend.call_recorder(lambda *a, **kw: "/the-redirect"),
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
         )
@@ -2261,6 +2348,9 @@ def test_delete_project_release_no_confirm(self):
         assert request.session.flash.calls == [
             pretend.call("Confirm the request", queue="error")
         ]
+        assert request.flags.enabled.calls == [
+            pretend.call(AdminFlagValue.DISALLOW_DELETION)
+        ]
         assert request.route_path.calls == [
             pretend.call(
                 "manage.project.release",
@@ -2275,6 +2365,7 @@ def test_delete_project_release_bad_confirm(self):
             POST={"confirm_version": "invalid"},
             method="POST",
             db=pretend.stub(delete=pretend.call_recorder(lambda a: None)),
+            flags=pretend.stub(enabled=pretend.call_recorder(lambda *a: False)),
             route_path=pretend.call_recorder(lambda *a, **kw: "/the-redirect"),
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
         )
@@ -2301,6 +2392,42 @@ def test_delete_project_release_bad_confirm(self):
             )
         ]
 
+    def test_delete_project_release_file_disallow_deletion(self):
+        release = pretend.stub(version="1.2.3", project=pretend.stub(name="foobar"))
+        request = pretend.stub(
+            method="POST",
+            flags=pretend.stub(enabled=pretend.call_recorder(lambda *a: True)),
+            route_path=pretend.call_recorder(lambda *a, **kw: "/the-redirect"),
+            session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
+        )
+        view = views.ManageProjectRelease(release, request)
+
+        result = view.delete_project_release_file()
+
+        assert isinstance(result, HTTPSeeOther)
+        assert result.headers["Location"] == "/the-redirect"
+
+        assert request.flags.enabled.calls == [
+            pretend.call(AdminFlagValue.DISALLOW_DELETION)
+        ]
+
+        assert request.session.flash.calls == [
+            pretend.call(
+                (
+                    "Project deletion temporarily disabled. "
+                    "See https://pypi.org/help#admin-intervention for details."
+                ),
+                queue="error",
+            )
+        ]
+        assert request.route_path.calls == [
+            pretend.call(
+                "manage.project.release",
+                project_name=release.project.name,
+                version=release.version,
+            )
+        ]
+
     def test_delete_project_release_file(self, db_request):
         user = UserFactory.create()
 
@@ -2359,6 +2486,7 @@ def test_delete_project_release_file_no_confirm(self):
             POST={"confirm_project_name": ""},
             method="POST",
             db=pretend.stub(delete=pretend.call_recorder(lambda a: None)),
+            flags=pretend.stub(enabled=pretend.call_recorder(lambda *a: False)),
             route_path=pretend.call_recorder(lambda *a, **kw: "/the-redirect"),
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
         )
@@ -2370,6 +2498,9 @@ def test_delete_project_release_file_no_confirm(self):
         assert result.headers["Location"] == "/the-redirect"
 
         assert request.db.delete.calls == []
+        assert request.flags.enabled.calls == [
+            pretend.call(AdminFlagValue.DISALLOW_DELETION)
+        ]
         assert request.session.flash.calls == [
             pretend.call("Confirm the request", queue="error")
         ]
@@ -2396,6 +2527,7 @@ def no_result_found():
                 filter=lambda *a: pretend.stub(one=no_result_found)
             ),
         )
+        db_request.flags = pretend.stub(enabled=pretend.call_recorder(lambda *a: False))
         db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")
         db_request.session = pretend.stub(
             flash=pretend.call_recorder(lambda *a, **kw: None)
@@ -2409,6 +2541,9 @@ def no_result_found():
         assert result.headers["Location"] == "/the-redirect"
 
         assert db_request.db.delete.calls == []
+        assert db_request.flags.enabled.calls == [
+            pretend.call(AdminFlagValue.DISALLOW_DELETION)
+        ]
         assert db_request.session.flash.calls == [
             pretend.call("Could not find file", queue="error")
         ]

warehouse/admin/flags.py

@@ -16,6 +16,7 @@
 
 
 class AdminFlagValue:
+    DISALLOW_DELETION = "disallow-deletion"
     DISALLOW_NEW_PROJECT_REGISTRATION = "disallow-new-project-registration"
     DISALLOW_NEW_USER_REGISTRATION = "disallow-new-user-registration"
     READ_ONLY = "read-only"

warehouse/manage/views.py

@@ -30,6 +30,7 @@
 from warehouse.accounts.interfaces import IPasswordBreachedService, IUserService
 from warehouse.accounts.models import Email, User
 from warehouse.accounts.views import logout
+from warehouse.admin.flags import AdminFlagValue
 from warehouse.email import (
     send_account_deletion_email,
     send_added_as_collaborator_email,
@@ -783,6 +784,18 @@ def manage_project_settings(project, request):
     permission="manage:project",
 )
 def delete_project(project, request):
+    if request.flags.enabled(AdminFlagValue.DISALLOW_DELETION):
+        request.session.flash(
+            (
+                "Project deletion temporarily disabled. "
+                "See https://pypi.org/help#admin-intervention for details."
+            ),
+            queue="error",
+        )
+        return HTTPSeeOther(
+            request.route_path("manage.project.settings", project_name=project.name)
+        )
+
     confirm_project(project, request, fail_route="manage.project.settings")
     remove_project(project, request)
 
@@ -872,6 +885,22 @@ def manage_project_release(self):
 
     @view_config(request_method="POST", request_param=["confirm_version"])
     def delete_project_release(self):
+        if self.request.flags.enabled(AdminFlagValue.DISALLOW_DELETION):
+            self.request.session.flash(
+                (
+                    "Project deletion temporarily disabled. "
+                    "See https://pypi.org/help#admin-intervention for details."
+                ),
+                queue="error",
+            )
+            return HTTPSeeOther(
+                self.request.route_path(
+                    "manage.project.release",
+                    project_name=self.release.project.name,
+                    version=self.release.version,
+                )
+            )
+
         version = self.request.POST.get("confirm_version")
         if not version:
             self.request.session.flash("Confirm the request", queue="error")
@@ -942,6 +971,13 @@ def _error(message):
                 )
             )
 
+        if self.request.flags.enabled(AdminFlagValue.DISALLOW_DELETION):
+            message = (
+                "Project deletion temporarily disabled. "
+                "See https://pypi.org/help#admin-intervention for details."
+            )
+            return _error(message)
+
         project_name = self.request.POST.get("confirm_project_name")
 
         if not project_name:

warehouse/migrations/versions/8650482fb903_add_disallow_deletion_adminflag.py

@@ -0,0 +1,41 @@
+# 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.
+"""
+Add disallow-deletion AdminFlag
+
+Revision ID: 8650482fb903
+Revises: 34b18e18775c
+Create Date: 2019-08-23 13:29:17.110252
+"""
+
+from alembic import op
+
+revision = "8650482fb903"
+down_revision = "34b18e18775c"
+
+
+def upgrade():
+    op.execute(
+        """
+        INSERT INTO admin_flags(id, description, enabled, notify)
+        VALUES (
+            'disallow-deletion',
+            'Disallow ALL project and release deletions',
+            FALSE,
+            FALSE
+        )
+    """
+    )
+
+
+def downgrade():
+    op.execute("DELETE FROM admin_flags WHERE id = 'disallow-deletion'")