Implement initial rollout of PEP 715

.github/workflows/ci.yml

@@ -93,7 +93,7 @@ jobs:
         with:
           path: |
               .mypy_cache
-          key: ${{ runner.os }}-${{ env.pythonLocation }}-${{ hashFiles('requirements.txt', 'requirements/*.txt') }}
+          key: ${{ runner.os }}-mypy-${{ env.pythonLocation }}-${{ hashFiles('requirements.txt', 'requirements/*.txt') }}
       - name: Restore built Python environment from deps
         uses: actions/cache/restore@v3
         with:

docs/blog/posts/2023-06-26-deprecate-egg-uploads.md

@@ -0,0 +1,37 @@
+---
+title: Deprecation of bdist_egg uploads to PyPI
+description: PyPI will stop accepting .egg uploads August 1, 2023.
+author: Ee Durbin
+publish_date: 2023-06-26
+date: "2023-06-26 00:00"
+tags:
+  - deprecation
+---
+
+[PEP 715](https://peps.python.org/pep-0715/), deprecating `bdist_egg`/`.egg`
+uploads to PyPI has been
+[accepted](https://discuss.python.org/t/pep-715-disabling-bdist-egg-distribution-uploads-on-pypi/27610/13).
+We'll begin the process of implementing this today.
+
+Please note that this does **NOT** remove any existing uploaded eggs from PyPI.
+
+The deprecation timeline is as follows:
+
+- Today, June 26, 2023: All maintainers of projects which have uploaded one or
+  more eggs since January 1, 2023 will recieve a one-time email informing them
+  of this change.
+- Today, June 26, 2023: Each upload of an egg to PyPI will result in a notice
+  being sent to all Owners and Maintainers for the project.
+- August 1, 2023: Uploads of eggs will be [**rejected**](https://www.youtube.com/watch?v=XNyUALnj8V0) by PyPI.
+
+You can read more detailed rationale in [PEP 715](https://peps.python.org/pep-0715/#rationale).
+Thanks to contributor [William Woodruff](https://blog.yossarian.net) for his
+work to author and propose PEP 715, as well as support the rollout of the
+implementation.
+
+---
+
+_Ee Durbin is the Director of Infrastructure at
+the Python Software Foundation.
+They have been contributing to keeping PyPI online, available, and
+secure since 2013._

tests/unit/cli/test_projects.py

@@ -0,0 +1,33 @@
+# 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 pretend
+
+from warehouse.cli import projects
+from warehouse.packaging.tasks import send_pep_715_notices
+
+
+def test_notify_pep_715(cli):
+    request = pretend.stub()
+    task = pretend.stub(
+        get_request=pretend.call_recorder(lambda: request),
+        run=pretend.call_recorder(lambda r: None),
+    )
+    config = pretend.stub(task=pretend.call_recorder(lambda f: task))
+
+    cli.invoke(projects.notify_pep_715, obj=config)
+    assert config.task.calls == [
+        pretend.call(send_pep_715_notices),
+        pretend.call(send_pep_715_notices),
+    ]
+    assert task.get_request.calls == [pretend.call()]
+    assert task.run.calls == [pretend.call(request)]

tests/unit/email/test_init.py

@@ -5822,3 +5822,92 @@ def test_trusted_publisher_emails(
                 },
             )
         ]
+
+
+class TestPEP715Emails:
+    @pytest.mark.parametrize(
+        "fn, template_name",
+        [
+            (email.send_egg_uploads_deprecated_email, "egg-uploads-deprecated"),
+            (
+                email.send_egg_uploads_deprecated_initial_email,
+                "egg-uploads-deprecated-initial-notice",
+            ),
+        ],
+    )
+    def test_pep_715_emails(
+        self, pyramid_request, pyramid_config, monkeypatch, fn, template_name
+    ):
+        stub_user = pretend.stub(
+            id="id",
+            username="username",
+            name="",
+            email="[email protected]",
+            primary_email=pretend.stub(email="[email protected]", verified=True),
+        )
+        subject_renderer = pyramid_config.testing_add_renderer(
+            f"email/{ template_name }/subject.txt"
+        )
+        subject_renderer.string_response = "Email Subject"
+        body_renderer = pyramid_config.testing_add_renderer(
+            f"email/{ template_name }/body.txt"
+        )
+        body_renderer.string_response = "Email Body"
+        html_renderer = pyramid_config.testing_add_renderer(
+            f"email/{ template_name }/body.html"
+        )
+        html_renderer.string_response = "Email HTML Body"
+
+        send_email = pretend.stub(
+            delay=pretend.call_recorder(lambda *args, **kwargs: None)
+        )
+        pyramid_request.task = pretend.call_recorder(lambda *args, **kwargs: send_email)
+        monkeypatch.setattr(email, "send_email", send_email)
+
+        pyramid_request.db = pretend.stub(
+            query=lambda a: pretend.stub(
+                filter=lambda *a: pretend.stub(
+                    one=lambda: pretend.stub(user_id=stub_user.id)
+                )
+            ),
+        )
+        pyramid_request.user = stub_user
+        pyramid_request.registry.settings = {"mail.sender": "[email protected]"}
+
+        project_name = "test_project"
+        result = fn(
+            pyramid_request,
+            stub_user,
+            project_name=project_name,
+        )
+
+        assert result == {
+            "project_name": project_name,
+        }
+        subject_renderer.assert_()
+        body_renderer.assert_(project_name=project_name)
+        html_renderer.assert_(project_name=project_name)
+        assert pyramid_request.task.calls == [pretend.call(send_email)]
+        assert send_email.delay.calls == [
+            pretend.call(
+                f"{stub_user.username} <{stub_user.email}>",
+                {
+                    "subject": "Email Subject",
+                    "body_text": "Email Body",
+                    "body_html": (
+                        "<html>\n<head></head>\n"
+                        "<body><p>Email HTML Body</p></body>\n</html>\n"
+                    ),
+                },
+                {
+                    "tag": "account:email:sent",
+                    "user_id": stub_user.id,
+                    "additional": {
+                        "from_": "[email protected]",
+                        "to": stub_user.email,
+                        "subject": "Email Subject",
+                        "redact_ip": False,
+                    },
+                },
+            )
+        ]

tests/unit/forklift/test_legacy.py

@@ -33,6 +33,14 @@
 from wtforms.form import Form
 from wtforms.validators import ValidationError
 
+from tests.common.db.organizations import (
+    OrganizationFactory,
+    OrganizationProjectFactory,
+    OrganizationRoleFactory,
+    TeamFactory,
+    TeamProjectRoleFactory,
+    TeamRoleFactory,
+)
 from warehouse.admin.flags import AdminFlag, AdminFlagValue
 from warehouse.classifiers.models import Classifier
 from warehouse.errors import BasicAuthTwoFactorEnabled
@@ -79,6 +87,13 @@ def _get_whl_testdata(name="fake_package", version="1.0"):
     return temp_f.getvalue()
 
 
+def _get_egg_testdata():
+    temp_f = io.BytesIO()
+    with zipfile.ZipFile(file=temp_f, mode="w") as zfp:
+        zfp.writestr("fake_package/PKG-INFO", "Fake metadata")
+    return temp_f.getvalue()
+
+
 def _storage_hash(data):
     return hashlib.blake2b(data, digest_size=256 // 8).hexdigest()
 
@@ -94,6 +109,12 @@ def _storage_hash(data):
 _TAR_BZ2_PKG_STORAGE_HASH = _storage_hash(_TAR_BZ2_PKG_TESTDATA)
 
 
+_EGG_PKG_TESTDATA = _get_egg_testdata()
+_EGG_PKG_MD5 = hashlib.md5(_EGG_PKG_TESTDATA).hexdigest()
+_EGG_PKG_SHA256 = hashlib.sha256(_EGG_PKG_TESTDATA).hexdigest()
+_EGG_PKG_STORAGE_HASH = _storage_hash(_EGG_PKG_TESTDATA)
+
+
 class TestExcWithMessage:
     def test_exc_with_message(self):
         exc = legacy._exc_with_message(HTTPBadRequest, "My Test Message.")
@@ -3677,6 +3698,111 @@ def test_fails_without_user(self, pyramid_config, pyramid_request):
             "See /the/help/url/ for more information."
         )
 
+    def test_egg_upload_sends_pep_715_notice(
+        self, pyramid_config, db_request, metrics, monkeypatch
+    ):
+        user = UserFactory.create()
+        EmailFactory.create(user=user)
+        project = ProjectFactory.create()
+        RoleFactory.create(user=user, project=project)
+
+        pyramid_config.testing_securitypolicy(identity=user)
+        db_request.user = user
+        db_request.user_agent = "warehouse-tests/6.6.6"
+        db_request.POST = MultiDict(
+            {
+                "metadata_version": "1.2",
+                "name": project.name,
+                "version": "1.0.0",
+                "summary": "This is my summary!",
+                "filetype": "bdist_egg",
+                "pyversion": "2.7",
+                "md5_digest": _EGG_PKG_MD5,
+                "content": pretend.stub(
+                    filename="{}-{}.egg".format(project.name, "1.0.0"),
+                    file=io.BytesIO(_EGG_PKG_TESTDATA),
+                    type="application/zip",
+                ),
+            }
+        )
+
+        send_email = pretend.call_recorder(lambda *a, **kw: None)
+        monkeypatch.setattr(legacy, "send_egg_uploads_deprecated_email", send_email)
+
+        storage_service = pretend.stub(store=lambda path, filepath, meta: None)
+        db_request.find_service = lambda svc, name=None, context=None: {
+            IFileStorage: storage_service,
+            IMetricsService: metrics,
+        }.get(svc)
+
+        resp = legacy.file_upload(db_request)
+
+        assert resp.status_code == 200
+        assert send_email.calls == [
+            pretend.call(db_request, user, project_name=project.name)
+        ]
+
+    def test_egg_upload_sends_pep_715_notice_org_roles(
+        self, pyramid_config, db_request, metrics, monkeypatch
+    ):
+        user = UserFactory.create()
+        EmailFactory.create(user=user)
+        project = ProjectFactory.create()
+        RoleFactory.create(user=user, project=project)
+
+        org = OrganizationFactory()
+        OrganizationProjectFactory(organization=org, project=project)
+        org_owner = UserFactory.create()
+        OrganizationRoleFactory.create(user=org_owner, organization=org)
+
+        org_member = UserFactory.create()
+        OrganizationRoleFactory.create(
+            user=org_member, organization=org, role_name="Member"
+        )
+        team = TeamFactory.create(organization=org)
+        TeamRoleFactory.create(team=team, user=org_member)
+        # Duplicate the role directly on the project to ensure only one email
+        RoleFactory.create(user=org_member, project=project, role_name="Maintainer")
+        TeamProjectRoleFactory.create(project=project, team=team)
+
+        pyramid_config.testing_securitypolicy(identity=user)
+        db_request.user = user
+        db_request.user_agent = "warehouse-tests/6.6.6"
+        db_request.POST = MultiDict(
+            {
+                "metadata_version": "1.2",
+                "name": project.name,
+                "version": "1.0.0",
+                "summary": "This is my summary!",
+                "filetype": "bdist_egg",
+                "pyversion": "2.7",
+                "md5_digest": _EGG_PKG_MD5,
+                "content": pretend.stub(
+                    filename="{}-{}.egg".format(project.name, "1.0.0"),
+                    file=io.BytesIO(_EGG_PKG_TESTDATA),
+                    type="application/zip",
+                ),
+            }
+        )
+
+        send_email = pretend.call_recorder(lambda *a, **kw: None)
+        monkeypatch.setattr(legacy, "send_egg_uploads_deprecated_email", send_email)
+
+        storage_service = pretend.stub(store=lambda path, filepath, meta: None)
+        db_request.find_service = lambda svc, name=None, context=None: {
+            IFileStorage: storage_service,
+            IMetricsService: metrics,
+        }.get(svc)
+
+        resp = legacy.file_upload(db_request)
+
+        assert resp.status_code == 200
+        assert set(send_email.calls) == {
+            pretend.call(db_request, user, project_name=project.name),
+            pretend.call(db_request, org_owner, project_name=project.name),
+            pretend.call(db_request, org_member, project_name=project.name),
+        }
+
 
 @pytest.mark.parametrize("status", [True, False])
 def test_legacy_purge(monkeypatch, status):