From c97df7c7f035ef5001adcfb7885371013c239480 Mon Sep 17 00:00:00 2001 From: Wu Clan Date: Mon, 18 Mar 2024 14:15:35 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E9=85=8D=E7=BD=AE=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E8=A7=A3=E6=9E=90=E9=94=99=E8=AF=AF=E6=8F=90=E7=A4=BA?= =?UTF-8?q?=20(#169)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- httpfpt/common/errors.py | 7 ++ httpfpt/core/get_conf.py | 183 ++++++++++++++-------------- httpfpt/core/path_conf.py | 2 + httpfpt/run.py | 5 +- httpfpt/utils/send_report/wechat.py | 1 - 5 files changed, 107 insertions(+), 91 deletions(-) diff --git a/httpfpt/common/errors.py b/httpfpt/common/errors.py index 4e0c02d..84a8f65 100644 --- a/httpfpt/common/errors.py +++ b/httpfpt/common/errors.py @@ -12,6 +12,13 @@ def __str__(self) -> str: return self.msg +class ConfigInitError(HttpFptErrorMixin, RuntimeError): + """配置初始化错误""" + + def __init__(self, msg: str) -> None: + super().__init__(msg) + + class AuthError(HttpFptErrorMixin, ValueError): """认证错误""" diff --git a/httpfpt/core/get_conf.py b/httpfpt/core/get_conf.py index 85d9f91..a1bae83 100644 --- a/httpfpt/core/get_conf.py +++ b/httpfpt/core/get_conf.py @@ -5,101 +5,106 @@ from glom import glom +from httpfpt.common.errors import ConfigInitError from httpfpt.common.toml_handler import read_toml +__all__ = ['httpfpt_config'] + class HttpFptConfig: def __init__(self) -> None: self.settings = read_toml(str(Path(__file__).resolve().parent), 'conf.toml') - - # 项目目录名 - self.PROJECT_NAME = glom(self.settings, 'project.name') - - # 测试报告 - self.TEST_REPORT_TITLE = glom(self.settings, 'report.title') - self.TESTER_NAME = glom(self.settings, 'report.tester_name') - - # mysql 数据库 - self.MYSQL_HOST = glom(self.settings, 'mysql.host') - self.MYSQL_PORT = glom(self.settings, 'mysql.port') - self.MYSQL_USER = glom(self.settings, 'mysql.user') - self.MYSQL_PASSWORD = glom(self.settings, 'mysql.password') - self.MYSQL_DATABASE = glom(self.settings, 'mysql.database') - self.MYSQL_CHARSET = glom(self.settings, 'mysql.charset') - - # redis 数据库 - self.REDIS_HOST = glom(self.settings, 'redis.host') - self.REDIS_PORT = glom(self.settings, 'redis.port') - self.REDIS_PASSWORD = glom(self.settings, 'redis.password') - self.REDIS_DATABASE = glom(self.settings, 'redis.database') - self.REDIS_TIMEOUT = glom(self.settings, 'redis.timeout') - - # 邮件 - self.EMAIL_SERVER = glom(self.settings, 'email.host') - self.EMAIL_PORT = glom(self.settings, 'email.port') - self.EMAIL_USER = glom(self.settings, 'email.user') - self.EMAIL_PASSWORD = glom(self.settings, 'email.password') - self.EMAIL_SEND_TO = glom(self.settings, 'email.receiver') - self.EMAIL_SSL = glom(self.settings, 'email.ssl') - self.EMAIL_SEND = glom(self.settings, 'email.send') - - # 钉钉 - self.DINGDING_WEBHOOK = glom(self.settings, 'dingding.webhook') - self.DINGDING_PROXY = { - 'http': glom(self.settings, 'dingding.proxies.http') - if glom(self.settings, 'dingding.proxies.http') != '' - else None, - 'https': glom(self.settings, 'dingding.proxies.https') - if glom(self.settings, 'dingding.proxies.https') != '' - else None, - } - self.DINGDING_SEND = glom(self.settings, 'dingding.send') - - # 飞书 - self.FEISHU_WEBHOOK = glom(self.settings, 'feishu.webhook') - self.FEISHU_PROXY = { - 'http': glom(self.settings, 'feishu.proxies.http') - if glom(self.settings, 'feishu.proxies.http') != '' - else None, - 'https': glom(self.settings, 'feishu.proxies.https') - if glom(self.settings, 'feishu.proxies.https') != '' - else None, - } - self.FEISHU_SEND = glom(self.settings, 'feishu.send') - - # 企业微信 - self.WECHAT_WEBHOOK = glom(self.settings, 'wechat.webhook') - self.WECHAT_PROXY = { - 'http': glom(self.settings, 'wechat.proxies.http') - if glom(self.settings, 'wechat.proxies.http') != '' - else None, - 'https': glom(self.settings, 'wechat.proxies.https') - if glom(self.settings, 'wechat.proxies.https') != '' - else None, - } - self.WECHAT_SEND = glom(self.settings, 'wechat.send') - - # 请求发送 - self.REQUEST_TIMEOUT = glom(self.settings, 'request.timeout') - self.REQUEST_VERIFY = glom(self.settings, 'request.verify') - self.REQUEST_REDIRECTS = glom(self.settings, 'request.redirects') - self.REQUEST_PROXIES_REQUESTS = { - 'http': glom(self.settings, 'request.proxies.http') - if glom(self.settings, 'request.proxies.http') != '' - else None, - 'https': glom(self.settings, 'request.proxies.https') - if glom(self.settings, 'request.proxies.https') != '' - else None, - } - self.REQUEST_PROXIES_HTTPX = { - 'http://': glom(self.settings, 'request.proxies.http') - if glom(self.settings, 'request.proxies.http') != '' - else None, - 'https://': glom(self.settings, 'request.proxies.https') - if glom(self.settings, 'request.proxies.https') != '' - else None, - } - self.REQUEST_RETRY = glom(self.settings, 'request.retry') + try: + # 项目目录名 + self.PROJECT_NAME = glom(self.settings, 'project.name') + + # 测试报告 + self.TEST_REPORT_TITLE = glom(self.settings, 'report.title') + self.TESTER_NAME = glom(self.settings, 'report.tester_name') + + # mysql 数据库 + self.MYSQL_HOST = glom(self.settings, 'mysql.host') + self.MYSQL_PORT = glom(self.settings, 'mysql.port') + self.MYSQL_USER = glom(self.settings, 'mysql.user') + self.MYSQL_PASSWORD = glom(self.settings, 'mysql.password') + self.MYSQL_DATABASE = glom(self.settings, 'mysql.database') + self.MYSQL_CHARSET = glom(self.settings, 'mysql.charset') + + # redis 数据库 + self.REDIS_HOST = glom(self.settings, 'redis.host') + self.REDIS_PORT = glom(self.settings, 'redis.port') + self.REDIS_PASSWORD = glom(self.settings, 'redis.password') + self.REDIS_DATABASE = glom(self.settings, 'redis.database') + self.REDIS_TIMEOUT = glom(self.settings, 'redis.timeout') + + # 邮件 + self.EMAIL_SERVER = glom(self.settings, 'email.host') + self.EMAIL_PORT = glom(self.settings, 'email.port') + self.EMAIL_USER = glom(self.settings, 'email.user') + self.EMAIL_PASSWORD = glom(self.settings, 'email.password') + self.EMAIL_SEND_TO = glom(self.settings, 'email.receiver') + self.EMAIL_SSL = glom(self.settings, 'email.ssl') + self.EMAIL_SEND = glom(self.settings, 'email.send') + + # 钉钉 + self.DINGDING_WEBHOOK = glom(self.settings, 'dingding.webhook') + self.DINGDING_PROXY = { + 'http': glom(self.settings, 'dingding.proxies.http') + if glom(self.settings, 'dingding.proxies.http') != '' + else None, + 'https': glom(self.settings, 'dingding.proxies.https') + if glom(self.settings, 'dingding.proxies.https') != '' + else None, + } + self.DINGDING_SEND = glom(self.settings, 'dingding.send') + + # 飞书 + self.FEISHU_WEBHOOK = glom(self.settings, 'feishu.webhook') + self.FEISHU_PROXY = { + 'http': glom(self.settings, 'feishu.proxies.http') + if glom(self.settings, 'feishu.proxies.http') != '' + else None, + 'https': glom(self.settings, 'feishu.proxies.https') + if glom(self.settings, 'feishu.proxies.https') != '' + else None, + } + self.FEISHU_SEND = glom(self.settings, 'feishu.send') + + # 企业微信 + self.WECHAT_WEBHOOK = glom(self.settings, 'wechat.webhook') + self.WECHAT_PROXY = { + 'http': glom(self.settings, 'wechat.proxies.http') + if glom(self.settings, 'wechat.proxies.http') != '' + else None, + 'https': glom(self.settings, 'wechat.proxies.https') + if glom(self.settings, 'wechat.proxies.https') != '' + else None, + } + self.WECHAT_SEND = glom(self.settings, 'wechat.send') + + # 请求发送 + self.REQUEST_TIMEOUT = glom(self.settings, 'request.timeout') + self.REQUEST_VERIFY = glom(self.settings, 'request.verify') + self.REQUEST_REDIRECTS = glom(self.settings, 'request.redirects') + self.REQUEST_PROXIES_REQUESTS = { + 'http': glom(self.settings, 'request.proxies.http') + if glom(self.settings, 'request.proxies.http') != '' + else None, + 'https': glom(self.settings, 'request.proxies.https') + if glom(self.settings, 'request.proxies.https') != '' + else None, + } + self.REQUEST_PROXIES_HTTPX = { + 'http://': glom(self.settings, 'request.proxies.http') + if glom(self.settings, 'request.proxies.http') != '' + else None, + 'https://': glom(self.settings, 'request.proxies.https') + if glom(self.settings, 'request.proxies.https') != '' + else None, + } + self.REQUEST_RETRY = glom(self.settings, 'request.retry') + except KeyError as e: + raise ConfigInitError(f'配置解析失败：缺失参数 {str(e)}，请核对项目配置文件') @lru_cache(maxsize=None) diff --git a/httpfpt/core/path_conf.py b/httpfpt/core/path_conf.py index 930f648..ed8b049 100644 --- a/httpfpt/core/path_conf.py +++ b/httpfpt/core/path_conf.py @@ -4,6 +4,8 @@ from functools import lru_cache +__all__ = ['httpfpt_path'] + class HttpFptPathConfig: @property diff --git a/httpfpt/run.py b/httpfpt/run.py index e7d6a39..a40a85f 100644 --- a/httpfpt/run.py +++ b/httpfpt/run.py @@ -73,7 +73,10 @@ def startup( if not os.path.exists(httpfpt_path.html_report_dir): os.makedirs(httpfpt_path.html_report_dir) run_args.extend( - (f'--html={os.path.join(httpfpt_path.html_report_dir, html_report_filename)}', '--self-contained-html') + ( + f'--html={os.path.join(httpfpt_path.html_report_dir, html_report_filename)}', + '--self-contained-html', + ) ) if allure: diff --git a/httpfpt/utils/send_report/wechat.py b/httpfpt/utils/send_report/wechat.py index 8b0e230..825251e 100644 --- a/httpfpt/utils/send_report/wechat.py +++ b/httpfpt/utils/send_report/wechat.py @@ -1,6 +1,5 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- - from httpfpt.common.log import log from httpfpt.core.get_conf import httpfpt_config