Potential fixes for 2 code quality findings

django_forms_workflows/form_builder_views.py

@@ -153,6 +153,7 @@ def form_builder_clone(request, form_id):
             # Copy group permissions
             cloned_form.submit_groups.set(original_form.submit_groups.all())
             cloned_form.view_groups.set(original_form.view_groups.all())
+            cloned_form.reviewer_groups.set(original_form.reviewer_groups.all())
             cloned_form.admin_groups.set(original_form.admin_groups.all())
 
             # Clone workflows, stages, and stage approval groups
@@ -403,6 +404,9 @@ def form_builder_save(request):
             }
         )
 
+    except FormBuilderPayloadError as exc:
+        # Raised inside the atomic block above, so the partial save is rolled back.
+        return JsonResponse({"success": False, "error": exc.client_message}, status=400)
     except json.JSONDecodeError:
         return JsonResponse(
             {"success": False, "error": "Invalid JSON data"}, status=400
@@ -414,6 +418,18 @@ def form_builder_save(request):
         )
 
 
+class FormBuilderPayloadError(ValueError):
+    """The builder payload is invalid.
+
+    ``client_message`` is the only text returned to the client, so exception
+    details can never reach the response by accident.
+    """
+
+    def __init__(self, client_message):
+        super().__init__(client_message)
+        self.client_message = client_message
+
+
 def save_form_definition_from_builder_data(data, user, form_definition=None):
     """
     Create or update a FormDefinition and its FormFields from form builder
@@ -527,6 +543,13 @@ def save_form_definition_from_builder_data(data, user, form_definition=None):
     updated_field_ids = set()
     field_id_mapping = {}  # Map old IDs to new IDs for frontend update
 
+    # Stages a field on this form may be attached to
+    allowed_workflow_stage_ids = set(
+        WorkflowStage.objects.filter(
+            workflow__form_definition=form_definition
+        ).values_list("id", flat=True)
+    )
+
     # Create or update fields
     for field_data in fields_data:
         field_id = field_data.get("id")
@@ -585,15 +608,32 @@ def save_form_definition_from_builder_data(data, user, form_definition=None):
 
         # Add workflow stage FK (for staged approval workflows)
         wf_stage_id = field_data.get("workflow_stage_id")
-        if wf_stage_id is not None:
-            field_props["workflow_stage_id"] = int(wf_stage_id) if wf_stage_id else None
+        if wf_stage_id:
+            try:
+                wf_stage_id = int(wf_stage_id)
+            except (TypeError, ValueError) as exc:
+                raise FormBuilderPayloadError(
+                    "Invalid workflow_stage_id for field "
+                    f"'{field_data.get('field_name', '')}'"
+                ) from exc
+            if wf_stage_id not in allowed_workflow_stage_ids:
+                raise FormBuilderPayloadError(
+                    f"workflow_stage_id {wf_stage_id} is not valid for this form"
+                )
+            field_props["workflow_stage_id"] = wf_stage_id
         else:
             field_props["workflow_stage_id"] = None
 
         # Create or update field
         if field_id and isinstance(field_id, int):
-            # Update existing field
-            FormField.objects.filter(id=field_id).update(**field_props)
+            # Update existing field, scoped to current form_definition
+            updated_count = FormField.objects.filter(
+                id=field_id, form_definition=form_definition
+            ).update(**field_props)
+            if updated_count == 0:
+                raise FormBuilderPayloadError(
+                    f"Invalid field id {field_id} for this form"
+                )
             updated_field_ids.add(field_id)
             field_id_mapping[str(old_id)] = field_id
         else:

tests/test_builders.py

@@ -13,10 +13,12 @@
 )
 from django_forms_workflows.models import (
     FormDefinition,
+    FormField,
     NotificationRule,
     PostSubmissionAction,
     StageApprovalGroup,
     WorkflowDefinition,
+    WorkflowStage,
 )
 from django_forms_workflows.workflow_builder_views import (
     convert_visual_to_workflow,
@@ -1388,6 +1390,146 @@ def test_keeps_fields_present_in_the_payload_by_id(self, user, form_with_fields)
 # ── Form builder view tests ─────────────────────────────────────────────────
 
 
+class TestFormBuilderSavePayloadValidation:
+    """form_builder_save must keep fields and stages scoped to the form being
+    saved, and reject bad references without persisting any of the save."""
+
+    @staticmethod
+    def _save(client, form, fields):
+        return client.post(
+            _fb_url("builder/api/save/"),
+            data=json.dumps(
+                {
+                    "id": form.id,
+                    "name": form.name,
+                    "slug": form.slug,
+                    "fields": fields,
+                }
+            ),
+            content_type="application/json",
+        )
+
+    @staticmethod
+    def _field_payload(field, **overrides):
+        return {
+            "id": field.id,
+            "field_name": field.field_name,
+            "field_label": "Changed Label",
+            "field_type": field.field_type,
+            "order": field.order,
+            **overrides,
+        }
+
+    @pytest.fixture
+    def other_form(self, db):
+        return FormDefinition.objects.create(
+            name="Other Form", slug="other-form", description="", instructions=""
+        )
+
+    @pytest.fixture
+    def field(self, form_with_fields):
+        return form_with_fields.fields.get(field_name="full_name")
+
+    def test_assigns_stage_from_same_form(
+        self, client, superuser, form_with_fields, field
+    ):
+        wf = WorkflowDefinition.objects.create(form_definition=form_with_fields)
+        stage = WorkflowStage.objects.create(workflow=wf, name="Review", order=1)
+        client.force_login(superuser)
+
+        resp = self._save(
+            client,
+            form_with_fields,
+            [self._field_payload(field, workflow_stage_id=str(stage.id))],
+        )
+
+        assert resp.status_code == 200
+        field.refresh_from_db()
+        assert field.workflow_stage_id == stage.id
+
+    def test_clears_stage_when_blank(self, client, superuser, form_with_fields, field):
+        wf = WorkflowDefinition.objects.create(form_definition=form_with_fields)
+        field.workflow_stage = WorkflowStage.objects.create(
+            workflow=wf, name="Review", order=1
+        )
+        field.save()
+        client.force_login(superuser)
+
+        resp = self._save(
+            client, form_with_fields, [self._field_payload(field, workflow_stage_id="")]
+        )
+
+        assert resp.status_code == 200
+        field.refresh_from_db()
+        assert field.workflow_stage_id is None
+
+    def test_rejects_stage_from_another_form_and_rolls_back(
+        self, client, superuser, form_with_fields, field, other_form
+    ):
+        wf = WorkflowDefinition.objects.create(form_definition=other_form)
+        foreign_stage = WorkflowStage.objects.create(
+            workflow=wf, name="Foreign", order=1
+        )
+        original_version = form_with_fields.version
+        original_label = field.field_label
+        client.force_login(superuser)
+
+        resp = self._save(
+            client,
+            form_with_fields,
+            [self._field_payload(field, workflow_stage_id=foreign_stage.id)],
+        )
+
+        assert resp.status_code == 400
+        assert resp.json()["success"] is False
+        assert "not valid for this form" in resp.json()["error"]
+        form_with_fields.refresh_from_db()
+        field.refresh_from_db()
+        assert form_with_fields.version == original_version
+        assert field.field_label == original_label
+        assert field.workflow_stage_id is None
+
+    def test_rejects_malformed_stage_id(
+        self, client, superuser, form_with_fields, field
+    ):
+        client.force_login(superuser)
+
+        resp = self._save(
+            client,
+            form_with_fields,
+            [self._field_payload(field, workflow_stage_id="not-a-number")],
+        )
+
+        assert resp.status_code == 400
+        assert "Invalid workflow_stage_id" in resp.json()["error"]
+
+    def test_rejects_field_from_another_form_and_rolls_back(
+        self, client, superuser, form_with_fields, field, other_form
+    ):
+        foreign_field = FormField.objects.create(
+            form_definition=other_form,
+            field_name="foreign",
+            field_label="Foreign Label",
+            field_type="text",
+            order=1,
+        )
+        original_field_count = form_with_fields.fields.count()
+        client.force_login(superuser)
+
+        resp = self._save(
+            client, form_with_fields, [self._field_payload(foreign_field)]
+        )
+
+        assert resp.status_code == 400
+        assert "Invalid field id" in resp.json()["error"]
+        foreign_field.refresh_from_db()
+        assert foreign_field.form_definition_id == other_form.id
+        assert foreign_field.field_label == "Foreign Label"
+        # The removal of this form's own fields must be rolled back as well.
+        assert form_with_fields.fields.count() == original_field_count
+
+
+@pytest.mark.django_db
 class TestFormBuilderViews:
     """Integration tests for the form builder HTTP endpoints."""
 
@@ -1477,6 +1619,27 @@ def test_clone_form(self, client, superuser, form_with_fields):
         assert cloned.id != form_with_fields.id
         assert cloned.fields.count() == form_with_fields.fields.count()
 
+    def test_clone_copies_all_permission_groups(
+        self, client, superuser, form_with_fields, approval_group
+    ):
+        for groups in (
+            form_with_fields.submit_groups,
+            form_with_fields.view_groups,
+            form_with_fields.reviewer_groups,
+            form_with_fields.admin_groups,
+        ):
+            groups.set([approval_group])
+        client.force_login(superuser)
+
+        resp = client.post(_fb_url(f"builder/api/clone/{form_with_fields.id}/"))
+
+        assert resp.status_code == 200
+        cloned = FormDefinition.objects.get(id=resp.json()["form_id"])
+        assert list(cloned.submit_groups.all()) == [approval_group]
+        assert list(cloned.view_groups.all()) == [approval_group]
+        assert list(cloned.reviewer_groups.all()) == [approval_group]
+        assert list(cloned.admin_groups.all()) == [approval_group]
+
     def test_clone_requires_staff(self, client, user, form_with_fields):
         client.force_login(user)
         url = _fb_url(f"builder/api/clone/{form_with_fields.id}/")