0001-workflows-allow-attachments-in-emails-8274.patch
| tests/test_workflows.py | ||
|---|---|---|
| 1 | 1 |
import datetime |
| 2 |
import os |
|
| 2 | 3 |
import pytest |
| 3 | 4 |
import shutil |
| 4 | 5 |
import StringIO |
| ... | ... | |
| 765 | 766 |
assert emails.count() == 1 |
| 766 | 767 |
assert emails.get('foobar').get('from') == 'foobar@localhost'
|
| 767 | 768 | |
| 769 | ||
| 770 |
def test_email_attachments(pub, emails): |
|
| 771 |
formdef = FormDef() |
|
| 772 |
formdef.name = 'baz' |
|
| 773 |
formdef.fields = [ |
|
| 774 |
FileField(id='3', label='File', type='file', varname='file'), |
|
| 775 |
] |
|
| 776 |
formdef.store() |
|
| 777 | ||
| 778 |
upload = PicklableUpload('test.jpeg', 'image/jpeg')
|
|
| 779 |
jpg = open(os.path.join(os.path.dirname(__file__), 'image-with-gps-data.jpeg')).read() |
|
| 780 |
upload.receive([jpg]) |
|
| 781 |
formdata = formdef.data_class()() |
|
| 782 |
formdata.data = {'3': upload}
|
|
| 783 |
formdata.just_created() |
|
| 784 |
formdata.store() |
|
| 785 |
pub.substitutions.feed(formdata) |
|
| 786 | ||
| 787 |
sendmail = SendmailWorkflowStatusItem() |
|
| 788 |
sendmail.subject = 'foobar' |
|
| 789 |
sendmail.body = '<p>force html</p>' |
|
| 790 |
sendmail.to = ['to@example.net'] |
|
| 791 |
sendmail.attachments = ['form_var_file_raw'] |
|
| 792 |
sendmail.perform(formdata) |
|
| 793 |
get_response().process_after_jobs() |
|
| 794 |
assert emails.count() == 1 |
|
| 795 |
assert emails.emails['foobar']['msg'].is_multipart() |
|
| 796 |
assert emails.emails['foobar']['msg'].get_content_subtype() == 'mixed' |
|
| 797 |
assert emails.emails['foobar']['msg'].get_payload()[0].get_content_type() == 'text/html' |
|
| 798 |
assert emails.emails['foobar']['msg'].get_payload()[1].get_content_type() == 'image/jpeg' |
|
| 799 | ||
| 800 |
# build a backoffice field |
|
| 801 |
Workflow.wipe() |
|
| 802 |
FormDef.wipe() |
|
| 803 |
wf = Workflow(name='email with attachments') |
|
| 804 |
wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(wf) |
|
| 805 |
wf.backoffice_fields_formdef.fields = [ |
|
| 806 |
FileField(id='bo1', label='bo field 1', type='file', varname='backoffice_file1'), |
|
| 807 |
FileField(id='bo2', label='bo field 2', type='file', varname='backoffice_file2'), |
|
| 808 |
] |
|
| 809 |
st1 = wf.add_status('Status1')
|
|
| 810 |
wf.store() |
|
| 811 |
formdef = FormDef() |
|
| 812 |
formdef.name = 'baz' |
|
| 813 |
formdef.fields = [ |
|
| 814 |
FileField(id='1', label='File', type='file', varname='frontoffice_file'), |
|
| 815 |
] |
|
| 816 |
formdef.workflow_id = wf.id |
|
| 817 |
formdef.store() |
|
| 818 |
formdata = formdef.data_class()() |
|
| 819 |
formdata.data = {'1': upload}
|
|
| 820 |
formdata.just_created() |
|
| 821 |
formdata.store() |
|
| 822 |
pub.substitutions.feed(formdata) |
|
| 823 |
# store file in backoffice field form_fbo1 / form_var_backoffice_file_raw |
|
| 824 |
setbo = SetBackofficeFieldsWorkflowStatusItem() |
|
| 825 |
setbo.parent = st1 |
|
| 826 |
setbo.fields = [{'field_id': 'bo1', 'value': '=form_var_frontoffice_file_raw'}]
|
|
| 827 |
setbo.perform(formdata) |
|
| 828 | ||
| 829 |
emails.empty() |
|
| 830 |
sendmail.attachments = ['form_fbo1'] |
|
| 831 |
sendmail.perform(formdata) |
|
| 832 |
get_response().process_after_jobs() |
|
| 833 |
assert emails.count() == 1 |
|
| 834 |
assert emails.emails['foobar']['msg'].is_multipart() |
|
| 835 |
assert emails.emails['foobar']['msg'].get_content_subtype() == 'mixed' |
|
| 836 |
assert emails.emails['foobar']['msg'].get_payload()[0].get_content_type() == 'text/html' |
|
| 837 |
assert emails.emails['foobar']['msg'].get_payload()[1].get_content_type() == 'image/jpeg' |
|
| 838 | ||
| 839 |
emails.empty() |
|
| 840 |
sendmail.attachments = ['form_var_backoffice_file1_raw'] |
|
| 841 |
sendmail.perform(formdata) |
|
| 842 |
get_response().process_after_jobs() |
|
| 843 |
assert emails.count() == 1 |
|
| 844 |
assert emails.emails['foobar']['msg'].is_multipart() |
|
| 845 |
assert emails.emails['foobar']['msg'].get_content_subtype() == 'mixed' |
|
| 846 |
assert emails.emails['foobar']['msg'].get_payload()[0].get_content_type() == 'text/html' |
|
| 847 |
assert emails.emails['foobar']['msg'].get_payload()[1].get_content_type() == 'image/jpeg' |
|
| 848 | ||
| 849 |
emails.empty() |
|
| 850 |
sendmail.attachments = ['form_var_backoffice_file1_raw', 'form_var_backoffice_file2_raw'] |
|
| 851 |
sendmail.perform(formdata) |
|
| 852 |
get_response().process_after_jobs() |
|
| 853 |
assert emails.count() == 1 |
|
| 854 |
assert emails.emails['foobar']['msg'].is_multipart() |
|
| 855 |
assert emails.emails['foobar']['msg'].get_content_subtype() == 'mixed' |
|
| 856 |
assert emails.emails['foobar']['msg'].get_payload()[0].get_content_type() == 'text/html' |
|
| 857 |
assert emails.emails['foobar']['msg'].get_payload()[1].get_content_type() == 'image/jpeg' |
|
| 858 |
# backoffice_file2 is unset, no more parts : |
|
| 859 |
assert len(emails.emails['foobar']['msg'].get_payload()) == 2 |
|
| 860 | ||
| 861 |
# set backoffice_file2 and retry |
|
| 862 |
setbo.fields = [{'field_id': 'bo2',
|
|
| 863 |
'value': '={"content": "blah", "filename": "hello.txt", '
|
|
| 864 |
'"content_type": "text/plain"}'}] |
|
| 865 |
setbo.perform(formdata) |
|
| 866 |
emails.empty() |
|
| 867 |
sendmail.perform(formdata) |
|
| 868 |
get_response().process_after_jobs() |
|
| 869 |
assert emails.count() == 1 |
|
| 870 |
assert emails.emails['foobar']['msg'].is_multipart() |
|
| 871 |
assert emails.emails['foobar']['msg'].get_content_subtype() == 'mixed' |
|
| 872 |
assert emails.emails['foobar']['msg'].get_payload()[0].get_content_type() == 'text/html' |
|
| 873 |
assert emails.emails['foobar']['msg'].get_payload()[1].get_content_type() == 'image/jpeg' |
|
| 874 |
assert emails.emails['foobar']['msg'].get_payload()[2].get_content_type() == 'text/plain' |
|
| 875 |
assert emails.emails['foobar']['msg'].get_payload()[2].get_payload() == 'blah' |
|
| 876 |
assert len(emails.emails['foobar']['msg'].get_payload()) == 3 |
|
| 877 | ||
| 878 | ||
| 768 | 879 |
def test_webservice_call(pub): |
| 769 | 880 |
pub.substitutions.feed(MockSubstitutionVariables()) |
| 770 | 881 | |
| wcs/workflows.py | ||
|---|---|---|
| 38 | 38 |
import qommon.errors |
| 39 | 39 | |
| 40 | 40 |
from wcs.roles import Role, logged_users_role, get_user_roles |
| 41 |
from wcs.fields import FileField |
|
| 41 | 42 |
from wcs.formdef import FormDef |
| 42 | 43 |
from wcs.formdata import Evolution |
| 43 | 44 | |
| ... | ... | |
| 2015 | 2016 |
subject = None |
| 2016 | 2017 |
body = None |
| 2017 | 2018 |
custom_from = None |
| 2019 |
attachments = None |
|
| 2018 | 2020 | |
| 2019 | 2021 |
comment = None |
| 2020 | 2022 | |
| ... | ... | |
| 2053 | 2055 |
return _('Send mail (not completed)')
|
| 2054 | 2056 | |
| 2055 | 2057 |
def get_parameters(self): |
| 2056 |
return ('to', 'subject', 'body', 'custom_from')
|
|
| 2058 |
return ('to', 'subject', 'body', 'attachments', 'custom_from')
|
|
| 2057 | 2059 | |
| 2058 | 2060 |
def fill_admin_form(self, form): |
| 2059 | 2061 |
self.add_parameters_widgets(form, self.get_parameters()) |
| ... | ... | |
| 2076 | 2078 |
value=self.body, cols=80, rows=10, |
| 2077 | 2079 |
validation_function=ComputedExpressionWidget.validate_ezt, |
| 2078 | 2080 |
hint=_('Available variables: url, url_status, details, name, number, comment, field_NAME'))
|
| 2081 | ||
| 2082 |
attachments_fields = [(None, '---', None)] |
|
| 2083 |
attachments_varnameless_fields = [] |
|
| 2084 |
for field in self.parent.parent.get_backoffice_fields(): |
|
| 2085 |
if field.key != 'file': |
|
| 2086 |
continue |
|
| 2087 |
if field.varname: |
|
| 2088 |
codename = 'form_var_%s_raw' % field.varname |
|
| 2089 |
else: |
|
| 2090 |
codename = 'form_f%s' % field.id # = form_fbo<n> |
|
| 2091 |
attachments_varnameless_fields.append(codename) |
|
| 2092 |
attachments_fields.append((codename, field.label, codename)) |
|
| 2093 |
# filter: don't show removed fields without varname |
|
| 2094 |
attachments = [attachment for attachment in self.attachments or [] |
|
| 2095 |
if ((not attachment.startswith('form_fbo')) or
|
|
| 2096 |
(attachment in attachments_varnameless_fields))] |
|
| 2097 |
if 'attachments' in parameters: |
|
| 2098 |
if len(attachments_fields) > 1: |
|
| 2099 |
form.add(WidgetList, '%sattachments' % prefix, title=_('Attachments'),
|
|
| 2100 |
element_type=SingleSelectWidgetWithOther, |
|
| 2101 |
value=attachments, |
|
| 2102 |
add_element_label=_('Add attachment'),
|
|
| 2103 |
element_kwargs={'render_br': False, 'options': attachments_fields})
|
|
| 2104 |
else: |
|
| 2105 |
form.add(WidgetList, '%sattachments' % prefix, |
|
| 2106 |
title=_('Attachments (Python expressions)'),
|
|
| 2107 |
element_type=StringWidget, |
|
| 2108 |
value=attachments, |
|
| 2109 |
add_element_label=_('Add attachment'),
|
|
| 2110 |
element_kwargs={'render_br': False, 'size': 50},
|
|
| 2111 |
advanced=not(bool(attachments))) |
|
| 2112 | ||
| 2079 | 2113 |
if 'custom_from' in parameters: |
| 2080 | 2114 |
form.add(ComputedExpressionWidget, '%scustom_from' % prefix, |
| 2081 | 2115 |
title=_('Custom From Address'), value=self.custom_from,
|
| ... | ... | |
| 2152 | 2186 |
if self.custom_from: |
| 2153 | 2187 |
email_from = self.compute(self.custom_from) |
| 2154 | 2188 | |
| 2189 |
attachments = [] |
|
| 2190 |
if self.attachments: |
|
| 2191 |
global_eval_dict = get_publisher().get_global_eval_dict() |
|
| 2192 |
local_eval_dict = get_publisher().substitutions.get_context_variables() |
|
| 2193 |
for attachment in self.attachments: |
|
| 2194 |
try: |
|
| 2195 |
picklableupload = eval(attachment, global_eval_dict, local_eval_dict) |
|
| 2196 |
except: |
|
| 2197 |
get_publisher().notify_of_exception(sys.exc_info(), |
|
| 2198 |
context='[Sendmail/Attachment]') |
|
| 2199 |
continue |
|
| 2200 |
if picklableupload: |
|
| 2201 |
try: |
|
| 2202 |
picklableupload = FileField.convert_value_from_anything(picklableupload) |
|
| 2203 |
except ValueError: |
|
| 2204 |
continue |
|
| 2205 |
attachments.append(picklableupload) |
|
| 2206 | ||
| 2155 | 2207 |
if len(addresses) > 1: |
| 2156 | 2208 |
emails.email(mail_subject, mail_body, email_rcpt=None, |
| 2157 | 2209 |
bcc=addresses, email_from=email_from, |
| 2158 | 2210 |
exclude_current_user=False, |
| 2211 |
attachments=attachments, |
|
| 2159 | 2212 |
fire_and_forget=True) |
| 2160 | 2213 |
else: |
| 2161 | 2214 |
emails.email(mail_subject, mail_body, email_rcpt=addresses, |
| 2162 | 2215 |
email_from=email_from, exclude_current_user=False, |
| 2216 |
attachments=attachments, |
|
| 2163 | 2217 |
fire_and_forget=True) |
| 2164 | 2218 |
register_item_class(SendmailWorkflowStatusItem) |
| 2165 | 2219 | |
| 2166 |
- |
|