25. 自动化测试框架开发之日志装饰器的开发

日志装饰器的开发

一、日志功能实现原理

1.1 日志装饰器核心代码

def logs(func):
    @wraps(func)  # ✅保持函数元数据
    def wrap_func(*args, **kwargs):
        # 捕获原始参数
        tuple_args = args
        dict_kwargs = kwargs
        
        # 执行目标函数（无异常处理）
        func(*args, **kwargs)  # ⚠️风险点
        
        # 记录DEBUG级别日志
        log.debug(
            f'{func.__name__}(*args:tuple = *{tuple_args}, **kwargs:dict = **{dict_kwargs})',
            extra={'status': 'PASS'}  # ⚠️状态硬编码
        )
    return wrap_func

1.2 功能特性对照表

二、测试验证流程

2.1 测试用例设计

class PaymentTest(unittest.TestCase):
    @logs
    @depend('user_login')
    def test_creditcard_pay(self, card="622588******1234", cvv=123):
        """正常支付测试"""
        self.process_payment(100)

    @logs
    def test_failed_case(self):
        """异常情况测试"""
        raise ValueError("模拟业务异常")

2.2 实际执行结果

# test_creditcard_pay 日志
14:35:22 - demo_log - DEBUG - PASS - test_creditcard_pay(*args:tuple = *(), **kwargs:dict = **{'card': '622588******1234', 'cvv': 123})

# test_failed_case 日志
14:35:23 - demo_log - DEBUG - PASS - test_failed_case(*args:tuple = *(), **kwargs:dict = **{})
ERROR: test_failed_case (__main__.PaymentTest)
Traceback...ValueError: 模拟业务异常

2.3 问题分析清单

1. 数据安全风险：CVV等敏感字段明文记录
2. 状态不准确：异常用例仍标记PASS
3. 异常丢失：错误堆栈未写入日志系统
4. 依赖缺陷：前置用例失败时仍执行日志记录

三、企业级改进方案

（基于头部互联网公司测试规范）

3.1 增强型日志装饰器

def secure_log(level=logging.INFO):
    """支持动态日志级别和参数过滤"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 参数脱敏处理
            safe_kwargs = {
                k: '***MASKED***' if 'cvv' in k else v
                for k, v in kwargs.items()
            }
            
            status = "FAIL"
            try:
                result = func(*args, **kwargs)
                status = "PASS"
                return result
            except Exception as e:
                log.error(f"{func.__name__}执行异常", 
                        exc_info=True,
                        extra={"status": "ERROR"})
                raise
            finally:
                log.log(level,
                       f"{func.__name__}执行状态: {status}",
                       extra={
                           "safe_args": args,
                           "safe_kwargs": safe_kwargs
                       })
        return wrapper
    return decorator

3.2 改进后执行效果

# 成功用例日志
14:40:15 - demo_log - INFO - PASS - test_creditcard_pay执行状态: PASS

# 失败用例日志
14:40:16 - demo_log - ERROR - ERROR - test_failed_case执行异常
Traceback (...完整错误堆栈...)
14:40:16 - demo_log - INFO - FAIL - test_failed_case执行状态: FAIL

四、大厂最佳实践经验

来自某金融科技公司的测试规范要求：

1. 日志记录必须包含唯一追踪ID（trace_id）实现全链路追踪
2. 敏感字段（密码、CVV等）需在前置处理层进行脱敏
3. 测试日志应与业务日志分离存储，保留周期不少于180天
4. 采用动态日志级别策略：成功用例INFO级别，失败用例ERROR级别
5. 每个测试用例的日志必须包含环境标识（env）、执行机节点（node）等元数据

# 大厂标准实现示例
class EnterpriseLogger:
    def __init__(self):
        self.log = logging.getLogger("secure_test")
        self.log.addFilter(SensitiveFilter())
        
    def record(self, case, status, metadata=None):
        log_data = {
            "trace_id": uuid.uuid4().hex,
            "env": os.getenv("DEPLOY_ENV"),
            "node": socket.gethostname(),
            "case": case.__name__,
            "status": status,
            "timestamp": datetime.utcnow().isoformat()
        }
        if metadata:
            log_data.update(metadata)
            
        self.log.info(json.dumps(log_data))

五、完整代码

"""
Python :3.13.3
Selenium: 4.31.0

decorators.py
"""

import unittest
from functools import wraps
from chap6.logs import log


class DependencyError(Exception):

    def __init__(self, _type):
        self._type = _type

    def __str__(self):
        if self._type == 0:
            return f'Dependency name of test is required!'
        if self._type == 1:
            return f'Dependency name of test can not the case self!'
        return None


def depend(case=''):
    if not case:
        raise DependencyError
    _mark = []

    def wrap_func(func):
        @wraps(func)
        def inner_func(self):
            if case == func.__name__:
                raise DependencyError(1)
            _r = self._outcome.result
            _f, _e, _s = _r.failures, _r.errors, _r.skipped

            if not (_f or _e or _s):
                func(self)

            if _f:
                _mark.extend([fail[0] for fail in _f])
            if _e:
                _mark.extend([error[0] for error in _e])
            if _s:
                _mark.extend([skip[0] for skip in _s])

            unittest.skipIf(
                case in str(_mark),
                f'The pre-depend case :{case} has failed! Skip the specified case!'
            )(func)(self)

        return inner_func

    return wrap_func


def logs(func):
    @wraps(func)
    def wrap_func(*args, **kwargs):
        tuple_args = args
        dict_kwargs = kwargs
        func(*args, **kwargs)
        log.debug(
            f'{func.__name__}(*args:tuple = *{tuple_args}, **kwargs:dict = **{dict_kwargs})',
            extra={'status': 'PASS'}
        )

    return wrap_func

「小贴士」：点击头像→【关注】按钮，获取更多软件测试的晋升认知不迷路！ 🚀