#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

93.33

/ckanext-hdx_smtp_assumerole/ckanext/hdx_smtp_assumerole/helpers/mailer_patches.py

# encoding: utf-8

import logging
import mimetypes
from typing import Dict, List, Optional, Tuple, Any, Union
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders

import ckan.lib.mailer as mailer
from ckan import model
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 functions
_original_mail_user = None
_original_mail_recipient = None
_patches_applied = False


def _build_mime_message_with_attachments(
    mail_from: str,
    recipient_email: str,
    recipient_name: Optional[str],
    subject: str,
    body: str,
    body_html: Optional[str],
    headers: Optional[Dict[str, str]],
    attachments: Optional[List[Union[Tuple[str, Any], Tuple[str, Any, str]]]]
) -> MIMEMultipart:
    """
    Build a MIME message with optional HTML body and attachments.

    Shared logic for both mail_user and mail_recipient to avoid code duplication.

    :param mail_from: Sender email address
    :param recipient_email: Recipient email address
    :param recipient_name: Recipient display name (optional)
    :param subject: Email subject
    :param body: Plain text body
    :param body_html: HTML body (optional)
    :param headers: Additional headers dict (optional)
    :param attachments: List of attachment tuples (optional)
    :return: MIMEMultipart message object
    """
    msg = MIMEMultipart()
    msg['From'] = mail_from
    msg['Subject'] = subject

    # Add To header with display name
    if recipient_name:
        msg['To'] = f'"{recipient_name}" <{recipient_email}>'
    else:
        msg['To'] = recipient_email

    # Add custom headers
    if headers:
        for key, value in headers.items():
            if key not in ['From', 'To', 'Subject'] and value:
                msg[key] = value

    # Add body
    if body_html:
        if body:
            # Both plain and HTML
            msg_alt = MIMEMultipart('alternative')
            msg_alt.attach(MIMEText(body, 'plain', 'utf-8'))
            msg_alt.attach(MIMEText(body_html, 'html', 'utf-8'))
            msg.attach(msg_alt)
        else:
            # HTML only
            msg.attach(MIMEText(body_html, 'html', 'utf-8'))
    else:
        # Plain text only
        msg.attach(MIMEText(body, 'plain', 'utf-8'))

    # Add attachments
    if attachments:
        for attachment in attachments:
            if len(attachment) == 3:
                filename, file_obj, media_type = attachment
            else:
                filename, file_obj = attachment
                media_type, _ = mimetypes.guess_type(filename)
                if not media_type:
                    media_type = 'application/octet-stream'

            maintype, subtype = media_type.split('/', 1)
            part = MIMEBase(maintype, subtype)
            part.set_payload(file_obj.read())
            encoders.encode_base64(part)
            part.add_header('Content-Disposition', f'attachment; filename={filename}')
            msg.attach(part)

    return msg


def patch_mailer_functions() -> None:
    """
    Apply monkey patches to ckan.lib.mailer functions.
    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_user, _original_mail_recipient, _patches_applied

    if _patches_applied:
        log.debug('Mailer patches already applied, skipping')
        return

    log.debug('Applying monkey patches to ckan.lib.mailer to use SES API')

    # Store original functions
    _original_mail_user = mailer.mail_user
    _original_mail_recipient = mailer.mail_recipient

    # Apply patches
    mailer.mail_user = patched_mail_user
    mailer.mail_recipient = patched_mail_recipient

    _patches_applied = True

    log.debug('Successfully patched ckan.lib.mailer to use SES API')


def patched_mail_user(
    recipient: Union[model.User, Dict[str, Any]],
    subject: str,
    body: str,
    body_html: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
    attachments: Optional[List[Union[Tuple[str, Any], Tuple[str, Any, str]]]] = None
) -> None:
    """
    Patched version of ckan.lib.mailer.mail_user that uses SES API.

    :param recipient: User object or user dict
    :param subject: Email subject
    :param body: Email body (plain text)
    :param body_html: Email body (HTML)
    :param headers: Additional headers dict
    :param attachments: List of attachment tuples (filename, file_object) or (filename, file_object, media_type)
    :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)

        # Extract recipient email and name
        if isinstance(recipient, model.User):
            recipient_email = recipient.email
            recipient_name = recipient.display_name or recipient.name
        else:
            recipient_email = recipient.get('email')
            recipient_name = recipient.get('display_name') or recipient.get('name')

        # Get mail_from config
        config = tk.config
        mail_from = config.get('smtp.mail_from') or config.get('mail_from')

        # If we have attachments or body_html, build a complete MIME message
        if attachments or body_html:
            msg = _build_mime_message_with_attachments(
                mail_from, recipient_email, recipient_name, subject,
                body, body_html, headers, attachments
            )
            send_email_via_ses(
                smtp_from=mail_from,
                recipients=[recipient_email],
                subject=subject,
                mime_message=msg,
                access_key=ses_creds['access_key'],
                secret_key=ses_creds['secret_key'],
                session_token=ses_creds['session_token'],
                region=ses_creds['region']
            )
        else:
            # Simple plain text email - use simple mode
            if not headers:
                headers = {}
            if recipient_name:
                headers['To'] = f'"{recipient_name}" <{recipient_email}>'
            else:
                headers['To'] = recipient_email

            send_email_via_ses(
                smtp_from=mail_from,
                recipients=[recipient_email],
                subject=subject,
                body=body,
                headers=headers,
                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 patched_mail_recipient(
    recipient_name: str,
    recipient_email: str,
    subject: str,
    body: str,
    body_html: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
    attachments: Optional[List[Union[Tuple[str, Any], Tuple[str, Any, str]]]] = None
) -> None:
    """
    Patched version of ckan.lib.mailer.mail_recipient that uses SES API.

    :param recipient_name: Recipient name
    :param recipient_email: Recipient email address
    :param subject: Email subject
    :param body: Email body (plain text)
    :param body_html: Email body (HTML)
    :param headers: Additional headers dict
    :param attachments: List of attachment tuples (filename, file_object) or (filename, file_object, media_type)
    :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)

        # Get mail_from config
        config = tk.config
        mail_from = config.get('smtp.mail_from') or config.get('mail_from')

        # If we have attachments or body_html, build a complete MIME message
        if attachments or body_html:
            msg = _build_mime_message_with_attachments(
                mail_from, recipient_email, recipient_name, subject,
                body, body_html, headers, attachments
            )
            send_email_via_ses(
                smtp_from=mail_from,
                recipients=[recipient_email],
                subject=subject,
                mime_message=msg,
                access_key=ses_creds['access_key'],
                secret_key=ses_creds['secret_key'],
                session_token=ses_creds['session_token'],
                region=ses_creds['region']
            )
        else:
            # Simple plain text email - use simple mode
            send_email_via_ses(
                smtp_from=mail_from,
                recipients=[recipient_email],
                subject=subject,
                body=body,
                headers=headers,
                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 unpatch_mailer_functions() -> None:
    """
    Remove monkey patches and restore original functions.
    Useful for testing or cleanup.
    """
    global _original_mail_user, _original_mail_recipient, _patches_applied

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

    log.info('Removing monkey patches from ckan.lib.mailer')

    # Restore original functions
    if _original_mail_user is not None:
        mailer.mail_user = _original_mail_user
    if _original_mail_recipient is not None:
        mailer.mail_recipient = _original_mail_recipient

    _patches_applied = False

    log.info('Successfully removed patches from ckan.lib.mailer')


def is_patched() -> bool:
    """
    Check if mailer functions are currently patched.

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

1	# encoding: utf-8
2
3	import logging	1✔
4	import mimetypes	1✔
5	from typing import Dict, List, Optional, Tuple, Any, Union	1✔
6	from email.mime.multipart import MIMEMultipart	1✔
7	from email.mime.text import MIMEText	1✔
8	from email.mime.base import MIMEBase	1✔
9	from email import encoders	1✔
10
11	import ckan.lib.mailer as mailer	1✔
12	from ckan import model	1✔
13	import ckan.plugins.toolkit as tk	1✔
14
15	from ckanext.hdx_smtp_assumerole.helpers.credentials_manager import SMTPCredentialsManager	1✔
16	from ckanext.hdx_smtp_assumerole.helpers.ses_sender import send_email_via_ses	1✔
17
18	log = logging.getLogger(__name__)	1✔
19
20	# Store original functions
21	_original_mail_user = None	1✔
22	_original_mail_recipient = None	1✔
23	_patches_applied = False	1✔
24
25
26	def _build_mime_message_with_attachments(	1✔
27	mail_from: str,
28	recipient_email: str,
29	recipient_name: Optional[str],
30	subject: str,
31	body: str,
32	body_html: Optional[str],
33	headers: Optional[Dict[str, str]],
34	attachments: Optional[List[Union[Tuple[str, Any], Tuple[str, Any, str]]]]
35	) -> MIMEMultipart:
36	"""
37	Build a MIME message with optional HTML body and attachments.
38
39	Shared logic for both mail_user and mail_recipient to avoid code duplication.
40
41	:param mail_from: Sender email address
42	:param recipient_email: Recipient email address
43	:param recipient_name: Recipient display name (optional)
44	:param subject: Email subject
45	:param body: Plain text body
46	:param body_html: HTML body (optional)
47	:param headers: Additional headers dict (optional)
48	:param attachments: List of attachment tuples (optional)
49	:return: MIMEMultipart message object
50	"""
51	msg = MIMEMultipart()	1✔
52	msg['From'] = mail_from	1✔
53	msg['Subject'] = subject	1✔
54
55	# Add To header with display name
56	if recipient_name:	1✔
57	msg['To'] = f'"{recipient_name}" <{recipient_email}>'	1✔
58	else:
59	msg['To'] = recipient_email	1✔
60
61	# Add custom headers
62	if headers:	1✔
63	for key, value in headers.items():	×
64	if key not in ['From', 'To', 'Subject'] and value:	×
65	msg[key] = value	×
66
67	# Add body
68	if body_html:	1✔
69	if body:	1✔
70	# Both plain and HTML
71	msg_alt = MIMEMultipart('alternative')	1✔
72	msg_alt.attach(MIMEText(body, 'plain', 'utf-8'))	1✔
73	msg_alt.attach(MIMEText(body_html, 'html', 'utf-8'))	1✔
74	msg.attach(msg_alt)	1✔
75	else:
76	# HTML only
77	msg.attach(MIMEText(body_html, 'html', 'utf-8'))	1✔
78	else:
79	# Plain text only
80	msg.attach(MIMEText(body, 'plain', 'utf-8'))	1✔
81
82	# Add attachments
83	if attachments:	1✔
84	for attachment in attachments:	1✔
85	if len(attachment) == 3:	1✔
86	filename, file_obj, media_type = attachment	1✔
87	else:
88	filename, file_obj = attachment	1✔
89	media_type, _ = mimetypes.guess_type(filename)	1✔
90	if not media_type:	1✔
91	media_type = 'application/octet-stream'	1✔
92
93	maintype, subtype = media_type.split('/', 1)	1✔
94	part = MIMEBase(maintype, subtype)	1✔
95	part.set_payload(file_obj.read())	1✔
96	encoders.encode_base64(part)	1✔
97	part.add_header('Content-Disposition', f'attachment; filename={filename}')	1✔
98	msg.attach(part)	1✔
99
100	return msg	1✔
101
102
103	def patch_mailer_functions() -> None:	1✔
104	"""
105	Apply monkey patches to ckan.lib.mailer functions.
106	Replaces SMTP-based email sending with SES API.
107
108	This should be called once at application startup.
109	Thread-safe - will only patch once even if called multiple times.
110	"""
111	global _original_mail_user, _original_mail_recipient, _patches_applied
112
113	if _patches_applied:	1✔
114	log.debug('Mailer patches already applied, skipping')	1✔
115	return	1✔
116
117	log.debug('Applying monkey patches to ckan.lib.mailer to use SES API')	1✔
118
119	# Store original functions
120	_original_mail_user = mailer.mail_user	1✔
121	_original_mail_recipient = mailer.mail_recipient	1✔
122
123	# Apply patches
124	mailer.mail_user = patched_mail_user	1✔
125	mailer.mail_recipient = patched_mail_recipient	1✔
126
127	_patches_applied = True	1✔
128
129	log.debug('Successfully patched ckan.lib.mailer to use SES API')	1✔
130
131
132	def patched_mail_user(	1✔
133	recipient: Union[model.User, Dict[str, Any]],
134	subject: str,
135	body: str,
136	body_html: Optional[str] = None,
137	headers: Optional[Dict[str, str]] = None,
138	attachments: Optional[List[Union[Tuple[str, Any], Tuple[str, Any, str]]]] = None
139	) -> None:
140	"""
141	Patched version of ckan.lib.mailer.mail_user that uses SES API.
142
143	:param recipient: User object or user dict
144	:param subject: Email subject
145	:param body: Email body (plain text)
146	:param body_html: Email body (HTML)
147	:param headers: Additional headers dict
148	:param attachments: List of attachment tuples (filename, file_object) or (filename, file_object, media_type)
149	:raises Exception: If SES credentials are not available or email sending fails
150	"""
151	try:	1✔
152	# Get credentials manager and ensure fresh credentials
153	manager = SMTPCredentialsManager.get_instance()	1✔
154	manager.ensure_fresh_credentials()	1✔
155
156	# Get SES credentials
157	ses_creds = manager.get_ses_credentials()	1✔
158	if not ses_creds:	1✔
159	error_msg = 'No SES credentials available - cannot send email'	1✔
160	log.error(error_msg)	1✔
161	raise Exception(error_msg)	1✔
162
163	# Extract recipient email and name
164	if isinstance(recipient, model.User):	1✔
165	recipient_email = recipient.email	×
166	recipient_name = recipient.display_name or recipient.name	×
167	else:
168	recipient_email = recipient.get('email')	1✔
169	recipient_name = recipient.get('display_name') or recipient.get('name')	1✔
170
171	# Get mail_from config
172	config = tk.config	1✔
173	mail_from = config.get('smtp.mail_from') or config.get('mail_from')	1✔
174
175	# If we have attachments or body_html, build a complete MIME message
176	if attachments or body_html:	1✔
177	msg = _build_mime_message_with_attachments(	1✔
178	mail_from, recipient_email, recipient_name, subject,
179	body, body_html, headers, attachments
180	)
181	send_email_via_ses(	1✔
182	smtp_from=mail_from,
183	recipients=[recipient_email],
184	subject=subject,
185	mime_message=msg,
186	access_key=ses_creds['access_key'],
187	secret_key=ses_creds['secret_key'],
188	session_token=ses_creds['session_token'],
189	region=ses_creds['region']
190	)
191	else:
192	# Simple plain text email - use simple mode
193	if not headers:	1✔
194	headers = {}	1✔
195	if recipient_name:	1✔
196	headers['To'] = f'"{recipient_name}" <{recipient_email}>'	1✔
197	else:
198	headers['To'] = recipient_email	×
199
200	send_email_via_ses(	1✔
201	smtp_from=mail_from,
202	recipients=[recipient_email],
203	subject=subject,
204	body=body,
205	headers=headers,
206	access_key=ses_creds['access_key'],
207	secret_key=ses_creds['secret_key'],
208	session_token=ses_creds['session_token'],
209	region=ses_creds['region']
210	)
211
212	except Exception as e:	1✔
213	log.error(f'Failed to send email via SES API: {str(e)}')	1✔
214	raise	1✔
215
216
217	def patched_mail_recipient(	1✔
218	recipient_name: str,
219	recipient_email: str,
220	subject: str,
221	body: str,
222	body_html: Optional[str] = None,
223	headers: Optional[Dict[str, str]] = None,
224	attachments: Optional[List[Union[Tuple[str, Any], Tuple[str, Any, str]]]] = None
225	) -> None:
226	"""
227	Patched version of ckan.lib.mailer.mail_recipient that uses SES API.
228
229	:param recipient_name: Recipient name
230	:param recipient_email: Recipient email address
231	:param subject: Email subject
232	:param body: Email body (plain text)
233	:param body_html: Email body (HTML)
234	:param headers: Additional headers dict
235	:param attachments: List of attachment tuples (filename, file_object) or (filename, file_object, media_type)
236	:raises Exception: If SES credentials are not available or email sending fails
237	"""
238	try:	1✔
239	# Get credentials manager and ensure fresh credentials
240	manager = SMTPCredentialsManager.get_instance()	1✔
241	manager.ensure_fresh_credentials()	1✔
242
243	# Get SES credentials
244	ses_creds = manager.get_ses_credentials()	1✔
245	if not ses_creds:	1✔
246	error_msg = 'No SES credentials available - cannot send email'	1✔
247	log.error(error_msg)	1✔
248	raise Exception(error_msg)	1✔
249
250	# Get mail_from config
251	config = tk.config	1✔
252	mail_from = config.get('smtp.mail_from') or config.get('mail_from')	1✔
253
254	# If we have attachments or body_html, build a complete MIME message
255	if attachments or body_html:	1✔
256	msg = _build_mime_message_with_attachments(	1✔
257	mail_from, recipient_email, recipient_name, subject,
258	body, body_html, headers, attachments
259	)
260	send_email_via_ses(	1✔
261	smtp_from=mail_from,
262	recipients=[recipient_email],
263	subject=subject,
264	mime_message=msg,
265	access_key=ses_creds['access_key'],
266	secret_key=ses_creds['secret_key'],
267	session_token=ses_creds['session_token'],
268	region=ses_creds['region']
269	)
270	else:
271	# Simple plain text email - use simple mode
272	send_email_via_ses(	1✔
273	smtp_from=mail_from,
274	recipients=[recipient_email],
275	subject=subject,
276	body=body,
277	headers=headers,
278	access_key=ses_creds['access_key'],
279	secret_key=ses_creds['secret_key'],
280	session_token=ses_creds['session_token'],
281	region=ses_creds['region']
282	)
283
284	except Exception as e:	1✔
285	log.error(f'Failed to send email via SES API: {str(e)}')	1✔
286	raise	1✔
287
288
289	def unpatch_mailer_functions() -> None:	1✔
290	"""
291	Remove monkey patches and restore original functions.
292	Useful for testing or cleanup.
293	"""
294	global _original_mail_user, _original_mail_recipient, _patches_applied
295
296	if not _patches_applied:	1✔
297	log.debug('Mailer patches not applied, nothing to unpatch')	×
298	return	×
299
300	log.info('Removing monkey patches from ckan.lib.mailer')	1✔
301
302	# Restore original functions
303	if _original_mail_user is not None:	1✔
304	mailer.mail_user = _original_mail_user	1✔
305	if _original_mail_recipient is not None:	1✔
306	mailer.mail_recipient = _original_mail_recipient	1✔
307
308	_patches_applied = False	1✔
309
310	log.info('Successfully removed patches from ckan.lib.mailer')	1✔
311
312
313	def is_patched() -> bool:	1✔
314	"""
315	Check if mailer functions are currently patched.
316
317	:return: True if patches are applied
318	:rtype: bool
319	"""
320	return _patches_applied	1✔

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