#6920

Committed 22 Nov 2025 06:13PM UTC coverage: 78.218% (+0.007%) from 78.211%

Build # #6920

Build Type

push

coveralls-python

Committed by

teodorescuserban

Commit Message

HDX-10015 Add production config and development tools

Production configuration:
- Add boto3 to requirements.txt for SES API support
- Configure plugin in prod.ini.tpl and common-config-ini.txt
- Add HDX_SMTP_DOMAIN to hdx_theme config declaration
- Update setup_py_helper.sh for plugin packaging

Development tools:
- run-plugin-tests.sh: Quick test runner with docker compose
  - Auto-initializes test environment if needed
  - Reuses running stack for fast iteration
  - Shows coverage for specified plugin only
- TESTING.md: Comprehensive testing documentation

Run Details

13915 of 17790 relevant lines covered (78.22%)

0.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

18.9

/ckanext-hdx_smtp_assumerole/ckanext/hdx_smtp_assumerole/helpers/hdx_users_mailer_patches.py

# encoding: utf-8

import logging
import cgi

from typing import Dict, List, Optional, Tuple, Any
from email import utils
from email.header import Header
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from time import time
from six import PY3
from email import encoders

import ckan
import ckan.plugins.toolkit as tk

from ckanext.hdx_smtp_assumerole.helpers.credentials_manager import SMTPCredentialsManager
from ckanext.hdx_smtp_assumerole.helpers.ses_sender import send_email_via_ses

log = logging.getLogger(__name__)

# Store original function
_original_mail_recipient_html = None
_patches_applied = False

CHARSET = 'utf-8'


def patch_hdx_users_mailer() -> None:
    """
    Apply monkey patches to ckanext.hdx_users.helpers.mailer.
    Replaces SMTP-based email sending with SES API.

    This should be called once at application startup.
    Thread-safe - will only patch once even if called multiple times.
    """
    global _original_mail_recipient_html, _patches_applied

    if _patches_applied:
        log.debug('HDX users mailer patches already applied, skipping')
        return

    try:
        # Import hdx_users mailer module
        from ckanext.hdx_users.helpers import mailer as hdx_mailer

        log.debug('Applying monkey patches to ckanext.hdx_users.helpers.mailer to use SES API')

        # Store original function
        _original_mail_recipient_html = hdx_mailer._mail_recipient_html

        # Apply patch
        hdx_mailer._mail_recipient_html = patched_mail_recipient_html

        _patches_applied = True

        log.debug('Successfully patched ckanext.hdx_users.helpers.mailer to use SES API')

    except ImportError:
        log.warning('ckanext.hdx_users not found, skipping HDX users mailer patches')
    except Exception as e:
        log.error(f'Error patching HDX users mailer: {str(e)}')


def patched_mail_recipient_html(
    sender_name: str = 'Humanitarian Data Exchange (HDX)',
    sender_email: str = 'hdx@humdata.org',
    recipients_list: Optional[List[Dict[str, str]]] = None,
    subject: Optional[str] = None,
    content_dict: Optional[Dict[str, Any]] = None,
    cc_recipients_list: Optional[List[Dict[str, str]]] = None,
    bcc_recipients_list: Optional[List[Dict[str, str]]] = None,
    footer: bool = True,
    headers: Dict[str, str] = {},
    reply_wanted: bool = False,
    snippet: str = 'email/email.html',
    file: Optional[Tuple[str, Any]] = None
) -> None:
    """
    Patched version of hdx_users._mail_recipient_html that uses SES API.

    :raises Exception: If SES credentials are not available or email sending fails
    """
    try:
        # Get credentials manager and ensure fresh credentials
        manager = SMTPCredentialsManager.get_instance()
        manager.ensure_fresh_credentials()

        # Get SES credentials
        ses_creds = manager.get_ses_credentials()
        if not ses_creds:
            error_msg = 'No SES credentials available - cannot send email'
            log.error(error_msg)
            raise Exception(error_msg)

        # Build email message (similar to original code)
        config = tk.config
        mail_from = config.get('smtp.mail_from')

        template_data = {
            'data': {
                'data': content_dict,
                'footer': footer,
                '_snippet': snippet,
                'logo_hdx_email': config.get('ckan.site_url', '#') + '/images/homepage/logo-hdx.png'
            },
        }
        body_html = tk.render('email/email.html', template_data)

        # Build MIME message
        msg = MIMEMultipart()
        for k, v in headers.items():
            msg[k] = v

        subject_header = Header(subject.encode(CHARSET), CHARSET)
        msg['Subject'] = subject_header
        msg['From'] = u'"{display_name}" <{email}>'.format(
            display_name=sender_name,
            email=mail_from
        )

        recipient_email_list = []
        recipients = None

        if recipients_list:
            for r in recipients_list:
                email = r.get('email')
                recipient_email_list.append(email)
                display_name = r.get('display_name')
                if display_name:
                    recipient = u'"{display_name}" <{email}>'.format(
                        display_name=_get_decoded_str(display_name),
                        email=email
                    )
                else:
                    recipient = u'{email}'.format(email=email)
                recipients = u', '.join([recipients, recipient]) if recipients else recipient

        msg['To'] = recipients if PY3 else Header(recipients, CHARSET)

        if bcc_recipients_list:
            for r in bcc_recipients_list:
                recipient_email_list.append(r.get('email'))

        cc_recipients = None
        if cc_recipients_list:
            for r in cc_recipients_list:
                recipient_email_list.append(r.get('email'))
                cc_recipient = u'"{display_name}" <{email}>'.format(
                    display_name=_get_decoded_str(r.get('display_name')),
                    email=r.get('email')
                )
                cc_recipients = u', '.join([cc_recipients, cc_recipient]) if cc_recipients else cc_recipient
            if cc_recipients:
                msg['Cc'] = cc_recipients if PY3 else Header(cc_recipients, CHARSET)
            else:
                msg['Cc'] = ''

        msg['Date'] = utils.formatdate(time())
        msg['X-Mailer'] = "CKAN %s" % ckan.__version__

        reply_to = u'"{display_name}" <{email}>'.format(
            display_name=_get_decoded_str(sender_name),
            email=sender_email
        )
        msg['Reply-To'] = reply_to if PY3 else Header(reply_to, CHARSET)

        part = MIMEText(body_html, 'html', CHARSET)
        msg.attach(part)

        if isinstance(file, cgi.FieldStorage):
            _part = MIMEBase('application', 'octet-stream')
            _part.set_payload(file.file.read())
            encoders.encode_base64(_part)
            extension = file.filename.split('.')[-1]
            header_value = 'attachment; filename=attachment.{0}'.format(extension)
            _part.add_header('Content-Disposition', header_value)
            msg.attach(_part)

        # Send via SES API with pre-built MIME message (includes attachments)
        send_email_via_ses(
            smtp_from=mail_from,
            recipients=recipient_email_list,
            subject=subject,
            mime_message=msg,  # Pass complete MIME message with attachments
            access_key=ses_creds['access_key'],
            secret_key=ses_creds['secret_key'],
            session_token=ses_creds['session_token'],
            region=ses_creds['region']
        )

    except Exception as e:
        log.error(f'Failed to send email via SES API: {str(e)}')
        raise


def _get_decoded_str(display_name: Optional[str]) -> str:
    """Helper function to decode display names (copied from hdx_users mailer)."""
    if display_name:
        try:
            decoded = ''
            for text, charset in Header(display_name).decode_header():
                if charset:
                    decoded += text.decode(charset)
                elif isinstance(text, bytes):
                    decoded += text.decode('utf-8')
                else:
                    decoded += text
            return decoded
        except Exception as e:
            log.warning(f'Error decoding display name: {e}')
            return display_name
    return ''


def unpatch_hdx_users_mailer() -> None:
    """
    Remove monkey patches and restore original functions.
    Useful for testing or cleanup.
    """
    global _original_mail_recipient_html, _patches_applied

    if not _patches_applied:
        log.debug('HDX users mailer patches not applied, nothing to unpatch')
        return

    try:
        from ckanext.hdx_users.helpers import mailer as hdx_mailer

        log.info('Removing monkey patches from ckanext.hdx_users.helpers.mailer')

        # Restore original function
        if _original_mail_recipient_html is not None:
            hdx_mailer._mail_recipient_html = _original_mail_recipient_html

        _patches_applied = False

        log.info('Successfully removed patches from ckanext.hdx_users.helpers.mailer')

    except ImportError:
        log.warning('ckanext.hdx_users not found during unpatch')
    except Exception as e:
        log.error(f'Error unpatching HDX users mailer: {str(e)}')


def is_patched() -> bool:
    """
    Check if hdx_users mailer is currently patched.

    :return: True if patches are applied
    :rtype: bool
    """
    return _patches_applied

1	# encoding: utf-8
2
3	import logging	1✔
4	import cgi	1✔
5
6	from typing import Dict, List, Optional, Tuple, Any	1✔
7	from email import utils	1✔
8	from email.header import Header	1✔
9	from email.mime.base import MIMEBase	1✔
10	from email.mime.multipart import MIMEMultipart	1✔
11	from email.mime.text import MIMEText	1✔
12	from time import time	1✔
13	from six import PY3	1✔
14	from email import encoders	1✔
15
16	import ckan	1✔
17	import ckan.plugins.toolkit as tk	1✔
18
19	from ckanext.hdx_smtp_assumerole.helpers.credentials_manager import SMTPCredentialsManager	1✔
20	from ckanext.hdx_smtp_assumerole.helpers.ses_sender import send_email_via_ses	1✔
21
22	log = logging.getLogger(__name__)	1✔
23
24	# Store original function
25	_original_mail_recipient_html = None	1✔
26	_patches_applied = False	1✔
27
28	CHARSET = 'utf-8'	1✔
29
30
31	def patch_hdx_users_mailer() -> None:	1✔
32	"""
33	Apply monkey patches to ckanext.hdx_users.helpers.mailer.
34	Replaces SMTP-based email sending with SES API.
35
36	This should be called once at application startup.
37	Thread-safe - will only patch once even if called multiple times.
38	"""
39	global _original_mail_recipient_html, _patches_applied
40
41	if _patches_applied:	×
42	log.debug('HDX users mailer patches already applied, skipping')	×
43	return	×
44
45	try:	×
46	# Import hdx_users mailer module
47	from ckanext.hdx_users.helpers import mailer as hdx_mailer	×
48
49	log.debug('Applying monkey patches to ckanext.hdx_users.helpers.mailer to use SES API')	×
50
51	# Store original function
52	_original_mail_recipient_html = hdx_mailer._mail_recipient_html	×
53
54	# Apply patch
55	hdx_mailer._mail_recipient_html = patched_mail_recipient_html	×
56
57	_patches_applied = True	×
58
59	log.debug('Successfully patched ckanext.hdx_users.helpers.mailer to use SES API')	×
60
61	except ImportError:	×
62	log.warning('ckanext.hdx_users not found, skipping HDX users mailer patches')	×
63	except Exception as e:	×
64	log.error(f'Error patching HDX users mailer: {str(e)}')	×
65
66
67	def patched_mail_recipient_html(	1✔
68	sender_name: str = 'Humanitarian Data Exchange (HDX)',
69	sender_email: str = 'hdx@humdata.org',
70	recipients_list: Optional[List[Dict[str, str]]] = None,
71	subject: Optional[str] = None,
72	content_dict: Optional[Dict[str, Any]] = None,
73	cc_recipients_list: Optional[List[Dict[str, str]]] = None,
74	bcc_recipients_list: Optional[List[Dict[str, str]]] = None,
75	footer: bool = True,
76	headers: Dict[str, str] = {},
77	reply_wanted: bool = False,
78	snippet: str = 'email/email.html',
79	file: Optional[Tuple[str, Any]] = None
80	) -> None:
81	"""
82	Patched version of hdx_users._mail_recipient_html that uses SES API.
83
84	:raises Exception: If SES credentials are not available or email sending fails
85	"""
86	try:	×
87	# Get credentials manager and ensure fresh credentials
88	manager = SMTPCredentialsManager.get_instance()	×
89	manager.ensure_fresh_credentials()	×
90
91	# Get SES credentials
92	ses_creds = manager.get_ses_credentials()	×
93	if not ses_creds:	×
94	error_msg = 'No SES credentials available - cannot send email'	×
95	log.error(error_msg)	×
96	raise Exception(error_msg)	×
97
98	# Build email message (similar to original code)
99	config = tk.config	×
100	mail_from = config.get('smtp.mail_from')	×
101
102	template_data = {	×
103	'data': {
104	'data': content_dict,
105	'footer': footer,
106	'_snippet': snippet,
107	'logo_hdx_email': config.get('ckan.site_url', '#') + '/images/homepage/logo-hdx.png'
108	},
109	}
110	body_html = tk.render('email/email.html', template_data)	×
111
112	# Build MIME message
113	msg = MIMEMultipart()	×
114	for k, v in headers.items():	×
115	msg[k] = v	×
116
117	subject_header = Header(subject.encode(CHARSET), CHARSET)	×
118	msg['Subject'] = subject_header	×
119	msg['From'] = u'"{display_name}" <{email}>'.format(	×
120	display_name=sender_name,
121	email=mail_from
122	)
123
124	recipient_email_list = []	×
125	recipients = None	×
126
127	if recipients_list:	×
128	for r in recipients_list:	×
129	email = r.get('email')	×
130	recipient_email_list.append(email)	×
131	display_name = r.get('display_name')	×
132	if display_name:	×
133	recipient = u'"{display_name}" <{email}>'.format(	×
134	display_name=_get_decoded_str(display_name),
135	email=email
136	)
137	else:
138	recipient = u'{email}'.format(email=email)	×
139	recipients = u', '.join([recipients, recipient]) if recipients else recipient	×
140
141	msg['To'] = recipients if PY3 else Header(recipients, CHARSET)	×
142
143	if bcc_recipients_list:	×
144	for r in bcc_recipients_list:	×
145	recipient_email_list.append(r.get('email'))	×
146
147	cc_recipients = None	×
148	if cc_recipients_list:	×
149	for r in cc_recipients_list:	×
150	recipient_email_list.append(r.get('email'))	×
151	cc_recipient = u'"{display_name}" <{email}>'.format(	×
152	display_name=_get_decoded_str(r.get('display_name')),
153	email=r.get('email')
154	)
155	cc_recipients = u', '.join([cc_recipients, cc_recipient]) if cc_recipients else cc_recipient	×
156	if cc_recipients:	×
157	msg['Cc'] = cc_recipients if PY3 else Header(cc_recipients, CHARSET)	×
158	else:
159	msg['Cc'] = ''	×
160
161	msg['Date'] = utils.formatdate(time())	×
162	msg['X-Mailer'] = "CKAN %s" % ckan.__version__	×
163
164	reply_to = u'"{display_name}" <{email}>'.format(	×
165	display_name=_get_decoded_str(sender_name),
166	email=sender_email
167	)
168	msg['Reply-To'] = reply_to if PY3 else Header(reply_to, CHARSET)	×
169
170	part = MIMEText(body_html, 'html', CHARSET)	×
171	msg.attach(part)	×
172
173	if isinstance(file, cgi.FieldStorage):	×
174	_part = MIMEBase('application', 'octet-stream')	×
175	_part.set_payload(file.file.read())	×
176	encoders.encode_base64(_part)	×
177	extension = file.filename.split('.')[-1]	×
178	header_value = 'attachment; filename=attachment.{0}'.format(extension)	×
179	_part.add_header('Content-Disposition', header_value)	×
180	msg.attach(_part)	×
181
182	# Send via SES API with pre-built MIME message (includes attachments)
183	send_email_via_ses(	×
184	smtp_from=mail_from,
185	recipients=recipient_email_list,
186	subject=subject,
187	mime_message=msg, # Pass complete MIME message with attachments
188	access_key=ses_creds['access_key'],
189	secret_key=ses_creds['secret_key'],
190	session_token=ses_creds['session_token'],
191	region=ses_creds['region']
192	)
193
194	except Exception as e:	×
195	log.error(f'Failed to send email via SES API: {str(e)}')	×
196	raise	×
197
198
199	def _get_decoded_str(display_name: Optional[str]) -> str:	1✔
200	"""Helper function to decode display names (copied from hdx_users mailer)."""
201	if display_name:	×
202	try:	×
203	decoded = ''	×
204	for text, charset in Header(display_name).decode_header():	×
205	if charset:	×
206	decoded += text.decode(charset)	×
207	elif isinstance(text, bytes):	×
208	decoded += text.decode('utf-8')	×
209	else:
210	decoded += text	×
211	return decoded	×
212	except Exception as e:	×
213	log.warning(f'Error decoding display name: {e}')	×
214	return display_name	×
215	return ''	×
216
217
218	def unpatch_hdx_users_mailer() -> None:	1✔
219	"""
220	Remove monkey patches and restore original functions.
221	Useful for testing or cleanup.
222	"""
223	global _original_mail_recipient_html, _patches_applied
224
225	if not _patches_applied:	×
226	log.debug('HDX users mailer patches not applied, nothing to unpatch')	×
227	return	×
228
229	try:	×
230	from ckanext.hdx_users.helpers import mailer as hdx_mailer	×
231
232	log.info('Removing monkey patches from ckanext.hdx_users.helpers.mailer')	×
233
234	# Restore original function
235	if _original_mail_recipient_html is not None:	×
236	hdx_mailer._mail_recipient_html = _original_mail_recipient_html	×
237
238	_patches_applied = False	×
239
240	log.info('Successfully removed patches from ckanext.hdx_users.helpers.mailer')	×
241
242	except ImportError:	×
243	log.warning('ckanext.hdx_users not found during unpatch')	×
244	except Exception as e:	×
245	log.error(f'Error unpatching HDX users mailer: {str(e)}')	×
246
247
248	def is_patched() -> bool:	1✔
249	"""
250	Check if hdx_users mailer is currently patched.
251
252	:return: True if patches are applied
253	:rtype: bool
254	"""
255	return _patches_applied	×

OCHA-DAP / hdx-ckan / #6920

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous