TCP可以正常使用
This commit is contained in:
694
test/ch395f_socket_test.py
Normal file
694
test/ch395f_socket_test.py
Normal file
@@ -0,0 +1,694 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
CH395F BSD Socket API 测试程序
|
||||
用于测试 STM32F4 + CH395F 的网络功能
|
||||
|
||||
使用方法:
|
||||
python ch395f_socket_test.py [模式] [选项]
|
||||
|
||||
模式:
|
||||
tcp_server - 作为 TCP 服务器（测试 CH395F TCP Client）
|
||||
tcp_client - 作为 TCP 客户端（测试 CH395F TCP Server）
|
||||
udp_server - 作为 UDP 服务器（测试 CH395F UDP Client）
|
||||
udp_client - 作为 UDP 客户端（测试 CH395F UDP Server）
|
||||
tcp_echo - TCP 回显服务器（测试双向通信）
|
||||
udp_echo - UDP 回显服务器（测试双向通信）
|
||||
stress - 压力测试（多连接并发测试）
|
||||
|
||||
选项:
|
||||
--ip 目标 IP 地址（默认: 192.168.1.100）
|
||||
--port 端口号（默认: 8080）
|
||||
--count 测试次数（默认: 10）
|
||||
--size 数据包大小（默认: 1024）
|
||||
|
||||
示例:
|
||||
# 测试 CH395F TCP Server
|
||||
python ch395f_socket_test.py tcp_client --ip 192.168.1.100 --port 8080
|
||||
|
||||
# 测试 CH395F TCP Client
|
||||
python ch395f_socket_test.py tcp_server --port 8080
|
||||
|
||||
# 测试 CH395F UDP Client
|
||||
python ch395f_socket_test.py udp_server --port 8888
|
||||
|
||||
# 测试 CH395F UDP Server
|
||||
python ch395f_socket_test.py udp_client --ip 192.168.1.100 --port 8888
|
||||
|
||||
# 压力测试（多连接）
|
||||
python ch395f_socket_test.py stress --ip 192.168.1.100 --port 8080 --count 7
|
||||
"""
|
||||
|
||||
import socket
|
||||
import sys
|
||||
import time
|
||||
import argparse
|
||||
import threading
|
||||
import struct
|
||||
import random
|
||||
import string
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class Colors:
|
||||
"""终端颜色"""
|
||||
HEADER = '\033[95m'
|
||||
BLUE = '\033[94m'
|
||||
CYAN = '\033[96m'
|
||||
GREEN = '\033[92m'
|
||||
YELLOW = '\033[93m'
|
||||
RED = '\033[91m'
|
||||
END = '\033[0m'
|
||||
BOLD = '\033[1m'
|
||||
|
||||
|
||||
def log_msg(msg, color=Colors.END):
|
||||
"""带时间戳的日志输出"""
|
||||
timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
||||
print(f"{Colors.CYAN}[{timestamp}]{Colors.END} {color}{msg}{Colors.END}")
|
||||
|
||||
|
||||
def log_success(msg):
|
||||
log_msg(f"✓ {msg}", Colors.GREEN)
|
||||
|
||||
|
||||
def log_error(msg):
|
||||
log_msg(f"✗ {msg}", Colors.RED)
|
||||
|
||||
|
||||
def log_info(msg):
|
||||
log_msg(f"ℹ {msg}", Colors.BLUE)
|
||||
|
||||
|
||||
def log_warn(msg):
|
||||
log_msg(f"⚠ {msg}", Colors.YELLOW)
|
||||
|
||||
|
||||
def generate_data(size):
|
||||
"""生成测试数据"""
|
||||
return ''.join(random.choices(string.ascii_letters + string.digits, k=size)).encode()
|
||||
|
||||
|
||||
def verify_data(data, expected_len):
|
||||
"""验证数据"""
|
||||
if len(data) != expected_len:
|
||||
return False, f"长度不匹配: 期望 {expected_len}, 实际 {len(data)}"
|
||||
return True, "OK"
|
||||
|
||||
|
||||
# ==================== TCP 测试函数 ====================
|
||||
|
||||
def tcp_server_test(args):
|
||||
"""TCP 服务器测试（测试 CH395F TCP Client）"""
|
||||
log_info("=" * 60)
|
||||
log_info("TCP 服务器测试 - 测试 CH395F TCP Client 功能")
|
||||
log_info("=" * 60)
|
||||
|
||||
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
server_socket.settimeout(30)
|
||||
|
||||
try:
|
||||
server_socket.bind(('0.0.0.0', args.port))
|
||||
server_socket.listen(1)
|
||||
log_success(f"服务器启动，监听端口 {args.port}")
|
||||
log_info("等待 CH395F TCP Client 连接...")
|
||||
|
||||
client_socket, client_addr = server_socket.accept()
|
||||
log_success(f"连接来自: {client_addr[0]}:{client_addr[1]}")
|
||||
client_socket.settimeout(5)
|
||||
|
||||
# 测试接收数据
|
||||
log_info("-" * 40)
|
||||
log_info("测试 1: 接收数据")
|
||||
for i in range(args.count):
|
||||
try:
|
||||
data = client_socket.recv(args.size)
|
||||
if data:
|
||||
log_success(f"收到数据 ({len(data)} bytes): {data[:50]}...")
|
||||
else:
|
||||
log_error("收到空数据")
|
||||
break
|
||||
except socket.timeout:
|
||||
log_error("接收超时")
|
||||
break
|
||||
except Exception as e:
|
||||
log_error(f"接收错误: {e}")
|
||||
break
|
||||
|
||||
# 测试发送数据
|
||||
log_info("-" * 40)
|
||||
log_info("测试 2: 发送数据")
|
||||
for i in range(args.count):
|
||||
try:
|
||||
test_data = generate_data(min(100, args.size))
|
||||
client_socket.send(test_data)
|
||||
log_success(f"发送数据 ({len(test_data)} bytes)")
|
||||
time.sleep(0.1)
|
||||
except Exception as e:
|
||||
log_error(f"发送错误: {e}")
|
||||
break
|
||||
|
||||
# 测试双向通信
|
||||
log_info("-" * 40)
|
||||
log_info("测试 3: 双向通信")
|
||||
for i in range(min(5, args.count)):
|
||||
try:
|
||||
# 发送
|
||||
test_data = f"PING_{i}".encode()
|
||||
client_socket.send(test_data)
|
||||
log_info(f"发送: {test_data.decode()}")
|
||||
|
||||
# 接收回显
|
||||
data = client_socket.recv(args.size)
|
||||
if data:
|
||||
log_info(f"收到: {data.decode()}")
|
||||
if data == test_data:
|
||||
log_success("回显数据匹配!")
|
||||
else:
|
||||
log_error("回显数据不匹配!")
|
||||
except Exception as e:
|
||||
log_error(f"通信错误: {e}")
|
||||
break
|
||||
|
||||
client_socket.close()
|
||||
log_success("测试完成，连接已关闭")
|
||||
|
||||
except socket.timeout:
|
||||
log_error("等待连接超时")
|
||||
except Exception as e:
|
||||
log_error(f"服务器错误: {e}")
|
||||
finally:
|
||||
server_socket.close()
|
||||
|
||||
|
||||
def tcp_client_test(args):
|
||||
"""TCP 客户端测试（测试 CH395F TCP Server）"""
|
||||
log_info("=" * 60)
|
||||
log_info("TCP 客户端测试 - 测试 CH395F TCP Server 功能")
|
||||
log_info("=" * 60)
|
||||
|
||||
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
client_socket.settimeout(10)
|
||||
|
||||
try:
|
||||
# 连接服务器
|
||||
log_info(f"正在连接 CH395F TCP Server: {args.ip}:{args.port}")
|
||||
client_socket.connect((args.ip, args.port))
|
||||
log_success("连接成功!")
|
||||
|
||||
# 测试发送数据
|
||||
log_info("-" * 40)
|
||||
log_info("测试 1: 发送数据")
|
||||
for i in range(args.count):
|
||||
try:
|
||||
test_data = generate_data(min(100, args.size))
|
||||
client_socket.send(test_data)
|
||||
log_success(f"发送数据 ({len(test_data)} bytes)")
|
||||
log_info(f"发送 hex: {' '.join(f'{b:02X}' for b in test_data[:16])}")
|
||||
|
||||
# 等待回显
|
||||
data = client_socket.recv(args.size)
|
||||
if data:
|
||||
log_info(f"接收 hex: {' '.join(f'{b:02X}' for b in data[:16])}")
|
||||
log_info(f"接收长度: {len(data)}, 期望长度: {len(test_data)}")
|
||||
if data == test_data:
|
||||
log_success("回显数据匹配!")
|
||||
else:
|
||||
log_error("回显数据不匹配!")
|
||||
# 找出第一个不同的字节
|
||||
for j in range(min(len(data), len(test_data))):
|
||||
if data[j] != test_data[j]:
|
||||
log_error(f"首个差异: 位置 {j}, 发送={test_data[j]:02X}, 接收={data[j]:02X}")
|
||||
break
|
||||
else:
|
||||
log_error("收到空数据")
|
||||
break
|
||||
|
||||
time.sleep(0.1)
|
||||
except Exception as e:
|
||||
log_error(f"发送/接收错误: {e}")
|
||||
break
|
||||
|
||||
# 测试大数据量
|
||||
log_info("-" * 40)
|
||||
log_info("测试 2: 大数据量传输")
|
||||
try:
|
||||
large_data = generate_data(min(1024, args.size))
|
||||
client_socket.send(large_data)
|
||||
log_success(f"发送大数据 ({len(large_data)} bytes)")
|
||||
|
||||
received = b""
|
||||
while len(received) < len(large_data):
|
||||
chunk = client_socket.recv(min(1024, len(large_data) - len(received)))
|
||||
if not chunk:
|
||||
break
|
||||
received += chunk
|
||||
|
||||
if received == large_data:
|
||||
log_success("大数据传输完整!")
|
||||
else:
|
||||
log_error(f"大数据传输不完整: 发送 {len(large_data)}, 接收 {len(received)}")
|
||||
except Exception as e:
|
||||
log_error(f"大数据传输错误: {e}")
|
||||
|
||||
# 测试快速收发
|
||||
log_info("-" * 40)
|
||||
log_info("测试 3: 快速收发")
|
||||
success_count = 0
|
||||
fail_count = 0
|
||||
start_time = time.time()
|
||||
|
||||
for i in range(args.count):
|
||||
try:
|
||||
test_data = f"FAST_{i:04d}".encode()
|
||||
client_socket.send(test_data)
|
||||
data = client_socket.recv(100)
|
||||
if data and data == test_data:
|
||||
success_count += 1
|
||||
else:
|
||||
fail_count += 1
|
||||
except Exception as e:
|
||||
fail_count += 1
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
log_info(f"快速收发结果: 成功 {success_count}, 失败 {fail_count}, 耗时 {elapsed:.2f}s")
|
||||
|
||||
client_socket.close()
|
||||
log_success("测试完成，连接已关闭")
|
||||
|
||||
except ConnectionRefusedError:
|
||||
log_error("连接被拒绝 - CH395F TCP Server 未运行")
|
||||
except socket.timeout:
|
||||
log_error("连接超时")
|
||||
except Exception as e:
|
||||
log_error(f"客户端错误: {e}")
|
||||
finally:
|
||||
client_socket.close()
|
||||
|
||||
|
||||
# ==================== UDP 测试函数 ====================
|
||||
|
||||
def udp_server_test(args):
|
||||
"""UDP 服务器测试（测试 CH395F UDP Client）"""
|
||||
log_info("=" * 60)
|
||||
log_info("UDP 服务器测试 - 测试 CH395F UDP Client 功能")
|
||||
log_info("=" * 60)
|
||||
|
||||
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
server_socket.settimeout(30)
|
||||
|
||||
try:
|
||||
server_socket.bind(('0.0.0.0', args.port))
|
||||
log_success(f"服务器启动，监听端口 {args.port}")
|
||||
log_info("等待 CH395F UDP Client 数据...")
|
||||
|
||||
# 接收数据
|
||||
for i in range(args.count):
|
||||
try:
|
||||
data, client_addr = server_socket.recvfrom(args.size)
|
||||
log_success(f"收到来自 {client_addr[0]}:{client_addr[1]} ({len(data)} bytes)")
|
||||
log_info(f"数据: {data[:50]}...")
|
||||
|
||||
# 回显数据
|
||||
server_socket.sendto(data, client_addr)
|
||||
log_info(f"回显数据到 {client_addr[0]}:{client_addr[1]}")
|
||||
except socket.timeout:
|
||||
log_warn("接收超时")
|
||||
break
|
||||
except Exception as e:
|
||||
log_error(f"接收错误: {e}")
|
||||
break
|
||||
|
||||
log_success("测试完成")
|
||||
|
||||
except Exception as e:
|
||||
log_error(f"服务器错误: {e}")
|
||||
finally:
|
||||
server_socket.close()
|
||||
|
||||
|
||||
def udp_client_test(args):
|
||||
"""UDP 客户端测试（测试 CH395F UDP Server）"""
|
||||
log_info("=" * 60)
|
||||
log_info("UDP 客户端测试 - 测试 CH395F UDP Server 功能")
|
||||
log_info("=" * 60)
|
||||
|
||||
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
client_socket.settimeout(5)
|
||||
|
||||
try:
|
||||
dest_addr = (args.ip, args.port)
|
||||
|
||||
# 发送数据
|
||||
log_info("-" * 40)
|
||||
log_info("测试 1: 发送 UDP 数据")
|
||||
for i in range(args.count):
|
||||
try:
|
||||
test_data = f"UDP_MSG_{i:04d}".encode()
|
||||
client_socket.sendto(test_data, dest_addr)
|
||||
log_success(f"发送数据到 {args.ip}:{args.port} ({len(test_data)} bytes)")
|
||||
|
||||
# 等待回显
|
||||
data, addr = client_socket.recvfrom(args.size)
|
||||
if data:
|
||||
if data == test_data:
|
||||
log_success(f"回显数据匹配! (from {addr[0]}:{addr[1]})")
|
||||
else:
|
||||
log_error("回显数据不匹配!")
|
||||
else:
|
||||
log_error("收到空数据")
|
||||
break
|
||||
|
||||
time.sleep(0.2)
|
||||
except socket.timeout:
|
||||
log_warn("接收超时")
|
||||
except Exception as e:
|
||||
log_error(f"发送/接收错误: {e}")
|
||||
break
|
||||
|
||||
# 测试大数据
|
||||
log_info("-" * 40)
|
||||
log_info("测试 2: 大数据 UDP 传输")
|
||||
try:
|
||||
large_data = generate_data(min(512, args.size))
|
||||
client_socket.sendto(large_data, dest_addr)
|
||||
log_success(f"发送大数据 ({len(large_data)} bytes)")
|
||||
|
||||
data, addr = client_socket.recvfrom(args.size)
|
||||
if data and data == large_data:
|
||||
log_success("大数据传输完整!")
|
||||
else:
|
||||
log_error("大数据传输失败")
|
||||
except Exception as e:
|
||||
log_error(f"大数据传输错误: {e}")
|
||||
|
||||
log_success("测试完成")
|
||||
|
||||
except Exception as e:
|
||||
log_error(f"客户端错误: {e}")
|
||||
finally:
|
||||
client_socket.close()
|
||||
|
||||
|
||||
# ==================== 回显测试 ====================
|
||||
|
||||
def tcp_echo_server_test(args):
|
||||
"""TCP 回显服务器测试"""
|
||||
log_info("=" * 60)
|
||||
log_info("TCP 回显服务器测试 - 测试双向通信功能")
|
||||
log_info("=" * 60)
|
||||
|
||||
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
server_socket.settimeout(60)
|
||||
|
||||
try:
|
||||
server_socket.bind(('0.0.0.0', args.port))
|
||||
server_socket.listen(1)
|
||||
log_success(f"回显服务器启动，端口 {args.port}")
|
||||
log_info("等待连接...")
|
||||
|
||||
client_socket, client_addr = server_socket.accept()
|
||||
log_success(f"连接来自: {client_addr[0]}:{client_addr[1]}")
|
||||
client_socket.settimeout(5)
|
||||
|
||||
# 循环回显
|
||||
echo_count = 0
|
||||
start_time = time.time()
|
||||
|
||||
log_info("开始回显测试 (按 Ctrl+C 停止)...")
|
||||
while echo_count < args.count:
|
||||
try:
|
||||
data = client_socket.recv(args.size)
|
||||
if not data:
|
||||
log_warn("客户端断开连接")
|
||||
break
|
||||
|
||||
echo_count += 1
|
||||
client_socket.send(data)
|
||||
|
||||
if echo_count % 10 == 0:
|
||||
elapsed = time.time() - start_time
|
||||
rate = echo_count / elapsed if elapsed > 0 else 0
|
||||
log_info(f"回显 {echo_count} 次, 速率: {rate:.1f} msg/s")
|
||||
|
||||
except socket.timeout:
|
||||
continue
|
||||
except Exception as e:
|
||||
log_error(f"回显错误: {e}")
|
||||
break
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
log_success(f"回显测试完成: {echo_count} 次, 耗时 {elapsed:.2f}s")
|
||||
|
||||
client_socket.close()
|
||||
|
||||
except Exception as e:
|
||||
log_error(f"服务器错误: {e}")
|
||||
finally:
|
||||
server_socket.close()
|
||||
|
||||
|
||||
def udp_echo_server_test(args):
|
||||
"""UDP 回显服务器测试"""
|
||||
log_info("=" * 60)
|
||||
log_info("UDP 回显服务器测试 - 测试双向通信功能")
|
||||
log_info("=" * 60)
|
||||
|
||||
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
server_socket.settimeout(60)
|
||||
|
||||
try:
|
||||
server_socket.bind(('0.0.0.0', args.port))
|
||||
log_success(f"UDP 回显服务器启动，端口 {args.port}")
|
||||
log_info("等待数据...")
|
||||
|
||||
echo_count = 0
|
||||
start_time = time.time()
|
||||
clients = set()
|
||||
|
||||
while echo_count < args.count:
|
||||
try:
|
||||
data, client_addr = server_socket.recvfrom(args.size)
|
||||
if not data:
|
||||
continue
|
||||
|
||||
clients.add(client_addr)
|
||||
echo_count += 1
|
||||
|
||||
# 回显数据
|
||||
server_socket.sendto(data, client_addr)
|
||||
|
||||
if echo_count % 10 == 0:
|
||||
elapsed = time.time() - start_time
|
||||
rate = echo_count / elapsed if elapsed > 0 else 0
|
||||
log_info(f"回显 {echo_count} 次, 客户端数: {len(clients)}, 速率: {rate:.1f} msg/s")
|
||||
|
||||
except socket.timeout:
|
||||
continue
|
||||
except Exception as e:
|
||||
log_error(f"回显错误: {e}")
|
||||
break
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
log_success(f"回显测试完成: {echo_count} 次, 耗时 {elapsed:.2f}s")
|
||||
|
||||
except Exception as e:
|
||||
log_error(f"服务器错误: {e}")
|
||||
finally:
|
||||
server_socket.close()
|
||||
|
||||
|
||||
# ==================== 压力测试 ====================
|
||||
|
||||
def stress_test_thread(client_id, args, results):
|
||||
"""压力测试线程"""
|
||||
thread_name = f"Client-{client_id}"
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.settimeout(10)
|
||||
|
||||
try:
|
||||
sock.connect((args.ip, args.port))
|
||||
log_success(f"[{thread_name}] 连接成功")
|
||||
|
||||
success_count = 0
|
||||
fail_count = 0
|
||||
|
||||
for i in range(args.count):
|
||||
try:
|
||||
# 发送
|
||||
test_data = f"T{client_id:02d}_{i:04d}".encode()
|
||||
sock.send(test_data)
|
||||
|
||||
# 接收
|
||||
data = sock.recv(100)
|
||||
if data and data == test_data:
|
||||
success_count += 1
|
||||
else:
|
||||
fail_count += 1
|
||||
|
||||
except Exception as e:
|
||||
fail_count += 1
|
||||
|
||||
results[client_id] = (success_count, fail_count)
|
||||
log_info(f"[{thread_name}] 完成: 成功 {success_count}, 失败 {fail_count}")
|
||||
|
||||
except Exception as e:
|
||||
log_error(f"[{thread_name}] 错误: {e}")
|
||||
results[client_id] = (0, args.count)
|
||||
finally:
|
||||
sock.close()
|
||||
|
||||
|
||||
def stress_test(args):
|
||||
"""压力测试（多连接并发）"""
|
||||
log_info("=" * 60)
|
||||
log_info("压力测试 - 多连接并发")
|
||||
log_info("=" * 60)
|
||||
|
||||
num_clients = min(args.count, 7) # CH395F 最多支持 7 个并发连接
|
||||
log_info(f"创建 {num_clients} 个并发连接...")
|
||||
|
||||
results = {}
|
||||
threads = []
|
||||
|
||||
# 创建并启动线程
|
||||
for i in range(num_clients):
|
||||
thread = threading.Thread(
|
||||
target=stress_test_thread,
|
||||
args=(i, args, results)
|
||||
)
|
||||
threads.append(thread)
|
||||
thread.start()
|
||||
time.sleep(0.1) # 错开连接时间
|
||||
|
||||
# 等待所有线程完成
|
||||
for thread in threads:
|
||||
thread.join()
|
||||
|
||||
# 汇总结果
|
||||
total_success = sum(r[0] for r in results.values())
|
||||
total_fail = sum(r[1] for r in results.values())
|
||||
|
||||
log_info("-" * 40)
|
||||
log_info("压力测试结果:")
|
||||
log_info(f" 并发连接数: {num_clients}")
|
||||
log_info(f" 每连接测试次数: {args.count}")
|
||||
log_info(f" 总成功: {total_success}")
|
||||
log_info(f" 总失败: {total_fail}")
|
||||
log_info(f" 成功率: {total_success/(total_success+total_fail)*100:.1f}%")
|
||||
|
||||
if total_fail == 0:
|
||||
log_success("压力测试通过!")
|
||||
else:
|
||||
log_error("压力测试有失败!")
|
||||
|
||||
|
||||
# ==================== 数据包大小测试 ====================
|
||||
|
||||
def packet_size_test(args):
|
||||
"""不同数据包大小测试"""
|
||||
log_info("=" * 60)
|
||||
log_info("数据包大小测试")
|
||||
log_info("=" * 60)
|
||||
|
||||
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
client_socket.settimeout(5)
|
||||
|
||||
try:
|
||||
client_socket.connect((args.ip, args.port))
|
||||
log_success("连接成功")
|
||||
|
||||
# 测试不同的数据包大小
|
||||
test_sizes = [1, 10, 100, 512, 1024, 2048, 4096]
|
||||
|
||||
for size in test_sizes:
|
||||
try:
|
||||
test_data = generate_data(size)
|
||||
start_time = time.time()
|
||||
client_socket.send(test_data)
|
||||
|
||||
received = b""
|
||||
while len(received) < size:
|
||||
chunk = client_socket.recv(size - len(received))
|
||||
if not chunk:
|
||||
break
|
||||
received += chunk
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
|
||||
if received == test_data:
|
||||
rate = size / elapsed / 1024 if elapsed > 0 else 0
|
||||
log_success(f"大小 {size:5d} bytes: 成功, 速率 {rate:.1f} KB/s")
|
||||
else:
|
||||
log_error(f"大小 {size:5d} bytes: 数据不匹配")
|
||||
except Exception as e:
|
||||
log_error(f"大小 {size:5d} bytes: 错误 - {e}")
|
||||
|
||||
time.sleep(0.1)
|
||||
|
||||
except Exception as e:
|
||||
log_error(f"测试错误: {e}")
|
||||
finally:
|
||||
client_socket.close()
|
||||
|
||||
|
||||
# ==================== 主程序 ====================
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description='CH395F BSD Socket API 测试程序',
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
示例:
|
||||
%(prog)s tcp_client --ip 192.168.1.100 --port 8080
|
||||
%(prog)s tcp_server --port 8080
|
||||
%(prog)s stress --ip 192.168.1.100 --port 8080 --count 5
|
||||
"""
|
||||
)
|
||||
|
||||
parser.add_argument('mode', choices=['tcp_server', 'tcp_client', 'udp_server', 'udp_client',
|
||||
'tcp_echo', 'udp_echo', 'stress', 'packet_size'],
|
||||
help='测试模式')
|
||||
parser.add_argument('--ip', default='192.168.1.100', help='目标 IP 地址')
|
||||
parser.add_argument('--port', type=int, default=8080, help='端口号')
|
||||
parser.add_argument('--count', type=int, default=10, help='测试次数')
|
||||
parser.add_argument('--size', type=int, default=1024, help='数据包大小')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
print()
|
||||
print(Colors.BOLD + "=" * 60 + Colors.END)
|
||||
print(Colors.BOLD + " CH395F BSD Socket API 测试程序" + Colors.END)
|
||||
print(Colors.BOLD + "=" * 60 + Colors.END)
|
||||
print()
|
||||
|
||||
try:
|
||||
if args.mode == 'tcp_server':
|
||||
tcp_server_test(args)
|
||||
elif args.mode == 'tcp_client':
|
||||
tcp_client_test(args)
|
||||
elif args.mode == 'udp_server':
|
||||
udp_server_test(args)
|
||||
elif args.mode == 'udp_client':
|
||||
udp_client_test(args)
|
||||
elif args.mode == 'tcp_echo':
|
||||
tcp_echo_server_test(args)
|
||||
elif args.mode == 'udp_echo':
|
||||
udp_echo_server_test(args)
|
||||
elif args.mode == 'stress':
|
||||
stress_test(args)
|
||||
elif args.mode == 'packet_size':
|
||||
packet_size_test(args)
|
||||
except KeyboardInterrupt:
|
||||
log_warn("用户中断")
|
||||
except Exception as e:
|
||||
log_error(f"未预期的错误: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user