#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

97.44

/ckanext-hdx_smtp_assumerole/ckanext/hdx_smtp_assumerole/helpers/ses_sender.py

"""
SES API email sender - sends emails using boto3 SES API instead of SMTP.
This allows using temporary credentials with session tokens from AssumeRole.
"""
import logging
import boto3
from typing import Dict, List, Optional, Union, Any
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

log = logging.getLogger(__name__)


def send_email_via_ses(
    smtp_from: str,
    recipients: Union[str, List[str]],
    subject: str,
    body: Optional[str] = None,
    body_html: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
    mime_message: Optional[Any] = None,
    access_key: Optional[str] = None,
    secret_key: Optional[str] = None,
    session_token: Optional[str] = None,
    region: Optional[str] = None
) -> Dict[str, Any]:
    """
    Send email using AWS SES API instead of SMTP.

    Args:
        smtp_from: Sender email address
        recipients: List of recipient email addresses (or single string)
        subject: Email subject
        body: Plain text body (can be None if body_html or mime_message is provided)
        body_html: HTML body (optional)
        headers: Additional email headers (dict)
        mime_message: Pre-built MIME message (email.mime object). If provided, body/body_html/subject/headers are ignored
        access_key: AWS access key ID
        secret_key: AWS secret access key
        session_token: AWS session token (for temporary credentials)
        region: AWS region

    Returns:
        dict: SES send_raw_email response

    Raises:
        Exception: If SES API call fails
    """
    if isinstance(recipients, str):
        recipients = [recipients]

    # Use pre-built MIME message if provided, otherwise build one
    if mime_message:
        msg = mime_message
    else:
        # Build MIME message
        if body_html:
            if body:
                # Both plain and HTML versions
                msg = MIMEMultipart('alternative')
                part1 = MIMEText(body, 'plain', 'utf-8')
                part2 = MIMEText(body_html, 'html', 'utf-8')
                msg.attach(part1)
                msg.attach(part2)
            else:
                # HTML only
                msg = MIMEMultipart()
                part = MIMEText(body_html, 'html', 'utf-8')
                msg.attach(part)
        else:
            # Plain text only
            msg = MIMEText(body, 'plain', 'utf-8')

        # Set standard headers (can be overridden by custom headers)
        msg['Subject'] = subject
        msg['From'] = smtp_from
        msg['To'] = ', '.join(recipients)

        # Add/override with custom headers
        if headers:
            for key, value in headers.items():
                # Skip empty headers and headers that are part of the message body
                if value and key not in ['Content-Type', 'MIME-Version', 'Content-Transfer-Encoding']:
                    # Replace existing header if it exists
                    if key in msg:
                        del msg[key]
                    msg[key] = value

    # Create SES client with temporary credentials
    ses_client = boto3.client(
        'ses',
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        aws_session_token=session_token,
        region_name=region
    )

    try:
        response = ses_client.send_raw_email(
            Source=smtp_from,
            Destinations=recipients,
            RawMessage={'Data': msg.as_string()}
        )

        log.info(f'Email sent via SES API to {recipients}, MessageId: {response["MessageId"]}')
        return response

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

1	"""
2	SES API email sender - sends emails using boto3 SES API instead of SMTP.
3	This allows using temporary credentials with session tokens from AssumeRole.
4	"""
5	import logging	1✔
6	import boto3	1✔
7	from typing import Dict, List, Optional, Union, Any	1✔
8	from email.mime.text import MIMEText	1✔
9	from email.mime.multipart import MIMEMultipart	1✔
10
11	log = logging.getLogger(__name__)	1✔
12
13
14	def send_email_via_ses(	1✔
15	smtp_from: str,
16	recipients: Union[str, List[str]],
17	subject: str,
18	body: Optional[str] = None,
19	body_html: Optional[str] = None,
20	headers: Optional[Dict[str, str]] = None,
21	mime_message: Optional[Any] = None,
22	access_key: Optional[str] = None,
23	secret_key: Optional[str] = None,
24	session_token: Optional[str] = None,
25	region: Optional[str] = None
26	) -> Dict[str, Any]:
27	"""
28	Send email using AWS SES API instead of SMTP.
29
30	Args:
31	smtp_from: Sender email address
32	recipients: List of recipient email addresses (or single string)
33	subject: Email subject
34	body: Plain text body (can be None if body_html or mime_message is provided)
35	body_html: HTML body (optional)
36	headers: Additional email headers (dict)
37	mime_message: Pre-built MIME message (email.mime object). If provided, body/body_html/subject/headers are ignored
38	access_key: AWS access key ID
39	secret_key: AWS secret access key
40	session_token: AWS session token (for temporary credentials)
41	region: AWS region
42
43	Returns:
44	dict: SES send_raw_email response
45
46	Raises:
47	Exception: If SES API call fails
48	"""
49	if isinstance(recipients, str):	1✔
50	recipients = [recipients]	1✔
51
52	# Use pre-built MIME message if provided, otherwise build one
53	if mime_message:	1✔
54	msg = mime_message	1✔
55	else:
56	# Build MIME message
57	if body_html:	1✔
58	if body:	1✔
59	# Both plain and HTML versions
60	msg = MIMEMultipart('alternative')	1✔
61	part1 = MIMEText(body, 'plain', 'utf-8')	1✔
62	part2 = MIMEText(body_html, 'html', 'utf-8')	1✔
63	msg.attach(part1)	1✔
64	msg.attach(part2)	1✔
65	else:
66	# HTML only
67	msg = MIMEMultipart()	1✔
68	part = MIMEText(body_html, 'html', 'utf-8')	1✔
69	msg.attach(part)	1✔
70	else:
71	# Plain text only
72	msg = MIMEText(body, 'plain', 'utf-8')	1✔
73
74	# Set standard headers (can be overridden by custom headers)
75	msg['Subject'] = subject	1✔
76	msg['From'] = smtp_from	1✔
77	msg['To'] = ', '.join(recipients)	1✔
78
79	# Add/override with custom headers
80	if headers:	1✔
81	for key, value in headers.items():	1✔
82	# Skip empty headers and headers that are part of the message body
83	if value and key not in ['Content-Type', 'MIME-Version', 'Content-Transfer-Encoding']:	1✔
84	# Replace existing header if it exists
85	if key in msg:	1✔
86	del msg[key]	×
87	msg[key] = value	1✔
88
89	# Create SES client with temporary credentials
90	ses_client = boto3.client(	1✔
91	'ses',
92	aws_access_key_id=access_key,
93	aws_secret_access_key=secret_key,
94	aws_session_token=session_token,
95	region_name=region
96	)
97
98	try:	1✔
99	response = ses_client.send_raw_email(	1✔
100	Source=smtp_from,
101	Destinations=recipients,
102	RawMessage={'Data': msg.as_string()}
103	)
104
105	log.info(f'Email sent via SES API to {recipients}, MessageId: {response["MessageId"]}')	1✔
106	return response	1✔
107
108	except Exception as e:	1✔
109	log.error(f'Failed to send email via SES API: {e}')	1✔
110	raise	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