Fix EmailFormProcessor ignoring block settings

backend/news/+testing-profile.tests

@@ -0,0 +1 @@
+Added a dedicated GenericSetup testing profile to configure mail and captcha settings for the test suite. @ericof

backend/news/19.bugfix

@@ -0,0 +1 @@
+Fixed the form email processor ignoring the block's configured settings: the selected mail template, admin info, mail header/footer and sender/recipient addresses are now honored when a form is submitted. @ericof

backend/news/19.internal

@@ -0,0 +1 @@
+Moved the mail templates setting to a ``JSONField`` (``schemaform.mail_templates_json``), deprecating ``schemaform.mail_templates``, with an upgrade step that migrates existing templates. @ericof

backend/src/plone/formblock/interfaces.py

@@ -1,9 +1,12 @@
 from plone.dexterity.content import DexterityContent
+from plone.schema import JSONField
+from typing import Any
+from typing import TypedDict
 from zope import schema
 from zope.interface import Attribute
 from zope.interface import Interface
+from zope.publisher.http import HTTPRequest
 from zope.publisher.interfaces.browser import IDefaultBrowserLayer
-from ZPublisher.BaseRequest import BaseRequest
 
 import dataclasses
 
@@ -46,12 +49,65 @@ def verify(data):
         """
 
 
+class FormSchema(TypedDict):
+    """A JSON Schema definition for a form block."""
+
+    properties: dict[str, dict]
+    fieldsets: list[dict]
+    required: list[str]
+
+
+class RichTextField(TypedDict):
+    data: str
+
+
+class SchemaFormBlock(TypedDict):
+    """A form block schema definition."""
+
+    schema: FormSchema
+    title: str
+    description: str
+    submit_label: str
+    show_cancel: bool
+    cancel_label: str
+    forward_user_to: str
+    success: str
+    thankyou: str
+    captcha: str
+    send: bool
+    recipients: str
+    bcc: str
+    admin_info: str
+    sender: str
+    sender_name: str
+    subject: str
+    mail_header: RichTextField
+    mail_footer: RichTextField
+    store: bool
+    data_wipe: int
+    send_confirmation: bool
+    confirmation_recipients: str
+    fixed_attachment: Any
+    mail_template: str
+
+
+@dataclasses.dataclass
+class AddressesFromBlock:
+    """Addresses extracted from a form block, with variable substitution."""
+
+    sender: str
+    admin_recipients: str
+    confirmation_recipients: str
+    bcc: str
+
+
 @dataclasses.dataclass
 class FormSubmissionContext:
     context: DexterityContent
-    block: dict
+    block: SchemaFormBlock
     form_data: dict
-    request: BaseRequest
+    request: HTTPRequest
+    addresses: AddressesFromBlock
 
     def get_records(self) -> list:
         """
@@ -109,8 +165,12 @@ def __call__():
 
 class IFormSettings(Interface):
     mail_templates = schema.Dict(
-        title="Email templates",
+        title="Email templates (Deprecated)",
         key_type=schema.TextLine(),
         value_type=schema.Text(),
+        default={},
+    )
+    mail_templates_json = JSONField(
+        title="Email templates",
         default={"default": DEFAULT_TEMPLATE},
     )

backend/src/plone/formblock/processors/email.py

@@ -1,41 +1,26 @@
 from email.message import EmailMessage
-from email.utils import formataddr
 from plone import api
-from plone.base.interfaces.controlpanel import IMailSchema
-from plone.formblock import _
 from plone.formblock.interfaces import FormSubmissionContext
 from plone.formblock.interfaces import IFormSubmissionProcessor
-from plone.formblock.utils.email import create_message
-from plone.registry.registry import Registry
+from plone.formblock.interfaces import SchemaFormBlock
+from plone.formblock.utils import email as utils
 from Products.MailHost.MailHost import MailHost
 from zExceptions import BadRequest
 from zope.component import adapter
-from zope.component import getMultiAdapter
 from zope.interface import implementer
 
-import re
 
+def template_vars_from_block(block: SchemaFormBlock, form_data: dict) -> dict:
+    """Extract template variables from a form block.
 
-def is_mailhost_configured() -> bool:
-    smtp_host = api.portal.get_registry_record("plone.smtp_host")
-    email_from_address = api.portal.get_registry_record("plone.email_from_address")
-    return bool(smtp_host and email_from_address)
-
-
-def substitute_variables(
-    value: str, context: dict | None = None, form_data: dict | None = None
-) -> str:
-    if context is None and form_data is not None:
-        context = form_data
-    elif context is None:
-        context = {}
-
-    def replace(match):
-        name = match.group(1)
-        return context.get(name, "")
-
-    pattern = r"\$\{([^}]+)\}"
-    return re.sub(pattern, replace, value)
+    Substituting variables if necessary.
+    """
+    attrs = ("mail_header", "mail_footer")
+    template_vars = {}
+    for attr in attrs:
+        value = block.get(attr, {}).get("data", "")
+        template_vars[attr] = utils.substitute_variables(value, context=form_data)
+    return template_vars
 
 
 @implementer(IFormSubmissionProcessor)
@@ -50,39 +35,19 @@ class EmailFormProcessor:
     def __init__(self, context: FormSubmissionContext):
         self.context = context.context
         self.request = context.request
+        self.addresses = context.addresses
         self.block = context.block
+        self.template = utils.get_template_from_block(self.block)
         self.form_data = context.form_data
         self.records = context.get_records()
         self.attachments = context.get_attachments()
         self.portal_transforms = api.portal.get_tool(name="portal_transforms")
-        self.templates: dict = (
-            api.portal.get_registry_record("schemaform.mail_templates") or {}
-        )
-
-        registry: Registry = api.portal.get_tool("portal_registry")
-        self.mail_settings = registry.forInterface(IMailSchema, prefix="plone")
-        self.charset: str = registry.get("plone.email_charset", "utf-8")
 
     def _body_as_plain_text(self, html: str) -> str:
         tool = self.portal_transforms
         text = tool.convertTo("text/plain", html, mimetype="text/html").getData()
         return text.strip()
 
-    def get_sender(self) -> str:
-        sender = self.block.get("sender", "")
-        sender = (
-            substitute_variables(sender, context=self.form_data)
-            or self.mail_settings.email_from_address
-        )
-
-        sender_name = self.block.get("sender_name", "")
-        sender_name = (
-            substitute_variables(sender_name, context=self.form_data)
-            or self.mail_settings.email_from_name
-        )
-
-        return formataddr((sender_name, sender))
-
     def get_subject(self) -> str:
         subject = self.block.get("subject")
         schema_properties = self.block.get("schema", {}).get("properties", {})
@@ -91,72 +56,24 @@ def get_subject(self) -> str:
                 subject = "${subject}"
             else:
                 subject = self.block.get("title") or "Form Submission"
-        subject = substitute_variables(subject, context=self.form_data)
+        subject = utils.substitute_variables(subject, context=self.form_data)
         return subject
 
     def get_value(self, field_id, default=None):
         return self.form_data.get(field_id, default)
 
-    def get_bcc(self) -> str:
-        bcc: str = self.block.get("bcc", "")
-        bcc = substitute_variables(bcc, context=self.form_data)
-        return bcc or ""
-
-    def get_confirmation_recipients(self) -> str:
-        confirmation_recipients = self.block.get("confirmation_recipients", "")
-        confirmation_recipients = substitute_variables(
-            confirmation_recipients, context=self.form_data
-        )
-        return confirmation_recipients
-
-    def prepare_message(self, admin=False):  # noqa: C901
-        template_name = self.block.get("email_template", "default")
-        admin_info = self.block.get("admin_info", "")
-        properties = self.block.get("schema").get("properties")
-        template = self.templates.get(template_name, "")
-        plone = getMultiAdapter((self.context, self.request), name="plone")
-        template_vars = {
-            "mail_header": substitute_variables(
-                self.block.get("mail_header", {}).get("data", ""),
-                context=self.form_data,
-            ),
-            "mail_footer": substitute_variables(
-                self.block.get("mail_footer", {}).get("data", ""),
-                context=self.form_data,
-            ),
-        }
-        form_fields = ""
-        if admin:
-            form_fields += admin_info.replace("\n", "<br/>") + "<br/><br/>"
-
-        def format_property(factory, value):
-            if factory == "label_boolean_field" or factory == "termsAccepted":
-                if value == True:  # noqa: E712
-                    return self.context.translate(
-                        _("Yes"),
-                        context=self.request,
-                    )
-                else:
-                    return self.context.translate(
-                        _("No"),
-                        context=self.request,
-                    )
-            elif factory == "checkbox_group":
-                if isinstance(value, list):
-                    return "<br/>".join(value)
-                else:
-                    return str(value)
-            elif factory == "label_date_field":
-                return plone.toLocalizedTime(value)
-            elif factory == "label_datetime_field":
-                return plone.toLocalizedTime(value, True)
-            else:
-                return str(value)
-
+    def prepare_message(self, admin: bool = False) -> str:
+        block = self.block
+        form_data = self.form_data
+        template = self.template
+        template_vars = template_vars_from_block(block, form_data)
+        admin_info = block.get("admin_info", "")
+        properties = block.get("schema", {}).get("properties")
+        form_fields = admin_info.replace("\n", "<br/>") + "<br/><br/>" if admin else ""
         form_fields += "<table>\n"
         for record in self.records:
             factory = properties[record["field_id"]].get("factory", "")
-            value = format_property(factory, record["value"])
+            value = utils.format_property(factory, record["value"])
             template_vars[record["field_id"]] = value
 
             if factory == "hidden" and not admin:
@@ -169,7 +86,7 @@ def format_property(factory, value):
             )
         form_fields += "\n</table>\n"
         template_vars["form_fields"] = form_fields
-        message = substitute_variables(template, template_vars)
+        message = utils.substitute_variables(template, template_vars)
         return message
 
     def send_mail(self, msg: EmailMessage, charset: str) -> None:
@@ -178,7 +95,9 @@ def send_mail(self, msg: EmailMessage, charset: str) -> None:
         # by default (False) exceptions are handled by MailHost and we can't catch them.
         host.send(msg, charset=charset, immediate=True)
 
-    def _prepare_msg_admin(self, mfrom: str, mto: str, subject: str) -> EmailMessage:
+    def _prepare_msg_admin(
+        self, mfrom: str, mto: str, subject: str, bcc: str
+    ) -> EmailMessage:
         body = self.prepare_message(True)
         body_text = self._body_as_plain_text(body)
 
@@ -189,20 +108,20 @@ def _prepare_msg_admin(self, mfrom: str, mto: str, subject: str) -> EmailMessage
             if header_value:
                 headers[header] = header_value
 
-        return create_message(
+        return utils.create_message(
             mfrom=mfrom,
             mto=mto,
             subject=subject,
             body=body,
             body_txt=body_text,
             reply_to=mfrom,
-            bcc=self.get_bcc(),
+            bcc=bcc,
             headers=headers,
             attachments=self.attachments,
         )
 
     def _prepare_msg_confirmation(
-        self, mfrom: str, mto: str, subject: str
+        self, mfrom: str, mto: str, subject: str, bcc: str = ""
     ) -> EmailMessage:
         body = self.prepare_message()
         body_text = self._body_as_plain_text(body)
@@ -211,38 +130,43 @@ def _prepare_msg_confirmation(
         if self.block.get("fixed_attachment"):
             attachments["fixed_attachment"] = self.block["fixed_attachment"]
 
-        return create_message(
+        return utils.create_message(
             mfrom=mfrom,
             mto=mto,
             subject=subject,
             body=body,
             body_txt=body_text,
             reply_to="",
+            bcc=bcc,
             headers={},
             attachments=attachments,
         )
 
     def __call__(self) -> None:
         queued: list[EmailMessage] = []
-        send_to_admin = bool(self.block.get("send"))
-        send_confirmation = bool(self.block.get("send_confirmation"))
+        block = self.block
+        addresses = self.addresses
+        send_to_admin = bool(block.get("send"))
+        send_confirmation = bool(block.get("send_confirmation"))
         if not send_to_admin and not send_confirmation:
             return
 
-        if not is_mailhost_configured():
+        if not utils.is_mailhost_configured():
             raise BadRequest("MailHost is not configured.")
 
         subject = self.get_subject()
-        mfrom = self.get_sender()
+        mfrom = addresses.sender
 
         if send_to_admin:
-            mto = self.block.get("recipients", self.mail_settings.email_from_address)
-            msg = self._prepare_msg_admin(mfrom, mto, subject)
+            mto = addresses.admin_recipients
+            bcc = addresses.bcc
+            msg = self._prepare_msg_admin(mfrom, mto, subject, bcc)
             queued.append(msg)
 
-        if send_confirmation and (mto := self.get_confirmation_recipients()):
+        if send_confirmation and (mto := addresses.confirmation_recipients.strip()):
             msg = self._prepare_msg_confirmation(mfrom, mto, subject)
             queued.append(msg)
 
+        charset = api.portal.get_registry_record("plone.email_charset", default="utf-8")
         for msg in queued:
-            self.send_mail(msg=msg, charset=self.charset)
+            self.send_mail(msg=msg, charset=charset)

backend/src/plone/formblock/profiles/default/metadata.xml

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="utf-8"?>
 <metadata>
-  <version>1000</version>
+  <version>1001</version>
   <dependencies>
     <dependency>profile-plone.restapi:default</dependency>
     <dependency>profile-plone.volto:default</dependency>