【pytest】自定製回撥事件
阿新 • • 發佈：2020-08-09
閱讀目錄
一、引言
二、監控外掛
三、基於pluggy自定製回撥
引言
採用pytest框架搭建的介面自動化測試專案，案例失敗以後新增自定製操作。文章以新增一個監控外掛為例，講解怎麼實現pytest自定製回撥。
監控外掛
目錄
|--monitor |└─ __init__.py |└─ config.py # 監控配置檔案 |└─ params.py # 引數管理檔案 |└─ request.py # requests庫封裝方法 |└─ stack.py # 偽棧，存放params物件，核心push 、 pop、get操作 |└─ plugin.py # pytest外掛
架構設計
引用外掛
from monitor.plugin import MonitorHookImpl
pytest.main([], [MonitorHookImpl()])
plugin.py
from pytest import hookimpl from _pytest.nodes import Item def error_callback(key, err): """失敗回撥""" print("\n******************") params_stack.print() # 除錯日誌 print("\n******************") params_obj = params_stack.get(key) if params_obj: # 註冊介面引數 params_obj.set_api_params() # 持久化儲存資料 params_obj.save() # 判斷是否限制傳送 if BaseConfig.SEND_DING_LIMIT and not params_obj.is_send: return content = params_obj.format() # user_mobiles = service_get_user_mobile_from_org(params_obj.users) user_mobiles = params_obj.get_user_mobiles() at_info = f"### 【後端服務介面異常通知】\n\n- 請**<font color=#FF0000>" \ f"{','.join(map(lambda x: f'@{x}', user_mobiles))}</font>**及時關注\n\n【斷言】{err}\n\n" send_ding(BaseConfig.DING_SUCC_WEBHOOK, title="介面監控", content=at_info + content, at_mobiles=user_mobiles, at_all=False) else: send_ding(BaseConfig.DING_ERR_WEBHOOK, title="專案監控", content=f"【{key}】未獲取到監控資訊") class MonitorHookImpl(object): """ 監控回撥 hook implementation (hookimpl) 是一個被恰當標記過的回撥函式。 hookimpls 通過register()方法載入。 """ @hookimpl(hookwrapper=True) def pytest_runtest_call(self, item: Item): """ pytest執行回撥，必須包含唯一yield，實現生成器函式 """ err = None # print(item.__dict__) try: # 觸發執行用例 _ = (yield).get_result() except AssertionError as e: # 捕獲斷言異常 err = e.__str__() except Exception as e: # 捕獲其它指令碼異常 err = e.__str__() if err: Log.info(f'異常原因:{err}') error_callback(item.name, err)
request.py
import json import requests import traceback from urllib import parse from monitor.params import Params from monitor.stack import params_stack # logger = logging.getLogger('monitor') # 全部session物件 session = None def get_case_info_from_traceback(trace_list: list): """從追述中獲取案例資訊""" for trace in trace_list: if trace.name.startswith("test_"): # mac系統 '/' 分割， windows會出錯 return trace.name, trace.filename.split("/") class RequestError(Exception): pass class Requests(object): """傳送請求 requests庫實現同步傳送，grequests修改實現非同步併發 """ def __init__(self, create_session): # 請求引數物件 self.url = None self.params = None self.method = None self.headers = None self.timeout = 1 self.cookies = None # session self._session = None self.init_session(create_session) def init_session(self, create_session): """初始化session物件""" global session if create_session or not session: # 例項化session物件 session = requests.session() self._session = session else: self._session = session def request(self, method, url, *args, **kwargs): """請求方法""" _params = kwargs.get('params') if method == 'get' else kwargs.get('data') or kwargs.get('json') call_method, f_list = get_case_info_from_traceback(traceback.extract_stack()) # 記錄請求資訊 params = Params(f_list[-2], f_list[-1], call_method) params.set_api_request_params({ "url": url, "params": _params, "headers": kwargs.get("headers"), "cookies": kwargs.get("cookies"), "timeout": self.timeout }) scheme = parse.urlsplit(url).scheme if scheme == 'https': kwargs['cert'] = config.client_cert kwargs['verify'] = False if hasattr(self._session, method): result = getattr(self._session, method)(url, *args, **kwargs) params.set_api_response_params(result) # 新增全域性引數stack params_stack.push(call_method, params) if result.status_code != 200: raise RequestError(f"請求異常: 【{result.status_code}】\n {result.text}") else: params_stack.push(call_method, params) raise RequestError(f"請求異常: 無效請求方式【{method}】") return result def get(self, *args, **kwargs): return self.request('get', *args, **kwargs) def post(self, *args, **kwargs): return self.request('post', *args, **kwargs) def delete(self, *args, **kwargs): return self.request('delete', *args, **kwargs) def put(self, *args, **kwargs): return self.request('put', *args, **kwargs) def session(self): """預設session呼叫 支援介面session方法呼叫 """ return self request = Requests(create_session=True)
stack.py
這裡為什麼要用棧維護？
當前程式碼中每次傳入的key會替換掉以前的值，這就導致瞭如果併發執行的情況下，這個key所對應的value會被覆蓋。所以這裡提出棧的概念是為以後的擴充套件做準備。
ps: Stack中註釋的程式碼是把key對應的value構造為一個list實現的棧。
class Local(object):
__slots__ = ("__storage__", )
def __init__(self):
object.__setattr__(self, '__storage__', {})
# object.__setattr__(self, '__uuid__', uuid)
def __setattr__(self, key, value):
# ident = self.__uuid__
# key => uuid唯一鍵值
try:
self.__storage__[key] = value
except KeyError:
self.__storage__ = {key: value}
def __setitem__(self, key, value):
# key => uuid唯一鍵值
# 只有在 Local()["aaa"] = aa時候觸發
try:
self.__storage__[key] = value
except KeyError:
self.__storage__ = {key: value}
def __getattr__(self, item):
try:
return self.__storage__[item]
except KeyError:
raise AttributeError(item)
def __delattr__(self, item):
try:
del self.__storage__[item]
except KeyError:
raise AttributeError(item)
def __delitem__(self, item):
try:
del self.__storage__[item]
except KeyError:
raise AttributeError(item)
def __repr__(self):
return str(self.__storage__)
class Stack(object):
"""
外部呼叫只初始化一次，然後呼叫
"""
def __init__(self):
self._local = Local()
def push(self, key, item):
self._local[key] = item
# rv = getattr(self._local, key, None)
# if not rv:
# self._local[key] = rv = []
# rv.append(item)
def get(self, key):
rv = getattr(self._local, key, None)
if rv is None:
return None
return rv
def pop(self, key):
del self._local[key]
# if rv is None:
# return None
# if len(rv) == 1:
# return rv[-1]
# else:
# return rv.pop()
def print(self):
print(self._local)
# 建立單例實體
params_stack = Stack()
params.py
class Params(object):
"""儲存執行引數"""
def __init__(self, module, script, method):
# 模組
self.module = module
# 指令碼名稱
self.script = script
# 方法名稱, 案例名稱
self.method = method
# 負責人
self.users = get_user_by_module(module) + BaseConfig.DEFAULT_AT_USER
# api_params操作物件
self.__api_params = ApiParams()
# 存放__api_params物件
self.api_params = None
self.now_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
def set_api_request_params(self, params: dict):
return self.__api_params.api_request.set_params(params)
def set_api_response_params(self, response):
return self.__api_params.api_response.set_params(response)
def set_api_params(self):
"""註冊介面引數"""
if self.api_params:
return
self.api_params = self.__api_params.get_params()
def save(self):
"""儲存資料
資料庫持久化儲存資料
"""
params_db.insert_api_monitor(
get_api_by_url(self.api_params['inParams']['url']), # api
self.module,
self.script,
self.method,
",".join(self.users),
json.dumps(self.api_params['inParams']['params']),
json.dumps(self.api_params['outParams']['json_result'])[:250] or self.api_params['outParams']['text_result'][:250],
self.now_time
)
def format(self):
"""格式化引數"""
msg = f"【介面】{get_api_by_url(self.api_params['inParams']['url'])}\n\n" \
f"【時間】{self.now_time}\n\n" \
f"【入參】\n\n{json.dumps(self.api_params['inParams']['params'], ensure_ascii=False)}\n\n***\n\n" \
f"【響應】\n\n{json.dumps(self.api_params['outParams']['json_result'], ensure_ascii=False)[:255] or self.api_params['outParams']['text_result'][:255]}\n\n***\n\n" \
f"【模組】{self.module}\n\n" \
f"【指令碼】{self.script}\n\n" \
f"【案例】{self.method}\n\n"
return msg
def is_send(self):
"""判斷是否傳送"""
number = params_db.get_api_number_of_repetitions(
get_api_by_url(self.api_params['inParams']['url']),
BaseConfig.IS_NOT_SEND_HOUR
)
if number > BaseConfig.IS_NOT_SEND_NUMBER:
# 超出限制了
return False
return True
def get_user_mobiles(self):
"""獲取使用者手機號"""
return params_db.get_user_phone_numbers(self.users)
class ApiRequestParamsError(Exception):
pass
class AbstractParams(object):
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def set_params(self, *args):
"""執行入口"""
raise NotImplementedError("Interface must be implemented")
@abc.abstractmethod
def get_params(self, *args):
"""執行入口"""
raise NotImplementedError("Interface must be implemented")
class ApiParams(object):
"""介面引數
負責協調介面請求引數 及 介面響應引數
"""
def __init__(self):
# 例項化請求引數，響應引數物件 -> 對應@api_request, @api_response
self.__reqParams = ApiRequestParams()
self.__resParams = ApiReposeParams()
@property
def api_response(self):
return self.__resParams
@property
def api_request(self):
return self.__reqParams
def get_params(self):
"""獲取引數值，此引數值會被記錄到資料副本"""
return {
"inParams": self.api_request.get_params(),
"outParams": self.api_response.get_params()
}
class ApiRequestParams(AbstractParams):
"""介面請求引數"""
def __init__(self):
# 儲存請求引數
self.params = None
# 入參編碼格式
self.coding_format = 'json'
def set_params(self, params: dict):
self.params = params
def get_params(self):
return self.params
class ApiReposeParams(AbstractParams):
"""介面響應引數"""
def __init__(self):
# 出參編碼格式
self.coding_format = 'json'
# 儲存響應引數
self.params = None
self.status_code = 200
def set_params(self, response):
"""負責引數，接受request返回的響應"""
self.status_code = response.status_code
self.params = {
"json_result": None,
"text_result": "",
"headers": response.headers,
'cookies': response.cookies.get_dict(),
"consuming": response.elapsed.total_seconds()
}
# 根據編碼方式獲取原始資料
if self.coding_format == 'json' and self.status_code == 200:
try:
self.params.update({
"json_result": response.json()
})
except UnicodeDecodeError:
self.params.update({
"text_result": response.text
})
else:
self.params.update({
"text_result": response.text
})
def get_params(self):
return self.params
基於pluggy自定製回撥
from pluggy import PluginManager, HookimplMarker, HookspecMarker
monitor_hook_spec = HookspecMarker("monitor")
monitor_hook_impl = HookimplMarker("monitor")
class MonitorHookSpec(object):
"""
hook specification (hookspec)用來validate每個hookimpl，保證hookimpl被正確的定義。
hookspec 通過 add_hookspecs()方法載入，一般在註冊hookimpl之前先載入；
"""
@monitor_hook_spec
def add_err_log(self, log):
""" log """
class MonitorHookImpl(object):
"""
監控回撥
hook implementation (hookimpl) 是一個被恰當標記過的回撥函式。
hookimpls 通過register()方法載入。
"""
# 同上，甚至可以不定義hookspec裡的引數
@monitor_hook_impl
def add_err_log(self, log):
"""Default implementation.
"""
return f"Logger：{log}"
def manage():
"""自定義註冊外掛"""
# 例項化外掛物件
pm = PluginManager("monitor")
pm.add_hookspecs(MonitorHookSpec)
pm.register(MonitorHookImpl())
return pm
# 也可以自己實現pytest_configure方法，呼叫的話把pytest.main()這個方法重新實現了，在內部呼叫pytest_configure即可
def pytest_configure(config) -> None:
"""pytest配置
:params: config / pytest.config配置，需要修改pytest.main函式邏輯在config.pluginmanager新增特定功能
"""
impl = MonitorHookImpl()
config.pluginmanager.add_hookspecs(MonitorHookSpec)
config.pluginmanager.register(impl)