One secured SMTP path, optional login, sender and reply address, failing health
The message endpoint used implicit TLS and the PDF endpoint STARTTLS on the same port, so one of them always failed. Both now share one connection path with an explicit SMTP_RELAY_SECURITY, log in only with credentials, and take SMTP_FROM and SMTP_REPLY_TO. /health answers 503 when the SMTP server refuses, so the container turns unhealthy instead of looking fine. The old tests never ran; eight new ones do, in CI too.
This commit is contained in:
6 files changed
+254
-190
No files matched your search
@@ -8,7 +8,20 @@ on:
|
||||
branches: [main]
|
||||
|
||||
jobs:
|
||||
test:
|
||||
runs-on: ubuntu-latest
|
||||
container:
|
||||
image: git.librete.ch/public/runner-image:v0.2.0@sha256:f60c587d3c0b0aac04a572db5349e27672bf76baec2ce547a3dcc28cebcf1b7e
|
||||
timeout-minutes: 10
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- name: unit tests in the image's Python
|
||||
run: |
|
||||
tar -c --exclude=.git . | docker run -i --rm python:3.13.1-alpine3.20 sh -c \
|
||||
'mkdir /w && tar -x -C /w && cd /w && pip install -q -r requirements.txt && PYTHONPATH=src python -m unittest discover -s src/tests'
|
||||
|
||||
build:
|
||||
needs: test
|
||||
runs-on: ubuntu-latest
|
||||
container:
|
||||
image: git.librete.ch/public/runner-image:v0.2.0@sha256:f60c587d3c0b0aac04a572db5349e27672bf76baec2ce547a3dcc28cebcf1b7e
|
||||
|
||||
@@ -2,6 +2,19 @@
|
||||
|
||||
All notable changes to libreshop/mail are documented here.
|
||||
|
||||
## v0.2.0 - 2026-10-15
|
||||
|
||||
- One SMTP connection path for `/v1/send/message`, `/v1/send/pdf` and `/health`. Before, the message endpoint always
|
||||
used implicit TLS and the PDF endpoint always STARTTLS on the same port, so one of them failed whatever the port.
|
||||
- `SMTP_RELAY_SECURITY` (`ssl`, `starttls`, `none`; default by port). A failed STARTTLS is an error instead of a
|
||||
plain-text login.
|
||||
- Login only when `SMTP_RELAY_USERNAME` and `SMTP_RELAY_PASSWORD` are both set, so a trusted relay works without
|
||||
credentials.
|
||||
- `SMTP_FROM` sets the sender (default `SMTP_RELAY_USERNAME`), `SMTP_REPLY_TO` the reply address. Every mail carries
|
||||
`Date` and `Message-ID`.
|
||||
- `/health` answers 503 when the SMTP server does not accept the connection, so Docker marks the container unhealthy.
|
||||
- Unit tests that run (8), and a CI test job.
|
||||
|
||||
## Unreleased
|
||||
|
||||
- Extracted from `mp/mail/` (2026-04-29). The component history before
|
||||
|
||||
@@ -18,6 +18,18 @@ pulls `git.librete.ch/libreshop/mail:<pin>` instead of building locally.
|
||||
docker build -t libreshop/mail:dev .
|
||||
```
|
||||
|
||||
## Settings
|
||||
|
||||
| variable | meaning |
|
||||
|---|---|
|
||||
| `SMTP_RELAY_HOST`, `SMTP_RELAY_PORT` | the SMTP server |
|
||||
| `SMTP_RELAY_SECURITY` | `ssl` (port 465), `starttls` (587 or 25) or `none` (a relay on a private network); default by port |
|
||||
| `SMTP_RELAY_USERNAME`, `SMTP_RELAY_PASSWORD` | login, used only when both are set |
|
||||
| `SMTP_FROM` | sender, e.g. `MUELLERPRINTS <shop@example.org>`; default `SMTP_RELAY_USERNAME` |
|
||||
| `SMTP_REPLY_TO` | optional reply address |
|
||||
|
||||
`GET /health` answers 503 when the SMTP server does not accept the connection.
|
||||
|
||||
## Adapter contract
|
||||
|
||||
See `docker-entrypoint.sh` and `Dockerfile` for the runtime surface.
|
||||
|
||||
+103
-68
@@ -1,6 +1,7 @@
|
||||
from flask import Flask, request, jsonify
|
||||
from prometheus_flask_exporter import PrometheusMetrics
|
||||
import smtplib
|
||||
from email.utils import formatdate, make_msgid, parseaddr
|
||||
from email.mime.text import MIMEText
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.application import MIMEApplication
|
||||
@@ -66,6 +67,82 @@ pdf_sent_counter = metrics.counter(
|
||||
)
|
||||
|
||||
|
||||
# --- SMTP --------------------------------------------------------------------
|
||||
# One connection path for every endpoint. SMTP_RELAY_SECURITY is "ssl" (TLS from the first byte, port 465), "starttls"
|
||||
# (port 587 or 25) or "none" (a relay on a private network, e.g. Mailu's smtp container trusting its own subnet).
|
||||
# Without it, port 465 means ssl and any other port starttls. The service logs in only when both SMTP_RELAY_USERNAME
|
||||
# and SMTP_RELAY_PASSWORD are set. SMTP_FROM is the sender (default: SMTP_RELAY_USERNAME); SMTP_REPLY_TO, if set,
|
||||
# is where replies go.
|
||||
SECURITIES = ("ssl", "starttls", "none")
|
||||
|
||||
|
||||
class SmtpConfigError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def smtp_settings():
|
||||
host = os.environ.get('SMTP_RELAY_HOST', '')
|
||||
port_text = os.environ.get('SMTP_RELAY_PORT', '')
|
||||
if not host or not port_text.isdigit():
|
||||
raise SmtpConfigError("SMTP_RELAY_HOST and a numeric SMTP_RELAY_PORT are required")
|
||||
port = int(port_text)
|
||||
security = os.environ.get('SMTP_RELAY_SECURITY') or ("ssl" if port == 465 else "starttls")
|
||||
if security not in SECURITIES:
|
||||
raise SmtpConfigError(f"SMTP_RELAY_SECURITY must be one of {', '.join(SECURITIES)}")
|
||||
username = os.environ.get('SMTP_RELAY_USERNAME', '')
|
||||
password = os.environ.get('SMTP_RELAY_PASSWORD', '')
|
||||
sender = os.environ.get('SMTP_FROM') or username
|
||||
if not parseaddr(sender)[1]:
|
||||
raise SmtpConfigError("SMTP_FROM (or SMTP_RELAY_USERNAME) must be a mail address")
|
||||
return {
|
||||
'host': host, 'port': port, 'security': security, 'username': username, 'password': password,
|
||||
'sender': sender, 'reply_to': os.environ.get('SMTP_REPLY_TO', '')
|
||||
}
|
||||
|
||||
|
||||
def open_smtp(settings, timeout=30):
|
||||
"""Connects, secures the connection as configured and logs in if credentials are set. A failed STARTTLS is an error:
|
||||
the credentials and the mail must not travel in the clear."""
|
||||
if settings['security'] == "ssl":
|
||||
smtp = smtplib.SMTP_SSL(settings['host'], settings['port'], timeout=timeout)
|
||||
else:
|
||||
smtp = smtplib.SMTP(settings['host'], settings['port'], timeout=timeout)
|
||||
try:
|
||||
smtp.ehlo()
|
||||
if settings['security'] == "starttls":
|
||||
smtp.starttls()
|
||||
smtp.ehlo()
|
||||
if settings['username'] and settings['password']:
|
||||
smtp.login(settings['username'], settings['password'])
|
||||
except Exception:
|
||||
smtp.close()
|
||||
raise
|
||||
return smtp
|
||||
|
||||
|
||||
def new_message(settings, to_email, subject):
|
||||
msg = MIMEMultipart('mixed')
|
||||
msg['From'] = settings['sender']
|
||||
msg['To'] = to_email
|
||||
msg['Subject'] = subject
|
||||
msg['Date'] = formatdate(localtime=False)
|
||||
msg['Message-ID'] = make_msgid(domain=parseaddr(settings['sender'])[1].rpartition('@')[2] or None)
|
||||
if settings['reply_to']:
|
||||
msg['Reply-To'] = settings['reply_to']
|
||||
return msg
|
||||
|
||||
|
||||
def deliver(settings, to_email, msg):
|
||||
smtp = open_smtp(settings)
|
||||
try:
|
||||
smtp.sendmail(parseaddr(settings['sender'])[1], [to_email], msg.as_string())
|
||||
finally:
|
||||
try:
|
||||
smtp.quit()
|
||||
except smtplib.SMTPException:
|
||||
smtp.close()
|
||||
|
||||
|
||||
@app.route('/v1')
|
||||
@metrics.do_not_track()
|
||||
def info():
|
||||
@@ -87,18 +164,9 @@ def send_message():
|
||||
logger.info(f"Sending text email to {to_email}")
|
||||
logger.debug(f"Subject: {subject}")
|
||||
|
||||
# SMTP Configuration
|
||||
smtp_server = os.environ.get('SMTP_RELAY_HOST')
|
||||
smtp_port = int(os.environ.get('SMTP_RELAY_PORT'))
|
||||
smtp_username = os.environ.get('SMTP_RELAY_USERNAME')
|
||||
smtp_password = os.environ.get('SMTP_RELAY_PASSWORD')
|
||||
|
||||
try:
|
||||
# Create a message
|
||||
msg = MIMEMultipart('mixed')
|
||||
msg['From'] = smtp_username
|
||||
msg['To'] = to_email
|
||||
msg['Subject'] = subject
|
||||
settings = smtp_settings()
|
||||
msg = new_message(settings, to_email, subject)
|
||||
|
||||
# Create the multipart/alternative part
|
||||
msgAlternative = MIMEMultipart('alternative')
|
||||
@@ -112,12 +180,8 @@ def send_message():
|
||||
logger.debug("HTML content included in email")
|
||||
msgAlternative.attach(MIMEText(html_content, 'html', 'utf-8'))
|
||||
|
||||
# Send email
|
||||
logger.debug(f"Connecting to SMTP server {smtp_server}:{smtp_port}")
|
||||
with smtplib.SMTP_SSL(smtp_server, smtp_port) as smtp:
|
||||
smtp.login(smtp_username, smtp_password)
|
||||
smtp.sendmail(smtp_username, to_email, msg.as_string())
|
||||
smtp.quit()
|
||||
logger.debug(f"Connecting to SMTP server {settings['host']}:{settings['port']} ({settings['security']})")
|
||||
deliver(settings, to_email, msg)
|
||||
|
||||
logger.info("Email sent successfully")
|
||||
return jsonify({"message": "Email sent successfully"}), 200
|
||||
@@ -140,18 +204,9 @@ def send_pdf():
|
||||
logger.info(f"Sending PDF email to {to_email}")
|
||||
logger.debug(f"Subject: {subject}")
|
||||
|
||||
# SMTP Configuration
|
||||
smtp_server = os.environ.get('SMTP_RELAY_HOST')
|
||||
smtp_port = int(os.environ.get('SMTP_RELAY_PORT'))
|
||||
smtp_username = os.environ.get('SMTP_RELAY_USERNAME')
|
||||
smtp_password = os.environ.get('SMTP_RELAY_PASSWORD')
|
||||
|
||||
try:
|
||||
# Create a message with mixed content (attachments + text)
|
||||
msg = MIMEMultipart('mixed')
|
||||
msg['From'] = smtp_username
|
||||
msg['To'] = to_email
|
||||
msg['Subject'] = subject
|
||||
settings = smtp_settings()
|
||||
msg = new_message(settings, to_email, subject)
|
||||
|
||||
# Create the multipart/alternative part for text and HTML
|
||||
msgAlternative = MIMEMultipart('alternative')
|
||||
@@ -174,19 +229,8 @@ def send_pdf():
|
||||
else:
|
||||
logger.warning("No PDF attachment provided")
|
||||
|
||||
# Send email
|
||||
logger.debug(f"Connecting to SMTP server {smtp_server}:{smtp_port}")
|
||||
with smtplib.SMTP(smtp_server, smtp_port) as smtp:
|
||||
logger.debug("SMTP connection established")
|
||||
try:
|
||||
smtp.starttls()
|
||||
logger.debug("STARTTLS successful")
|
||||
except Exception as e:
|
||||
logger.error(f"STARTTLS failed: {str(e)}")
|
||||
|
||||
smtp.login(smtp_username, smtp_password)
|
||||
smtp.sendmail(smtp_username, to_email, msg.as_string())
|
||||
logger.debug("Email sent via SMTP")
|
||||
logger.debug(f"Connecting to SMTP server {settings['host']}:{settings['port']} ({settings['security']})")
|
||||
deliver(settings, to_email, msg)
|
||||
|
||||
logger.info("PDF email sent successfully")
|
||||
return jsonify({"message": "Email sent successfully"}), 200
|
||||
@@ -197,43 +241,34 @@ def send_pdf():
|
||||
|
||||
@app.route('/health', methods=['GET'])
|
||||
def health_check():
|
||||
"""503 unless the configured SMTP server answers and accepts the connection's security and login, so Docker's
|
||||
healthcheck marks the container unhealthy when no mail can leave."""
|
||||
logger = logging.getLogger('app:health-check')
|
||||
health_status = {'status': 'ok'}
|
||||
health_status = {'status': 'ok', 'smtp_config': True, 'smtp_responding': False}
|
||||
|
||||
# Check SMTP configuration
|
||||
smtp_server = os.environ.get('SMTP_RELAY_HOST')
|
||||
smtp_port = os.environ.get('SMTP_RELAY_PORT')
|
||||
smtp_username = os.environ.get('SMTP_RELAY_USERNAME')
|
||||
|
||||
if not all([smtp_server, smtp_port, smtp_username]):
|
||||
logger.error("SMTP configuration incomplete")
|
||||
health_status['smtp_config'] = False
|
||||
health_status['status'] = 'error'
|
||||
else:
|
||||
health_status['smtp_config'] = True
|
||||
|
||||
# Try to connect to SMTP server
|
||||
try:
|
||||
smtp_port_int = int(smtp_port)
|
||||
use_ssl = smtp_port_int == 465
|
||||
settings = smtp_settings()
|
||||
except SmtpConfigError as e:
|
||||
logger.error(f"SMTP configuration incomplete: {e}")
|
||||
health_status.update(status='error', smtp_config=False)
|
||||
return jsonify(health_status), 503
|
||||
|
||||
if use_ssl:
|
||||
with smtplib.SMTP_SSL(smtp_server, smtp_port_int, timeout=5) as smtp:
|
||||
smtp.ehlo()
|
||||
try:
|
||||
smtp = open_smtp(settings, timeout=5)
|
||||
try:
|
||||
smtp.noop()
|
||||
health_status['smtp_responding'] = True
|
||||
else:
|
||||
with smtplib.SMTP(smtp_server, smtp_port_int, timeout=5) as smtp:
|
||||
smtp.ehlo()
|
||||
health_status['smtp_responding'] = True
|
||||
|
||||
finally:
|
||||
try:
|
||||
smtp.quit()
|
||||
except smtplib.SMTPException:
|
||||
smtp.close()
|
||||
except Exception as e:
|
||||
logger.error(f"SMTP connection test failed: {str(e)}")
|
||||
health_status['smtp_responding'] = False
|
||||
health_status['status'] = 'error'
|
||||
|
||||
logger.debug(f"Health check performed: {health_status}")
|
||||
return jsonify(health_status)
|
||||
|
||||
return jsonify(health_status), (200 if health_status['status'] == 'ok' else 503)
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(host='0.0.0.0', port=2222, debug=False)
|
||||
+113
-78
@@ -1,92 +1,127 @@
|
||||
import email
|
||||
import os
|
||||
import smtplib
|
||||
import unittest
|
||||
from flask import Flask
|
||||
from flask.testing import FlaskClient
|
||||
from unittest.mock import patch, MagicMock
|
||||
from io import BytesIO
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from app import app
|
||||
|
||||
class TestApp(unittest.TestCase):
|
||||
RELAY = {'SMTP_RELAY_HOST': 'smtp', 'SMTP_RELAY_PORT': '25', 'SMTP_RELAY_SECURITY': 'none',
|
||||
'SMTP_RELAY_USERNAME': '', 'SMTP_RELAY_PASSWORD': '',
|
||||
'SMTP_FROM': 'MUELLERPRINTS <shop@pilot.mail.example>', 'SMTP_REPLY_TO': 'paperwork@shop.example'}
|
||||
SUBMISSION = {'SMTP_RELAY_HOST': 'smtp.example', 'SMTP_RELAY_PORT': '587',
|
||||
'SMTP_RELAY_USERNAME': 'paperwork@shop.example', 'SMTP_RELAY_PASSWORD': 'pw'}
|
||||
IMPLICIT_TLS = dict(SUBMISSION, SMTP_RELAY_PORT='465')
|
||||
|
||||
|
||||
def message(smtp):
|
||||
"""The envelope sender, the recipients and the parsed message of the one sendmail call."""
|
||||
smtp.sendmail.assert_called_once()
|
||||
sender, recipients, text = smtp.sendmail.call_args.args
|
||||
return sender, recipients, email.message_from_string(text)
|
||||
|
||||
|
||||
class SendTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
app.testing = True
|
||||
self.app = app.test_client()
|
||||
self.client = app.test_client()
|
||||
|
||||
def test_info_endpoint(self):
|
||||
response = self.app.get('/v1')
|
||||
def post_message(self):
|
||||
return self.client.post('/v1/send/message', json={'subject': 'Bestellung', 'message': 'Danke', 'to_email': 'kunde@example.com'})
|
||||
|
||||
def post_pdf(self):
|
||||
return self.client.post('/v1/send/pdf', data={'subject': 'Rechnung', 'message': 'Anbei', 'to_email': 'kunde@example.com',
|
||||
'pdf_blob': (BytesIO(b'%PDF-1.4'), 'document.pdf')}, content_type='multipart/form-data')
|
||||
|
||||
@patch.dict(os.environ, RELAY, clear=True)
|
||||
@patch('smtplib.SMTP_SSL')
|
||||
@patch('smtplib.SMTP')
|
||||
def test_relay_without_login_or_tls(self, smtp_cls, ssl_cls):
|
||||
smtp = smtp_cls.return_value
|
||||
self.assertEqual(self.post_message().status_code, 200)
|
||||
smtp_cls.assert_called_once_with('smtp', 25, timeout=30)
|
||||
ssl_cls.assert_not_called()
|
||||
smtp.starttls.assert_not_called()
|
||||
smtp.login.assert_not_called()
|
||||
sender, recipients, msg = message(smtp)
|
||||
self.assertEqual(sender, 'shop@pilot.mail.example')
|
||||
self.assertEqual(recipients, ['kunde@example.com'])
|
||||
self.assertEqual(msg['From'], 'MUELLERPRINTS <shop@pilot.mail.example>')
|
||||
self.assertEqual(msg['Reply-To'], 'paperwork@shop.example')
|
||||
self.assertTrue(msg['Date'])
|
||||
self.assertTrue(msg['Message-ID'].endswith('@pilot.mail.example>'))
|
||||
smtp.quit.assert_called_once()
|
||||
|
||||
@patch.dict(os.environ, SUBMISSION, clear=True)
|
||||
@patch('smtplib.SMTP')
|
||||
def test_pdf_over_starttls_with_login(self, smtp_cls):
|
||||
smtp = smtp_cls.return_value
|
||||
self.assertEqual(self.post_pdf().status_code, 200)
|
||||
smtp.starttls.assert_called_once()
|
||||
smtp.login.assert_called_once_with('paperwork@shop.example', 'pw')
|
||||
sender, _, msg = message(smtp)
|
||||
self.assertEqual(sender, 'paperwork@shop.example')
|
||||
self.assertEqual(msg['From'], 'paperwork@shop.example')
|
||||
self.assertIsNone(msg['Reply-To'])
|
||||
self.assertEqual([p.get_content_type() for p in msg.walk()][-1], 'application/pdf')
|
||||
|
||||
@patch.dict(os.environ, IMPLICIT_TLS, clear=True)
|
||||
@patch('smtplib.SMTP')
|
||||
@patch('smtplib.SMTP_SSL')
|
||||
def test_port_465_uses_implicit_tls_for_both_endpoints(self, ssl_cls, smtp_cls):
|
||||
self.assertEqual(self.post_message().status_code, 200)
|
||||
self.assertEqual(self.post_pdf().status_code, 200)
|
||||
self.assertEqual(ssl_cls.call_count, 2)
|
||||
smtp_cls.assert_not_called()
|
||||
ssl_cls.return_value.starttls.assert_not_called()
|
||||
|
||||
@patch.dict(os.environ, SUBMISSION, clear=True)
|
||||
@patch('smtplib.SMTP')
|
||||
def test_failed_starttls_sends_nothing(self, smtp_cls):
|
||||
smtp = smtp_cls.return_value
|
||||
smtp.starttls.side_effect = smtplib.SMTPNotSupportedError('STARTTLS extension not supported by server.')
|
||||
self.assertEqual(self.post_message().status_code, 500)
|
||||
smtp.login.assert_not_called()
|
||||
smtp.sendmail.assert_not_called()
|
||||
smtp.close.assert_called_once()
|
||||
|
||||
@patch.dict(os.environ, SUBMISSION, clear=True)
|
||||
@patch('smtplib.SMTP')
|
||||
def test_failed_login_sends_nothing(self, smtp_cls):
|
||||
smtp = smtp_cls.return_value
|
||||
smtp.login.side_effect = smtplib.SMTPAuthenticationError(535, b'Authentication failed')
|
||||
self.assertEqual(self.post_pdf().status_code, 500)
|
||||
smtp.sendmail.assert_not_called()
|
||||
|
||||
|
||||
class HealthTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
app.testing = True
|
||||
self.client = app.test_client()
|
||||
|
||||
@patch.dict(os.environ, RELAY, clear=True)
|
||||
@patch('smtplib.SMTP')
|
||||
def test_healthy_when_the_server_answers(self, smtp_cls):
|
||||
response = self.client.get('/health')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn(b"mail v1.0.0", response.data)
|
||||
self.assertEqual(response.json, {'status': 'ok', 'smtp_config': True, 'smtp_responding': True})
|
||||
smtp_cls.return_value.noop.assert_called_once()
|
||||
|
||||
@patch.dict(os.environ, SUBMISSION, clear=True)
|
||||
@patch('smtplib.SMTP')
|
||||
def test_send_message_endpoint_success(self, mock_smtp):
|
||||
mock_smtp_instance = MagicMock()
|
||||
mock_smtp.return_value = mock_smtp_instance
|
||||
def test_unhealthy_when_the_server_refuses(self, smtp_cls):
|
||||
smtp_cls.side_effect = smtplib.SMTPConnectError(421, b'IP=192.0.2.1 - Your address is blocked')
|
||||
response = self.client.get('/health')
|
||||
self.assertEqual(response.status_code, 503)
|
||||
self.assertEqual(response.json['smtp_responding'], False)
|
||||
|
||||
with app.app_context():
|
||||
response = self.app.post('/v1/send/message', json={
|
||||
'subject': 'Test Subject',
|
||||
'message': 'Test Message',
|
||||
'to_email': 'test@example.com'
|
||||
})
|
||||
@patch.dict(os.environ, {'SMTP_RELAY_HOST': 'smtp'}, clear=True)
|
||||
def test_unhealthy_without_configuration(self):
|
||||
response = self.client.get('/health')
|
||||
self.assertEqual(response.status_code, 503)
|
||||
self.assertEqual(response.json['smtp_config'], False)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
mock_smtp_instance.starttls.assert_called_once()
|
||||
mock_smtp_instance.login.assert_called_once()
|
||||
mock_smtp_instance.sendmail.assert_called_once()
|
||||
mock_smtp_instance.quit.assert_called_once()
|
||||
|
||||
@patch('smtplib.SMTP')
|
||||
def test_send_message_endpoint_failure(self, mock_smtp):
|
||||
mock_smtp_instance = MagicMock()
|
||||
mock_smtp_instance.login.side_effect = smtplib.SMTPAuthenticationError("Authentication failed")
|
||||
mock_smtp.return_value = mock_smtp_instance
|
||||
|
||||
with app.app_context():
|
||||
response = self.app.post('/v1/send/message', json={
|
||||
'subject': 'Test Subject',
|
||||
'message': 'Test Message',
|
||||
'to_email': 'test@example.com'
|
||||
})
|
||||
|
||||
self.assertEqual(response.status_code, 500)
|
||||
mock_smtp_instance.starttls.assert_called_once()
|
||||
mock_smtp_instance.login.assert_called_once()
|
||||
mock_smtp_instance.sendmail.assert_not_called()
|
||||
mock_smtp_instance.quit.assert_called_once()
|
||||
|
||||
@patch('smtplib.SMTP')
|
||||
def test_send_pdf_endpoint_success(self, mock_smtp):
|
||||
mock_smtp_instance = MagicMock()
|
||||
mock_smtp.return_value = mock_smtp_instance
|
||||
|
||||
with app.app_context():
|
||||
response = self.app.post('/v1/send/pdf', data={
|
||||
'subject': 'Test Subject',
|
||||
'message': 'Test Message',
|
||||
'to_email': 'test@example.com',
|
||||
}, content_type='multipart/form-data', files={'pdf_blob': ('document.pdf', b'PDF_CONTENT')})
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
mock_smtp_instance.starttls.assert_called_once()
|
||||
mock_smtp_instance.login.assert_called_once()
|
||||
mock_smtp_instance.sendmail.assert_called_once()
|
||||
mock_smtp_instance.quit.assert_called_once()
|
||||
|
||||
@patch('smtplib.SMTP')
|
||||
def test_send_pdf_endpoint_failure(self, mock_smtp):
|
||||
mock_smtp_instance = MagicMock()
|
||||
mock_smtp_instance.login.side_effect = smtplib.SMTPAuthenticationError("Authentication failed")
|
||||
mock_smtp.return_value = mock_smtp_instance
|
||||
|
||||
with app.app_context():
|
||||
response = self.app.post('/v1/send/pdf', data={
|
||||
'subject': 'Test Subject',
|
||||
'message': 'Test Message',
|
||||
'to_email': 'test@example.com',
|
||||
}, content_type='multipart/form-data', files={'pdf_blob': ('document.pdf', b'PDF_CONTENT')})
|
||||
|
||||
self.assertEqual(response.status_code, 500)
|
||||
mock_smtp_instance.starttls.assert_called_once()
|
||||
mock_smtp_instance.login.assert_called_once()
|
||||
mock_smtp_instance.sendmail.assert_not_called()
|
||||
mock_smtp_instance.quit.assert_called_once()
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -1,44 +0,0 @@
|
||||
import smtplib
|
||||
import os
|
||||
|
||||
# Get SMTP settings from environment variables
|
||||
smtp_server = os.environ.get('SMTP_RELAY_HOST')
|
||||
smtp_port = int(os.environ.get('SMTP_RELAY_PORT'))
|
||||
smtp_username = os.environ.get('SMTP_RELAY_USERNAME')
|
||||
smtp_password = os.environ.get('SMTP_RELAY_PASSWORD')
|
||||
|
||||
print(f"Testing connection to {smtp_server}:{smtp_port}")
|
||||
|
||||
# Create connection
|
||||
server = smtplib.SMTP(smtp_server, smtp_port)
|
||||
|
||||
# Enable debug output
|
||||
server.set_debuglevel(1)
|
||||
|
||||
# Test EHLO and STARTTLS
|
||||
try:
|
||||
print("Sending EHLO...")
|
||||
server.ehlo()
|
||||
print("EHLO response received. Checking for STARTTLS support...")
|
||||
|
||||
server.starttls()
|
||||
print("STARTTLS successful")
|
||||
|
||||
print("Attempting login...")
|
||||
server.login(smtp_username, smtp_password)
|
||||
print("Login successful")
|
||||
|
||||
server.quit()
|
||||
print("Connection test completed successfully")
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
print("Testing SSL connection instead...")
|
||||
|
||||
try:
|
||||
ssl_server = smtplib.SMTP_SSL(smtp_server, smtp_port)
|
||||
ssl_server.set_debuglevel(1)
|
||||
ssl_server.login(smtp_username, smtp_password)
|
||||
print("SSL connection successful")
|
||||
ssl_server.quit()
|
||||
except Exception as ssl_e:
|
||||
print(f"SSL Error: {ssl_e}")
|
||||
Reference in new issue
Block a user