One secured SMTP path, optional login, sender and reply address, failing health (#3)
Co-authored-by: Michael Czechowski <mail@librete.ch>
This commit was merged in pull request #3.
This commit is contained in:
1 parent
22e642e5bb
commit
2efca9415e
6 files changed
+251
-187
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.
|
||||
|
||||
+104
-69
@@ -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
|
||||
|
||||
if use_ssl:
|
||||
with smtplib.SMTP_SSL(smtp_server, smtp_port_int, timeout=5) as smtp:
|
||||
smtp.ehlo()
|
||||
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
|
||||
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
|
||||
|
||||
try:
|
||||
smtp = open_smtp(settings, timeout=5)
|
||||
try:
|
||||
smtp.noop()
|
||||
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)
|
||||
+109
-74
@@ -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):
|
||||
def setUp(self):
|
||||
app.testing = True
|
||||
self.app = app.test_client()
|
||||
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 test_info_endpoint(self):
|
||||
response = self.app.get('/v1')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn(b"mail v1.0.0", response.data)
|
||||
|
||||
@patch('smtplib.SMTP')
|
||||
def test_send_message_endpoint_success(self, mock_smtp):
|
||||
mock_smtp_instance = MagicMock()
|
||||
mock_smtp.return_value = mock_smtp_instance
|
||||
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)
|
||||
|
||||
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, 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()
|
||||
class SendTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
app.testing = True
|
||||
self.client = app.test_client()
|
||||
|
||||
@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
|
||||
def post_message(self):
|
||||
return self.client.post('/v1/send/message', json={'subject': 'Bestellung', 'message': 'Danke', 'to_email': 'kunde@example.com'})
|
||||
|
||||
with app.app_context():
|
||||
response = self.app.post('/v1/send/message', json={
|
||||
'subject': 'Test Subject',
|
||||
'message': 'Test Message',
|
||||
'to_email': 'test@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')
|
||||
|
||||
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.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('smtplib.SMTP')
|
||||
def test_send_pdf_endpoint_success(self, mock_smtp):
|
||||
mock_smtp_instance = MagicMock()
|
||||
mock_smtp.return_value = mock_smtp_instance
|
||||
@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')
|
||||
|
||||
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')})
|
||||
@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()
|
||||
|
||||
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.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('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
|
||||
@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()
|
||||
|
||||
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()
|
||||
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.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_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)
|
||||
|
||||
@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)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.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