fiddler模拟弱网延时请求

场景：

由于网络原因，单据部分推送失败。

使用fiddler+Python重现：

1、打开fiddler，按 Ctrl+R 打开 CustomRules.js
2、找到 static function OnBeforeResponse(oSession: Session) {
3、在函数内添加延迟代码
4、保存（Ctrl+S）
5、清空 Fiddler
6、执行Python代码

备注：有时 CustomRules.js添加延时代码不生效；需要在 AutoResponder 中点击 Add Rule，在 Rule Editor 中粘贴 URL，前面加上 EXACT；或者（去掉复杂的域名匹配，只匹配 URL 路径）。

注意：规则前面的空格去掉。

"""
使用 Fiddler 代理进行非销推送接口弱网测试
文件名：fiddler_proxy_test.py功能说明：
1. 通过 Fiddler 代理发送请求，配合 Fiddler 自定义规则模拟弱网环境
2. 支持启用/禁用代理切换
3. 支持补偿重试机制测试
4. 自动处理 HTTPS 证书验证问题
"""import requests
import json
import time
from datetime import datetime
import configparser
import os
import urllib3# 禁用 SSL 警告（因为 Fiddler 会拦截 HTTPS 流量）
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)class FiddlerProxyConfig:"""Fiddler 代理配置类"""# Fiddler 默认监听端口FIDDLER_HOST = "127.0.0.1"FIDDLER_PORT = 8888@classmethoddef get_proxies(cls, enabled=True):"""获取代理配置参数:enabled: 是否启用代理，True=使用 Fiddler 代理，False=直连返回:代理字典或 None"""if enabled:proxy_url = f"http://{cls.FIDDLER_HOST}:{cls.FIDDLER_PORT}"return {'http': proxy_url,'https': proxy_url,}return None@classmethoddef check_fiddler_running(cls):"""检查 Fiddler 是否正在运行返回:True=Fiddler 正在运行，False=未运行"""try:response = requests.get("http://ipv4.fiddler:8888/",  # Fiddler 特殊域名timeout=2,proxies=cls.get_proxies(True))return Trueexcept:# 尝试直接连接端口import socketsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)sock.settimeout(1)result = sock.connect_ex((cls.FIDDLER_HOST, cls.FIDDLER_PORT))sock.close()return result == 0def load_config():"""从配置文件加载配置信息"""config = configparser.ConfigParser()config_path = os.path.join(os.path.dirname(__file__), 'config.ini')if not os.path.exists(config_path):raise FileNotFoundError(f"配置文件不存在: {config_path}")config.read(config_path, encoding='utf-8')# 读取单号ydOaNonsalesNo = config.get('TEST_DATA', 'ydOaNonsalesNo')# ccfox 参数可选ccfox = config.get('TEST_DATA', 'ccfox', fallback=ydOaNonsalesNo)# 读取超时配置timeouts = {}if config.has_section('TIMEOUT'):for key, value in config.items('TIMEOUT'):if not key.startswith('#'):try:timeouts[key] = float(value)except ValueError:print(f"[WARNING] 无法解析超时配置 {key}={value}，已跳过")# 读取重试配置retry_config = {'auto_retry': config.getboolean('RETRY', 'auto_retry', fallback=False),'retry_wait': config.getfloat('RETRY', 'retry_wait', fallback=2),'retry_timeout': config.getfloat('RETRY', 'retry_timeout', fallback=30)}return {'url': config.get('API', 'url'),'access_token': config.get('API', 'access_token'),'test_data': {'ydOaNonsalesNo': ydOaNonsalesNo,'ccfox': ccfox},'timeouts': timeouts,'retry': retry_config}def test_single_request(url, test_data, headers, timeout, scenario_name, use_proxy=True):"""执行单次请求测试（支持 Fiddler 代理）参数:url: 接口URLtest_data: 请求数据headers: 请求头timeout: 超时时间（秒）scenario_name: 场景名称use_proxy: 是否使用 Fiddler 代理返回: (成功标志, 结果详情, 耗时)"""start_time = time.time()# 获取代理配置proxies = FiddlerProxyConfig.get_proxies(use_proxy)try:response = requests.post(url,json=test_data,headers=headers,timeout=timeout,proxies=proxies,          # 使用 Fiddler 代理verify=False               # 禁用 SSL 证书验证)elapsed = time.time() - start_time# 检查是否业务成功if response.status_code == 200:try:result = response.json()error_code = result.get('errorCode', '')status = result.get('status', False)message = result.get('message', '')if error_code in ['', '0'] and status != False:# 业务成功return True, {"type": "BUSINESS_SUCCESS","status_code": response.status_code,"message": message or "推送成功","result": result}, elapsedelse:# 业务失败return False, {"type": "BUSINESS_FAILURE","status_code": response.status_code,"error_code": error_code,"message": message or "业务失败","result": result}, elapsedexcept Exception as e:return False, {"type": "PARSE_ERROR","status_code": response.status_code,"message": f"响应解析失败: {str(e)}","content": response.text[:200]}, elapsedelse:return False, {"type": "HTTP_ERROR","status_code": response.status_code,"message": f"HTTP错误: {response.status_code}"}, elapsedexcept requests.exceptions.ConnectTimeout as e:elapsed = time.time() - start_timereturn False, {"type": "CONNECT_TIMEOUT","message": "连接超时","error": str(e)}, elapsedexcept requests.exceptions.ReadTimeout as e:elapsed = time.time() - start_timereturn False, {"type": "READ_TIMEOUT","message": "读取超时","error": str(e)}, elapsedexcept requests.exceptions.Timeout as e:elapsed = time.time() - start_timereturn False, {"type": "TIMEOUT","message": "请求超时","error": str(e)}, elapsedexcept requests.exceptions.ProxyError as e:elapsed = time.time() - start_timereturn False, {"type": "PROXY_ERROR","message": "代理连接失败（Fiddler 未运行？）","error": str(e)}, elapsedexcept Exception as e:elapsed = time.time() - start_timereturn False, {"type": "EXCEPTION","message": f"异常: {type(e).__name__}","error": str(e)}, elapseddef test_with_fiddler():"""使用 Fiddler 代理进行弱网测试"""print("="*60)print("  非销推送接口 - Fiddler 代理弱网测试工具")print("="*60)# 加载配置try:config = load_config()print("[OK] 配置文件加载成功")print(f"  接口URL: {config['url']}")print(f"  测试单号: {config['test_data']['ydOaNonsalesNo']}")print(f"  Token: {config['access_token'][:30]}...")except Exception as e:print(f"[ERROR] 配置文件加载失败: {str(e)}")return# 检查 Fiddler 是否运行print("\n[检查] Fiddler 代理状态...")if FiddlerProxyConfig.check_fiddler_running():print(f"[OK] Fiddler 正在运行 ({FiddlerProxyConfig.FIDDLER_HOST}:{FiddlerProxyConfig.FIDDLER_PORT})")use_proxy = Trueelse:print(f"[WARNING] Fiddler 未运行或端口 {FiddlerProxyConfig.FIDDLER_PORT} 未监听")print("  >> 提示：请先启动 Fiddler，否则将直连测试（无法模拟弱网）")# 询问是否继续user_input = input("\n是否继续测试？(y=继续直连测试, n=退出): ").strip().lower()if user_input != 'y':print("测试已取消")returnuse_proxy = Falseprint("[INFO] 将使用直连模式（不经过 Fiddler）")url = config['url']test_data = config['test_data']# 准备请求头headers = {'Content-Type': 'application/json','access_token': config['access_token']}# 生成测试场景timeout_scenarios = []timeouts_config = config['timeouts']if not timeouts_config:print("\n[WARNING] 配置文件中没有配置任何超时场景，将使用默认场景")timeout_scenarios = [{"name": "默认超时(30秒)", "timeout": 30}]else:name_mapping = {'extreme_short': '极短超时','short': '短超时','one_second': '1秒超时','three_seconds': '3秒超时','server_timeout_test': '服务端超时测试',}for key, timeout_value in timeouts_config.items():friendly_name = name_mapping.get(key, key)scenario_name = f"{friendly_name}({timeout_value}秒)"timeout_scenarios.append({"name": scenario_name,"timeout": timeout_value,"config_key": key})# 获取重试配置retry_config = config['retry']auto_retry = retry_config['auto_retry']retry_wait = retry_config['retry_wait']retry_timeout = retry_config['retry_timeout']proxy_status = "启用 Fiddler 代理" if use_proxy else "直连模式"print(f"\n{'='*60}")print(f"测试配置: {proxy_status}")if auto_retry:print(f"将测试 {len(timeout_scenarios)} 个场景（带自动补偿重试）")print(f"重试配置: 等待{retry_wait}秒后重试，重试超时{retry_timeout}秒")else:print(f"将测试 {len(timeout_scenarios)} 个场景（不自动重试）")if use_proxy:print("\n[注意] Fiddler 延迟配置:")print("  请在 Fiddler CustomRules.js 中设置延迟时间")print("  例如: oSession['response-trickle-delay'] = '5000'")print("="*60)results = []for idx, scenario in enumerate(timeout_scenarios, 1):print(f"\n{'='*60}")print(f"[{idx}/{len(timeout_scenarios)}] 测试场景: {scenario['name']}")print(f"{'='*60}")# 第一次请求print(f"\n【第1次推送】超时设置: {scenario['timeout']}秒")print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]}")success1, result1, elapsed1 = test_single_request(url, test_data, headers, scenario['timeout'], scenario['name'], use_proxy)if success1:print(f"[成功] 第1次推送成功")print(f"  耗时: {elapsed1:.3f}秒")print(f"  消息: {result1.get('message', '')}")results.append({"scenario": scenario['name'],"first_attempt": "SUCCESS","retry_needed": False,"elapsed": elapsed1,"result": result1})else:# 第一次失败failure_type = result1.get('type', 'UNKNOWN')if failure_type in ['READ_TIMEOUT', 'CONNECT_TIMEOUT', 'TIMEOUT']:print(f"[预期失败] 第1次推送超时")print(f"  类型: {result1.get('message', '')}")print(f"  耗时: {elapsed1:.3f}秒")if auto_retry:print(f"\n  >> 模拟补偿任务，等待{retry_wait}秒后重试...")time.sleep(retry_wait)# 第二次请求print(f"\n【第2次推送(补偿)】超时设置: {retry_timeout}秒")print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]}")success2, result2, elapsed2 = test_single_request(url, test_data, headers, retry_timeout, scenario['name'], use_proxy)if success2:print(f"[成功] 第2次推送成功（补偿任务生效）")print(f"  耗时: {elapsed2:.3f}秒")print(f"  消息: {result2.get('message', '')}")results.append({"scenario": scenario['name'],"first_attempt": "TIMEOUT","first_elapsed": elapsed1,"retry_needed": True,"retry_result": "SUCCESS","retry_elapsed": elapsed2,"total_elapsed": elapsed1 + retry_wait + elapsed2,"result": result2})else:failure_type2 = result2.get('type', 'UNKNOWN')print(f"[失败] 第2次推送也失败")print(f"  类型: {result2.get('type', '')}")print(f"  耗时: {elapsed2:.3f}秒")print(f"  错误: {result2.get('message', '')}")results.append({"scenario": scenario['name'],"first_attempt": "TIMEOUT","first_elapsed": elapsed1,"retry_needed": True,"retry_result": "FAILURE","retry_type": failure_type2,"retry_elapsed": elapsed2,"total_elapsed": elapsed1 + retry_wait + elapsed2,"error": result2})else:print(f"\n  >> 补偿重试已禁用，不执行第2次推送")results.append({"scenario": scenario['name'],"first_attempt": "TIMEOUT","first_elapsed": elapsed1,"retry_needed": False,"elapsed": elapsed1,"error": result1})elif failure_type == 'PROXY_ERROR':print(f"[代理错误] 无法连接到 Fiddler 代理")print(f"  耗时: {elapsed1:.3f}秒")print(f"  错误: {result1.get('message', '')}")print(f"\n  >> 请确保 Fiddler 正在运行并监听端口 {FiddlerProxyConfig.FIDDLER_PORT}")results.append({"scenario": scenario['name'],"first_attempt": "PROXY_ERROR","retry_needed": False,"elapsed": elapsed1,"error": result1})# 如果是代理错误，询问是否继续user_input = input("\n是否继续测试其他场景？(y/n): ").strip().lower()if user_input != 'y':print("测试已中止")breakelse:# 业务错误print(f"[业务失败] 第1次推送返回业务错误")print(f"  类型: {failure_type}")print(f"  耗时: {elapsed1:.3f}秒")print(f"  错误: {result1.get('message', '')}")if 'error_code' in result1:print(f"  错误码: {result1['error_code']}")results.append({"scenario": scenario['name'],"first_attempt": failure_type,"retry_needed": False,"elapsed": elapsed1,"error": result1})# 场景之间间隔if idx < len(timeout_scenarios):time.sleep(2)# 输出测试总结print(f"\n{'='*60}")print("测试总结")print(f"{'='*60}")print(f"代理模式: {proxy_status}")print(f"{'场景':<30s} | {'第1次':<12s} | {'重试':<8s} | {'最终结果':<12s} | {'总耗时'}")print("-" * 85)for result in results:scenario = result['scenario'][:28]first = result.get('first_attempt', 'N/A')[:12]retry_needed = '是' if result.get('retry_needed', False) else '否'if result.get('retry_needed'):final_result = result.get('retry_result', 'N/A')[:12]total_time = result.get('total_elapsed', 0)else:final_result = first[:12]total_time = result.get('elapsed', 0)print(f"{scenario:<30s} | {first:<12s} | {retry_needed:<8s} | {final_result:<12s} | {total_time:.3f}秒")# 保存测试结果log_dir = os.path.join(os.path.dirname(__file__), 'logs')if not os.path.exists(log_dir):os.makedirs(log_dir)proxy_suffix = "_proxy" if use_proxy else "_direct"log_file = os.path.join(log_dir, f"fiddler_test{proxy_suffix}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json")with open(log_file, 'w', encoding='utf-8') as f:json.dump({"test_mode": "fiddler_proxy" if use_proxy else "direct","fiddler_config": {"host": FiddlerProxyConfig.FIDDLER_HOST,"port": FiddlerProxyConfig.FIDDLER_PORT},"test_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),"results": results}, f, indent=2, ensure_ascii=False)print(f"\n测试结果已保存到: {log_file}")if __name__ == "__main__":test_with_fiddler()