Skip to content

Commit

Permalink
添加配置文件解析错误提示 (#169)
Browse files Browse the repository at this point in the history
  • Loading branch information
@wu-clan
wu-clan authored Mar 18, 2024
1 parent 8247f89 commit c97df7c
Show file tree
Hide file tree
Showing 5 changed files with 107 additions and 91 deletions.
7 changes: 7 additions & 0 deletions httpfpt/common/errors.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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):
"""认证错误"""

Expand Down
183 changes: 94 additions & 89 deletions httpfpt/core/get_conf.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
2 changes: 2 additions & 0 deletions httpfpt/core/path_conf.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

from functools import lru_cache

__all__ = ['httpfpt_path']


class HttpFptPathConfig:
@property
Expand Down
5 changes: 4 additions & 1 deletion httpfpt/run.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
1 change: 0 additions & 1 deletion httpfpt/utils/send_report/wechat.py
View file
Original file line number Diff line number Diff line change
@@ -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

Expand Down

0 comments on commit c97df7c

Please sign in to comment.