diff --git a/django_forms_workflows/form_builder_views.py b/django_forms_workflows/form_builder_views.py index b7c3157..95db09e 100644 --- a/django_forms_workflows/form_builder_views.py +++ b/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: diff --git a/poetry.lock b/poetry.lock index 66656e0..05c26d3 100644 --- a/poetry.lock +++ b/poetry.lock @@ -2636,15 +2636,15 @@ files = [ [[package]] name = "pyasn1" -version = "0.6.4" +version = "0.6.5" description = "Pure-Python implementation of ASN.1 types and DER/BER/CER codecs (X.208)" optional = true python-versions = ">=3.8" groups = ["main"] markers = "extra == \"ldap\" or extra == \"all\" or extra == \"gmail\"" files = [ - {file = "pyasn1-0.6.4-py3-none-any.whl", hash = "sha256:deda9277cfd454080ec40b207fb6df82206a3a2688735233cdcd8d3d565f088b"}, - {file = "pyasn1-0.6.4.tar.gz", hash = "sha256:9c447d8431c947fe4c8febc4ed9e760bc29011a5b01e5c74b67025bd9fb8ce81"}, + {file = "pyasn1-0.6.5-py3-none-any.whl", hash = "sha256:4017b0fce125096b8573b39c989fdb1a4b19cf761e683398184550fea0e66587"}, + {file = "pyasn1-0.6.5.tar.gz", hash = "sha256:6b11616a6bcf3a40e635f082bb232fa57b5e0dd7f517bd7ddd69ec3794508a47"}, ] [[package]] diff --git a/tests/test_builders.py b/tests/test_builders.py index 9ac7ec2..e4a3fc5 100644 --- a/tests/test_builders.py +++ b/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}/")