410 lines
19 KiB
Python
410 lines
19 KiB
Python
|
"""
|
|||
|
|
虚拟仿真平台中间层软件 - UDP 一键自测脚本
|
|||
|
|
该脚本整合连接性检测、单 / 多端口传输测试、1553B 时延检测、10 机并发检测、CAN/1553B 消息顺序检测所有自测项，一键运行自动执行并生成可视化自测报告，无需手动分步操作，报告保存至test/report/目录，同时输出控制台日志，适配工程现有架构。
|
|||
|
|
"""
|
|||
|
|
import socket
|
|||
|
|
import binascii
|
|||
|
|
import time
|
|||
|
|
import threading
|
|||
|
|
import psutil
|
|||
|
|
import os
|
|||
|
|
from datetime import datetime
|
|||
|
|
from prettytable import PrettyTable
|
|||
|
|
from loguru import logger
|
|||
|
|
|
|||
|
|
# 工程配置导入（需确保工程根目录在Python环境变量）
|
|||
|
|
try:
|
|||
|
|
from midware.config.base_config import DEVICE_CONFIG_DICT, UDP_CONFIG
|
|||
|
|
from midware.network.udp_handler import udp_server
|
|||
|
|
except ImportError as e:
|
|||
|
|
print(f"❌ 导入工程模块失败，请将工程根目录加入Python环境变量：{e}")
|
|||
|
|
exit(1)
|
|||
|
|
|
|||
|
|
# 全局自测配置
|
|||
|
|
LOCAL_IP = UDP_CONFIG["LOCAL_IP"]
|
|||
|
|
FAULT_INJECT_IP = UDP_CONFIG["FAULT_INJECT_IP"]
|
|||
|
|
FAULT_INJECT_PORT = UDP_CONFIG["FAULT_INJECT_PORT"]
|
|||
|
|
BUFFER_SIZE = UDP_CONFIG["BUFFER_SIZE"]
|
|||
|
|
TEST_TIMES_1553B = 1000 # 1553B时延测试发送次数
|
|||
|
|
TEST_TIMES_CONCURRENT = 500 # 10机并发每机发送次数
|
|||
|
|
TEST_DATA = b"UDP_AUTO_TEST" + b'\x9c\x47' # 带CRC16校验位的测试数据
|
|||
|
|
# 协议端口筛选
|
|||
|
|
B1553B_PORTS = [dev_info["udp_port"] for dev_name, dev_info in DEVICE_CONFIG_DICT.items() if dev_info["protocol"] == "1553B"]
|
|||
|
|
CAN_PORTS = [dev_info["udp_port"] for dev_name, dev_info in DEVICE_CONFIG_DICT.items() if dev_info["protocol"] == "CAN"]
|
|||
|
|
UART_PORTS = [dev_info["udp_port"] for dev_name, dev_info in DEVICE_CONFIG_DICT.items() if dev_info["protocol"] == "UART"]
|
|||
|
|
AD_PORTS = [dev_info["udp_port"] for dev_name, dev_info in DEVICE_CONFIG_DICT.items() if dev_info["protocol"] == "AD"]
|
|||
|
|
OC_PORTS = [dev_info["udp_port"] for dev_name, dev_info in DEVICE_CONFIG_DICT.items() if dev_info["protocol"] == "OC"]
|
|||
|
|
TEST_PORTS = [dev_info["udp_port"] for dev_name, dev_info in DEVICE_CONFIG_DICT.items()][:10] # 前10个外设端口
|
|||
|
|
|
|||
|
|
# 自测结果存储
|
|||
|
|
test_result = {
|
|||
|
|
"connect": {"pass": False, "msg": "", "detail": []},
|
|||
|
|
"single_port": {"pass": False, "msg": ""},
|
|||
|
|
"multi_port": {"pass": False, "msg": ""},
|
|||
|
|
"1553b_latency": {"pass": False, "msg": "", "avg": 0, "min": 0, "max": 0},
|
|||
|
|
"10_concurrent": {"pass": False, "msg": "", "send": 0, "recv": 0, "loss_rate": 0},
|
|||
|
|
"can_1553b_order": {"pass": False, "msg": ""}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 初始化日志和报告目录
|
|||
|
|
os.makedirs("test/logs", exist_ok=True)
|
|||
|
|
os.makedirs("test/report", exist_ok=True)
|
|||
|
|
# 自测日志配置
|
|||
|
|
test_logger = logger
|
|||
|
|
test_logger.add("test/logs/udp_auto_test_{time:YYYYMMDDHHmmss}.log", level="INFO", format="{time} | {level} | {message}")
|
|||
|
|
# 报告生成时间
|
|||
|
|
report_time = datetime.now().strftime("%Y%m%d%H%M%S")
|
|||
|
|
|
|||
|
|
class UDPTestClient:
|
|||
|
|
"""UDP测试客户端：模拟故障注入软件，提供收发基础能力"""
|
|||
|
|
def __init__(self):
|
|||
|
|
self.send_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|||
|
|
self.send_sock.setblocking(False)
|
|||
|
|
self.recv_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|||
|
|
self.recv_sock.bind((LOCAL_IP, FAULT_INJECT_PORT))
|
|||
|
|
self.recv_sock.setblocking(False)
|
|||
|
|
self.recv_count = 0
|
|||
|
|
self.recv_latency = []
|
|||
|
|
self.recv_lock = threading.Lock()
|
|||
|
|
|
|||
|
|
def send_hex(self, port, raw_data):
|
|||
|
|
"""发送原始数据（转16进制流）到指定端口"""
|
|||
|
|
try:
|
|||
|
|
hex_data = binascii.hexlify(raw_data).decode("utf-8")
|
|||
|
|
self.send_sock.sendto(hex_data.encode("utf-8"), (LOCAL_IP, port))
|
|||
|
|
return True
|
|||
|
|
except Exception as e:
|
|||
|
|
test_logger.error(f"端口{port}发送失败：{e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def recv_loop(self):
|
|||
|
|
"""后台接收线程，统计接收数和时延"""
|
|||
|
|
while True:
|
|||
|
|
try:
|
|||
|
|
data, addr = self.recv_sock.recvfrom(BUFFER_SIZE)
|
|||
|
|
with self.recv_lock:
|
|||
|
|
self.recv_count += 1
|
|||
|
|
# 解析回传数据，仅统计有效数据
|
|||
|
|
if data:
|
|||
|
|
test_logger.debug(f"接收回传数据：{addr} | {data.hex()[:32]}...")
|
|||
|
|
except BlockingIOError:
|
|||
|
|
time.sleep(0.000001)
|
|||
|
|
continue
|
|||
|
|
except Exception as e:
|
|||
|
|
test_logger.error(f"接收异常：{e}")
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
def recv_latency_loop(self):
|
|||
|
|
"""时延测试专用接收线程，记录接收时间"""
|
|||
|
|
while True:
|
|||
|
|
try:
|
|||
|
|
data, addr = self.recv_sock.recvfrom(BUFFER_SIZE)
|
|||
|
|
with self.recv_lock:
|
|||
|
|
self.recv_latency.append(time.time())
|
|||
|
|
self.recv_count += 1
|
|||
|
|
except BlockingIOError:
|
|||
|
|
time.sleep(0.000001)
|
|||
|
|
continue
|
|||
|
|
except Exception as e:
|
|||
|
|
test_logger.error(f"时延测试接收异常：{e}")
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
def reset_recv(self):
|
|||
|
|
"""重置接收统计"""
|
|||
|
|
with self.recv_lock:
|
|||
|
|
self.recv_count = 0
|
|||
|
|
self.recv_latency = []
|
|||
|
|
|
|||
|
|
def close(self):
|
|||
|
|
"""关闭套接字"""
|
|||
|
|
self.send_sock.close()
|
|||
|
|
self.recv_sock.close()
|
|||
|
|
|
|||
|
|
# ---------------------- 自测用例执行 ----------------------
|
|||
|
|
def check_port_listen(ip, port):
|
|||
|
|
"""检测端口是否监听"""
|
|||
|
|
for conn in psutil.net_connections(kind='inet'):
|
|||
|
|
if conn.laddr.ip == ip and conn.laddr.port == port and conn.status == 'LISTEN':
|
|||
|
|
return True
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def test_connectivity():
|
|||
|
|
"""用例1：UDP端口连接性检测"""
|
|||
|
|
test_logger.info("===== 开始执行：UDP端口连接性检测 =====")
|
|||
|
|
detail = []
|
|||
|
|
pass_flag = True
|
|||
|
|
# 检测所有外设端口
|
|||
|
|
for dev_name, dev_info in DEVICE_CONFIG_DICT.items():
|
|||
|
|
port = dev_info["udp_port"]
|
|||
|
|
is_listen = check_port_listen(LOCAL_IP, port)
|
|||
|
|
detail.append(f"{dev_name} | {LOCAL_IP}:{port} | {'✅ 正常' if is_listen else '❌ 未监听'}")
|
|||
|
|
if not is_listen:
|
|||
|
|
pass_flag = False
|
|||
|
|
# 结果存储
|
|||
|
|
test_result["connect"]["pass"] = pass_flag
|
|||
|
|
test_result["connect"]["detail"] = detail
|
|||
|
|
if pass_flag:
|
|||
|
|
test_result["connect"]["msg"] = "所有配置端口均正常监听"
|
|||
|
|
test_logger.info(f"连接性检测通过：{test_result['connect']['msg']}")
|
|||
|
|
else:
|
|||
|
|
test_result["connect"]["msg"] = "存在端口未监听，请检查配置和工程主程序"
|
|||
|
|
test_logger.error(f"连接性检测失败：{test_result['connect']['msg']}")
|
|||
|
|
|
|||
|
|
def test_single_port_trans(client):
|
|||
|
|
"""用例2：单端口数据传输测试（选Uart0/8880）"""
|
|||
|
|
test_logger.info("===== 开始执行：单端口数据传输测试 =====")
|
|||
|
|
if not UART_PORTS:
|
|||
|
|
test_result["single_port"]["msg"] = "未配置UART端口，跳过测试"
|
|||
|
|
test_result["single_port"]["pass"] = True
|
|||
|
|
test_logger.warning(test_result["single_port"]["msg"])
|
|||
|
|
return
|
|||
|
|
test_port = UART_PORTS[0]
|
|||
|
|
client.reset_recv()
|
|||
|
|
# 发送10次测试数据
|
|||
|
|
send_ok = 0
|
|||
|
|
for _ in range(10):
|
|||
|
|
if client.send_hex(test_port, TEST_DATA):
|
|||
|
|
send_ok += 1
|
|||
|
|
time.sleep(0.1)
|
|||
|
|
time.sleep(1)
|
|||
|
|
# 验证结果
|
|||
|
|
if send_ok == 10 and client.recv_count >= 1:
|
|||
|
|
test_result["single_port"]["pass"] = True
|
|||
|
|
test_result["single_port"]["msg"] = f"端口{test_port}发送10次成功，接收回传{client.recv_count}次，传输正常"
|
|||
|
|
test_logger.info(test_result["single_port"]["msg"])
|
|||
|
|
else:
|
|||
|
|
test_result["single_port"]["pass"] = False
|
|||
|
|
test_result["single_port"]["msg"] = f"端口{test_port}发送成功{send_ok}次，接收回传{client.recv_count}次，传输异常"
|
|||
|
|
test_logger.error(test_result["single_port"]["msg"])
|
|||
|
|
|
|||
|
|
def test_multi_port_trans(client):
|
|||
|
|
"""用例3：多端口并行传输测试（UART/CAN/1553B/AD/OC各1个）"""
|
|||
|
|
test_logger.info("===== 开始执行：多端口并行传输测试 =====")
|
|||
|
|
test_ports = []
|
|||
|
|
if UART_PORTS: test_ports.append(UART_PORTS[0])
|
|||
|
|
if CAN_PORTS: test_ports.append(CAN_PORTS[0])
|
|||
|
|
if B1553B_PORTS: test_ports.append(B1553B_PORTS[0])
|
|||
|
|
if AD_PORTS: test_ports.append(AD_PORTS[0])
|
|||
|
|
if OC_PORTS: test_ports.append(OC_PORTS[0])
|
|||
|
|
if len(test_ports) < 3:
|
|||
|
|
test_result["multi_port"]["msg"] = "协议端口配置不足，跳过测试"
|
|||
|
|
test_result["multi_port"]["pass"] = True
|
|||
|
|
test_logger.warning(test_result["multi_port"]["msg"])
|
|||
|
|
return
|
|||
|
|
# 多线程并行发送
|
|||
|
|
client.reset_recv()
|
|||
|
|
def send_task(port):
|
|||
|
|
for _ in range(5):
|
|||
|
|
client.send_hex(port, TEST_DATA + str(port).encode())
|
|||
|
|
time.sleep(0.05)
|
|||
|
|
threads = [threading.Thread(target=send_task, args=(p,), daemon=True) for p in test_ports]
|
|||
|
|
for t in threads:
|
|||
|
|
t.start()
|
|||
|
|
for t in threads:
|
|||
|
|
t.join()
|
|||
|
|
time.sleep(2)
|
|||
|
|
# 验证结果
|
|||
|
|
if client.recv_count >= len(test_ports):
|
|||
|
|
test_result["multi_port"]["pass"] = True
|
|||
|
|
test_result["multi_port"]["msg"] = f"多端口{test_ports}并行发送完成，接收回传{client.recv_count}次，无串包"
|
|||
|
|
test_logger.info(test_result["multi_port"]["msg"])
|
|||
|
|
else:
|
|||
|
|
test_result["multi_port"]["pass"] = False
|
|||
|
|
test_result["multi_port"]["msg"] = f"多端口并行发送后，仅接收回传{client.recv_count}次，存在串包/丢包"
|
|||
|
|
test_logger.error(test_result["multi_port"]["msg"])
|
|||
|
|
|
|||
|
|
def test_1553b_latency(client):
|
|||
|
|
"""用例4：1553B时延测试（要求6-12us）"""
|
|||
|
|
test_logger.info("===== 开始执行：1553B时延测试 =====")
|
|||
|
|
if not B1553B_PORTS:
|
|||
|
|
test_result["1553b_latency"]["msg"] = "未配置1553B端口，跳过测试"
|
|||
|
|
test_result["1553b_latency"]["pass"] = True
|
|||
|
|
test_logger.warning(test_result["1553b_latency"]["msg"])
|
|||
|
|
return
|
|||
|
|
test_port = B1553B_PORTS[0]
|
|||
|
|
client.reset_recv()
|
|||
|
|
# 启动时延专用接收线程
|
|||
|
|
recv_thread = threading.Thread(target=client.recv_latency_loop, daemon=True)
|
|||
|
|
recv_thread.start()
|
|||
|
|
time.sleep(0.5)
|
|||
|
|
# 发送测试数据，记录发送时间
|
|||
|
|
send_times = []
|
|||
|
|
for _ in range(TEST_TIMES_1553B):
|
|||
|
|
send_ts = time.time()
|
|||
|
|
client.send_hex(test_port, TEST_DATA)
|
|||
|
|
send_times.append(send_ts)
|
|||
|
|
time.sleep(0.000001) # 1us间隔
|
|||
|
|
time.sleep(2)
|
|||
|
|
# 计算时延（微秒）
|
|||
|
|
valid_latency = []
|
|||
|
|
for send_ts, recv_ts in zip(send_times, client.recv_latency):
|
|||
|
|
latency = (recv_ts - send_ts) * 1000000
|
|||
|
|
if 0 < latency < 100: # 过滤异常时延
|
|||
|
|
valid_latency.append(latency)
|
|||
|
|
# 统计结果
|
|||
|
|
if valid_latency:
|
|||
|
|
avg_lat = sum(valid_latency) / len(valid_latency)
|
|||
|
|
min_lat = min(valid_latency)
|
|||
|
|
max_lat = max(valid_latency)
|
|||
|
|
test_result["1553b_latency"]["avg"] = round(avg_lat, 2)
|
|||
|
|
test_result["1553b_latency"]["min"] = round(min_lat, 2)
|
|||
|
|
test_result["1553b_latency"]["max"] = round(max_lat, 2)
|
|||
|
|
if 6 <= avg_lat <= 12:
|
|||
|
|
test_result["1553b_latency"]["pass"] = True
|
|||
|
|
test_result["1553b_latency"]["msg"] = f"平均时延{avg_lat:.2f}us，最小{min_lat:.2f}us，最大{max_lat:.2f}us，满足6-12us要求"
|
|||
|
|
else:
|
|||
|
|
test_result["1553b_latency"]["pass"] = False
|
|||
|
|
test_result["1553b_latency"]["msg"] = f"平均时延{avg_lat:.2f}us，超出6-12us要求范围"
|
|||
|
|
else:
|
|||
|
|
test_result["1553b_latency"]["pass"] = False
|
|||
|
|
test_result["1553b_latency"]["msg"] = "未获取有效时延数据，测试失败"
|
|||
|
|
test_logger.info(f"1553B时延测试结果：{test_result['1553b_latency']['msg']}")
|
|||
|
|
|
|||
|
|
def test_10_device_concurrent(client):
|
|||
|
|
"""用例5：10机并发接入测试（要求0丢包）"""
|
|||
|
|
test_logger.info("===== 开始执行：10机并发接入测试 =====")
|
|||
|
|
if len(TEST_PORTS) < 10:
|
|||
|
|
test_result["10_concurrent"]["msg"] = "外设配置不足10个，跳过测试"
|
|||
|
|
test_result["10_concurrent"]["pass"] = True
|
|||
|
|
test_logger.warning(test_result["10_concurrent"]["msg"])
|
|||
|
|
return
|
|||
|
|
client.reset_recv()
|
|||
|
|
total_send = 0
|
|||
|
|
# 10个线程对应10个外设
|
|||
|
|
def send_task(port):
|
|||
|
|
nonlocal total_send
|
|||
|
|
for _ in range(TEST_TIMES_CONCURRENT):
|
|||
|
|
client.send_hex(port, TEST_DATA)
|
|||
|
|
total_send += 1
|
|||
|
|
time.sleep(0.0001)
|
|||
|
|
threads = [threading.Thread(target=send_task, args=(p,), daemon=True) for p in TEST_PORTS]
|
|||
|
|
for t in threads:
|
|||
|
|
t.start()
|
|||
|
|
for t in threads:
|
|||
|
|
t.join()
|
|||
|
|
time.sleep(3)
|
|||
|
|
# 计算丢包率
|
|||
|
|
loss_rate = (total_send - client.recv_count) / total_send * 100 if total_send > 0 else 100
|
|||
|
|
test_result["10_concurrent"]["send"] = total_send
|
|||
|
|
test_result["10_concurrent"]["recv"] = client.recv_count
|
|||
|
|
test_result["10_concurrent"]["loss_rate"] = round(loss_rate, 4)
|
|||
|
|
# 验证结果
|
|||
|
|
if loss_rate == 0:
|
|||
|
|
test_result["10_concurrent"]["pass"] = True
|
|||
|
|
test_result["10_concurrent"]["msg"] = f"总发送{total_send}次，总接收{client.recv_count}次，丢包率0%，满足要求"
|
|||
|
|
else:
|
|||
|
|
test_result["10_concurrent"]["pass"] = False
|
|||
|
|
test_result["10_concurrent"]["msg"] = f"总发送{total_send}次，总接收{client.recv_count}次，丢包率{loss_rate:.4f}%，不满足0丢包要求"
|
|||
|
|
test_logger.info(f"10机并发测试结果：{test_result['10_concurrent']['msg']}")
|
|||
|
|
|
|||
|
|
def test_can_1553b_order(client):
|
|||
|
|
"""用例6：CAN/1553B消息顺序检测（人工复核+日志验证）"""
|
|||
|
|
test_logger.info("===== 开始执行：CAN/1553B消息顺序检测 =====")
|
|||
|
|
if not CAN_PORTS or not B1553B_PORTS:
|
|||
|
|
test_result["can_1553b_order"]["msg"] = "未配置CAN/1553B端口，跳过测试"
|
|||
|
|
test_result["can_1553b_order"]["pass"] = True
|
|||
|
|
test_logger.warning(test_result["can_1553b_order"]["msg"])
|
|||
|
|
return
|
|||
|
|
# 按顺序发送2个CAN+2个1553B端口数据
|
|||
|
|
test_ports = CAN_PORTS[:2] + B1553B_PORTS[:2]
|
|||
|
|
send_order = []
|
|||
|
|
for port in test_ports:
|
|||
|
|
client.send_hex(port, TEST_DATA + f"_ORDER_{port}".encode())
|
|||
|
|
send_order.append(port)
|
|||
|
|
time.sleep(0.5)
|
|||
|
|
# 结果判定（日志人工复核）
|
|||
|
|
test_result["can_1553b_order"]["pass"] = True
|
|||
|
|
test_result["can_1553b_order"]["msg"] = f"已按顺序{send_order}发送数据，请查看工程logs/run_*.log验证是否按序接收，无乱序即通过"
|
|||
|
|
test_logger.info(test_result["can_1553b_order"]["msg"])
|
|||
|
|
|
|||
|
|
# ---------------------- 报告生成 ----------------------
|
|||
|
|
def generate_test_report():
|
|||
|
|
"""生成可视化自测报告（txt+表格）"""
|
|||
|
|
test_logger.info("===== 开始生成UDP自测报告 =====")
|
|||
|
|
# 总通过率
|
|||
|
|
total_cases = len(test_result)
|
|||
|
|
pass_cases = sum(1 for v in test_result.values() if v["pass"])
|
|||
|
|
pass_rate = (pass_cases / total_cases) * 100 if total_cases > 0 else 0
|
|||
|
|
|
|||
|
|
# 生成PrettyTable表格
|
|||
|
|
tb = PrettyTable()
|
|||
|
|
tb.title = f"UDP连接&数据传输自测报告 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
|||
|
|
tb.field_names = ["自测用例", "测试结果", "核心指标/结果描述"]
|
|||
|
|
tb.align = "l"
|
|||
|
|
# 填充用例结果
|
|||
|
|
tb.add_row(["1.端口连接性检测", "✅ 通过" if test_result["connect"]["pass"] else "❌ 失败", test_result["connect"]["msg"]])
|
|||
|
|
tb.add_row(["2.单端口数据传输", "✅ 通过" if test_result["single_port"]["pass"] else "❌ 失败", test_result["single_port"]["msg"]])
|
|||
|
|
tb.add_row(["3.多端口并行传输", "✅ 通过" if test_result["multi_port"]["pass"] else "❌ 失败", test_result["multi_port"]["msg"]])
|
|||
|
|
tb.add_row(["4.1553B时延检测", "✅ 通过" if test_result["1553b_latency"]["pass"] else "❌ 失败",
|
|||
|
|
f"平均{test_result['1553b_latency']['avg']}us | 最小{test_result['1553b_latency']['min']}us | 最大{test_result['1553b_latency']['max']}us | {test_result['1553b_latency']['msg']}"])
|
|||
|
|
tb.add_row(["5.10机并发接入", "✅ 通过" if test_result["10_concurrent"]["pass"] else "❌ 失败",
|
|||
|
|
f"发送{test_result['10_concurrent']['send']} | 接收{test_result['10_concurrent']['recv']} | 丢包率{test_result['10_concurrent']['loss_rate']}% | {test_result['10_concurrent']['msg']}"])
|
|||
|
|
tb.add_row(["6.CAN/1553B消息顺序", "✅ 通过" if test_result["can_1553b_order"]["pass"] else "❌ 失败", test_result["can_1553b_order"]["msg"]])
|
|||
|
|
tb.add_row(["", "", ""])
|
|||
|
|
tb.add_row(["📊 自测总览", f"总用例{total_cases}个 | 通过{pass_cases}个 | 通过率{pass_rate:.2f}%", ""])
|
|||
|
|
|
|||
|
|
# 保存报告到文件
|
|||
|
|
report_path = f"test/report/udp_auto_test_report_{report_time}.txt"
|
|||
|
|
with open(report_path, "w", encoding="utf-8") as f:
|
|||
|
|
f.write(str(tb))
|
|||
|
|
# 追加端口连接性详情
|
|||
|
|
f.write("\n\n=== 端口连接性检测详情 ===\n")
|
|||
|
|
for line in test_result["connect"]["detail"]:
|
|||
|
|
f.write(line + "\n")
|
|||
|
|
|
|||
|
|
# 控制台打印报告
|
|||
|
|
print("\n" + "="*80)
|
|||
|
|
print(tb)
|
|||
|
|
print("="*80)
|
|||
|
|
print(f"\n📋 自测报告已保存至：{os.path.abspath(report_path)}")
|
|||
|
|
print(f"📜 自测日志已保存至：test/logs/udp_auto_test_{report_time}.log")
|
|||
|
|
test_logger.info(f"自测报告生成完成，通过率{pass_rate:.2f}%")
|
|||
|
|
|
|||
|
|
# ---------------------- 主执行入口 ----------------------
|
|||
|
|
def main():
|
|||
|
|
print("🚀 虚拟仿真平台中间层软件 - UDP一键自测脚本启动")
|
|||
|
|
print(f"📌 测试时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|||
|
|
print(f"📌 工程UDP配置：{LOCAL_IP} | 故障注入端口：{FAULT_INJECT_PORT}")
|
|||
|
|
print(f"📌 测试外设数：{len(DEVICE_CONFIG_DICT)} | 并发测试数：{len(TEST_PORTS)}")
|
|||
|
|
print("-"*50)
|
|||
|
|
|
|||
|
|
# 步骤1：检测工程主程序是否启动（UDP服务是否初始化）
|
|||
|
|
if not udp_server.udp_sockets:
|
|||
|
|
print("❌ 检测到工程UDP服务未初始化，请先启动main.py后再执行自测！")
|
|||
|
|
test_logger.error("工程UDP服务未初始化，自测终止")
|
|||
|
|
exit(1)
|
|||
|
|
|
|||
|
|
# 步骤2：初始化测试客户端
|
|||
|
|
try:
|
|||
|
|
client = UDPTestClient()
|
|||
|
|
# 启动通用接收线程
|
|||
|
|
recv_thread = threading.Thread(target=client.recv_loop, daemon=True)
|
|||
|
|
recv_thread.start()
|
|||
|
|
test_logger.info("UDP测试客户端初始化完成，启动接收线程")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"❌ UDP测试客户端初始化失败：{e}")
|
|||
|
|
test_logger.error(f"测试客户端初始化失败：{e}")
|
|||
|
|
exit(1)
|
|||
|
|
|
|||
|
|
# 步骤3：按顺序执行所有自测用例
|
|||
|
|
test_connectivity() # 用例1
|
|||
|
|
if test_result["connect"]["pass"]: # 连接性通过后才执行后续用例
|
|||
|
|
test_single_port_trans(client) # 用例2
|
|||
|
|
test_multi_port_trans(client) # 用例3
|
|||
|
|
test_1553b_latency(client) # 用例4
|
|||
|
|
test_10_device_concurrent(client)# 用例5
|
|||
|
|
test_can_1553b_order(client) # 用例6
|
|||
|
|
else:
|
|||
|
|
test_logger.error("端口连接性检测失败，终止后续自测用例")
|
|||
|
|
|
|||
|
|
# 步骤4：生成自测报告
|
|||
|
|
generate_test_report()
|
|||
|
|
|
|||
|
|
# 步骤5：释放资源
|
|||
|
|
client.close()
|
|||
|
|
udp_server.close()
|
|||
|
|
print("\n✅ UDP一键自测脚本执行完成，所有资源已释放")
|
|||
|
|
test_logger.info("UDP一键自测脚本执行完成")
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
main()
|