添加运行异常时发送邮件通知

httpfpt/conftest.py

@@ -26,11 +26,9 @@ def session_fixture(tmp_path_factory, worker_id):
 
 @pytest.fixture(scope='package', autouse=True)
 def package_fixture():
-    log.info('🚀 START')
     yield
-    log.info('')  # 预留空行
-    log.info('🏁 FINISH')
-
+    # 预留空行
+    log.info('')
     # 清理临时变量
     VariableCache().clear()
 

httpfpt/db/redis_db.py

@@ -1,6 +1,5 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-import sys
 from typing import Any
 
 from redis import Redis, AuthenticationError
@@ -26,13 +25,10 @@ def init(self) -> None:
             self.redis.ping()
         except TimeoutError:
             log.error('数据库 redis 连接超时')
-            sys.exit(1)
         except AuthenticationError:
             log.error('数据库 redis 授权认证错误')
-            sys.exit(1)
         except Exception as e:
             log.error(f'数据库 redis 连接异常: {e}')
-            sys.exit(1)
         else:
             log.info('数据库 redis 连接成功')
 

httpfpt/enums/email_type.py

@@ -0,0 +1,8 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from httpfpt.enums import IntEnum
+
+
+class EmailType(IntEnum):
+    REPORT = 0
+    ERROR = 1

httpfpt/run.py

@@ -154,13 +154,15 @@ def run(
 
     log.info(f'开始运行项目：{PROJECT_NAME}' if run_path == default_case_path else f'开始运行：{run_path}')
     log.info(f'Pytest 命令: pytest {format_run_args_to_pytest_command}')
+    log.info('🚀 START')
     pytest.main(run_args)
+    log.info('🏁 FINISH')
 
     yaml_report_files = os.listdir(YAML_REPORT_PATH)
     yaml_report_files.sort()
     test_result = read_yaml(YAML_REPORT_PATH, filename=yaml_report_files[-1])
 
-    SendMail(is_html_report_file.split('=')[1], test_result).send() if EMAIL_REPORT_SEND and html_report else ...
+    SendMail(test_result, is_html_report_file.split('=')[1]).send_report() if EMAIL_REPORT_SEND and html_report else ...
 
     DingTalk(test_result).send() if DING_TALK_REPORT_SEND else ...
 
@@ -175,29 +177,39 @@ def run(
     ) if allure and allure_serve else ...
 
 
+def main(*args, **kwargs) -> None:
+    try:
+        logo = """\n
+         /$$   /$$ /$$$$$$$$ /$$$$$$$$ /$$$$$$$  /$$$$$$$$ /$$$$$$$  /$$$$$$$$
+        | $$  | $$|__  $$__/|__  $$__/| $$__  $$| $$_____/| $$__  $$|__  $$__/
+        | $$  | $$   | $$      | $$   | $$  | $$| $$      | $$  | $$   | $$   
+        | $$$$$$$$   | $$      | $$   | $$$$$$$/| $$$$$   | $$$$$$$/   | $$   
+        | $$__  $$   | $$      | $$   | $$____/ | $$__/   | $$____/    | $$   
+        | $$  | $$   | $$      | $$   | $$      | $$      | $$         | $$   
+        | $$  | $$   | $$      | $$   | $$      | $$      | $$         | $$   
+        |__/  |__/   |__/      |__/   |__/      |__/      |__/         |__/   
+    
+            """
+        print(logo)
+        log.info(logo)
+
+        # 初始化 redis 数据库 (必选)
+        redis_client.init()
+
+        # 用例数据唯一 case_id 检测（可选）
+        get_all_testcase_id(get_all_testcase_data())
+
+        # 用例数据完整架构 pydantic 快速检测（可选）
+        get_all_testcase_data(pydantic_verify=True)
+
+        # 执行程序 (必选)
+        run(*args, **kwargs)
+    except Exception as e:
+        log.error(f'运行异常：{e}')
+        import traceback
+
+        SendMail({'error': traceback.format_exc()}).send_error()
+
+
 if __name__ == '__main__':
-    logo = """\n
-     /$$   /$$ /$$$$$$$$ /$$$$$$$$ /$$$$$$$  /$$$$$$$$ /$$$$$$$  /$$$$$$$$
-    | $$  | $$|__  $$__/|__  $$__/| $$__  $$| $$_____/| $$__  $$|__  $$__/
-    | $$  | $$   | $$      | $$   | $$  | $$| $$      | $$  | $$   | $$   
-    | $$$$$$$$   | $$      | $$   | $$$$$$$/| $$$$$   | $$$$$$$/   | $$   
-    | $$__  $$   | $$      | $$   | $$____/ | $$__/   | $$____/    | $$   
-    | $$  | $$   | $$      | $$   | $$      | $$      | $$         | $$   
-    | $$  | $$   | $$      | $$   | $$      | $$      | $$         | $$   
-    |__/  |__/   |__/      |__/   |__/      |__/      |__/         |__/   
-
-        """
-    print(logo)
-    log.info(logo)
-
-    # 初始化 redis 数据库 (必选)
-    redis_client.init()
-
-    # 用例数据唯一 case_id 检测（可选）
-    get_all_testcase_id(get_all_testcase_data())
-
-    # 用例数据完整架构 pydantic 快速检测（可选）
-    get_all_testcase_data(pydantic_verify=True)
-
-    # 执行程序 (必选)
-    run()
+    main()

httpfpt/templates/email_notification.html

@@ -0,0 +1,33 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <title> HttpFpt Run Exception </title>
+</head>
+<body>
+    <div style="font-family: 'Helvetica Neue',Helvetica,Arial,sans-serif;
+        box-sizing: border-box;
+        display: inline-block;
+        font-size: 14px;
+        overflow: hidden;
+        border-radius: 7px;
+        margin: 0;
+        border: 1px solid #e9e9e9;
+        width: 50%;
+        background-color: #f5f9fc;">
+        <div style="font-size: 16px;
+            vertical-align: top;
+            color: #fff;
+            background-color: #ff5757;
+            padding: 20px;">
+            <span style="margin:20px;display:block;"> HttpFpt 运行异常通知 </span>
+        </div>
+        <div style="margin: 30px 20px;">
+            <p>测试运行异常，报错信息如下：</p>
+            <pre style="font-family:auto,emoji;font-size:14px;margin:0;color:#ff5757">
+                {{ error }}
+            </pre>
+        </div>
+    </div>
+</body>
+</html>

httpfpt/templates/email_report.html

@@ -20,7 +20,7 @@
             color: #fff;
             background-color: #57b9ff;
             padding: 20px;">
-            <span style="margin-top:20px;display:block"> Fastpt Test Report </span>
+            <span style="margin:20px;display:block"> {{ test_title }} 自动化测试报告 </span>
         </div>
         <div style="margin: 30px 20px;">
             <table style="font-size:14px;margin:0">

httpfpt/utils/send_report/send_email.py

@@ -12,23 +12,24 @@
 from httpfpt.common.log import log
 from httpfpt.core import get_conf
 from httpfpt.core.path_conf import HTML_EMAIL_REPORT_PATH
+from httpfpt.enums.email_type import EmailType
 from httpfpt.utils.file_control import get_file_property
 
 
 class SendMail:
-    def __init__(self, filename: str, content: dict):
-        self.filename = filename
+    def __init__(self, content: dict, filename: str | None = None):
         self.content = content
+        self.filename = filename
 
-    def take_messages(self) -> MIMEMultipart:
+    def take_report(self) -> MIMEMultipart:
         """
-        生成邮件内容，和html报告附件
+        获取报告
         """
         msg = MIMEMultipart()
         msg['Subject'] = get_conf.TEST_REPORT_TITLE
         msg['From'] = get_conf.EMAIL_USER
         msg['date'] = time.strftime('%a, %d %b %Y %H:%M:%S %z')
-        self.content.update({'test_title': get_conf.PROJECT_NAME})
+        self.content.update({'test_title': get_conf.TEST_REPORT_TITLE})
         self.content.update({'tester_name': get_conf.TESTER_NAME})
 
         # 邮件正文
@@ -39,30 +40,64 @@ def take_messages(self) -> MIMEMultipart:
         msg.attach(mail_body)
 
         # 读取要发送的附件
-        with open(self.filename, 'rb') as f:
-            annex_body = f.read()
+        if self.filename:
+            with open(self.filename, 'rb') as f:
+                annex_body = f.read()
+                # 邮件附件
+                att1 = MIMEApplication(annex_body)
+                att1['Content-Type'] = 'application/octet-stream'
+                att1['Content-Disposition'] = f'attachment; filename={get_file_property(self.filename)[0]}'
+                msg.attach(att1)
 
-        # 邮件附件
-        att1 = MIMEApplication(annex_body)
-        att1['Content-Type'] = 'application/octet-stream'
-        att1['Content-Disposition'] = f'attachment; filename={get_file_property(self.filename)[0]}'
-        msg.attach(att1)
+        return msg
+
+    def take_error(self) -> MIMEMultipart:
+        """
+        获取错误信息
+        """
+        msg = MIMEMultipart()
+        msg['Subject'] = 'HttpFpt 运行异常通知'
+        msg['From'] = get_conf.EMAIL_USER
+        msg['date'] = time.strftime('%a, %d %b %Y %H:%M:%S %z')
+
+        # 邮件正文
+        with open(os.path.join(HTML_EMAIL_REPORT_PATH, 'email_notification.html'), 'r', encoding='utf-8') as f:
+            html = Template(f.read())
+
+        mail_body = MIMEText(html.render(**self.content), _subtype='html', _charset='utf-8')
+        msg.attach(mail_body)
 
         return msg
 
-    def send(self) -> None:
+    def _send(self, msg_type: int) -> None:
         """
         发送邮件
         """
+        msg = ''
+        if msg_type == EmailType.REPORT:
+            msg = self.take_report().as_string()
+        elif msg_type == EmailType.ERROR:
+            msg = self.take_error().as_string()
+        if get_conf.EMAIL_SSL:
+            smtp = smtplib.SMTP_SSL(host=get_conf.EMAIL_SERVER, port=get_conf.EMAIL_PORT)
+        else:
+            smtp = smtplib.SMTP(host=get_conf.EMAIL_SERVER, port=get_conf.EMAIL_PORT)
+        smtp.login(get_conf.EMAIL_USER, get_conf.EMAIL_PASSWORD)
+        smtp.sendmail(get_conf.EMAIL_USER, get_conf.EMAIL_SEND_TO, msg)
+        smtp.quit()
+
+    def send_report(self) -> None:
         try:
-            if get_conf.EMAIL_SSL:
-                smtp = smtplib.SMTP_SSL(host=get_conf.EMAIL_SERVER, port=get_conf.EMAIL_PORT)
-            else:
-                smtp = smtplib.SMTP(host=get_conf.EMAIL_SERVER, port=get_conf.EMAIL_PORT)
-            smtp.login(get_conf.EMAIL_USER, get_conf.EMAIL_PASSWORD)
-            smtp.sendmail(get_conf.EMAIL_USER, get_conf.EMAIL_SEND_TO, self.take_messages().as_string())
-            smtp.quit()
+            self._send(0)
         except Exception as e:
             log.error(f'测试报告邮件发送失败: {e}')
         else:
-            log.success('测试报告邮件发送成功')
+            log.info('测试报告邮件发送成功')
+
+    def send_error(self) -> None:
+        try:
+            self._send(1)
+        except Exception as e:
+            log.error(f'运行异常通知邮件发送失败: {e}')
+        else:
+            log.info('运行异常通知邮件发送成功')

requirements.txt

@@ -13,7 +13,6 @@ pymysql==0.9.3
 pyright==1.1.315
 pytest==7.4.1
 pytest-html==4.0.0
-pytest-loguru==0.2.0
 pytest-metadata==3.0.0
 pytest-pretty==1.2.0
 pytest-xdist==2.5.0