Implement debug_mode for email delivery (vibe-kanban f32190a0)

The EmailDeliveryConfig in api/core/workflow/nodes/human\_input/entities.py has a field `debug_mode`. When this field is set to `True` in `node_data`, the test run (InvokeFrom.DEBUGGER) and the delivery test should only sent test emails to the current user, instead of sending to the specified recipients.

Please implement this logic, write correspond test cases to ensure that the logic works as expected.
This commit is contained in:
QuantumGhost
2026-01-15 10:21:32 +08:00
parent ea90746ed7
commit e50d849913
5 changed files with 166 additions and 7 deletions

View File

@ -62,6 +62,14 @@ class EmailDeliveryConfig(BaseModel):
# Body is the content of email, it may contain the speical placeholder `{{#url#}}`, which
# represent the url to submit the form.
body: str
debug_mode: bool = False
def with_debug_recipient(self, user_id: str) -> "EmailDeliveryConfig":
if not user_id:
debug_recipients = EmailRecipients(whole_workspace=False, items=[])
return self.model_copy(update={"recipients": debug_recipients})
debug_recipients = EmailRecipients(whole_workspace=False, items=[MemberRecipient(user_id=user_id)])
return self.model_copy(update={"recipients": debug_recipients})
@classmethod
def replace_url_placeholder(cls, body: str, url: str | None) -> str:
@ -98,6 +106,22 @@ class EmailDeliveryMethod(_DeliveryMethodBase):
DeliveryChannelConfig = Annotated[WebAppDeliveryMethod | EmailDeliveryMethod, Field(discriminator="type")]
def apply_debug_email_recipient(
method: DeliveryChannelConfig,
*,
enabled: bool,
user_id: str,
) -> DeliveryChannelConfig:
if not enabled:
return method
if not isinstance(method, EmailDeliveryMethod):
return method
if not method.config.debug_mode:
return method
debug_config = method.config.with_debug_recipient(user_id or "")
return method.model_copy(update={"config": debug_config})
class FormInputPlaceholder(BaseModel):
"""Placeholder configuration for form inputs."""

View File

@ -20,7 +20,7 @@ from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
from extensions.ext_database import db
from libs.datetime_utils import naive_utc_now
from .entities import DeliveryChannelConfig, HumanInputNodeData
from .entities import DeliveryChannelConfig, HumanInputNodeData, apply_debug_email_recipient
from .enums import DeliveryMethodType, HumanInputFormStatus, PlaceholderType
if TYPE_CHECKING:
@ -169,8 +169,15 @@ class HumanInputNode(Node[HumanInputNodeData]):
def _effective_delivery_methods(self) -> Sequence[DeliveryChannelConfig]:
enabled_methods = [method for method in self._node_data.delivery_methods if method.enabled]
if self.invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE}:
return [method for method in enabled_methods if method.type != DeliveryMethodType.WEBAPP]
return enabled_methods
enabled_methods = [method for method in enabled_methods if method.type != DeliveryMethodType.WEBAPP]
return [
apply_debug_email_recipient(
method,
enabled=self.invoke_from == InvokeFrom.DEBUGGER,
user_id=self.user_id or "",
)
for method in enabled_methods
]
def _human_input_required_event(self, form_entity: HumanInputFormEntity) -> HumanInputRequired:
node_data = self._node_data