``.
+ The HTML part replaces the URL with ``
``; the plain-
+ text part keeps the URL as a clickable link. When the template doesn't
+ reference ``{finish_photo_url}`` (or image bytes aren't available), the
+ original single-part text shape is used — no attachment, no surprise
+ inline image (#1792).
+ """
smtp_server = config.get("smtp_server", "").strip()
smtp_port = int(config.get("smtp_port", 587))
username = config.get("username", "").strip()
@@ -429,12 +450,48 @@ class NotificationService:
if auth_enabled and not all([username, password]):
return False, "Username and password are required when authentication is enabled"
+ # Template-driven: only inline-embed when the user's template explicitly
+ # referenced {finish_photo_url} (so the URL appears in the rendered body)
+ # AND the photo bytes are available. Falls back to text-only otherwise.
+ inline_photo = bool(image_data and finish_photo_url and finish_photo_url in body)
+
try:
- msg = MIMEMultipart()
- msg["From"] = from_email
- msg["To"] = to_email
- msg["Subject"] = f"[Bambuddy] {subject}"
- msg.attach(MIMEText(body, "plain"))
+ if inline_photo:
+ # multipart/related → (multipart/alternative → text, html) + inline image
+ msg = MIMEMultipart("related")
+ msg["From"] = from_email
+ msg["To"] = to_email
+ msg["Subject"] = f"[Bambuddy] {subject}"
+
+ alt = MIMEMultipart("alternative")
+ alt.attach(MIMEText(body, "plain"))
+ # Build HTML body: escape the rendered body, then swap the
+ # escaped URL substring for an inline
referencing the
+ # MIMEImage we attach below. Done AFTER escape so the cid: URL
+ # we inject isn't re-escaped.
+ escaped_body = html.escape(body).replace("\n", "
\n")
+ escaped_url = html.escape(finish_photo_url)
+ img_tag = (
+ '
'
+ )
+ html_body = f"{escaped_body.replace(escaped_url, img_tag)}
"
+ alt.attach(MIMEText(html_body, "html"))
+ msg.attach(alt)
+
+ img = MIMEImage(image_data, _subtype="jpeg")
+ # Angle-bracketed Content-ID per RFC 2392, referenced from HTML
+ # without the brackets via ``cid:bambuddy-finish-photo``.
+ img.add_header("Content-ID", "")
+ img.add_header("Content-Disposition", "inline", filename="finish-photo.jpg")
+ msg.attach(img)
+ else:
+ msg = MIMEMultipart()
+ msg["From"] = from_email
+ msg["To"] = to_email
+ msg["Subject"] = f"[Bambuddy] {subject}"
+ msg.attach(MIMEText(body, "plain"))
if security == "ssl":
# Direct SSL connection (typically port 465)
@@ -682,7 +739,13 @@ class NotificationService:
elif provider.provider_type == "telegram":
return await self._send_telegram(config, f"*{title}*\n{message}", image_data=image_data)
elif provider.provider_type == "email":
- return await self._send_email(config, title, message)
+ # finish_photo_url is pulled from the rendered template variables
+ # so _send_email can detect whether the template referenced the
+ # URL and inline-embed the photo only in that case.
+ finish_photo_url = (variables or {}).get("finish_photo_url")
+ return await self._send_email(
+ config, title, message, image_data=image_data, finish_photo_url=finish_photo_url
+ )
elif provider.provider_type == "discord":
return await self._send_discord(config, title, message, image_data=image_data)
elif provider.provider_type == "webhook":
diff --git a/backend/tests/unit/services/test_notification_service.py b/backend/tests/unit/services/test_notification_service.py
index b52372099..a65db277e 100644
--- a/backend/tests/unit/services/test_notification_service.py
+++ b/backend/tests/unit/services/test_notification_service.py
@@ -2346,3 +2346,199 @@ class TestNtfyOutbound:
assert ok is False
assert "Cloudflare" in detail
+
+
+class TestEmailProvider:
+ """Tests for SMTP email provider, including #1792 finish-photo inline embed.
+
+ Embed is opt-in via the template: only when the user's template referenced
+ ``{finish_photo_url}`` (so the URL appears in the rendered body) AND the
+ photo bytes are available does ``_send_email`` build the multipart/related
+ shape. Otherwise it stays single-part text — no surprise inline image.
+ """
+
+ PHOTO_URL = "https://printer.local/api/v1/archives/42/photos/finish.jpg"
+
+ @pytest.fixture
+ def service(self):
+ return NotificationService()
+
+ @pytest.fixture
+ def smtp_config(self):
+ return {
+ "smtp_server": "smtp.example.com",
+ "smtp_port": "587",
+ "username": "alice",
+ "password": "secret",
+ "from_email": "bambuddy@example.com",
+ "to_email": "alice@example.com",
+ "security": "starttls",
+ "auth_enabled": "true",
+ }
+
+ @staticmethod
+ def _fake_smtp_class(captured: dict):
+ class FakeSMTP:
+ def __init__(self, host, port):
+ captured["host"] = host
+ captured["port"] = port
+
+ def starttls(self):
+ captured["starttls"] = True
+
+ def login(self, u, p):
+ captured["login"] = (u, p)
+
+ def sendmail(self, frm, to, body):
+ captured["from"] = frm
+ captured["to"] = to
+ captured["raw"] = body
+
+ def quit(self):
+ captured["quit"] = True
+
+ return FakeSMTP
+
+ @pytest.mark.asyncio
+ async def test_email_without_image_or_url_stays_text_only(self, service, smtp_config):
+ """No image_data and no URL in body → original single-part text shape."""
+ captured: dict = {}
+ with patch("backend.app.services.notification_service.smtplib.SMTP", self._fake_smtp_class(captured)):
+ ok, _ = await service._send_email(smtp_config, "Print Failed", "Reason: unknown")
+
+ assert ok is True
+ assert "image/jpeg" not in captured["raw"]
+ assert "multipart/related" not in captured["raw"]
+ assert "cid:bambuddy-finish-photo" not in captured["raw"]
+ assert "Reason: unknown" in captured["raw"]
+
+ @pytest.mark.asyncio
+ async def test_email_image_without_template_reference_stays_text_only(self, service, smtp_config):
+ """image_data present but template didn't include {finish_photo_url} → no embed.
+
+ Pins the template-driven contract: a user whose body is just
+ "Print failed. Reason: unknown" does NOT get a surprise inline image
+ stapled to the bottom, even though the photo bytes are available
+ upstream from the archive.
+ """
+ captured: dict = {}
+ with patch("backend.app.services.notification_service.smtplib.SMTP", self._fake_smtp_class(captured)):
+ ok, _ = await service._send_email(
+ smtp_config,
+ "Print Failed",
+ "Reason: unknown",
+ image_data=b"\xff\xd8\xff\xe0jpeg",
+ finish_photo_url=self.PHOTO_URL,
+ )
+
+ assert ok is True
+ raw = captured["raw"]
+ assert "image/jpeg" not in raw
+ assert "multipart/related" not in raw
+ assert "cid:bambuddy-finish-photo" not in raw
+
+ @pytest.mark.asyncio
+ async def test_email_inlines_when_template_uses_finish_photo_url(self, service, smtp_config):
+ """URL in body + image_data present → multipart/related + cid embed; HTML swaps URL for
."""
+ captured: dict = {}
+ body = f"Print failed. Reason: unknown\n\nSnapshot: {self.PHOTO_URL}"
+
+ with patch("backend.app.services.notification_service.smtplib.SMTP", self._fake_smtp_class(captured)):
+ ok, _ = await service._send_email(
+ smtp_config,
+ "Print Failed",
+ body,
+ image_data=b"\xff\xd8\xff\xe0fake-jpeg-bytes",
+ finish_photo_url=self.PHOTO_URL,
+ )
+
+ assert ok is True
+ raw = captured["raw"]
+ # multipart/related shape with both alt parts and an image part
+ assert "multipart/related" in raw
+ assert "multipart/alternative" in raw
+ assert "text/plain" in raw
+ assert "text/html" in raw
+ assert "image/jpeg" in raw
+ # HTML references the exact cid the Content-ID header registers
+ assert "Content-ID: " in raw
+ assert 'src="cid:bambuddy-finish-photo"' in raw
+ # Inline disposition so renders embedded, not as download attachment
+ assert 'Content-Disposition: inline; filename="finish-photo.jpg"' in raw
+ # Plain-text body keeps the URL so non-HTML clients still get a clickable link
+ assert self.PHOTO_URL in raw
+
+ @pytest.mark.asyncio
+ async def test_email_image_data_without_url_arg_stays_text_only(self, service, smtp_config):
+ """image_data passed but finish_photo_url=None → defence-in-depth, no embed.
+
+ Even if a future caller forgets to thread the URL through but does pass
+ the bytes, the conservative default is no embed (avoids attaching an
+ unreferenced image to an unrelated event type).
+ """
+ captured: dict = {}
+ with patch("backend.app.services.notification_service.smtplib.SMTP", self._fake_smtp_class(captured)):
+ ok, _ = await service._send_email(
+ smtp_config,
+ "Print Failed",
+ f"Snapshot: {self.PHOTO_URL}",
+ image_data=b"\xff\xd8\xff\xe0jpeg",
+ finish_photo_url=None,
+ )
+
+ assert ok is True
+ assert "image/jpeg" not in captured["raw"]
+ assert "multipart/related" not in captured["raw"]
+
+ @pytest.mark.asyncio
+ async def test_email_html_body_escapes_user_content(self, service, smtp_config):
+ """Template-rendered body must not be injected raw into the HTML part."""
+ captured: dict = {}
+ body = f"Filename: \nLine 2\nSnapshot: {self.PHOTO_URL}"
+
+ with patch("backend.app.services.notification_service.smtplib.SMTP", self._fake_smtp_class(captured)):
+ ok, _ = await service._send_email(
+ smtp_config,
+ "Print Failed",
+ body,
+ image_data=b"\xff\xd8\xff\xe0jpeg",
+ finish_photo_url=self.PHOTO_URL,
+ )
+
+ assert ok is True
+ raw = captured["raw"]
+ # Raw HTML must NOT round-trip into the HTML part — verify escaped form is present.
+ assert "<script>alert(1)</script>" in raw
+ # Newlines in the body become
in HTML
+ assert "Line 2" in raw
+ assert "
" in raw
+
+ @pytest.mark.asyncio
+ async def test_email_html_swaps_url_for_img_tag(self, service, smtp_config):
+ """In the HTML part, the URL substring is replaced with the
tag.
+
+ Plain text keeps the URL; HTML clients see the inline image where the
+ URL was. The URL must NOT appear inside an wrapping the image
+ — we replace the URL outright with the img tag (renderers don't need
+ the URL twice in the HTML part when the image is already inline).
+ """
+ captured: dict = {}
+ body = f"See: {self.PHOTO_URL} for the snapshot."
+
+ with patch("backend.app.services.notification_service.smtplib.SMTP", self._fake_smtp_class(captured)):
+ ok, _ = await service._send_email(
+ smtp_config,
+ "Print Failed",
+ body,
+ image_data=b"\xff\xd8\xff\xe0jpeg",
+ finish_photo_url=self.PHOTO_URL,
+ )
+
+ assert ok is True
+ raw = captured["raw"]
+ # The
tag appears in the HTML part
+ assert 'src="cid:bambuddy-finish-photo"' in raw
+ # The escaped URL is the marker we replaced — the HTML part should not
+ # contain BOTH the escaped URL AND the cid img (we swapped, not duplicated).
+ # The plain-text part still has the URL; check it's there at least once.
+ assert self.PHOTO_URL in raw
diff --git a/backend/tests/unit/test_user_print_template_rename_migration.py b/backend/tests/unit/test_user_print_template_rename_migration.py
new file mode 100644
index 000000000..337655a79
--- /dev/null
+++ b/backend/tests/unit/test_user_print_template_rename_migration.py
@@ -0,0 +1,146 @@
+"""Regression test for the user_print_* notification template rename migration (#1792).
+
+The four ``user_print_*`` notification templates seeded with names like
+"User Print Completed" looked indistinguishable from the provider-level
+"Print Completed" template in the Message Templates list (the EVENT_NAMES
+display map in routes/notification_templates.py already used the disambiguated
+"User Print Completed Email" label, but the seed wrote the short name to the
+DB, so the UI rendered the ambiguous one).
+
+The migration appends " Email" to those four template names IF AND ONLY IF
+the row still has the old default name — admins who renamed the template
+themselves keep their custom name. This test verifies both branches.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from backend.app.core.database import _migrate_rename_user_print_template_names
+
+
+@pytest.fixture
+async def engine():
+ """In-memory SQLite with just the notification_templates table.
+
+ The migration is a single UPDATE on one table, so the fixture only needs
+ that table — avoids the brittleness of registering every model in the
+ project just to satisfy run_migrations's broader DDL surface.
+ """
+ from backend.app.models.notification_template import NotificationTemplate
+
+ engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+ async with engine.begin() as conn:
+ await conn.run_sync(NotificationTemplate.__table__.create)
+ try:
+ yield engine
+ finally:
+ await engine.dispose()
+
+
+_OLD_DEFAULTS = {
+ "user_print_start": "User Print Started",
+ "user_print_complete": "User Print Completed",
+ "user_print_failed": "User Print Failed",
+ "user_print_stopped": "User Print Stopped",
+}
+_NEW_DEFAULTS = {
+ "user_print_start": "User Print Started Email",
+ "user_print_complete": "User Print Completed Email",
+ "user_print_failed": "User Print Failed Email",
+ "user_print_stopped": "User Print Stopped Email",
+}
+
+
+async def _insert_template(conn, event_type: str, name: str) -> None:
+ await conn.execute(
+ text(
+ "INSERT INTO notification_templates "
+ "(event_type, name, title_template, body_template, is_default) "
+ "VALUES (:et, :n, 't', 'b', 1)"
+ ),
+ {"et": event_type, "n": name},
+ )
+
+
+async def _name_for(conn, event_type: str) -> str:
+ return (
+ await conn.execute(
+ text("SELECT name FROM notification_templates WHERE event_type = :et"),
+ {"et": event_type},
+ )
+ ).scalar_one()
+
+
+async def test_migration_renames_default_named_user_print_rows(engine):
+ """Rows with the old default name get the new disambiguated name."""
+ async with engine.begin() as conn:
+ for event_type, old_name in _OLD_DEFAULTS.items():
+ await _insert_template(conn, event_type, old_name)
+
+ async with engine.begin() as conn:
+ await _migrate_rename_user_print_template_names(conn)
+
+ async with engine.begin() as conn:
+ for event_type, new_name in _NEW_DEFAULTS.items():
+ assert await _name_for(conn, event_type) == new_name
+
+
+async def test_migration_preserves_user_edited_names(engine):
+ """An admin who renamed a template keeps their custom name across the migration."""
+ async with engine.begin() as conn:
+ await _insert_template(conn, "user_print_complete", "My Custom Renamed Template")
+ await _insert_template(conn, "user_print_failed", "User Print Failed") # still default
+
+ async with engine.begin() as conn:
+ await _migrate_rename_user_print_template_names(conn)
+
+ async with engine.begin() as conn:
+ # Custom name preserved
+ assert await _name_for(conn, "user_print_complete") == "My Custom Renamed Template"
+ # Default name renamed
+ assert await _name_for(conn, "user_print_failed") == "User Print Failed Email"
+
+
+async def test_migration_does_not_touch_provider_templates(engine):
+ """The non-user provider templates with similar names must not be renamed."""
+ async with engine.begin() as conn:
+ await _insert_template(conn, "print_complete", "Print Completed")
+ await _insert_template(conn, "print_failed", "Print Failed")
+
+ async with engine.begin() as conn:
+ await _migrate_rename_user_print_template_names(conn)
+
+ async with engine.begin() as conn:
+ assert await _name_for(conn, "print_complete") == "Print Completed"
+ assert await _name_for(conn, "print_failed") == "Print Failed"
+
+
+async def test_migration_is_idempotent(engine):
+ """Running the migration twice must not double-suffix already-renamed rows."""
+ async with engine.begin() as conn:
+ for event_type, old_name in _OLD_DEFAULTS.items():
+ await _insert_template(conn, event_type, old_name)
+
+ async with engine.begin() as conn:
+ await _migrate_rename_user_print_template_names(conn)
+ async with engine.begin() as conn:
+ await _migrate_rename_user_print_template_names(conn)
+
+ async with engine.begin() as conn:
+ for event_type, new_name in _NEW_DEFAULTS.items():
+ current = await _name_for(conn, event_type)
+ assert current == new_name
+ assert "Email Email" not in current
+
+
+async def test_migration_handles_empty_table(engine):
+ """Migration on an empty table must be a safe no-op (fresh install path)."""
+ async with engine.begin() as conn:
+ await _migrate_rename_user_print_template_names(conn)
+
+ async with engine.begin() as conn:
+ count = (await conn.execute(text("SELECT COUNT(*) FROM notification_templates"))).scalar_one()
+ assert count == 0