Add Mailpit SMTP test server for Notification Service email testing
Adds a fourth Docker service (Mailpit) to capture outgoing emails without delivery, with CLI tool for sending test emails, listing/reading captured messages, and clearing the inbox. Supports BCC pattern matching ScadaLink's notification delivery model.
This commit is contained in:
Joseph Doherty
189
infra/tools/smtp_tool.py
Normal file
189
infra/tools/smtp_tool.py
Normal file
@@ -0,0 +1,189 @@
|
||||
#!/usr/bin/env python3
|
||||
"""SMTP/Mailpit client tool for ScadaLink test infrastructure."""
|
||||
|
||||
import argparse
|
||||
import email.mime.text
|
||||
import json
|
||||
import smtplib
|
||||
import sys
|
||||
import urllib.request
|
||||
|
||||
|
||||
DEFAULT_SMTP_HOST = "localhost"
|
||||
DEFAULT_SMTP_PORT = 1025
|
||||
DEFAULT_API_URL = "http://localhost:8025/api"
|
||||
|
||||
|
||||
def cmd_check(args):
|
||||
"""Test SMTP connectivity and report Mailpit status."""
|
||||
# Test SMTP connection
|
||||
try:
|
||||
server = smtplib.SMTP(args.host, args.port, timeout=5)
|
||||
server.ehlo()
|
||||
smtp_ok = True
|
||||
server.quit()
|
||||
except Exception as e:
|
||||
smtp_ok = False
|
||||
smtp_err = str(e)
|
||||
|
||||
# Test API/web UI
|
||||
try:
|
||||
req = urllib.request.Request(f"{args.api}/v1/info")
|
||||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||||
info = json.loads(resp.read())
|
||||
api_ok = True
|
||||
except Exception as e:
|
||||
api_ok = False
|
||||
api_err = str(e)
|
||||
|
||||
print(f"SMTP ({args.host}:{args.port}): {'OK' if smtp_ok else 'FAILED - ' + smtp_err}")
|
||||
print(f"Web UI/API ({args.api}): {'OK' if api_ok else 'FAILED - ' + api_err}")
|
||||
|
||||
if api_ok:
|
||||
print(f"\nMailpit version: {info.get('Version', 'unknown')}")
|
||||
print(f"Database path: {info.get('DatabasePath', 'unknown')}")
|
||||
messages = info.get('Messages', 0)
|
||||
print(f"Stored messages: {messages}")
|
||||
|
||||
if smtp_ok and api_ok:
|
||||
print("\nSMTP server is healthy.")
|
||||
else:
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def cmd_send(args):
|
||||
"""Send a test email via SMTP."""
|
||||
msg = email.mime.text.MIMEText(args.body, "plain")
|
||||
msg["Subject"] = args.subject
|
||||
msg["From"] = args.sender
|
||||
msg["To"] = args.to
|
||||
|
||||
if args.bcc:
|
||||
recipients = [args.to] + [b.strip() for b in args.bcc.split(",")]
|
||||
else:
|
||||
recipients = [args.to]
|
||||
|
||||
try:
|
||||
server = smtplib.SMTP(args.host, args.port, timeout=5)
|
||||
server.ehlo()
|
||||
server.sendmail(args.sender, recipients, msg.as_string())
|
||||
server.quit()
|
||||
bcc_note = f" (BCC: {args.bcc})" if args.bcc else ""
|
||||
print(f"Sent: {args.sender} -> {args.to}{bcc_note}")
|
||||
print(f"Subject: {args.subject}")
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def cmd_list(args):
|
||||
"""List messages in Mailpit inbox."""
|
||||
try:
|
||||
req = urllib.request.Request(f"{args.api}/v1/messages?limit={args.limit}")
|
||||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||||
data = json.loads(resp.read())
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
total = data.get("messages_count", 0)
|
||||
messages = data.get("messages", [])
|
||||
|
||||
print(f"Messages: {len(messages)} shown / {total} total\n")
|
||||
|
||||
if not messages:
|
||||
print("(inbox empty)")
|
||||
return
|
||||
|
||||
print(f"{'ID':<24} {'Date':<22} {'From':<30} {'To':<30} {'Subject'}")
|
||||
print("-" * 130)
|
||||
|
||||
for msg in messages:
|
||||
msg_id = msg["ID"]
|
||||
date = msg.get("Date", "")[:21]
|
||||
from_addr = msg.get("From", {}).get("Address", "")[:28]
|
||||
to_list = msg.get("To", [])
|
||||
to_addr = (to_list[0].get("Address", "") if to_list else "")[:28]
|
||||
subject = msg.get("Subject", "")[:40]
|
||||
print(f"{msg_id:<24} {date:<22} {from_addr:<30} {to_addr:<30} {subject}")
|
||||
|
||||
|
||||
def cmd_read(args):
|
||||
"""Read a specific message by ID."""
|
||||
try:
|
||||
req = urllib.request.Request(f"{args.api}/v1/message/{args.id}")
|
||||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||||
msg = json.loads(resp.read())
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
print(f"ID: {msg['ID']}")
|
||||
print(f"Date: {msg.get('Date', '')}")
|
||||
from_info = msg.get("From", {})
|
||||
print(f"From: {from_info.get('Name', '')} <{from_info.get('Address', '')}>")
|
||||
|
||||
for field in ["To", "Cc", "Bcc"]:
|
||||
addrs = msg.get(field, [])
|
||||
if addrs:
|
||||
formatted = ", ".join(f"{a.get('Name', '')} <{a.get('Address', '')}>" for a in addrs)
|
||||
print(f"{field}: {formatted}")
|
||||
|
||||
print(f"Subject: {msg.get('Subject', '')}")
|
||||
print()
|
||||
print(msg.get("Text", "(no text body)"))
|
||||
|
||||
|
||||
def cmd_clear(args):
|
||||
"""Delete all messages in Mailpit."""
|
||||
try:
|
||||
req = urllib.request.Request(f"{args.api}/v1/messages", method="DELETE")
|
||||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||||
resp.read()
|
||||
print("All messages deleted.")
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="SMTP/Mailpit client tool for ScadaLink test infrastructure")
|
||||
parser.add_argument("--host", default=DEFAULT_SMTP_HOST, help=f"SMTP host (default: {DEFAULT_SMTP_HOST})")
|
||||
parser.add_argument("--port", type=int, default=DEFAULT_SMTP_PORT, help=f"SMTP port (default: {DEFAULT_SMTP_PORT})")
|
||||
parser.add_argument("--api", default=DEFAULT_API_URL, help=f"Mailpit API URL (default: {DEFAULT_API_URL})")
|
||||
|
||||
sub = parser.add_subparsers(dest="command", required=True)
|
||||
|
||||
sub.add_parser("check", help="Test SMTP connectivity and Mailpit status")
|
||||
|
||||
send_p = sub.add_parser("send", help="Send a test email")
|
||||
send_p.add_argument("--sender", default="scada-notifications@company.com",
|
||||
help="From address (default: scada-notifications@company.com)")
|
||||
send_p.add_argument("--to", required=True, help="Recipient address")
|
||||
send_p.add_argument("--bcc", help="Comma-separated BCC addresses")
|
||||
send_p.add_argument("--subject", default="Test notification", help="Subject line")
|
||||
send_p.add_argument("--body", default="This is a test notification from ScadaLink.", help="Message body")
|
||||
|
||||
list_p = sub.add_parser("list", help="List messages in Mailpit inbox")
|
||||
list_p.add_argument("--limit", type=int, default=20, help="Number of messages to show (default: 20)")
|
||||
|
||||
read_p = sub.add_parser("read", help="Read a specific message by ID")
|
||||
read_p.add_argument("--id", required=True, help="Message ID (from list command)")
|
||||
|
||||
sub.add_parser("clear", help="Delete all messages")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
commands = {
|
||||
"check": cmd_check,
|
||||
"send": cmd_send,
|
||||
"list": cmd_list,
|
||||
"read": cmd_read,
|
||||
"clear": cmd_clear,
|
||||
}
|
||||
|
||||
commands[args.command](args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user