feat(admin): add organization-level file size and total size limits

tests/unit/admin/views/test_projects.py

@@ -112,6 +112,29 @@ def test_non_normalized_name(self, db_request):
         with pytest.raises(HTTPMovedPermanently):
             views.project_detail(project, db_request)
 
+    def test_with_organization(self, db_request):
+        from ....common.db.organizations import (
+            OrganizationFactory,
+            OrganizationProjectFactory,
+        )
+
+        organization = OrganizationFactory.create(
+            upload_limit=150 * views.ONE_MIB,
+            total_size_limit=100 * views.ONE_GIB,
+        )
+        org_project = OrganizationProjectFactory.create(organization=organization)
+        project = org_project.project
+        project.upload_limit = 50 * views.ONE_MIB
+        project.total_size_limit = 50 * views.ONE_GIB
+        db_request.matchdict["project_name"] = str(project.normalized_name)
+        result = views.project_detail(project, db_request)
+
+        assert result["project"] == project
+        assert project.organization == organization
+        # Verify that the organization limits are accessible through the project
+        assert project.organization.upload_limit == 150 * views.ONE_MIB
+        assert project.organization.total_size_limit == 100 * views.ONE_GIB
+
 
 class TestReleaseDetail:
     def test_gets_release(self, db_request):

tests/unit/forklift/test_legacy.py

@@ -5796,6 +5796,151 @@ def test_upload_for_company_organization_owned_project_fails_without_subscriptio
             "Please contact [email protected]."
         )
 
+    def test_upload_with_organization_file_size_limit_succeeds(
+        self, pyramid_config, db_request, monkeypatch
+    ):
+        organization = OrganizationFactory.create(
+            orgtype="Company", upload_limit=120 * (1024**2)  # 120 MiB
+        )
+        user = UserFactory.create(with_verified_primary_email=True)
+        OrganizationRoleFactory.create(organization=organization, user=user)
+        OrganizationStripeSubscriptionFactory.create(organization=organization)
+        project = OrganizationProjectFactory.create(
+            organization=organization,
+            project__upload_limit=70 * (1024**2),  # 70 MiB
+        ).project
+        version = "1.0.0"
+
+        filename = (
+            f"{project.normalized_name.replace('-', '_')}-{version}-py3-none-any.whl"
+        )
+        # Create a small file representing a 110 MiB file
+        filebody = _get_whl_testdata(
+            name=project.normalized_name.replace("-", "_"), version=version
+        )
+
+        @pretend.call_recorder
+        def storage_service_store(path, file_path, *, meta):
+            with open(file_path, "rb") as fp:
+                if file_path.endswith(".metadata"):
+                    assert fp.read() == b"Fake metadata"
+                else:
+                    assert fp.read() == filebody
+
+        storage_service = pretend.stub(store=storage_service_store)
+
+        db_request.find_service = pretend.call_recorder(
+            lambda svc, name=None, context=None: {
+                IFileStorage: storage_service,
+            }.get(svc)
+        )
+
+        monkeypatch.setattr(
+            legacy, "_is_valid_dist_file", lambda *a, **kw: (True, None)
+        )
+
+        # Mock the file size to be 110 MiB
+        class MockFieldStorage:
+            def __init__(self, data, filename):
+                self.file = io.BytesIO(data)
+                self.filename = filename
+                self.type = "application/x-wheel+zip"
+                self.length = 110 * (1024**2)  # 110 MiB
+
+        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 a summary",
+                "filetype": "bdist_wheel",
+                "pyversion": "py3",
+                "md5_digest": hashlib.md5(filebody).hexdigest(),
+                "content": MockFieldStorage(filebody, filename),
+            }
+        )
+
+        resp = legacy.file_upload(db_request)
+
+        assert resp.status_code == 200
+
+    def test_upload_with_organization_total_size_limit_succeeds(
+        self, pyramid_config, db_request, monkeypatch
+    ):
+        organization = OrganizationFactory.create(
+            orgtype="Company", total_size_limit=100 * (1024**3)  # 100 GiB
+        )
+        user = UserFactory.create(with_verified_primary_email=True)
+        OrganizationRoleFactory.create(organization=organization, user=user)
+        OrganizationStripeSubscriptionFactory.create(organization=organization)
+
+        # Create project with 90 GiB already used, project limit of 50 GiB
+        project = OrganizationProjectFactory.create(
+            organization=organization,
+            project__total_size_limit=50 * (1024**3),  # 50 GiB
+            project__total_size=90 * (1024**3),  # 90 GiB already used
+        ).project
+        version = "1.0.0"
+
+        filename = (
+            f"{project.normalized_name.replace('-', '_')}-{version}-py3-none-any.whl"
+        )
+        # Create a small file representing a 5 GiB file
+        filebody = _get_whl_testdata(
+            name=project.normalized_name.replace("-", "_"), version=version
+        )
+
+        @pretend.call_recorder
+        def storage_service_store(path, file_path, *, meta):
+            with open(file_path, "rb") as fp:
+                if file_path.endswith(".metadata"):
+                    assert fp.read() == b"Fake metadata"
+                else:
+                    assert fp.read() == filebody
+
+        storage_service = pretend.stub(store=storage_service_store)
+
+        db_request.find_service = pretend.call_recorder(
+            lambda svc, name=None, context=None: {
+                IFileStorage: storage_service,
+            }.get(svc)
+        )
+
+        monkeypatch.setattr(
+            legacy, "_is_valid_dist_file", lambda *a, **kw: (True, None)
+        )
+
+        # Mock the file size to be 5 GiB
+        class MockFieldStorage:
+            def __init__(self, data, filename):
+                self.file = io.BytesIO(data)
+                self.filename = filename
+                self.type = "application/x-wheel+zip"
+                self.length = 5 * (1024**3)  # 5 GiB
+
+        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 a summary",
+                "filetype": "bdist_wheel",
+                "pyversion": "py3",
+                "md5_digest": hashlib.md5(filebody).hexdigest(),
+                "content": MockFieldStorage(filebody, filename),
+            }
+        )
+
+        resp = legacy.file_upload(db_request)
+
+        assert resp.status_code == 200
+
     def test_upload_for_company_organization_owned_project_suceeds_with_subscription(
         self, pyramid_config, db_request, monkeypatch
     ):

warehouse/admin/routes.py

@@ -24,6 +24,16 @@ def includeme(config):
         "/admin/organizations/{organization_id}/rename/",
         domain=warehouse,
     )
+    config.add_route(
+        "admin.organization.set_upload_limit",
+        "/admin/organizations/{organization_id}/set_upload_limit/",
+        domain=warehouse,
+    )
+    config.add_route(
+        "admin.organization.set_total_size_limit",
+        "/admin/organizations/{organization_id}/set_total_size_limit/",
+        domain=warehouse,
+    )
     config.add_route(
         "admin.organization.add_role",
         "/admin/organizations/{organization_id}/add_role/",

warehouse/admin/templates/admin/organizations/detail.html

@@ -166,6 +166,66 @@ <h3 class="card-title">Organization Details</h3>
         </div>
       </div>
 
+      <div class="card" id="organization-limits">
+        <div class="card-header with-border">
+          <h3 class="card-title">Organization Limits</h3>
+        </div>
+        <div class="card-body">
+          <table class="table table-hover">
+            <tbody>
+              <tr>
+                <form method="POST" action="{{ request.route_path('admin.organization.set_upload_limit', organization_id=organization.id) }}">
+                  <input name="csrf_token" type="hidden" value="{{ request.session.get_csrf_token() }}">
+                  <td>Upload limit</td>
+                  <td>
+                    {% if organization.upload_limit %}
+                    {{ organization.upload_limit|filesizeformat(binary=True) }}
+                    {% else %}
+                    Default ({{ MAX_FILESIZE|filesizeformat(binary=True) }})
+                    {% endif %}
+                  </td>
+                  <td>
+                    {% if organization.upload_limit %}
+                    {% set upload_limit_value = organization.upload_limit / ONE_MIB %}
+                    {% else %}
+                    {% set upload_limit_value = '' %}
+                    {% endif %}
+                    <input type="number" name="upload_limit" id="uploadLimit" min="{{ MAX_FILESIZE / ONE_MIB }}" max="{{ UPLOAD_LIMIT_CAP / ONE_MIB }}" step=1 value="{{ upload_limit_value|int }}"> MiB
+                  </td>
+                  <td>
+                    <button type="submit" class="btn btn-primary btn-sm" {{ 'disabled' if not request.has_permission(Permissions.AdminOrganizationsWrite) }}>Submit</button>
+                  </td>
+                </form>
+              </tr>
+              <tr>
+                <form method="POST" action="{{ request.route_path('admin.organization.set_total_size_limit', organization_id=organization.id) }}">
+                  <input name="csrf_token" type="hidden" value="{{ request.session.get_csrf_token() }}">
+                  <td>Total size limit</td>
+                  <td>
+                    {% if organization.total_size_limit %}
+                    {{ organization.total_size_limit|filesizeformat(binary=True) }}
+                    {% else %}
+                    Default ({{ MAX_PROJECT_SIZE|filesizeformat(binary=True) }})
+                    {% endif %}
+                  </td>
+                  <td>
+                    {% if organization.total_size_limit %}
+                    {% set total_size_limit_value = organization.total_size_limit // ONE_GIB %}
+                    {% else %}
+                    {% set total_size_limit_value = '' %}
+                    {% endif %}
+                    <input type="number" name="total_size_limit" id="totalSizeLimit" min="{{ MAX_PROJECT_SIZE / ONE_GIB }}" step=1 value="{{ total_size_limit_value|int }}"> GiB
+                  </td>
+                  <td>
+                    <button type="submit" class="btn btn-primary btn-sm" {{ 'disabled' if not request.has_permission(Permissions.AdminOrganizationsWrite) }}>Submit</button>
+                  </td>
+                </form>
+              </tr>
+            </tbody>
+          </table>
+        </div>
+      </div>
+
       <div class="card">
         <div class="card-header with-border">
           <h3 class="card-title">Projects</h3>