diff --git a/emailer_lib/egress.py b/emailer_lib/egress.py
index c4e3178..24c7bd1 100644
--- a/emailer_lib/egress.py
+++ b/emailer_lib/egress.py
@@ -284,31 +284,33 @@ def send_intermediate_email_with_smtp(
msg_alt.attach(MIMEText(i_email.text, "plain"))
# Attach inline images
- for image_name, image_base64 in i_email.inline_attachments.items():
- img_bytes = base64.b64decode(image_base64)
- img = MIMEImage(img_bytes, _subtype="png", name=f"{image_name}")
+ if i_email.inline_attachments:
+ for image_name, image_base64 in i_email.inline_attachments.items():
+ img_bytes = base64.b64decode(image_base64)
+ img = MIMEImage(img_bytes, _subtype="png", name=f"{image_name}")
- img.add_header("Content-ID", f"<{image_name}>")
- img.add_header("Content-Disposition", "inline", filename=f"{image_name}")
+ img.add_header("Content-ID", f"<{image_name}>")
+ img.add_header("Content-Disposition", "inline", filename=f"{image_name}")
- msg.attach(img)
+ msg.attach(img)
# Attach external files (any type)
- for filename in i_email.external_attachments:
- with open(filename, "rb") as f:
- file_data = f.read()
-
- # Guess MIME type based on file extension
- mime_type, _ = mimetypes.guess_type(filename)
- if mime_type is None:
- mime_type = "application/octet-stream"
- main_type, sub_type = mime_type.split("/", 1)
-
- part = MIMEBase(main_type, sub_type)
- part.set_payload(file_data)
- encoders.encode_base64(part)
- part.add_header("Content-Disposition", "attachment", filename=filename)
- msg.attach(part)
+ if i_email.external_attachments:
+ for filename in i_email.external_attachments:
+ with open(filename, "rb") as f:
+ file_data = f.read()
+
+ # Guess MIME type based on file extension
+ mime_type, _ = mimetypes.guess_type(filename)
+ if mime_type is None:
+ mime_type = "application/octet-stream"
+ main_type, sub_type = mime_type.split("/", 1)
+
+ part = MIMEBase(main_type, sub_type)
+ part.set_payload(file_data)
+ encoders.encode_base64(part)
+ part.add_header("Content-Disposition", "attachment", filename=filename)
+ msg.attach(part)
# Send via SMTP with appropriate security protocol
if security == "ssl":
diff --git a/emailer_lib/ingress.py b/emailer_lib/ingress.py
index e7aad6d..e92e09a 100644
--- a/emailer_lib/ingress.py
+++ b/emailer_lib/ingress.py
@@ -1,12 +1,14 @@
from __future__ import annotations
from base64 import b64encode
import json
-import re
from email.message import EmailMessage
from mjml import mjml2html
from .structs import IntermediateEmail
+from .mjml import MJMLTag
+from .mjml.image_processor import _process_mjml_images
+import warnings
__all__ = [
"redmail_to_intermediate_email",
@@ -43,31 +45,33 @@ def yagmail_to_intermediate_email():
pass
-def mjml_to_intermediate_email(mjml_content: str) -> IntermediateEmail:
+def mjml_to_intermediate_email(
+ mjml_content: str | MJMLTag,
+) -> IntermediateEmail:
"""
Convert MJML markup to an IntermediateEmail
Parameters
- ------
+ ----------
mjml_content
- MJML markup string
+ MJML markup string or MJMLTag object
Returns
------
An Intermediate Email object
"""
- email_content = mjml2html(mjml_content)
+ # Handle MJMLTag objects by preprocessing images
+ if isinstance(mjml_content, MJMLTag):
+ processed_mjml, inline_attachments = _process_mjml_images(mjml_content)
+ mjml_markup = processed_mjml._to_mjml()
+ else:
+ # String-based MJML, no preprocessing needed
+ warnings.warn("MJMLTag not detected; treating input as plaintext MJML markup", UserWarning)
+ mjml_markup = mjml_content
+ inline_attachments = {}
- # Find all
tags and extract their src attributes
- pattern = r'
]+src="([^"\s]+)"[^>]*>'
- matches = re.findall(pattern, email_content)
- inline_attachments = {}
- for src in matches:
- # in theory, retrieve the externally hosted images and save to bytes
- # the user would need to pass CID-referenced images directly somehow,
- # as mjml doesn't handle them
- raise NotImplementedError("mj-image tags are not yet supported")
+ email_content = mjml2html(mjml_markup)
i_email = IntermediateEmail(
html=email_content,
diff --git a/emailer_lib/mjml/__init__.py b/emailer_lib/mjml/__init__.py
index 959139c..ea3b2f1 100644
--- a/emailer_lib/mjml/__init__.py
+++ b/emailer_lib/mjml/__init__.py
@@ -39,6 +39,7 @@
navbar_link,
social_element,
)
+# _process_mjml_images is called internally by mjml_to_intermediate_email
__all__ = (
"MJMLTag",
diff --git a/emailer_lib/mjml/_core.py b/emailer_lib/mjml/_core.py
index d2068f2..d55913e 100644
--- a/emailer_lib/mjml/_core.py
+++ b/emailer_lib/mjml/_core.py
@@ -1,38 +1,46 @@
# MJML core classes adapted from py-htmltools
## TODO: make sure Ending tags are rendered as needed
-# https://documentation.mjml.io/#ending-tags
+# https://documentation.mjml.io/#ending-tags
from typing import Dict, Mapping, Optional, Sequence, Union
import warnings
+from io import BytesIO
from mjml import mjml2html
# Types for MJML
-TagAttrValue = Union[str, float, bool, None]
+
+TagAttrValue = Union[str, float, bool, None, bytes, BytesIO]
TagAttrs = Union[Dict[str, TagAttrValue], "TagAttrDict"]
TagChild = Union["MJMLTag", str, float, None, Sequence["TagChild"]]
-class TagAttrDict(Dict[str, str]):
+class TagAttrDict(Dict[str, Union[str, bytes, BytesIO]]):
"""
- MJML attribute dictionary. All values are stored as strings.
+ MJML attribute dictionary. Most values are stored as strings, but bytes/BytesIO are preserved.
"""
- def __init__(
- self, *args: Mapping[str, TagAttrValue]
- ) -> None:
+ def __init__(self, *args: Mapping[str, TagAttrValue]) -> None:
super().__init__()
for mapping in args:
for k, v in mapping.items():
if v is not None:
- self[k] = str(v)
+ # Preserve bytes and BytesIO objects as-is for image processing
+ if isinstance(v, (bytes, BytesIO)):
+ self[k] = v
+ else:
+ self[k] = str(v)
def update(self, *args: Mapping[str, TagAttrValue]) -> None:
for mapping in args:
for k, v in mapping.items():
if v is not None:
- self[k] = str(v)
+ # Preserve bytes and BytesIO objects as-is for image processing
+ if isinstance(v, (bytes, BytesIO)):
+ self[k] = v
+ else:
+ self[k] = str(v)
class MJMLTag:
@@ -52,7 +60,7 @@ def __init__(
self.attrs = TagAttrDict()
self.children = []
self._is_leaf = _is_leaf
-
+
# Runtime validation for leaf tags
if self._is_leaf:
# For leaf tags, treat the first positional argument as content if provided
@@ -64,56 +72,67 @@ def __init__(
self.content = args[0]
else:
self.content = content
-
+
# Validate content type
- if self.content is not None and not isinstance(self.content, (str, int, float)):
+ if self.content is not None and not isinstance(
+ self.content, (str, int, float)
+ ):
raise TypeError(
f"<{tagName}> content must be a string, int, or float, "
f"got {type(self.content).__name__}"
)
-
+
# Validate attributes parameter type
- if attributes is not None and not isinstance(attributes, (dict, TagAttrDict)):
+ if attributes is not None and not isinstance(
+ attributes, (dict, TagAttrDict)
+ ):
raise TypeError(
f"attributes must be a dict or TagAttrDict, got {type(attributes).__name__}."
)
-
+
# Process attributes
if attributes is not None:
self.attrs.update(attributes)
else:
# For container tags
self.content = content
-
+
# Validate attributes parameter type
- if attributes is not None and not isinstance(attributes, (dict, TagAttrDict)):
+ if attributes is not None and not isinstance(
+ attributes, (dict, TagAttrDict)
+ ):
raise TypeError(
f"attributes must be a dict or TagAttrDict, got {type(attributes).__name__}. "
f"If you meant to pass children, use positional arguments for container tags."
)
-
+
# Collect children (for non-leaf tags only)
for arg in args:
if (
- isinstance(arg, (str, float)) or arg is None or isinstance(arg, MJMLTag)
+ isinstance(arg, (str, float))
+ or arg is None
+ or isinstance(arg, MJMLTag)
):
self.children.append(arg)
elif isinstance(arg, Sequence) and not isinstance(arg, str):
self.children.extend(arg)
-
+
# Process attributes
if attributes is not None:
self.attrs.update(attributes)
-
+
# TODO: confirm if this is the case... I don't think it is
# # If content is provided, children should be empty
# if self.content is not None:
# self.children = []
- def render_mjml(self, indent: int = 0, eol: str = "\n") -> str:
+ def _to_mjml(self, indent: int = 0, eol: str = "\n") -> str:
"""
Render MJMLTag and its children to MJML markup.
Ported from htmltools Tag rendering logic.
+
+ Note: BytesIO/bytes in image src attributes are not supported by _to_mjml().
+ Pass the MJMLTag directly to mjml_to_intermediate_email() instead.
"""
def _flatten(children):
@@ -125,6 +144,16 @@ def _flatten(children):
elif isinstance(c, (str, float)):
yield c
+ # Check for BytesIO/bytes in mj-image tags and raise clear error
+ if self.tagName == "mj-image" and "src" in self.attrs:
+ src_value = self.attrs["src"]
+ if isinstance(src_value, (bytes, BytesIO)):
+ raise ValueError(
+ "Cannot render MJML with BytesIO/bytes in image src attribute. "
+ "Pass the MJMLTag object directly to mjml_to_intermediate_email() instead of calling _to_mjml() first. "
+ "Example: i_email = mjml_to_intermediate_email(doc)"
+ )
+
# Build attribute string
attr_str = ""
if self.attrs:
@@ -138,7 +167,7 @@ def _flatten(children):
child_strs = []
for child in _flatten(self.children):
if isinstance(child, MJMLTag):
- child_strs.append(child.render_mjml(indent + 2, eol))
+ child_strs.append(child._to_mjml(indent + 2, eol))
else:
child_strs.append(str(child))
if child_strs:
@@ -152,31 +181,42 @@ def _flatten(children):
return f"{pad}<{self.tagName}{attr_str}>{self.tagName}>"
def _repr_html_(self):
- return self.to_html()
+ from ..ingress import mjml_to_intermediate_email
+ return mjml_to_intermediate_email(self)._repr_html_()
+ # TODO: make something deliberate
def __repr__(self) -> str:
- return self.render_mjml()
-
- def to_html(self, **mjml2html_kwargs):
+ warnings.warn(
+ f"__repr__ not yet fully implemented for MJMLTag({self.tagName})",
+ UserWarning,
+ stacklevel=2,
+ )
+ return f""
+
+ # warning explain that they are not to pass this to intermediate email
+ def to_html(self, **mjml2html_kwargs) -> str:
"""
Render MJMLTag to HTML using mjml2html.
-
+
If this is not a top-level tag, it will be automatically wrapped
in ... with a warning.
-
+
+ Note: This method embeds all images as inline data URIs in the HTML.
+ For email composition with inline attachments, use mjml_to_intermediate_email() instead.
+
Parameters
----------
**mjml2html_kwargs
Additional keyword arguments to pass to mjml2html
-
+
Returns
-------
str
- Result from mjml2html containing html content
+ Result from `mjml-python.mjml2html()` containing html content
"""
if self.tagName == "mjml":
# Already a complete MJML document
- mjml_markup = self.render_mjml()
+ mjml_markup = self._to_mjml()
elif self.tagName == "mj-body":
# Wrap only in mjml tag
warnings.warn(
@@ -184,10 +224,10 @@ def to_html(self, **mjml2html_kwargs):
"Automatically wrapping in .... "
"For full control, create a complete MJML document with the mjml() tag.",
UserWarning,
- stacklevel=2
+ stacklevel=2,
)
wrapped = MJMLTag("mjml", self)
- mjml_markup = wrapped.render_mjml()
+ mjml_markup = wrapped._to_mjml()
else:
# Warn and wrap in mjml/mj-body
warnings.warn(
@@ -195,10 +235,10 @@ def to_html(self, **mjml2html_kwargs):
"Automatically wrapping in .... "
"For full control, create a complete MJML document with the mjml() tag.",
UserWarning,
- stacklevel=2
+ stacklevel=2,
)
# Wrap in mjml and mj-body
wrapped = MJMLTag("mjml", MJMLTag("mj-body", self))
- mjml_markup = wrapped.render_mjml()
-
+ mjml_markup = wrapped._to_mjml()
+
return mjml2html(mjml_markup, **mjml2html_kwargs)
diff --git a/emailer_lib/mjml/image_processor.py b/emailer_lib/mjml/image_processor.py
new file mode 100644
index 0000000..02c61ed
--- /dev/null
+++ b/emailer_lib/mjml/image_processor.py
@@ -0,0 +1,164 @@
+"""
+Image preprocessing for MJML tags.
+
+This module handles conversion of image bytes to CID-referenced inline
+attachments for email embedding without using globals.
+"""
+
+from __future__ import annotations
+from typing import Any, Dict, Tuple
+import base64
+import uuid
+from io import BytesIO
+
+from ._core import MJMLTag
+
+__all__ = []
+
+
+def _convert_to_bytes(obj: Any) -> bytes:
+ """
+ Convert bytes or BytesIO to bytes.
+
+ Parameters
+ ----------
+ obj
+ The object to convert (bytes or BytesIO)
+
+ Returns
+ -------
+ bytes
+ The binary representation of the object
+
+ Raises
+ ------
+ TypeError
+ If the object type is not supported
+ """
+ if isinstance(obj, BytesIO):
+ return obj.getvalue()
+
+ if isinstance(obj, bytes):
+ return obj
+
+ raise TypeError(
+ f"Unsupported image type: {type(obj).__name__}. "
+ "Expected bytes or BytesIO."
+ )
+
+
+def _process_mjml_images(mjml_tag: MJMLTag) -> Tuple[MJMLTag, Dict[str, str]]:
+ """
+ Extract inline attachments from MJML tree and convert bytes/BytesIO to CID references.
+
+ This is a private function. Users should not call it directly.
+ It is called automatically by mjml_to_intermediate_email().
+
+ This function recursively walks through the MJML tag tree and finds mj-image tags
+ with BytesIO or bytes in their src attribute. It converts these to CID references
+ and extracts the base64 data for the inline_attachments dictionary.
+
+ Parameters
+ ----------
+ mjml_tag
+ The MJML tag tree to process
+
+ Returns
+ -------
+ Tuple[MJMLTag, Dict[str, str]]
+ A tuple of:
+ - The modified MJML tag tree with BytesIO/bytes converted to CID references
+ - Dictionary mapping CID filenames to base64-encoded image data
+
+ Examples
+ --------
+ ```python
+ from emailer_lib.mjml import mjml, body, section, column, image
+ from emailer_lib import mjml_to_intermediate_email
+ from io import BytesIO
+
+ # Create BytesIO with image data
+ buf = BytesIO(b'...png binary data...')
+
+ # Create MJML using regular image() with BytesIO as src
+ email = mjml(
+ body(
+ section(
+ column(
+ image(attributes={
+ "src": buf,
+ "alt": "My Plot",
+ "width": "600px"
+ })
+ )
+ )
+ )
+ )
+
+ # Pass directly to mjml_to_intermediate_email (calls _process_mjml_images internally)
+ i_email = mjml_to_intermediate_email(email)
+
+ # Result: i_email.inline_attachments = {"plot_1.png": "iVBORw0KGgo..."}
+ ```
+ """
+ inline_attachments: Dict[str, str] = {}
+
+ def _process_tag(tag: MJMLTag) -> MJMLTag:
+ """Recursively process a tag and its children."""
+
+ # Handle mj-image tags with BytesIO/bytes in src attribute
+ if tag.tagName == "mj-image" and "src" in tag.attrs:
+ src_value = tag.attrs["src"]
+
+ # Check if src is BytesIO or bytes
+ if isinstance(src_value, (bytes, BytesIO)):
+ # Convert to bytes and encode to base64
+ image_bytes_data = _convert_to_bytes(src_value)
+ b64_string = base64.b64encode(image_bytes_data).decode("utf-8")
+
+ # Generate CID filename using UUID
+ cid_id = uuid.uuid4().hex[:8]
+ cid_filename = f"plot_{cid_id}.png"
+
+ # Store in attachments
+ inline_attachments[cid_filename] = b64_string
+
+ # Create new tag with CID reference instead of BytesIO
+ new_attrs = dict(tag.attrs)
+ new_attrs["src"] = f"cid:{cid_filename}"
+ new_tag = MJMLTag(
+ tag.tagName,
+ attributes=new_attrs,
+ content=tag.content,
+ _is_leaf=tag._is_leaf
+ )
+
+ # Process children
+ for child in tag.children:
+ if isinstance(child, MJMLTag):
+ new_tag.children.append(_process_tag(child))
+ else:
+ new_tag.children.append(child)
+
+ return new_tag
+
+ # For all other tags, process recursively
+ new_tag = MJMLTag(
+ tag.tagName,
+ attributes=dict(tag.attrs),
+ content=tag.content,
+ _is_leaf=tag._is_leaf
+ )
+
+ for child in tag.children:
+ if isinstance(child, MJMLTag):
+ new_tag.children.append(_process_tag(child))
+ else:
+ new_tag.children.append(child)
+
+ return new_tag
+
+ # Process the entire tag tree
+ processed_tag = _process_tag(mjml_tag)
+
+ return processed_tag, inline_attachments
diff --git a/emailer_lib/mjml/tests/test_core.py b/emailer_lib/mjml/tests/test_core.py
index 935d955..6069c15 100644
--- a/emailer_lib/mjml/tests/test_core.py
+++ b/emailer_lib/mjml/tests/test_core.py
@@ -1,4 +1,6 @@
import pytest
+from io import BytesIO
+from emailer_lib.ingress import mjml_to_intermediate_email
from emailer_lib.mjml._core import MJMLTag, TagAttrDict
@@ -36,7 +38,7 @@ def test_tag_with_dict_attributes():
def test_tag_filters_none_children():
tag = MJMLTag("mj-column", MJMLTag("mj-text", content="Text"), None)
- mjml_content = tag.render_mjml()
+ mjml_content = tag._to_mjml()
# None should not appear in output
assert mjml_content.count("") == 1
@@ -44,25 +46,25 @@ def test_tag_filters_none_children():
def test_render_empty_tag():
tag = MJMLTag("mj-spacer")
- mjml_content = tag.render_mjml()
+ mjml_content = tag._to_mjml()
assert mjml_content == ""
def test_render_with_attributes():
tag = MJMLTag("mj-spacer", attributes={"height": "20px"})
- mjml_content = tag.render_mjml()
+ mjml_content = tag._to_mjml()
assert mjml_content == ''
def test_render_with_custom_indent():
tag = MJMLTag("mj-text", content="Hello")
- mjml_content = tag.render_mjml(indent=4)
+ mjml_content = tag._to_mjml(indent=4)
assert mjml_content.startswith(" ")
def test_render_with_custom_eol():
tag = MJMLTag("mj-text", content="Hello")
- mjml_content = tag.render_mjml(eol="\r\n")
+ mjml_content = tag._to_mjml(eol="\r\n")
assert "\r\n" in mjml_content
@@ -70,7 +72,7 @@ def test_render_nested_tags():
tag = MJMLTag(
"mj-section", MJMLTag("mj-column", MJMLTag("mj-text", content="Nested"))
)
- mjml_content = tag.render_mjml()
+ mjml_content = tag._to_mjml()
assert "" in mjml_content
assert "" in mjml_content
@@ -81,17 +83,17 @@ def test_render_nested_tags():
def test_render_with_string_and_tag_children():
child_tag = MJMLTag("mj-text", content="Tagged")
tag = MJMLTag("mj-column", "Plain text", child_tag, "More text")
- mjml_content = tag.render_mjml()
+ mjml_content = tag._to_mjml()
assert "Plain text" in mjml_content
assert "" in mjml_content
assert "More text" in mjml_content
-def test_repr_returns_mjml():
+def test_repr_returns_simple_string():
tag = MJMLTag("mj-text", content="Hello")
- assert repr(tag) == tag.render_mjml()
+ assert repr(tag) == ""
def test_to_html_with_complete_mjml_document():
@@ -120,12 +122,14 @@ def test_to_html_warns_and_wraps_other_tags():
assert "html" in html_result
-def test_repr_html_calls_to_html():
+def test_repr_html_returns_intermediate_email_repr_html():
tag = MJMLTag("mjml", MJMLTag("mj-body"))
html_from_repr = tag._repr_html_()
- html_from_method = tag.to_html()
- assert html_from_repr == html_from_method
+ # _repr_html_() should return the HTML representation from mjml_to_intermediate_email
+ assert "") == 3
assert "Text 1" in mjml_content
assert "Text 2" in mjml_content
assert "Text 3" in mjml_content
+
+
+def test_to_mjml_raises_on_bytesio_in_image_src():
+ image_data = BytesIO(b"fake image data")
+ image_tag = MJMLTag(
+ "mj-image",
+ attributes={"src": image_data, "alt": "Test"}
+ )
+
+ with pytest.raises(ValueError, match="Cannot render MJML with BytesIO/bytes"):
+ image_tag._to_mjml()
+
+
+def test_to_mjml_raises_on_bytes_in_image_src():
+ image_data = b"fake image data"
+ image_tag = MJMLTag(
+ "mj-image",
+ attributes={"src": image_data, "alt": "Test"}
+ )
+
+ with pytest.raises(ValueError, match="Cannot render MJML with BytesIO/bytes"):
+ image_tag._to_mjml()
+
+
+def test_tagattr_dict_stores_bytesio():
+ """Test that TagAttrDict can store BytesIO values."""
+ image_data = BytesIO(b"test data")
+ attrs = TagAttrDict({"src": image_data, "alt": "Test"})
+
+ # Verify BytesIO is stored as-is
+ assert isinstance(attrs["src"], BytesIO)
+ assert attrs["alt"] == "Test"
+
+
+def test_tagattr_dict_stores_bytes():
+ """Test that TagAttrDict can store bytes values."""
+ image_data = b"test data"
+ attrs = TagAttrDict({"src": image_data, "alt": "Test"})
+
+ # Verify bytes are stored as-is
+ assert isinstance(attrs["src"], bytes)
+ assert attrs["alt"] == "Test"
diff --git a/emailer_lib/mjml/tests/test_image_processor.py b/emailer_lib/mjml/tests/test_image_processor.py
new file mode 100644
index 0000000..5fe2f5c
--- /dev/null
+++ b/emailer_lib/mjml/tests/test_image_processor.py
@@ -0,0 +1,148 @@
+import pytest
+from io import BytesIO
+
+from emailer_lib.mjml.image_processor import _convert_to_bytes, _process_mjml_images
+from emailer_lib.mjml._core import MJMLTag
+
+
+def test_convert_to_bytes_from_bytesio():
+ data = b"test image data"
+ bytesio_obj = BytesIO(data)
+
+ result = _convert_to_bytes(bytesio_obj)
+
+ assert isinstance(result, bytes)
+ assert result == data
+
+
+def test_convert_to_bytes_from_bytes():
+ data = b"test image data"
+
+ result = _convert_to_bytes(data)
+
+ assert isinstance(result, bytes)
+ assert result == data
+
+
+def test_convert_to_bytes_raises_on_invalid_type():
+ with pytest.raises(TypeError, match="Unsupported image type"):
+ _convert_to_bytes("string")
+
+ with pytest.raises(TypeError, match="Unsupported image type"):
+ _convert_to_bytes(123)
+
+ with pytest.raises(TypeError, match="Unsupported image type"):
+ _convert_to_bytes([b"data"])
+
+
+def test_process_mjml_images_with_bytesio():
+ image_data = b"fake png data"
+ bytesio_obj = BytesIO(image_data)
+
+ # Create MJML tag with BytesIO image
+ image_tag = MJMLTag(
+ "mj-image",
+ attributes={"src": bytesio_obj, "alt": "Test"}
+ )
+
+ processed_tag, inline_attachments = _process_mjml_images(image_tag)
+
+ assert len(inline_attachments) == 1
+
+ cid_filename = list(inline_attachments.keys())[0]
+ assert cid_filename.startswith("plot_") and cid_filename.endswith(".png")
+ assert processed_tag.attrs["src"] == f"cid:{cid_filename}"
+ assert isinstance(inline_attachments[cid_filename], str)
+ assert len(inline_attachments[cid_filename]) > 0
+
+
+def test_process_mjml_images_with_bytes():
+ image_data = b"fake png data"
+
+ # Create MJML tag with bytes image
+ image_tag = MJMLTag(
+ "mj-image",
+ attributes={"src": image_data, "alt": "Test"}
+ )
+
+ processed_tag, inline_attachments = _process_mjml_images(image_tag)
+
+ assert len(inline_attachments) == 1
+
+ cid_filename = list(inline_attachments.keys())[0]
+ assert cid_filename.startswith("plot_") and cid_filename.endswith(".png")
+ assert processed_tag.attrs["src"] == f"cid:{cid_filename}"
+ assert isinstance(inline_attachments[cid_filename], str)
+ assert len(inline_attachments[cid_filename]) > 0
+
+
+def test_process_mjml_images_multiple_images():
+ image_data_1 = b"fake png data 1"
+ image_data_2 = b"fake png data 2"
+
+ # Create MJML structure with multiple images
+ col = MJMLTag(
+ "mj-column",
+ MJMLTag("mj-image", attributes={"src": BytesIO(image_data_1), "alt": "Img1"}),
+ MJMLTag("mj-image", attributes={"src": BytesIO(image_data_2), "alt": "Img2"}),
+ )
+
+ processed_tag, inline_attachments = _process_mjml_images(col)
+
+ assert len(inline_attachments) == 2
+
+ cid_filenames = list(inline_attachments.keys())
+ assert all(f.startswith("plot_") and f.endswith(".png") for f in cid_filenames)
+ assert cid_filenames[0] != cid_filenames[1]
+
+ children_with_images = [c for c in processed_tag.children if isinstance(c, MJMLTag)]
+ assert children_with_images[0].attrs["src"] == f"cid:{cid_filenames[0]}"
+ assert children_with_images[1].attrs["src"] == f"cid:{cid_filenames[1]}"
+
+
+def test_process_mjml_images_preserves_other_attributes():
+ image_data = b"fake png data"
+
+ image_tag = MJMLTag(
+ "mj-image",
+ attributes={
+ "src": BytesIO(image_data),
+ "alt": "Test Image",
+ "width": "600px",
+ "padding": "10px"
+ }
+ )
+
+ processed_tag, _ = _process_mjml_images(image_tag)
+
+ assert processed_tag.attrs["alt"] == "Test Image"
+ assert processed_tag.attrs["width"] == "600px"
+ assert processed_tag.attrs["padding"] == "10px"
+
+
+def test_process_mjml_images_preserves_non_image_tags():
+ # Create MJML structure with mixed content
+ section = MJMLTag(
+ "mj-section",
+ MJMLTag("mj-column",
+ MJMLTag("mj-text", content="Hello"),
+ MJMLTag("mj-image", attributes={"src": BytesIO(b"data"), "alt": "Img"}),
+ MJMLTag("mj-text", content="World"),
+ )
+ )
+
+ processed_tag, inline_attachments = _process_mjml_images(section)
+
+ assert processed_tag.tagName == "mj-section"
+ column = processed_tag.children[0]
+ assert column.tagName == "mj-column"
+ assert len(column.children) == 3
+
+ assert column.children[0].tagName == "mj-text"
+ assert column.children[0].content == "Hello"
+ assert column.children[2].tagName == "mj-text"
+ assert column.children[2].content == "World"
+
+ assert column.children[1].tagName == "mj-image"
+ cid_filename = list(inline_attachments.keys())[0]
+ assert column.children[1].attrs["src"] == f"cid:{cid_filename}"
diff --git a/emailer_lib/mjml/tests/test_tags.py b/emailer_lib/mjml/tests/test_tags.py
index ea10c32..3ea9305 100644
--- a/emailer_lib/mjml/tests/test_tags.py
+++ b/emailer_lib/mjml/tests/test_tags.py
@@ -39,7 +39,7 @@ def test_container_tag_accepts_children():
assert len(sec.children) == 1
assert sec.children[0].tagName == "mj-column"
- mjml_content = sec.render_mjml()
+ mjml_content = sec._to_mjml()
assert "" in mjml_content
assert "" in mjml_content
assert "" in mjml_content
@@ -53,7 +53,7 @@ def test_container_tag_accepts_attributes():
assert sec.attrs["background-color"] == "#fff"
assert sec.attrs["padding"] == "20px"
- mjml_content = sec.render_mjml()
+ mjml_content = sec._to_mjml()
assert '' in mjml_content
@@ -66,7 +66,7 @@ def test_container_tag_accepts_children_and_attrs():
assert len(sec.children) == 2
assert sec.attrs["background-color"] == "#f0f0f0"
- mjml_content = sec.render_mjml()
+ mjml_content = sec._to_mjml()
assert 'background-color="#f0f0f0"' in mjml_content
assert "Col 1" in mjml_content
assert "Col 2" in mjml_content
@@ -79,7 +79,7 @@ def test_leaf_tag_accepts_content():
assert txt.tagName == "mj-text"
assert txt.content == "Hello World"
- mjml_content = txt.render_mjml()
+ mjml_content = txt._to_mjml()
assert mjml_content == "\nHello World\n"
@@ -89,7 +89,7 @@ def test_leaf_tag_accepts_attributes():
assert txt.attrs["color"] == "red"
assert txt.attrs["font-size"] == "16px"
- mjml_content = txt.render_mjml()
+ mjml_content = txt._to_mjml()
assert 'color="red"' in mjml_content
assert 'font-size="16px"' in mjml_content
assert "Hello" in mjml_content
@@ -108,7 +108,7 @@ def test_button_is_leaf_tag():
assert btn.content == "Click Me"
assert btn.attrs["href"] == "https://example.com"
- mjml_content = btn.render_mjml()
+ mjml_content = btn._to_mjml()
assert 'href="https://example.com"' in mjml_content
assert "Click Me" in mjml_content
assert "Custom HTML"
- mjml_content = r.render_mjml()
+ mjml_content = r._to_mjml()
assert mjml_content == "\nCustom HTML
\n"
@@ -167,7 +167,7 @@ def test_table_tag():
assert tbl.tagName == "mj-table"
assert "" in tbl.content
- mjml_content = tbl.render_mjml()
+ mjml_content = tbl._to_mjml()
assert "" in mjml_content
assert "" in mjml_content
@@ -195,7 +195,7 @@ def test_image_tag():
assert img.attrs["src"] == "https://example.com/image.jpg"
assert img.attrs["alt"] == "Test Image"
- mjml_content = img.render_mjml()
+ mjml_content = img._to_mjml()
assert 'src="https://example.com/image.jpg"' in mjml_content
assert 'alt="Test Image"' in mjml_content
assert "" in mjml_content
assert "" in mjml_content
assert ' str:
+ """
+ Generate preview HTML with inline attachments embedded as base64 data URIs.
+
+ This internal method converts `cid:` references in the HTML to base64 data URIs,
+ making the HTML self-contained for preview purposes. This is distinct from the
+ HTML used in egress.py where cid references are kept and images are attached
+ as separate MIME parts.
+
+ Returns
+ -------
+ str
+ HTML content with inline attachments embedded as base64 data URIs.
+ """
+ html_with_inline = re.sub(
+ r'src="cid:([^"\s]+)"',
+ _add_base_64_to_inline_attachments(self.inline_attachments),
+ self.html,
+ )
+ return html_with_inline
+
+ def _add_subject_header(self, html: str) -> str:
+ """
+ Add subject line as a header in the HTML.
+
+ Parameters
+ ----------
+ html
+ The HTML content to add the subject to
+
+ Returns
+ -------
+ str
+ HTML with subject header added
+ """
+ if "]*>)",
+ r'\1\nSubject: {}
'.format(self.subject),
+ html,
+ count=1,
+ flags=re.IGNORECASE,
+ )
+ else:
+ # Fallback: prepend if no tag found
+ html = f'Subject: {self.subject}
\n' + html
+
+ return html
+
+ def _repr_html_(self) -> str:
+ """
+ Return HTML representation with inline attachments for rich display.
+
+ This method enables rich display of the IntermediateEmail in Jupyter notebooks
+ and other IPython-compatible environments. It converts cid: references to
+ base64 data URIs so the email can be previewed directly in the notebook.
+
+ Returns
+ -------
+ str
+ HTML content with inline attachments embedded as base64 data URIs.
+
+ Examples
+ --------
+ ```python
+ # In a Jupyter notebook, simply display the email object:
+ email = IntermediateEmail(
+ html='Hello 
',
+ subject="Test Email",
+ inline_attachments={"img1.png": "iVBORw0KGgo..."}
+ )
+ email # This will automatically call _repr_html_() for rich display
+ ```
+ """
+ html_with_inline = self._generate_preview_html()
+ return self._add_subject_header(html_with_inline)
+
def write_preview_email(self, out_file: str = "preview_email.html") -> None:
"""
Write a preview HTML file with inline attachments embedded.
This method replaces image sources in the HTML with base64-encoded data from
inline attachments, allowing you to preview the email as it would appear to recipients.
+ The generated HTML is self-contained with base64 data URIs, distinct from the email
+ sent via egress.py which uses cid references with MIME attachments.
Parameters
----------
@@ -96,24 +175,10 @@ def write_preview_email(self, out_file: str = "preview_email.html") -> None:
------
Raises ValueError if external attachments are present, as preview does not support them.
"""
- html_with_inline = re.sub(
- r'src="cid:([^"\s]+)"',
- _add_base_64_to_inline_attachments(self.inline_attachments),
- self.html,
- )
-
- # Insert subject as after the opening tag, if present
- if "]*>)",
- r'\1\nSubject: {}
'.format(self.subject),
- html_with_inline,
- count=1,
- flags=re.IGNORECASE,
- )
- else:
- # Fallback: prepend if no tag found
- html_with_inline = f'Subject: {self.subject}
\n' + html_with_inline
+ # Generate the preview HTML with inline base64 images
+ html_with_inline = self._generate_preview_html()
+ # Add subject header
+ html_with_inline = self._add_subject_header(html_with_inline)
with open(out_file, "w", encoding="utf-8") as f:
f.write(html_with_inline)
diff --git a/emailer_lib/tests/test_ingress.py b/emailer_lib/tests/test_ingress.py
index 113d304..b66abfd 100644
--- a/emailer_lib/tests/test_ingress.py
+++ b/emailer_lib/tests/test_ingress.py
@@ -135,7 +135,7 @@ def test_mjml_to_intermediate_email_no_images():
assert result.inline_attachments == {}
-def test_mjml_to_intermediate_email_with_image_raises():
+def test_mjml_to_intermediate_email_with_string_url():
mjml_content = """
@@ -148,8 +148,73 @@ def test_mjml_to_intermediate_email_with_image_raises():
"""
- with pytest.raises(NotImplementedError, match="mj-image tags are not yet supported"):
- mjml_to_intermediate_email(mjml_content)
+ result = mjml_to_intermediate_email(mjml_content)
+
+ assert isinstance(result, IntermediateEmail)
+ assert result.inline_attachments == {}
+ assert "https://example.com/image.jpg" in result.html
+
+
+def test_mjml_to_intermediate_email_with_bytesio():
+ from io import BytesIO
+ from emailer_lib.mjml import mjml, body, section, column, image
+
+ buf = BytesIO(b'\x89PNG\r\n\x1a\n')
+
+ mjml_tag = mjml(
+ body(
+ section(
+ column(
+ image(attributes={
+ "src": buf,
+ "alt": "Test Plot",
+ "width": "600px"
+ })
+ )
+ )
+ )
+ )
+
+ result = mjml_to_intermediate_email(mjml_tag)
+
+ assert isinstance(result, IntermediateEmail)
+ assert len(result.inline_attachments) == 1
+
+ cid_filename = list(result.inline_attachments.keys())[0]
+ assert cid_filename.endswith(".png")
+ assert f"cid:{cid_filename}" in result.html
+ assert result.inline_attachments[cid_filename] != ""
+
+
+def test_mjml_to_mjml_with_bytesio_raises_error():
+ from io import BytesIO
+ from emailer_lib.mjml import mjml, body, section, column, image
+
+ # Create a simple BytesIO object with fake image data
+ buf = BytesIO(b'\x89PNG\r\n\x1a\n')
+
+ mjml_tag = mjml(
+ body(
+ section(
+ column(
+ image(attributes={
+ "src": buf,
+ "alt": "Test Plot",
+ "width": "600px"
+ })
+ )
+ )
+ )
+ )
+
+ # Calling _to_mjml() should raise an error with a helpful message
+ with pytest.raises(ValueError, match="Cannot render MJML with BytesIO/bytes"):
+ mjml_tag._to_mjml()
+
+ # But passing the tag directly to mjml_to_intermediate_email should work
+ result = mjml_to_intermediate_email(mjml_tag)
+ assert isinstance(result, IntermediateEmail)
+ assert len(result.inline_attachments) == 1
def test_quarto_json_to_intermediate_email_basic(tmp_path):
@@ -217,4 +282,4 @@ def test_quarto_json_to_intermediate_email_empty_lists(tmp_path):
result = quarto_json_to_intermediate_email(str(json_file))
assert result.external_attachments == []
- assert result.inline_attachments == {}
\ No newline at end of file
+ assert result.inline_attachments == {}