从性能测试比较简单的角度入手ai全自动化

import osimport reimport subprocessimport timefrom urllib import responseimport requestsimport jsonimport matplotlib.pyplot as pltfrom prometheus_api_client import PrometheusConnectfrom jmx_utils import modify_jmx_threads, extract_jmx_metrics# 配置参数DEEPSEEK_API_KEY = ""JMETER_PATH = "/path/to/apache-jmeter-5.6.2/bin/jmeter"TARGET_SERVER = "192.168.1.100"TEST_DURATION = 300  # 测试持续时间(秒)PROMETHEUS_URL = "http://localhost:9090"GRAFANA_URL = "http://localhost:3000"defgenerate_jmx_with_deepseek(test_description):    """使用DeepSeek API生成JMeter脚本"""    headers = {        "Authorization": f"Bearer {DEEPSEEK_API_KEY}",        "Content-Type": "application/json"    }    prompt = f"""    请创建一个JMeter测试计划(JMX文件)，用于测试以下场景：    {test_description}    要求：    1. 基础配置    协议:[HTTP/HTTPS]    域名/IP:{TARGET_SERVER}    端口:[443]    headers:{{}}    2. 线程组设置    -并发用户数:[100]    -启动时间(秒):[30]    -持续时长(秒):[600]    -循环次数:[永久/具体数字]    3. 测试场景步骤（按顺序说明）    -步骤1:[事务名称] (例：用户登录)     请求类型:[POST]     路径:[/login]     请求体:{{"username":"","password":""}}     请求体中:json表达式中username和password的值使用变量     关联提取：     提取变量名:[token]     提取方式:[JSON提取器]/[正则表达式]     表达式:[$.access_token]    -步骤2:[事务名称] (例：查询商品)     请求类型:[GET]     路径:[/products/product_id]     路径中:product_id使用变量形式     断言：     响应码:[200]     响应时间阈值(ms):[500]     响应内容包含:["success":true]    4. 参数化需求    -CSV文件1：    -文件名:[users.csv]    -变量名:username,password    -循环方式:[顺序读取]/[随机读取]    6. 监控需求    -配置Prometheus Metrics    -必采集指标：    -响应时间分布    -错误率    -吞吐量(TPS)    -服务器资源(CPU/MEM)    """    payload = {        "model": "deepseek-chat",        "messages": [            {"role":"system", "content":"你是一个专业的性能测试工程师，专门生成JMeter测试脚本"},            {"role":"user", "content": prompt}        ],        "temperature": 0.3,        "max_tokens": 2000    }    response = requests.post(        "https://api.deepseek.com/v1/chat/completions",        headers=headers,        json=payload    )    if response.status_code != 200:        raise Exception(f"DeepSeek API错误: {response.text}")    response_content = response.json()    return response_contentdefextract_jmx_from_response(response_content):    """    从DeepSeek响应中提取JMX文件内容    支持多种格式：纯JMX、XML代码块、带注释的JMX等    """    # 1. 尝试匹配XML代码块（带标签的）    xml_block_match = re.search(r'```xmln(.*?)n```', response_content, re.DOTALL)    if xml_block_match:        return xml_block_match.group(1).strip()    # 2. 尝试匹配通用代码块（无标签的）    code_block_match = re.search(r'```n(.*?)n```', response_content, re.DOTALL)    if code_block_match:        return code_block_match.group(1).strip()    # 3. 尝试匹配纯JMX内容（无代码块标记）    jmx_match = re.search(r'<jmeterTestPlan.*?</jmeterTestPlan>', response_content, re.DOTALL)    if jmx_match:        return jmx_match.group(0).strip()    # 4. 尝试匹配关键JMX片段    jmx_fragment_match = re.search(r'(<TestPlan.*?</TestPlan>|<ThreadGroup.*?</ThreadGroup>)', response_content, re.DOTALL)    if jmx_fragment_match:        # 构建完整的JMX框架        returnf"""<?xml version="1.0" encoding="UTF-8"?><jmeterTestPlan version="1.2" properties="5.0" jmeter="5.6.2">  <hashTree>    {jmx_fragment_match.group(0)}  </hashTree></jmeterTestPlan>"""    # 5. 最后尝试：如果都没有匹配到，返回整个响应内容（可能是纯JMX）    return response_content.strip()defgenerate_jmx():    # 获取响应内容    result = response.json()    raw_content = result['choices'][0]['message']['content']    # 提取JMX内容    jmx_content = extract_jmx_from_response(raw_content)    # 保存JMX文件    withopen("generated_test.jmx", "w") as f:        f.write(jmx_content)    print("✅ JMX脚本已通过DeepSeek生成并提取")    # 添加后处理确保变量正确        return"generated_test.jmx"defrun_jmeter_test(jmx_file):    """执行JMeter性能测试"""    print("🚀 开始性能测试...")    cmd = [        JMETER_PATH,        "-n",  # 非GUI模式        "-t", jmx_file,        "-l", "test_results.jtl",        "-e", "-o", "html_report",        "-JTARGET_HOST=" + TARGET_SERVER,        "-JTEST_DURATION=" + str(TEST_DURATION)    ]    process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)    stdout, stderr = process.communicate()    if process.returncode != 0:        print(f"❌ JMeter执行错误: {stderr.decode()}")        returnFalse    print("✅ 性能测试完成")    returnTruedefstart_monitoring():    """启动监控系统（需提前安装Docker）"""    # 启动Prometheus    subprocess.run([        "docker", "run", "-d", "-p", "9090:9090",        "-v", f"{os.getcwd()}/prometheus.yml:/etc/prometheus/prometheus.yml",        "--name", "prometheus", "prom/prometheus"    ])    # 启动Grafana    subprocess.run([        "docker", "run", "-d", "-p", "3000:3000",        "--name", "grafana", "grafana/grafana"    ])    print("✅ 监控系统已启动")    print(f"Prometheus: {PROMETHEUS_URL}")    print(f"Grafana: {GRAFANA_URL} (初始账号: admin/admin)")defcollect_metrics():    """收集性能指标"""    print("📊 收集性能指标...")    prom = PrometheusConnect(url=PROMETHEUS_URL, disable_ssl=True)    # 收集服务器指标    cpu_query = '100 - (avg by(instance)(irate(node_cpu_seconds_total{mode="idle"}[1m])) * 100'    mem_query = '100 * (1 - ((node_memory_MemAvailable_bytes) / (node_memory_MemTotal_bytes)))'    net_query = 'irate(node_network_receive_bytes_total{device="eth0"}[1m])'    metrics = {        "CPU_Usage": prom.custom_query(cpu_query),        "Memory_Usage": prom.custom_query(mem_query),        "Network_IO": prom.custom_query(net_query),        "Response_Time": prom.custom_query('jmeter_response_time_seconds{quantile="0.95"}'),        "Throughput": prom.custom_query('rate(jmeter_requests_total[1m])')    }    # 保存指标数据    withopen("metrics.json", "w") as f:        json.dump(metrics, f)    # 生成可视化图表    plt.figure(figsize=(12, 8))    plt.subplot(2, 2, 1)    plt.plot([m['value'][1] for m in metrics['CPU_Usage']], label='CPU %')    plt.title("CPU Usage")    plt.subplot(2, 2, 2)    plt.plot([m['value'][1] for m in metrics['Memory_Usage']], label='Memory %')    plt.title("Memory Usage")    plt.subplot(2, 2, 3)    plt.plot([m['value'][1] for m in metrics['Response_Time']], label='Response Time (s)')    plt.title("95% Response Time")    plt.subplot(2, 2, 4)    plt.plot([m['value'][1] for m in metrics['Throughput']], label='Requests/s')    plt.title("Throughput")    plt.tight_layout()    plt.savefig("metrics_chart.png")    print("✅ 性能指标已收集并可视化")defanalyze_with_deepseek():    """使用DeepSeek分析测试结果并生成调优建议"""    # 读取测试结果摘要    withopen("html_report/statistics.json") as f:        stats = json.load(f)    # 读取指标数据    withopen("metrics.json") as f:        metrics = json.load(f)    # 准备分析提示    prompt = f"""    作为性能测试专家，请分析以下性能测试结果并提供调优建议：    ### 测试配置    - 目标服务器: {TARGET_SERVER}    - 持续时间: {TEST_DURATION}秒    - 初始线程数: 50    ### 测试结果摘要    {json.dumps(stats['Total'], indent=2)}    ### 关键性能指标趋势    - 平均CPU使用率: {sum(m['value'][1] for m in metrics['CPU_Usage'])/len(metrics['CPU_Usage']):.2f}%    - 峰值响应时间: {max(m['value'][1] for m in metrics['Response_Time']):.2f}秒    - 平均吞吐量: {sum(m['value'][1] for m in metrics['Throughput'])/len(metrics['Throughput']):.2f} 请求/秒    ### 问题分析    1. 识别性能瓶颈    2. 分析错误率高的原因    3. 评估资源利用情况    ### 调优建议    1. JMeter配置优化建议    2. 服务器配置优化建议    3. 应用代码优化建议    4. 下次测试参数调整建议    """    headers = {        "Authorization": f"Bearer {DEEPSEEK_API_KEY}",        "Content-Type": "application/json"    }    payload = {        "model": "deepseek-chat",        "messages": [            {"role": "system", "content": "你是一个资深的性能测试优化专家"},            {"role": "user", "content": prompt}        ],        "temperature": 0.5,        "max_tokens": 1500    }    response = requests.post(        "https://api.deepseek.com/v1/chat/completions",        headers=headers,        json=payload    )    if response.status_code != 200:        raise Exception(f"DeepSeek分析失败: {response.text}")    result = response.json()    analysis = result['choices'][0]['message']['content']    # 保存分析报告    withopen("performance_analysis.md", "w") as f:        f.write(f"# DeepSeek性能分析报告nn")        f.write(analysis)    print("✅ 性能分析报告已生成")    return analysisdefauto_tune(analysis):    """根据分析结果自动调优"""    # 从分析中提取关键建议    if"减少线程数"in analysis or"降低并发"in analysis:        print("🛠️ 根据建议减少线程数...")        modify_jmx_threads("generated_test.jmx", new_threads=30)    if"增加线程数"in analysis or"提高并发"in analysis:        print("🛠️ 根据建议增加线程数...")        modify_jmx_threads("generated_test.jmx", new_threads=70)    # 提取其他调优参数    ramp_up = extract_jmx_metrics("generated_test.jmx", "ThreadGroup.ramp_time")    if"增加渐变时间"in analysis:        new_ramp = int(ramp_up) * 1.5        print(f"🛠️ 增加渐变时间到 {new_ramp}秒")        # 实际实现需要修改JMX文件    # 这里可以添加更多自动调优逻辑    print("🔄 配置已更新，准备重新测试...")defmain():    # 1. 生成Prometheus配置文件    withopen("prometheus.yml", "w") as f:        f.write(f"""global:  scrape_interval: 5sscrape_configs:  - job_name: 'node_exporter'    static_configs:      - targets: ['{TARGET_SERVER}:9100']  - job_name: 'jmeter'    static_configs:      - targets: ['localhost:9270']        """)    # 2. 使用DeepSeek生成JMeter脚本    test_description = """    测试用户登录API的性能：    - API端点：/api/v1/login    - 请求方法：POST    - 请求体：{{"username": "testuser", "password": "Pass1234"}}    - 验证响应状态码200和响应时间<500ms    """    response_contente = generate_jmx_with_deepseek(test_description)    jmx_file = generate_jmx()    # 3. 启动监控    start_monitoring()    time.sleep(10)  # 等待监控服务启动    # 4. 执行性能测试    ifnot run_jmeter_test(jmx_file):        return    # 5. 收集指标    collect_metrics()    # 6. 使用DeepSeek分析结果    analysis = analyze_with_deepseek()    print("n=== DeepSeek分析结果 ===")    print(analysis)    # 7. 自动调优    auto_tune(analysis)    # 8. 执行调优后的测试    print("n🚀 开始调优后的性能测试...")    run_jmeter_test(jmx_file)    collect_metrics()    print("n🎉 全流程完成！")    print(f"测试报告: file://{os.getcwd()}/html_report/index.html")    print(f"性能分析: file://{os.getcwd()}/performance_analysis.md")if __name__ == "__main__":    main()

# 安装依赖pip install requests prometheus-api-client matplotlib# 安装Node Exporter (目标服务器)docker run -d -p 9100:9100 --name node_exporter prom/node-exporter# 安装JMeter Prometheus插件wget https://repo1.maven.org/maven2/kg/apc/jmeter-plugins-manager/1.9/jmeter-plugins-manager-1.9.jar# 放入JMeter的lib/ext目录