落地后执行效率的对比图加入缓存系统之前运行时间62分钟加入后运行时间35分钟一、背景实际项目中脚本都是使用的数据驱动基于公司历史原因测试数据都存在系统A中脚本需要调用这个系统的接口/getCase/来拿每一个用例文件都通过参数来拿去指定的接口的测试用例和测试数据。例如def get_case(): # 调用系统A接口拿到测试用例和测试数据 pass # ------------------用例文件 pytest.mark.parametrize(test_data,get_case()) def test_func(test_data,pro_client): # 登录 # 验证业务接口 resp pro_client.post(url 业务接口请求地址, json{} ) assert resp.status_code 200基于这个背景所以每次pytest在collect阶段都要逐一调用接口/getCase/collect阶段1400条用例一共耗时1400S共23分钟极大的影响了执行的效率。原因分析pytest.mark.parametrize中的表达式会在收集阶段立即求值因为数据驱动的来源是接口所以会在收集阶段就调用/getCase/非常耗时二、思考解决方案案例的增量和维护的周期大多数时候不会高于24小时。也就是24小时内不会变动这些案例和数据可以类似于redis做一个高速读写的缓存系统从而将这23分钟变成1分钟大大的提高这个效率。方案1申请一个服务器部署redis或者本地部署redis或者执行机部署redis。方案2python工程内本地做一个简化版本的缓存系统即可。基于资源角度考虑选择方案2三、实现代码讲解import json import hashlib import time import os from functools import wraps # 缓存配置兼容 Windows/Linux current_dir os.path.dirname(os.path.abspath(__file__)) CACHE_DIR os.path.join(current_dir, .pytest_cache, cases) os.makedirs(CACHE_DIR, exist_okTrue) CACHE_EXPIRE_SECONDS 24 * 60 * 60 # 通用缓存key生成函数替换原来的 def get_cache_key_general(func_name, args, kwargs): 通用的缓存key生成函数支持任意函数和任意参数 Args: func_name: 函数名称 args: 位置参数元组 kwargs: 关键字参数字典 Returns: 32位MD5字符串 # 将所有参数组合成唯一标识 params { func_name: func_name, args: args, kwargs: kwargs } # sort_keysTrue 确保参数顺序不影响结果 # defaultstr 处理非JSON类型的参数如datetime等 params_str json.dumps(params, sort_keysTrue, defaultstr, ensure_asciiFalse) return hashlib.md5(params_str.encode(utf-8)).hexdigest() # 缓存文件操作函数不变 def get_cache_file_path(cache_key): 获取缓存文件路径 return os.path.join(CACHE_DIR, f{cache_key}.json) def is_cache_valid(cache_file): 检查缓存是否在有效期内 if not os.path.exists(cache_file): return False mtime os.path.getmtime(cache_file) current_time time.time() return (current_time - mtime) CACHE_EXPIRE_SECONDS def save_to_cache(cache_key, data): 保存数据到缓存 cache_file get_cache_file_path(cache_key) with open(cache_file, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) return cache_file def load_from_cache(cache_key): 从缓存加载数据 cache_file get_cache_file_path(cache_key) if is_cache_valid(cache_file): with open(cache_file, r, encodingutf-8) as f: return json.load(f) return None def clear_all_cache(): 清理所有缓存文件 count 0 if os.path.exists(CACHE_DIR): for filename in os.listdir(CACHE_DIR): if filename.endswith(.json): file_path os.path.join(CACHE_DIR, filename) os.remove(file_path) count 1 return count def clear_expired_cache(): 清理过期的缓存文件 expired_count 0 if not os.path.exists(CACHE_DIR): return 0 for filename in os.listdir(CACHE_DIR): if filename.endswith(.json): file_path os.path.join(CACHE_DIR, filename) if not is_cache_valid(file_path): os.remove(file_path) expired_count 1 return expired_count # 通用装饰器使用新的缓存key函数 def cache_24h(force_refreshFalse): 24小时缓存的装饰器通用版本支持任意函数和任意参数 用法: cache_24h() def my_func(a, b, c): return result cache_24h(force_refreshTrue) def my_func2(x, y): return result def decorator(func): wraps(func) def wrapper(*args, **kwargs): # 使用通用函数生成缓存key cache_key get_cache_key_general(func.__name__, args, kwargs) if not force_refresh: cached_data load_from_cache(cache_key) if cached_data is not None: return cached_data result func(*args, **kwargs) save_to_cache(cache_key, result) return result return wrapper return decorator # pytest 集成不变 def pytest_addoption(parser): parser.addoption( --refresh-cases, actionstore_true, help强制刷新所有测试用例缓存 ) parser.addoption( --cache-hours, typeint, default24, help缓存有效期小时默认24小时 ) parser.addoption( --clear-cache, actionstore_true, help清理所有缓存文件后退出 ) def pytest_configure(config): global CACHE_EXPIRE_SECONDS cache_hours config.getoption(--cache-hours) CACHE_EXPIRE_SECONDS cache_hours * 60 * 60 if config.getoption(--clear-cache): count clear_all_cache() print(f✅ 已清理 {count} 个缓存文件) import sys sys.exit(0) if config.getoption(--refresh-cases): count clear_all_cache() print(f 已清理 {count} 个旧缓存将重新请求接口获取最新数据) def pytest_sessionstart(session): auto_clean os.environ.get(AUTO_CLEAN_CACHE, 0) if auto_clean 1: expired clear_expired_cache() if expired: print(f 自动清理了 {expired} 个过期缓存文件)用装饰器装饰调用系统A的接口/getCase/即可在调用前会自动判断本地是否有缓存文件从而实现collect阶段的时效问题。这样只能解决24小时内的第二次执行的时效问题那如何解决第一次的问题呢使用定时任务做一次collect only的pytest执行即可例如每天凌晨3点执行刷新本地缓存。命令如下pytest.main([--collect-only])如果你是使用的CI/CD那么这个方案解决不了你的问题你需要在jenkinsfile中拉取这个缓存文件或者使用隔离的redis方案去完成缓存方案。四、落地遇到问题的再次解决1、24小时内有修改用例并且24小时内需要回归这些改动后的用例此时需要指定刷新缓存需要加入刷新机制2、定时任务在24小时内指定执行但是过期时间是24小时可能存在无法刷新的情况会导致执行时判定缓存过期导致落地效果不如预期解决方案加入刷新机制并且定时任务中的入口强制刷新缓存即可。def pytest_addoption(parser): parser.addoption( --refresh-cases, actionstore_true, help强制刷新所有测试用例缓存 ) parser.addoption( --cache-hours, typeint, default24, help缓存有效期小时默认24小时 ) parser.addoption( --clear-cache, actionstore_true, help清理所有缓存文件后退出 ) def pytest_configure(config): global CACHE_EXPIRE_SECONDS cache_hours config.getoption(--cache-hours) CACHE_EXPIRE_SECONDS cache_hours * 60 * 60 if config.getoption(--clear-cache): count clear_all_cache() print(f✅ 已清理 {count} 个缓存文件) import sys sys.exit(0) if config.getoption(--refresh-cases): count clear_all_cache() print(f 已清理 {count} 个旧缓存将重新请求接口获取最新数据) def pytest_sessionstart(session): auto_clean os.environ.get(AUTO_CLEAN_CACHE, 0) if auto_clean 1: expired clear_expired_cache() if expired: print(f 自动清理了 {expired} 个过期缓存文件)后续只需要在pytest命令中加入pytest --refresh-cases这个参数即可强制刷新缓存。