feat: limit number of environments per project (#3875)

api/environments/permissions/permissions.py

@@ -42,15 +42,31 @@ def has_permission(self, request, view):  # type: ignore[no-untyped-def]
 
             try:
                 project = Project.objects.get(id=project_id)
-                return request.user.has_project_permission(CREATE_ENVIRONMENT, project)
             except Project.DoesNotExist:
                 return False
 
+            if (
+                project.environments.count() >= project.max_environments_allowed
+                and getattr(request, "is_e2e", False) is not True
+            ):
+                raise exceptions.ValidationError(
+                    "The project has reached the maximum number of allowed environments."
+                )
+
+            return request.user.has_project_permission(CREATE_ENVIRONMENT, project)
+
         # return true as all users can list and obj permissions will be handled later
         return True
 
     def has_object_permission(self, request, view, obj):  # type: ignore[no-untyped-def]
         if view.action == "clone":
+            if (
+                obj.project.environments.count() >= obj.project.max_environments_allowed
+                and getattr(request, "is_e2e", False) is not True
+            ):
+                raise exceptions.ValidationError(
+                    "The project has reached the maximum number of allowed environments."
+                )
             return request.user.has_project_permission(CREATE_ENVIRONMENT, obj.project)
         elif view.action in ("get_document", "retrieve", "trait_keys"):
             return request.user.has_environment_permission(VIEW_ENVIRONMENT, obj)

api/projects/admin.py

@@ -71,6 +71,7 @@ class ProjectAdmin(admin.ModelAdmin):  # type: ignore[type-arg]
         "max_segment_overrides_allowed",
         "edge_v2_migration_status",
         "edge_v2_migration_read_capacity_budget",
+        "max_environments_allowed",
     )
 
     @admin.action(

api/projects/migrations/0029_add_max_environments_allowed.py

@@ -0,0 +1,18 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("projects", "0028_add_enforce_feature_owners_to_project"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="project",
+            name="max_environments_allowed",
+            field=models.IntegerField(
+                default=100,
+                help_text="Max environments allowed for this project",
+            ),
+        ),
+    ]

api/projects/models.py

@@ -87,6 +87,10 @@ class Project(LifecycleModelMixin, SoftDeleteExportableModel):  # type: ignore[d
         default=100,
         help_text="Max segments overrides allowed for any (one) environment within this project",
     )
+    max_environments_allowed = models.IntegerField(
+        default=100,
+        help_text="Max environments allowed for this project",
+    )
     edge_v2_migration_status = models.CharField(
         max_length=50,
         choices=EdgeV2MigrationStatus.choices,
@@ -125,6 +129,7 @@ def is_too_large(self) -> bool:
         return (
             self.features.count() > self.max_features_allowed
             or self.live_segment_count() > self.max_segments_allowed
+            or self.environments.count() > self.max_environments_allowed
             or self.environments.annotate(
                 segment_override_count=Count("feature_segments")
             )

api/projects/serializers.py

@@ -45,6 +45,7 @@ class Meta:
             "edge_v2_migration_status",
             "minimum_change_request_approvals",
             "enforce_feature_owners",
+            "max_environments_allowed",
         )
         read_only_fields = (
             "enable_dynamo_db",
@@ -128,6 +129,7 @@ class Meta(ProjectListSerializer.Meta):
             "max_segments_allowed",
             "max_features_allowed",
             "max_segment_overrides_allowed",
+            "max_environments_allowed",
             "total_features",
             "total_segments",
         )

api/tests/unit/environments/permissions/test_unit_environments_permissions.py

@@ -5,6 +5,7 @@
     CREATE_ENVIRONMENT,
 )
 from pytest_mock import MockerFixture
+from rest_framework.exceptions import ValidationError
 
 from environments.identities.models import Identity
 from environments.models import Environment
@@ -399,3 +400,109 @@ def test_nested_environment_permissions__regular_user_destroys__returns_false(
 
     # Then
     assert result is False
+
+
+def test_environment_permissions__create_at_limit__raises_validation_error(
+    admin_user: FFAdminUser,
+    project: Project,
+    environment: Environment,
+) -> None:
+    # Given
+    project.max_environments_allowed = 1
+    project.save()
+
+    mock_view.action = "create"
+    mock_view.detail = False
+    mock_request.user = admin_user
+    mock_request.data = {"project": project.id, "name": "Another environment"}
+    mock_request.is_e2e = False
+
+    # When / Then
+    with pytest.raises(ValidationError, match="maximum number of allowed environments"):
+        environment_permissions.has_permission(mock_request, mock_view)  # type: ignore[no-untyped-call]
+
+
+def test_environment_permissions__create_below_limit__returns_true(
+    admin_user: FFAdminUser,
+    project: Project,
+    environment: Environment,
+) -> None:
+    # Given
+    project.max_environments_allowed = 100
+    project.save()
+
+    mock_view.action = "create"
+    mock_view.detail = False
+    mock_request.user = admin_user
+    mock_request.data = {"project": project.id, "name": "Another environment"}
+
+    # When
+    result = environment_permissions.has_permission(mock_request, mock_view)  # type: ignore[no-untyped-call]
+
+    # Then
+    assert result is True
+
+
+def test_environment_permissions__clone_at_limit__raises_validation_error(
+    admin_user: FFAdminUser,
+    project: Project,
+    environment: Environment,
+) -> None:
+    # Given
+    project.max_environments_allowed = 1
+    project.save()
+
+    mock_view.action = "clone"
+    mock_view.detail = True
+    mock_request.user = admin_user
+    mock_request.is_e2e = False
+
+    # When / Then
+    with pytest.raises(ValidationError, match="maximum number of allowed environments"):
+        environment_permissions.has_object_permission(  # type: ignore[no-untyped-call]
+            mock_request, mock_view, environment
+        )
+
+
+def test_environment_permissions__clone_below_limit__returns_true(
+    admin_user: FFAdminUser,
+    project: Project,
+    environment: Environment,
+) -> None:
+    # Given
+    project.max_environments_allowed = 100
+    project.save()
+
+    mock_view.action = "clone"
+    mock_view.detail = True
+    mock_request.user = admin_user
+
+    # When
+    result = environment_permissions.has_object_permission(  # type: ignore[no-untyped-call]
+        mock_request, mock_view, environment
+    )
+
+    # Then
+    assert result is True
+
+
+def test_environment_permissions__create_at_limit_with_increased_limit__returns_true(
+    admin_user: FFAdminUser,
+    project: Project,
+    environment: Environment,
+) -> None:
+    """Grandfathering: projects with a higher limit can still create environments."""
+    # Given
+    project.max_environments_allowed = 200
+    project.save()
+
+    mock_view.action = "create"
+    mock_view.detail = False
+    mock_request.user = admin_user
+    mock_request.data = {"project": project.id, "name": "Another environment"}
+
+    # When
+    result = environment_permissions.has_permission(mock_request, mock_view)  # type: ignore[no-untyped-call]
+
+    # Then
+    assert result is True

api/tests/unit/projects/test_unit_projects_models.py

@@ -6,6 +6,7 @@
 from django.utils import timezone
 from pytest_django.fixtures import SettingsWrapper
 
+from environments.models import Environment
 from organisations.models import Organisation
 from projects.models import EdgeV2MigrationStatus, Project
 from segments.models import Segment
@@ -209,3 +210,38 @@ def test_create_project__edge_enabled__sets_edge_v2_migration_complete(
 
     # Then
     assert project.edge_v2_migration_status == EdgeV2MigrationStatus.COMPLETE
+
+
+def test_is_too_large__environments_exceed_limit__returns_true(
+    project: Project,
+) -> None:
+    # Given
+
+    project.max_environments_allowed = 1
+    project.save()
+
+    Environment.objects.create(name="Env 1", project=project)
+    Environment.objects.create(name="Env 2", project=project)
+
+    # When
+    result = project.is_too_large
+
+    # Then
+    assert result is True
+
+
+def test_is_too_large__environments_within_limit__returns_false(
+    project: Project,
+) -> None:
+    # Given
+
+    project.max_environments_allowed = 100
+    project.save()
+
+    Environment.objects.create(name="Env 1", project=project)
+
+    # When
+    result = project.is_too_large
+
+    # Then
+    assert result is False