【python自动化实战-引入trace】在项目中引入trace跟踪每条testcase执行过程中的日志+日志推送es-Toy模板网

这篇具有很好参考价值的文章主要介绍了【python自动化实战-引入trace】在项目中引入trace跟踪每条testcase执行过程中的日志+日志推送es。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

背景

最近公司在跑自动化case过程中有很多异常case，有的很难依据现有的日志定位问题(日志多且乱，无关联性)，时常需要本地debug才能发现问题，很影响写自动化的效率以及自动化运行的通过率。

因此领导下发了一个需求：需要给每一条case绑定一个trace，通过trace关联case执行过程中的日志信息。这样通过查看日志即可快速知道case运行的问题点。经过一番思考与实践，就有了下面的方案。

分析

trace的引入：

在自定义的log_tool中添加trace_id的生成与写入
使用pytest的钩子函数给每条case添加不同的trace_id
trace_id 采用uuid 来生成
日志推送es(或者写入本地日志文件中)

实现

log_tool.py(自定义封装的日志工具)

# log_tool.py 中
import logging
import os
import time
import uuid
from functools import wraps
from loguru import logger

class MyLog:
    my_logger = logger
    """
    使用loguru封装日志类
    """
    path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    # 判断当前时间的目录在日志目录结构中是否存在，存在就不创建，不存在就创建
    dir_time = time.strftime("%Y-%m-%d", time.localtime(time.time()))
    log_file = os.path.join(os.path.join(os.path.join(path, "Log"), "log"), '{}.log'.format(dir_time))

    # 设置loguru写入日志方式
    log_handler = my_logger.add(log_file, level="INFO",
                                format="{time:YYYY-MM-DD HH:mm:ss.SSS} {level} {line} {message}", encoding="utf-8",
                                retention="30 days", backtrace=True)

    trace_id = []  # 定义trace_id，支持多级嵌套

    @classmethod
    def _log(cls, lev: str, log_msg):
        if cls.trace_id:
            log_msg = "[" + '|'.join([str(i) for i in MyLog.trace_id]) + "]" + str(log_msg)
        cls.my_logger.opt(lazy=True).log(lev.upper(), log_msg)

    @classmethod
    def debug(cls, log_msg):
        cls._log('debug', log_msg)

    @classmethod
    def info(cls, log_msg):
        cls._log('info', log_msg)

    @classmethod
    def warning(cls, log_msg):
        cls._log('warning', log_msg)

    @classmethod
    def error(cls, log_msg):
        cls._log('error', log_msg)

    @classmethod
    def critical(cls, log_msg):
        cls._log('critical', log_msg)

    @classmethod
    def traceback(cls):
        import traceback
        cls._log('error', f'执行失败！！！失败信息：\n {traceback.format_exc()}')

    def __del__(self):
        try:
            self.my_logger.remove(self.log_handler)
            del self.my_logger
        except:
            pass

conftest.py(利用钩子函数给每条case添加trace_id)

# conftest.py 中
import uuid
from utils.log_tool import MyLog # 自定义的日志类
from pytest import Item
	
def pytest_runtest_setup(item: Item):
    """case运行前执行"""
    MyLog.trace_id.append("case_" + str(uuid.uuid4().hex[:10]))

def pytest_runtest_teardown(item: Item):
    """case运行后执行"""
    MyLog.trace_id.pop()

效果：
python trace模块,python自动化,python,自动化,开发语言

conftest.py(利用fixture给指定的case添加trace_id；优点是可以指定某些case添加，缺点是需要侵入代码)

# conftest.py中定义
import uuid
import pytest
from utils.log_tool import MyLog

@pytest.fixture(scope="function")
def log_trace():
    MyLog.trace_id.append(uuid.uuid4())
    yield MyLog.trace_id
    MyLog.trace_id.pop()

# testcase中使用
def test_interface(self, log_trace):
    assert 1==2

拓展

除了testcase中使用trace_id外，还可以利用装饰器给一些长链路的场景单独设置trace

# 装饰器的实现方式
# log_tool.py 中
# 定义装饰器方法，可以使用在测试用例方法上
def log_trace(prefix=""):  # 可自定义添加前缀，默认前缀为空
    def wrapper_log(func):
        @wraps(func)  # wraps使inner装的更像一个func
        def inner(*args, **kwargs):
            try:
                MyLog.trace_id.append(str(prefix)+str(uuid.uuid4().hex[:10]))
                return func(*args, **kwargs)
            except Exception as e:
                MyLog.error(f'用例执行失败，失败原因：{e}')
                MyLog.traceback()
                raise e
            finally:
                MyLog.trace_id.pop()

        return inner
    return wrapper_log
 
# testcase中也可以使用
@log_trace("interface1_")
def test_interface1():
    with allure.step("步骤1"):
        MyLog.error(234444)
        assert 223 == 223
        MyLog.info("测试日志1112232323232322")
        assert "123" == 123

日志推送es的代码改造 (本地未进行验证，待检查)

# log_tool.py
from cmreslogging.handlers import CMRESHandler  # 安装elasticsearch==7.13.4和CMRESHandler==1.0.0

# 对_log方法进行改造
class MyLog:
    my_logger = logger
	...
	
    @classmethod
    def _log(cls, lev: str, log_msg):
        if cls.trace_id:
            log_msg = "[" + '|'.join([str(i) for i in MyLog.trace_id]) + "]" + str(log_msg)
        try:
            # 优先推送es,es推送失败则写入文件
            es_handler = CMRESHandler(hosts=[{"host": "127.0.0.1", "port": 9200}],
                                      auth_type=CMRESHandler.AuthType.NO_AUTH,
                                      es_index_name="autotest",
                                      es_doc_type=None,
                                      raise_on_indexing_exceptions=True)
            es_sink = cls.my_logger.add(es_handler, level="INFO",
                                        format="{message}")
        except:
            cls.my_logger.remove()
            file_sink = cls.my_logger.add(cls.log_file, level="INFO",
                                          format="{time:YYYY-MM-DD HH:mm:ss.SSS} {level} {line} {message}",
                                          encoding="utf-8",
                                          retention="30 days", backtrace=True)
        finally:
            cls.my_logger.opt(lazy=True).log(lev.upper(), log_msg)