Skip to content

Fix multi-recipient email delivery and SMTP cleanup #118

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ai-yang wants to merge 1 commit into
base: main
Choose a base branch
from
+145 −130
Open

Fix multi-recipient email delivery and SMTP cleanup #118

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
178 changes: 90 additions & 88 deletions tests/test_send_email.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,101 +1,103 @@
#!/usr/bin/python
# -*- coding: UTF-8 -*-

# ****************************************************************************
# Description: Send email unit test
# How to use: python -m unittest -v test_send_email.TestSendEmail
#
# Copyright 2022 Charmve. All Rights Reserved.
# Licensed under the MIT License.
# ****************************************************************************


import datetime
import os
import sys
import smtplib
import unittest
from email.mime.image import MIMEImage
from email.mime.text import MIMEText
from pathlib import Path

TOP_DIR = Path(__file__).parent.parent.joinpath("utils")
sys.path.append(TOP_DIR)
from unittest.mock import ANY, Mock, call, patch

from utils.send_email import send_email


class TestSendEmail(unittest.TestCase):
SRC_DIR = Path(__file__).parent.parent.joinpath("qbot/gui")

# 发件人邮箱
mail_sender = "1144262839@qq.com"
# 邮箱授权码,注意这里不是邮箱密码,如何获取邮箱授权码,请看本文最后教程
mail_license = os.getenv("MAIL_LICENSE")
# 收件人邮箱，可以为多个收件人
mail_receivers = ["yidazhang1@gmail.com"]
# 邮件主题
subject = """Python邮件测试"""

def test_sendtext(self, mail_sender=mail_sender, mail_receivers=mail_receivers):
# 邮件正文内容
body_content = """你好，这是一个测试邮件!"""
# 构造文本,参数1：正文内容，参数2：文本格式，参数3：编码方式
message_text = MIMEText(body_content, "plain", "utf-8")
self.assertTrue(send_email(mail_sender, mail_receivers, message_text))

def test_sendattachment(self):
# 构造附件
attachment = MIMEText(
open(self.SRC_DIR.joinpath("bkt_result/bkt_result.html"), "rb").read(),
"base64",
"utf-8",
)
# 设置附件信息
attachment["Content-Disposition"] = 'attachment; filename="bkt_result.html"'
self.assertTrue(send_email(self.mail_sender, self.mail_receivers, attachment))

def test_sendimage(self):
# 二进制读取图片
image_data = open(self.SRC_DIR.joinpath("imgs/UFund.png"), "rb")
# 设置读取获取的二进制数据
message_image = MIMEImage(image_data.read())
# 关闭刚才打开的文件
image_data.close()
self.assertTrue(send_email(self.mail_sender, self.mail_receivers, message_image))

def test_sendhtml(self):
# 发送 html 格式的邮件
now_time = datetime.datetime.now()
year = now_time.year
month = now_time.month
day = now_time.day
mytime = str(year) + " 年 " + str(month) + " 月 " + str(day) + " 日 "
fayanren = "爱因斯坦"
zhuchiren = "牛顿"

# 构造HTML
html_content = """
<html>
<body>
<h1 align="center">这个是标题，xxxx通知</h1>
<p><strong>您好：</strong></p>
<blockquote><p><strong>以下内容是本次会议的纪要,请查收！</strong></p></blockquote>
<blockquote><p><strong>发言人：{fayanren}</strong></p></blockquote>
<blockquote><p><strong>主持人：{zhuchiren}</strong></p></blockquote>
<p align="right">{mytime}</p>
<body>
<html>
""".format(
fayanren=fayanren, zhuchiren=zhuchiren, mytime=mytime
def setUp(self):
self.sender = "sender@qq.com"
self.receivers = ["first@example.com", "second@example.com"]
self.content = MIMEText("offline mock only", "plain", "utf-8")

smtp_patcher = patch("utils.send_email.smtplib.SMTP_SSL")
self.addCleanup(smtp_patcher.stop)
self.smtp_ssl = smtp_patcher.start()
self.server = self.smtp_ssl.return_value

def test_sends_to_every_recipient_and_quits(self):
result = send_email(self.sender, self.receivers, self.content)

self.assertTrue(result)
self.assertEqual(
self.server.sendmail.call_args_list,
[
call(self.sender, self.receivers[0], ANY),
call(self.sender, self.receivers[1], ANY),
],
)
message_html = MIMEText(html_content, "html", "utf-8")
self.assertTrue(send_email(self.mail_sender, self.mail_receivers, message_html))
self.server.quit.assert_called_once_with()

def test_quits_and_reraises_when_second_send_fails(self):
send_error = smtplib.SMTPException("second send failed")
self.server.sendmail.side_effect = [None, send_error]

with self.assertRaises(smtplib.SMTPException) as raised:
send_email(self.sender, self.receivers, self.content)

self.assertIs(raised.exception, send_error)
self.assertEqual(self.server.sendmail.call_count, 2)
self.server.quit.assert_called_once_with()

def test_quit_failure_does_not_replace_send_failure(self):
send_error = smtplib.SMTPException("second send failed")
self.server.sendmail.side_effect = [None, send_error]
self.server.quit.side_effect = smtplib.SMTPException("quit failed")

with self.assertRaises(smtplib.SMTPException) as raised:
send_email(self.sender, self.receivers, self.content)

self.assertIs(raised.exception, send_error)
self.assertEqual(self.server.sendmail.call_count, 2)
self.server.quit.assert_called_once_with()

def test_quit_failure_does_not_replace_login_failure(self):
login_error = smtplib.SMTPException("login failed")
self.server.login.side_effect = login_error
self.server.quit.side_effect = smtplib.SMTPException("quit failed")

with self.assertRaises(smtplib.SMTPException) as raised:
send_email(self.sender, self.receivers, self.content)

self.assertIs(raised.exception, login_error)
self.server.sendmail.assert_not_called()
self.server.quit.assert_called_once_with()

def test_quit_failure_does_not_replace_starttls_failure(self):
starttls_error = smtplib.SMTPException("starttls failed")
server = Mock()
server.starttls.side_effect = starttls_error
server.quit.side_effect = smtplib.SMTPException("quit failed")

with patch("utils.send_email.smtplib.SMTP", return_value=server):
with self.assertRaises(smtplib.SMTPException) as raised:
send_email("sender@gmail.com", self.receivers, self.content)

self.assertIs(raised.exception, starttls_error)
server.login.assert_not_called()
server.quit.assert_called_once_with()

def test_quit_failure_does_not_replace_connect_failure(self):
connect_error = smtplib.SMTPException("connect failed")
server = Mock()
server.connect.side_effect = connect_error
server.quit.side_effect = smtplib.SMTPException("quit failed")

with patch("utils.send_email.smtplib.SMTP", return_value=server):
with self.assertRaises(smtplib.SMTPException) as raised:
send_email("sender@163.com", self.receivers, self.content)

self.assertIs(raised.exception, connect_error)
server.login.assert_not_called()
server.quit.assert_called_once_with()

def test_qmail(self):
self.assertTrue(self.test_sendtext())
def test_empty_receivers_return_none_without_connecting(self):
result = send_email(self.sender, [], self.content)

def test_gmail(self):
self.assertTrue(self.test_sendtext())
self.assertIsNone(result)
self.smtp_ssl.assert_not_called()


if __name__ == "__main__":
Expand Down
97 changes: 55 additions & 42 deletions utils/send_email.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,49 +76,62 @@

def send_email(mail_sender, mail_receivers, content):

# 创建SMTP对象
if "qq" in mail_sender:
server = smtplib.SMTP_SSL("smtp.qq.com", 465)
elif "gmail" in mail_sender:
server = smtplib.SMTP("smtp.gmail.com", 587) # Connect to the server
server.starttls()
elif "163" in mail_sender:
server = smtplib.SMTP()
# 设置发件人邮箱的域名和端口，端口地址为25
server.connect("smtp.163.com", 25)
else:
print("Please check your sender email.")

# set_debuglevel(1)可以打印出和SMTP服务器交互的所有信息
# server.set_debuglevel(1)

# Connect and login to the email server
server.login(mail_sender, mail_license)

# Loop over each email to send to
for mail_receiver in mail_receivers:
# Setup MIMEMultipart for each email address (if we don't do this, the emails will concatenate on each email sent)
msg = MIMEMultipart()
msg["From"] = mail_sender
msg["To"] = mail_receiver
msg["Subject"] = subject

print("Send to: ", mail_receiver)

# Attach the message to the MIMEMultipart object
# msg.attach(message_text)
# msg.attach(message_image)
# Attach the attachment file
# msg.attach(attachment)
msg.attach(content)

# Send the email to this specific email address
server.sendmail(mail_sender, mail_receiver, msg.as_string())
print("邮件发送成功")
return True

# Quit the email server when everything is done
if not mail_receivers:
return None

server = None
try:
# 创建SMTP对象
if "qq" in mail_sender:
server = smtplib.SMTP_SSL("smtp.qq.com", 465)
elif "gmail" in mail_sender:
server = smtplib.SMTP("smtp.gmail.com", 587) # Connect to the server
server.starttls()
elif "163" in mail_sender:
server = smtplib.SMTP()
# 设置发件人邮箱的域名和端口，端口地址为25
server.connect("smtp.163.com", 25)
else:
print("Please check your sender email.")
return None

# set_debuglevel(1)可以打印出和SMTP服务器交互的所有信息
# server.set_debuglevel(1)

# Connect and login to the email server
server.login(mail_sender, mail_license)

# Loop over each email to send to
for mail_receiver in mail_receivers:
# Setup MIMEMultipart for each email address (if we don't do this, the emails will concatenate on each email sent)
msg = MIMEMultipart()
msg["From"] = mail_sender
msg["To"] = mail_receiver
msg["Subject"] = subject

print("Send to: ", mail_receiver)

# Attach the message to the MIMEMultipart object
# msg.attach(message_text)
# msg.attach(message_image)
# Attach the attachment file
# msg.attach(attachment)
msg.attach(content)

# Send the email to this specific email address
server.sendmail(mail_sender, mail_receiver, msg.as_string())
print("邮件发送成功")
except Exception:
if server is not None:
try:
server.quit()
except Exception:
# Preserve the original connection, login, or sending error.
pass
raise

server.quit()
return True


if __name__ == "__main__":
Expand Down
Loading