import subprocess import json import time from datetime import datetime def run_iperf_test(config): """运行iperf测试""" cmd = ["iperf3", "-c", config["server"], "-t", str(config["duration"])] if config.get("parallel"): cmd.extend(["-P", str(config["parallel"])]) if config.get("bandwidth"): cmd.extend(["-b", config["bandwidth"]]) # 运行测试 result = subprocess.run( cmd, capture_output=True, text=True, timeout=config["duration"] + 10 ) return parse_iperf_output(result.stdout) def parse_iperf_output(output): """解析iperf输出""" lines = output.split('\n') results = { "bandwidth": 0, "jitter": 0, "loss": 0, "timestamp": datetime.now().isoformat() } for line in lines: if "bits/sec" in line and "sender" in line: # 解析带宽 if "Gbits/sec" in line: match = re.search(r'([\d\.]+)\s+Gbits/sec', line) if match: results["bandwidth"] = float(match.group(1)) * 1000 elif "Mbits/sec" in line: # ... 类似处理 # 解析其他指标... return results def main(): # 测试配置 tests = [ {"server": "192.168.1.100", "duration": 30, "parallel": 1}, {"server": "192.168.1.100", "duration": 30, "parallel": 4}, {"server": "192.168.1.100", "duration": 30, "parallel": 8}, ] all_results = [] for test_config in tests: print(f"运行测试: {test_config}") result = run_iperf_test(test_config) all_results.append(result) time.sleep(5) # 测试间隔 # 保存结果 with open("test_results.json", "w") as f: json.dump(all_results, f, indent=2) print("测试完成，结果已保存") if __name__ == "__main__": main()

#!/usr/bin/env python3 """ 网络质量定期监控脚本 每天定时测试并记录 """ import schedule import time from iperf_automation import run_iperf_test def daily_monitor(): """每日监控任务""" print(f"开始每日网络监控 - {time.ctime()}") # 测试到多个目标 targets = [ {"name": "本地服务器", "server": "192.168.1.1", "port": 5201}, {"name": "互联网测试点1", "server": "ping.online.net", "port": 5201}, {"name": "互联网测试点2", "server": "iperf.he.net", "port": 5201}, ] results = [] for target in targets: try: result = run_iperf_test({ "server": target["server"], "duration": 60, "parallel": 4 }) result["target"] = target["name"] results.append(result) except Exception as e: print(f"测试{target['name']}失败: {e}") # 保存结果 save_results(results) print("每日监控完成") def save_results(results): """保存结果到数据库或文件""" # 这里可以保存到数据库、文件或发送到监控系统 pass # 设置定时任务 schedule.every().day.at("02:00").do(daily_monitor) # 每天凌晨2点 schedule.every().hour.do(lambda: run_quick_test()) # 每小时快速测试 def run_quick_test(): """快速测试""" pass if __name__ == "__main__": print("网络监控服务启动...") while True: schedule.run_pending() time.sleep(60)

from prometheus_client import start_http_server, Gauge import time # 创建指标 BANDWIDTH_GAUGE = Gauge('network_bandwidth_mbps', 'Network bandwidth in Mbps', ['direction', 'server']) LATENCY_GAUGE = Gauge('network_latency_ms', 'Network latency in ms', ['server']) LOSS_GAUGE = Gauge('network_packet_loss_percent', 'Packet loss percentage', ['server']) def update_metrics(server, results): """更新Prometheus指标""" BANDWIDTH_GAUGE.labels(direction='upload', server=server).set( results.get('upload_bandwidth', 0) ) BANDWIDTH_GAUGE.labels(direction='download', server=server).set( results.get('download_bandwidth', 0) ) LATENCY_GAUGE.labels(server=server).set( results.get('latency', 0) ) LOSS_GAUGE.labels(server=server).set( results.get('loss', 0) ) def monitoring_loop(): """监控循环""" start_http_server(8000) # Prometheus metrics端口 servers = ["server1", "server2", "server3"] while True: for server in servers: try: # 运行iperf测试 results = run_iperf_test_to(server) # 更新指标 update_metrics(server, results) except Exception as e: print(f"监控{server}失败: {e}") time.sleep(300) # 每5分钟一次

class BatchTester: def __init__(self, config_file): self.configs = self.load_configs(config_file) self.results = [] def load_configs(self, config_file): """加载测试配置""" # 从JSON/YAML文件加载 pass def run_all_tests(self): """运行所有测试""" for test_name, config in self.configs.items(): print(f"运行测试: {test_name}") try: result = self.run_single_test(config) result["test_name"] = test_name result["timestamp"] = datetime.now().isoformat() self.results.append(result) # 可选：测试间等待 time.sleep(config.get("interval", 10)) except Exception as e: print(f"测试{test_name}失败: {e}") self.results.append({ "test_name": test_name, "error": str(e), "status": "failed" }) def run_single_test(self, config): """运行单个测试""" # 具体测试逻辑 pass def generate_report(self, format="html"): """生成报告""" if format == "html": return self.generate_html_report() elif format == "pdf": return self.generate_pdf_report() elif format == "markdown": return self.generate_markdown_report() def generate_html_report(self): """生成HTML报告""" template = """  <html>  <head>  <title>网络测试报告</title>  <style>  table { border-collapse: collapse; width: 100%; }  th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }  th { background-color: #f2f2f2; }  .pass { background-color: #d4edda; }  .fail { background-color: #f8d7da; }  .warning { background-color: #fff3cd; }  </style>  </head>  <body>  <h1>网络测试报告</h1>  <p>生成时间: {{ timestamp }}</p>    <h2>测试概览</h2>  <table>  <tr>  <th>测试名称</th>  <th>带宽(Mbps)</th>  <th>延迟(ms)</th>  <th>丢包率(%)</th>  <th>状态</th>  </tr>  {% for result in results %}  <tr class="{{ result.status }}">  <td>{{ result.test_name }}</td>  <td>{{ result.bandwidth | default('N/A') }}</td>  <td>{{ result.latency | default('N/A') }}</td>  <td>{{ result.loss | default('N/A') }}</td>  <td>{{ result.status }}</td>  </tr>  {% endfor %}  </table>    <h2>详细结果</h2>  {% for result in results %}  <h3>{{ result.test_name }}</h3>  <pre>{{ result.raw_output }}</pre>  {% endfor %}  </body>  </html>  """ # 使用模板引擎渲染 # ... return html_content

class NetworkQualityScorer: def __init__(self, weights=None): # 默认权重 self.weights = weights or { "bandwidth": 0.4, # 带宽权重 40% "latency": 0.3, # 延迟权重 30% "loss": 0.2, # 丢包权重 20% "jitter": 0.1, # 抖动权重 10% } # 基准值（可根据网络类型调整） self.baselines = { "lan": { "bandwidth": 1000, # 1Gbps "latency": 1, # 1ms "loss": 0, # 0% "jitter": 0.1, # 0.1ms }, "wan": { "bandwidth": 100, # 100Mbps "latency": 50, # 50ms "loss": 0.1, # 0.1% "jitter": 5, # 5ms }, "internet": { "bandwidth": 50, # 50Mbps "latency": 100, # 100ms "loss": 0.5, # 0.5% "jitter": 10, # 10ms } } def calculate_score(self, results, network_type="wan"): """计算网络质量分数（0-100）""" baseline = self.baselines[network_type] scores = {} # 带宽得分（越高越好） bandwidth_score = min( 100, (results.get("bandwidth", 0) / baseline["bandwidth"]) * 100 ) # 延迟得分（越低越好，需要转换） latency = results.get("latency", float('inf')) if latency <= baseline["latency"]: latency_score = 100 else: # 延迟加倍，分数减半 latency_score = max( 0, 100 * (baseline["latency"] / latency) ) # 丢包得分（0丢包得100分） loss = results.get("loss", 100) # 默认100%丢包 if loss <= baseline["loss"]: loss_score = 100 else: loss_score = max( 0, 100 * (1 - loss / 100) # 转换为比例 ) # 抖动得分 jitter = results.get("jitter", float('inf')) if jitter <= baseline["jitter"]: jitter_score = 100 else: jitter_score = max( 0, 100 * (baseline["jitter"] / jitter) ) # 加权总分 total_score = ( bandwidth_score * self.weights["bandwidth"] + latency_score * self.weights["latency"] + loss_score * self.weights["loss"] + jitter_score * self.weights["jitter"] ) return { "total": round(total_score, 1), "components": { "bandwidth": round(bandwidth_score, 1), "latency": round(latency_score, 1), "loss": round(loss_score, 1), "jitter": round(jitter_score, 1), }, "details": { "bandwidth_actual": results.get("bandwidth"), "latency_actual": results.get("latency"), "loss_actual": results.get("loss"), "jitter_actual": results.get("jitter"), } } def get_quality_level(self, score): """根据分数获取质量等级""" if score >= 90: return {"level": "优秀", "color": "green", "emoji": "✅"} elif score >= 80: return {"level": "良好", "color": "blue", "emoji": "👍"} elif score >= 60: return {"level": "一般", "color": "yellow", "emoji": "⚠️"} elif score >= 40: return {"level": "较差", "color": "orange", "emoji": "🔶"} else: return {"level": "差", "color": "red", "emoji": "❌"} def generate_quality_report(self, test_results): """生成质量报告""" report = { "timestamp": datetime.now().isoformat(), "tests": [], "summary": { "average_score": 0, "best_test": None, "worst_test": None, "recommendations": [] } } total_score = 0 best_score = -1 worst_score = 101 for test in test_results: score_result = self.calculate_score(test["metrics"]) quality = self.get_quality_level(score_result["total"]) test_report = { "name": test["name"], "score": score_result["total"], "quality": quality, "details": score_result } report["tests"].append(test_report) # 更新统计 total_score += score_result["total"] if score_result["total"] > best_score: best_score = score_result["total"] report["summary"]["best_test"] = test["name"] if score_result["total"] < worst_score: worst_score = score_result["total"] report["summary"]["worst_test"] = test["name"] # 计算平均分 if report["tests"]: report["summary"]["average_score"] = total_score / len(report["tests"]) # 生成建议 report["summary"]["recommendations"] = self.generate_recommendations(report) return report def generate_recommendations(self, report): """根据测试结果生成改进建议""" recommendations = [] for test in report["tests"]: details = test["details"]["details"] components = test["details"]["components"] # 带宽相关建议 if components["bandwidth"] < 80: actual = details.get("bandwidth_actual", 0) expected = self.baselines["wan"]["bandwidth"] recommendations.append( f"测试 '{test['name']}' 带宽较低 ({actual}Mbps vs 预期{expected}Mbps)。" "建议检查：1.网线质量 2.网络设备规格 3.是否有带宽限制" ) # 延迟相关建议 if components["latency"] < 80: actual = details.get("latency_actual", 0) recommendations.append( f"测试 '{test['name']}' 延迟较高 ({actual}ms)。" "建议：1.优化路由 2.减少网络跳数 3.检查网络设备负载" ) # 丢包相关建议 if components["loss"] < 80: actual = details.get("loss_actual", 0) recommendations.append( f"测试 '{test['name']}' 丢包率较高 ({actual}%)。" "建议：1.检查网络线路 2.优化MTU设置 3.检查网络设备错误计数" ) # 去重 return list(set(recommendations))

def create_quality_dashboard(report): """创建质量仪表板""" import matplotlib.pyplot as plt import pandas as pd # 准备数据 tests = [t["name"] for t in report["tests"]] scores = [t["score"] for t in report["tests"]] colors = [t["quality"]["color"] for t in report["tests"]] # 创建图表 fig, axes = plt.subplots(2, 2, figsize=(15, 10)) # 1. 总分柱状图 axes[0, 0].bar(tests, scores, color=colors) axes[0, 0].set_title("各测试总分") axes[0, 0].set_ylabel("分数 (0-100)") axes[0, 0].set_ylim(0, 100) axes[0, 0].tick_params(axis='x', rotation=45) # 2. 各指标雷达图 categories = ['带宽', '延迟', '丢包', '抖动'] N = len(categories) for i, test in enumerate(report["tests"]): values = [ test["details"]["components"]["bandwidth"], test["details"]["components"]["latency"], test["details"]["components"]["loss"], test["details"]["components"]["jitter"] ] values += values[:1] # 闭合雷达图 angles = [n / float(N) * 2 * 3.14159 for n in range(N)] angles += angles[:1] ax = axes[0, 1] ax.plot(angles, values, label=test["name"]) ax.fill(angles, values, alpha=0.1) axes[0, 1].set_title("各测试指标雷达图") axes[0, 1].legend() # 3. 时间趋势图（如果有历史数据） # ... # 4. 质量分布饼图 quality_counts = {} for test in report["tests"]: level = test["quality"]["level"] quality_counts[level] = quality_counts.get(level, 0) + 1 axes[1, 0].pie( quality_counts.values(), labels=quality_counts.keys(), autopct='%1.1f%%', colors=['green', 'blue', 'yellow', 'orange', 'red'][:len(quality_counts)] ) axes[1, 0].set_title("质量等级分布") # 5. 详细数据表格 axes[1, 1].axis('tight') axes[1, 1].axis('off') table_data = [] for test in report["tests"]: table_data.append([ test["name"], f"{test['score']:.1f}", test["quality"]["level"], f"{test['details']['details'].get('bandwidth_actual', 'N/A')}", f"{test['details']['details'].get('latency_actual', 'N/A')}ms", f"{test['details']['details'].get('loss_actual', 'N/A')}%" ]) table = axes[1, 1].table( cellText=table_data, colLabels=["测试", "分数", "等级", "带宽", "延迟", "丢包"], loc='center', cellLoc='center' ) table.auto_set_font_size(False) table.set_fontsize(9) table.scale(1, 1.5) plt.tight_layout() plt.savefig("network_quality_report.png", dpi=300, bbox_inches='tight') plt.show() return fig

# plugin_base.py import abc from typing import Dict, Any class IperfPlugin(abc.ABC): """插件基类""" def __init__(self, name, version): self.name = name self.version = version @abc.abstractmethod def before_test(self, config: Dict[str, Any]) -> Dict[str, Any]: """测试前调用，可以修改配置""" pass @abc.abstractmethod def during_test(self, output_line: str) -> None: """测试过程中每行输出调用""" pass @abc.abstractmethod def after_test(self, results: Dict[str, Any]) -> Dict[str, Any]: """测试后调用，可以处理结果""" pass @abc.abstractmethod def get_ui_widget(self): """返回插件的UI部件""" pass # plugin_manager.py class PluginManager: def __init__(self): self.plugins = {} self.load_plugins() def load_plugins(self): """动态加载插件""" import importlib import pkgutil plugin_packages = ['plugins'] for package_name in plugin_packages: try: package = importlib.import_module(package_name) for _, module_name, _ in pkgutil.iter_modules(package.__path__): module = importlib.import_module(f"{package_name}.{module_name}") for attr_name in dir(module): attr = getattr(module, attr_name) if (isinstance(attr, type) and issubclass(attr, IperfPlugin) and attr != IperfPlugin): plugin_instance = attr() self.plugins[plugin_instance.name] = plugin_instance except ImportError: continue def run_before_hooks(self, config): """运行所有before_test钩子""" for plugin in self.plugins.values(): config = plugin.before_test(config) return config def run_during_hooks(self, output_line): """运行所有during_test钩子""" for plugin in self.plugins.values(): plugin.during_test(output_line) def run_after_hooks(self, results): """运行所有after_test钩子""" for plugin in self.plugins.values(): results = plugin.after_test(results) return results

# plugins/bandwidth_monitor.py import time from typing import Dict, Any from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel, QProgressBar class BandwidthMonitorPlugin(IperfPlugin): """带宽监控插件""" def __init__(self): super().__init__("带宽监控", "1.0") self.start_time = None self.total_bytes = 0 self.max_bandwidth = 0 self.ui_widget = None def before_test(self, config: Dict[str, Any]) -> Dict[str, Any]: self.start_time = time.time() self.total_bytes = 0 self.max_bandwidth = 0 return config def during_test(self, output_line: str) -> None: # 解析输出行，提取带宽信息 if "Mbits/sec" in output_line or "Gbits/sec" in output_line: # 这里添加解析逻辑 pass def after_test(self, results: Dict[str, Any]) -> Dict[str, Any]: if self.start_time: duration = time.time() - self.start_time avg_bandwidth = (self.total_bytes * 8) / duration / 1e6 # Mbps results["plugin_stats"] = { "total_bytes": self.total_bytes, "average_bandwidth_mbps": avg_bandwidth, "max_bandwidth_mbps": self.max_bandwidth, "duration": duration } return results def get_ui_widget(self): if not self.ui_widget: self.ui_widget = QWidget() layout = QVBoxLayout() self.current_label = QLabel("当前带宽: --") self.average_label = QLabel("平均带宽: --") self.max_label = QLabel("最大带宽: --") self.progress_bar = QProgressBar() layout.addWidget(self.current_label) layout.addWidget(self.average_label) layout.addWidget(self.max_label) layout.addWidget(self.progress_bar) self.ui_widget.setLayout(layout) return self.ui_widget

# api_server.py from flask import Flask, request, jsonify import threading import time app = Flask(__name__) class APIServer: def __init__(self, iperf_gui): self.gui = iperf_gui self.test_thread = None self.current_test = None def run(self, host='0.0.0.0', port=5000): """启动API服务器""" app.run(host=host, port=port, threaded=True) @app.route('/api/test/start', methods=['POST']) def start_test(): """启动测试""" data = request.json if self.current_test and self.current_test.is_alive(): return jsonify({"error": "测试正在进行中"}), 409 # 准备测试参数 params = self.prepare_params(data) # 在新线程中运行测试 self.test_thread = threading.Thread( target=self.run_test_in_thread, args=(params,) ) self.test_thread.start() return jsonify({ "status": "started", "test_id": str(int(time.time())) }) @app.route('/api/test/stop', methods=['POST']) def stop_test(): """停止测试""" if self.current_test: self.current_test.stop() return jsonify({"status": "stopped"}) else: return jsonify({"error": "没有正在运行的测试"}), 404 @app.route('/api/test/status', methods=['GET']) def get_status(): """获取测试状态""" if self.current_test and self.current_test.is_alive(): return jsonify({ "status": "running", "progress": self.current_test.get_progress(), "results": self.current_test.get_partial_results() }) else: return jsonify({"status": "stopped"}) @app.route('/api/config', methods=['GET', 'POST']) def config(): """配置管理""" if request.method == 'GET': return jsonify(self.gui.get_current_config()) else: config = request.json self.gui.apply_config(config) return jsonify({"status": "applied"}) def run_test_in_thread(self, params): """在线程中运行测试""" self.current_test = self.gui.start_test_with_params(params) while self.current_test.is_alive(): time.sleep(0.1) self.current_test = None

# websocket_server.py from flask_socketio import SocketIO, emit import json socketio = SocketIO(app, cors_allowed_origins="*") @socketio.on('connect') def handle_connect(): print('客户端连接') emit('connected', {'data': '连接成功'}) @socketio.on('start_test') def handle_start_test(data): """通过WebSocket启动测试""" test_id = start_test_async(data) emit('test_started', {'test_id': test_id}) # 实时推送结果 def send_updates(): while test_is_running(test_id): results = get_latest_results(test_id) emit('test_update', results) socketio.sleep(0.5) socketio.start_background_task(send_updates) @socketio.on('stop_test') def handle_stop_test(data): test_id = data.get('test_id') stop_test(test_id) emit('test_stopped', {'test_id': test_id})

# report_generator.py from jinja2 import Environment, FileSystemLoader import pdfkit import pandas as pd class ReportGenerator: def __init__(self, template_dir='templates'): self.env = Environment( loader=FileSystemLoader(template_dir), extensions=['jinja2.ext.do'] ) # 注册自定义过滤器 self.env.filters['format_bandwidth'] = self.format_bandwidth self.env.filters['format_time'] = self.format_time self.env.filters['quality_color'] = self.quality_color def generate_report(self, data, template_name, output_format='html'): """生成报告""" template = self.env.get_template(template_name) # 渲染模板 html_content = template.render(**data) if output_format == 'html': return html_content elif output_format == 'pdf': return self.html_to_pdf(html_content) elif output_format == 'docx': return self.html_to_docx(html_content) else: raise ValueError(f"不支持的格式: {output_format}") def format_bandwidth(self, value, unit='auto'): """格式化带宽显示""" if unit == 'auto': if value >= 1000: return f"{value/1000:.2f} Gbps" elif value >= 1: return f"{value:.2f} Mbps" else: return f"{value*1000:.2f} Kbps" elif unit == 'G': return f"{value/1000:.2f} Gbps" elif unit == 'M': return f"{value:.2f} Mbps" elif unit == 'K': return f"{value*1000:.2f} Kbps" def format_time(self, seconds): """格式化时间显示""" if seconds < 60: return f"{seconds:.1f} 秒" elif seconds < 3600: minutes = int(seconds // 60) secs = seconds % 60 return f"{minutes} 分 {secs:.0f} 秒" else: hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) return f"{hours} 小时 {minutes} 分" def quality_color(self, score): """根据分数返回颜色""" if score >= 90: return "#28a745" # 绿色 elif score >= 80: return "#17a2b8" # 蓝色 elif score >= 60: return "#ffc107" # 黄色 elif score >= 40: return "#fd7e14" # 橙色 else: return "#dc3545" # 红色 def html_to_pdf(self, html_content): """HTML转PDF""" options = { 'page-size': 'A4', 'margin-top': '0.75in', 'margin-right': '0.75in', 'margin-bottom': '0.75in', 'margin-left': '0.75in', 'encoding': "UTF-8", 'no-outline': None, 'enable-local-file-access': None } pdf = pdfkit.from_string(html_content, False, options=options) return pdf def generate_comparison_report(self, results_list, baseline=None): """生成对比报告""" # 创建DataFrame df = pd.DataFrame(results_list) # 计算统计 stats = { '测试次数': len(df), '平均带宽': df['bandwidth'].mean(), '最大带宽': df['bandwidth'].max(), '最小带宽': df['bandwidth'].min(), '带宽标准差': df['bandwidth'].std(), '平均延迟': df['latency'].mean(), '平均丢包': df['loss'].mean(), } if baseline: stats['与基线差异'] = { '带宽': ((df['bandwidth'].mean() - baseline['bandwidth']) / baseline['bandwidth']) * 100, '延迟': ((df['latency'].mean() - baseline['latency']) / baseline['latency']) * 100, } # 生成报告数据 report_data = { 'statistics': stats, 'results': df.to_dict('records'), 'charts': { 'bandwidth_trend': self.create_bandwidth_chart(df), 'quality_distribution': self.create_quality_distribution(df), 'comparison_chart': self.create_comparison_chart(df, baseline) if baseline else None } } return self.generate_report(report_data, 'comparison_report.html') def create_bandwidth_chart(self, df): """创建带宽趋势图数据""" # 这里生成图表数据，可以是Plotly、Chart.js等格式 chart_data = { 'labels': df['timestamp'].tolist(), 'datasets': [{ 'label': '带宽 (Mbps)', 'data': df['bandwidth'].tolist(), 'borderColor': 'rgb(75, 192, 192)', 'backgroundColor': 'rgba(75, 192, 192, 0.2)', }] } return chart_data

# test_scenarios.py class TestScenarioLibrary: """测试场景库""" SCENARIOS = { # 家庭网络场景 "home_network_basic": { "name": "家庭网络基础测试", "description": "测试家庭网络基本性能", "tests": [ { "name": "局域网有线测试", "server": "192.168.1.1", "duration": 30, "parallel": 1, "expected_bandwidth": 900 # Mbps }, { "name": "Wi-Fi近距离测试", "server": "192.168.1.1", "duration": 30, "parallel": 4, "wireless": True, "expected_bandwidth": 300 # Mbps，根据Wi-Fi规格调整 }, { "name": "互联网下载测试", "server": "ping.online.net", "duration": 60, "parallel": 8, "expected_bandwidth": "80% of ISP speed" } ] }, # 企业网络场景 "enterprise_network_validation": { "name": "企业网络验收测试", "description": "企业网络部署后的全面验证", "tests": [ { "name": "核心交换机性能测试", "duration": 300, "parallel": 32, "bidirectional": True, "expected_bandwidth": "line rate" }, { "name": "跨部门通信测试", "duration": 180, "parallel": 16, "multiple_servers": True, "expected_loss": 0 }, { "name": "互联网出口测试", "duration": 600, "parallel": 8, "long_running": True, "monitor_stability": True } ] }, # 云服务场景 "cloud_migration_validation": { "name": "云迁移网络验证", "description": "验证本地到云的网络性能", "tests": [ { "name": "本地到云单向带宽", "direction": "upload", "duration": 300, "parallel": 16, "monitor": ["bandwidth", "loss", "latency"] }, { "name": "云到本地单向带宽", "direction": "download", "duration": 300, "parallel": 16 }, { "name": "双向并发测试", "duration": 600, "parallel": 8, "bidirectional": True, "stress_test": True }, { "name": "不同时段稳定性", "schedule": ["00:00", "06:00", "12:00", "18:00"], "duration": 1800, "monitor_variation": True } ] }, # 实时应用场景 "realtime_application_simulation": { "name": "实时应用模拟测试", "description": "模拟VoIP、视频会议等实时应用", "tests": [ { "name": "VoIP质量测试", "protocol": "udp", "bandwidth": "0.1M", # 100Kbps，典型VoIP "duration": 300, "monitor": ["jitter", "loss", "latency"], "expected_jitter": 20, # ms "expected_loss": 1, # % "expected_latency": 150 # ms }, { "name": "视频会议测试", "protocol": "udp", "bandwidth": "4M", # 720p视频 "duration": 600, "parallel": 4, # 模拟多参会者 "expected_loss": 0.5 }, { "name": "在线游戏测试", "protocol": "udp", "bandwidth": "0.5M", "duration": 1800, "packet_size": 512, # 典型游戏包大小 "expected_latency": 50 } ] }, # 存储网络场景 "storage_network_performance": { "name": "存储网络性能测试", "description": "测试NAS、SAN等存储网络性能", "tests": [ { "name": "顺序读写测试", "protocol": "tcp", "duration": 300, "window_size": "1M", "parallel": 1, "test_pattern": "sequential" }, { "name": "随机读写测试", "protocol": "tcp", "duration": 300, "parallel": 32, "test_pattern": "random", "io_size": "4K" # 模拟数据库IO }, { "name": "混合读写测试", "protocol": "tcp", "duration": 600, "parallel": 16, "bidirectional": True, "test_pattern": "mixed" } ] }, # 网络安全场景 "network_security_validation": { "name": "网络安全策略验证", "description": "验证防火墙、IPS等安全设备的影响", "tests": [ { "name": "基础策略测试", "protocol": "tcp", "port": 80, "duration": 60, "baseline": True }, { "name": "IPS深度检测影响", "protocol": "tcp", "port": 80, "duration": 60, "ips_enabled": True, "compare_with": "基础策略测试" }, { "name": "不同包大小测试", "protocol": "tcp", "duration": 60, "packet_sizes": [64, 512, 1500, 9000], "compare_performance": True }, { "name": "并发连接数测试", "protocol": "tcp", "duration": 300, "parallel": [1, 10, 100, 1000], "test_connection_limit": True } ] } } @classmethod def get_scenario(cls, scenario_id): """获取测试场景""" return cls.SCENARIOS.get(scenario_id) @classmethod def list_scenarios(cls): """列出所有测试场景""" return [ { "id": scenario_id, "name": config["name"], "description": config["description"], "test_count": len(config.get("tests", [])) } for scenario_id, config in cls.SCENARIOS.items() ] @classmethod def execute_scenario(cls, scenario_id, custom_params=None): """执行测试场景""" scenario = cls.get_scenario(scenario_id) if not scenario: raise ValueError(f"场景不存在: {scenario_id}") results = [] all_passed = True for test_config in scenario["tests"]: # 合并自定义参数 if custom_params: test_config.update(custom_params) print(f"执行测试: {test_config['name']}") try: # 执行单个测试 result = cls.execute_single_test(test_config) # 验证期望值 if "expected_bandwidth" in test_config: expected = test_config["expected_bandwidth"] actual = result.get("bandwidth", 0) if isinstance(expected, str) and "%" in expected: # 百分比期望，如"80% of ISP speed" pass # 需要额外逻辑 elif actual < expected * 0.9: # 允许10%误差 result["status"] = "failed" result["message"] = f"带宽未达预期: {actual} < {expected}" all_passed = False else: result["status"] = "passed" results.append(result) except Exception as e: results.append({ "name": test_config.get("name", "未知测试"), "status": "error", "error": str(e) }) all_passed = False return { "scenario_id": scenario_id, "scenario_name": scenario["name"], "all_passed": all_passed, "results": results, "summary": cls.generate_scenario_summary(results) } @classmethod def execute_single_test(cls, test_config): """执行单个测试""" # 这里调用iperf3执行测试 # 简化示例，实际需要调用iperf3 return { "name": test_config["name"], "bandwidth": 950, # 模拟结果 "latency": 12, "loss": 0, "duration": test_config.get("duration", 30) } @classmethod def generate_scenario_summary(cls, results): """生成场景测试摘要""" if not results: return {} total = len(results) passed = sum(1 for r in results if r.get("status") == "passed") failed = sum(1 for r in results if r.get("status") == "failed") errors = sum(1 for r in results if r.get("status") == "error") bandwidths = [r.get("bandwidth", 0) for r in results if r.get("bandwidth")] latencies = [r.get("latency", 0) for r in results if r.get("latency")] return { "total_tests": total, "passed": passed, "failed": failed, "errors": errors, "pass_rate": (passed / total * 100) if total > 0 else 0, "avg_bandwidth": sum(bandwidths) / len(bandwidths) if bandwidths else 0, "avg_latency": sum(latencies) / len(latencies) if latencies else 0, "min_bandwidth": min(bandwidths) if bandwidths else 0, "max_bandwidth": max(bandwidths) if bandwidths else 0 }

# scenario_executor.py import time from datetime import datetime from concurrent.futures import ThreadPoolExecutor, as_completed class ScenarioExecutor: """场景执行器""" def __init__(self, max_workers=3): self.max_workers = max_workers self.executor = ThreadPoolExecutor(max_workers=max_workers) self.running_tests = {} self.results = {} def execute_scenario_async(self, scenario_id, params=None, callback=None): """异步执行测试场景""" future = self.executor.submit( self._execute_scenario_sync, scenario_id, params ) test_id = f"test_{int(time.time())}_{scenario_id}" self.running_tests[test_id] = { "future": future, "scenario_id": scenario_id, "start_time": datetime.now(), "callback": callback } # 添加完成回调 future.add_done_callback( lambda f: self._on_scenario_complete(test_id, f) ) return test_id def _execute_scenario_sync(self, scenario_id, params): """同步执行测试场景""" from test_scenarios import TestScenarioLibrary return TestScenarioLibrary.execute_scenario(scenario_id, params) def _on_scenario_complete(self, test_id, future): """场景完成回调""" if test_id in self.running_tests: test_info = self.running_tests.pop(test_id) try: result = future.result() result["test_id"] = test_id result["end_time"] = datetime.now() self.results[test_id] = result # 调用用户回调 if test_info["callback"]: test_info["callback"](result) # 触发事件 self._emit_event("scenario_completed", result) except Exception as e: error_result = { "test_id": test_id, "status": "error", "error": str(e), "end_time": datetime.now() } self.results[test_id] = error_result self._emit_event("scenario_failed", error_result) def execute_scenario_sequence(self, scenario_ids, params_list=None, interval=30, stop_on_error=False): """按顺序执行多个场景""" if params_list is None: params_list = [{}] * len(scenario_ids) sequence_id = f"seq_{int(time.time())}" sequence_results = { "sequence_id": sequence_id, "start_time": datetime.now(), "scenarios": [], "status": "running" } all_passed = True for i, (scenario_id, params) in enumerate(zip(scenario_ids, params_list)): print(f"执行场景 {i+1}/{len(scenario_ids)}: {scenario_id}") try: result = self._execute_scenario_sync(scenario_id, params) result["order"] = i + 1 sequence_results["scenarios"].append(result) if not result.get("all_passed", True): all_passed = False if stop_on_error: print(f"场景 {scenario_id} 失败，停止序列执行") break # 场景间间隔 if i < len(scenario_ids) - 1: print(f"等待 {interval} 秒后执行下一个场景...") time.sleep(interval) except Exception as e: error_result = { "scenario_id": scenario_id, "order": i + 1, "status": "error", "error": str(e) } sequence_results["scenarios"].append(error_result) all_passed = False if stop_on_error: break sequence_results["end_time"] = datetime.now() sequence_results["status"] = "completed" if all_passed else "failed" sequence_results["all_passed"] = all_passed # 保存序列结果 self.results[sequence_id] = sequence_results self._emit_event("sequence_completed", sequence_results) return sequence_results def get_test_status(self, test_id): """获取测试状态""" if test_id in self.running_tests: future = self.running_tests[test_id]["future"] if future.running(): return {"status": "running"} elif future.done(): return {"status": "done"} else: return {"status": "pending"} elif test_id in self.results: result = self.results[test_id] return { "status": "completed", "result": result } else: return {"status": "not_found"} def get_all_results(self): """获取所有结果""" return self.results def clear_results(self, older_than_days=7): """清理旧结果""" cutoff_time = datetime.now().timestamp() - (older_than_days * 24 * 3600) to_delete = [] for test_id, result in self.results.items(): end_time = result.get("end_time") if end_time and end_time.timestamp() < cutoff_time: to_delete.append(test_id) for test_id in to_delete: del self.results[test_id] return len(to_delete) def _emit_event(self, event_type, data): """触发事件（可扩展为事件总线）""" # 这里可以实现事件监听器模式 # 例如：self.event_listeners[event_type].append(callback) pass def stop_all(self): """停止所有测试""" self.executor.shutdown(wait=False) self.running_tests.clear()