162 lines
7.4 KiB
Python
162 lines
7.4 KiB
Python
'''
|
||
功能对等客户端：模拟故障注入软件，发送/接收UDP数据
|
||
author：Tan Mingyan
|
||
更新日期：2025/2/26
|
||
'''
|
||
|
||
|
||
'''
|
||
2. 功能自测步骤
|
||
步骤 1：启动工程主程序
|
||
运行main.py，查看日志确认：
|
||
外设配置成功加载（10 个外设）；
|
||
UDP 服务初始化完成，所有端口正常监听；
|
||
无套接字绑定、配置解析错误。
|
||
步骤 2：运行 UDP 测试客户端
|
||
运行test/udp_test_client.py，执行单端口发送和多端口并行发送，验证以下核心功能：
|
||
发送成功：客户端打印📤 发送数据成功，无端口连接失败；
|
||
中间层接收成功：查看工程logs/run_*.log，打印对应端口的数据接收日志，无 16 进制解析错误；
|
||
端口隔离性：多端口并行发送时，各端口数据独立接收，无串包、丢包（日志中各外设数据互不干扰）；
|
||
校验和检测：若发送数据包含正确校验位，日志打印校验和检测通过；若修改校验位为错误值，日志打印校验和检测失败并记录到logs/error_*.log；
|
||
回传数据接收：若在工程中调用udp_server.send_to_fault()（如协议解析后回传），客户端能打印📥 接收中间层回传数据，数据格式正确。
|
||
3. 常见问题排查
|
||
表格
|
||
问题现象 排查方向
|
||
客户端提示发送数据失败 1. 中间层主程序是否未启动；2. 目标端口是否在 CSV 配置中；3. 端口是否被占用
|
||
中间层未接收数据 1. 客户端发送的 IP 是否为127.0.0.1；2. 16 进制流格式是否正确（无乱码）；3. 查看logs/error_*.log是否有数据源过滤（非故障注入 IP）
|
||
校验和检测失败 1. 发送数据的校验位是否与error_check.py的算法匹配；2. 数据是否包含校验位（长度是否足够）
|
||
多端口串包 1. CSV 配置中是否有重复端口；2. 工程udp_handler.py的_get_dev_by_port是否正确映射外设
|
||
'''
|
||
import socket
|
||
import binascii
|
||
import threading
|
||
import time
|
||
import sys
|
||
import os
|
||
import io
|
||
|
||
# 设置标准输出和标准错误的编码为UTF-8
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
|
||
|
||
from pathlib import Path
|
||
# 获取当前文件的目录
|
||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||
# 获取上级目录（项目根目录）
|
||
parent_dir = os.path.dirname(current_dir)
|
||
# 将项目根目录添加到系统路径
|
||
sys.path.append(parent_dir)
|
||
#sys.path.append(str(Path(__file__).parent.parent.parent))
|
||
from midware.config.base_config import UDP_CONFIG
|
||
|
||
class UDPTestClient:
|
||
"""UDP测试客户端：模拟故障注入软件，发送/接收UDP数据"""
|
||
def __init__(self):
|
||
self.local_ip = UDP_CONFIG["LOCAL_IP"]
|
||
self.fault_port = UDP_CONFIG["FAULT_INJECT_PORT"] # 故障注入软件接收端口（客户端监听）
|
||
self.buffer_size = UDP_CONFIG["BUFFER_SIZE"]
|
||
# 发送套接字（向中间层外设端口发包）
|
||
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((self.local_ip, self.fault_port))
|
||
self.recv_sock.setblocking(False)
|
||
# 接收线程标识
|
||
self.recv_running = True
|
||
# 启动接收线程（后台接收，不阻塞发包）
|
||
self.recv_thread = threading.Thread(target=self._recv_loop, daemon=True)
|
||
self.recv_thread.start()
|
||
print(f"UDP测试客户端初始化完成 | 监听回传数据：{self.local_ip}:{self.fault_port}")
|
||
|
||
def _recv_loop(self):
|
||
"""后台接收线程：接收中间层回传的16进制流数据"""
|
||
while self.recv_running:
|
||
try:
|
||
data, addr = self.recv_sock.recvfrom(self.buffer_size)
|
||
hex_data = data.decode("utf-8").strip()
|
||
raw_data = binascii.unhexlify(hex_data) if hex_data else b""
|
||
print(f"\n📥 接收中间层回传数据 | 来源：{addr} | 16进制流：{hex_data[:64]}... | 原始数据长度：{len(raw_data)}字节")
|
||
except BlockingIOError:
|
||
time.sleep(0.001)
|
||
continue
|
||
except Exception as e:
|
||
print(f"\n❌ 接收数据异常：{str(e)}")
|
||
continue
|
||
|
||
def send_hex_data(self, dev_port, raw_data):
|
||
"""
|
||
向中间层指定外设端口发送16进制流数据（模拟故障注入软件发包）
|
||
:param dev_port: 外设UDP端口（中间层监听端口）
|
||
:param raw_data: 原始二进制数据，自动转为16进制流
|
||
"""
|
||
try:
|
||
# 原始数据转16进制流（符合工程数据格式要求）
|
||
hex_data = binascii.hexlify(raw_data).decode("utf-8")
|
||
self.send_sock.sendto(hex_data.encode("utf-8"), (self.local_ip, dev_port))
|
||
print(f"\n📤 发送数据成功 | 目标端口：{dev_port} | 16进制流：{hex_data[:64]}... | 原始数据长度：{len(raw_data)}字节")
|
||
return True
|
||
except Exception as e:
|
||
print(f"\n❌ 发送数据失败 | 目标端口：{dev_port} | 异常：{str(e)}")
|
||
return False
|
||
|
||
def close(self):
|
||
"""关闭套接字，释放资源"""
|
||
self.recv_running = False
|
||
self.send_sock.close()
|
||
self.recv_sock.close()
|
||
print("\n✅ UDP测试客户端已关闭")
|
||
|
||
# 多端口并行发送测试（线程版）
|
||
def multi_port_send(client, port_data_list):
|
||
"""
|
||
多端口并行发送数据，验证端口隔离性
|
||
:param port_data_list: 列表，元素为(端口, 原始数据)
|
||
"""
|
||
def send_task(port, data):
|
||
client.send_hex_data(port, data)
|
||
# 模拟不同单机的发包间隔
|
||
time.sleep(0.001)
|
||
|
||
threads = []
|
||
for port, data in port_data_list:
|
||
t = threading.Thread(target=send_task, args=(port, data), daemon=True)
|
||
threads.append(t)
|
||
t.start()
|
||
# 等待所有线程完成
|
||
for t in threads:
|
||
t.join()
|
||
print("\n===== 多端口并行发送任务完成 =====")
|
||
|
||
if __name__ == "__main__":
|
||
# 初始化测试客户端
|
||
client = UDPTestClient()
|
||
# 测试用原始数据（可自定义，建议包含校验位，适配error_check.py的校验逻辑）
|
||
test_data1 = b"UART_TEST_DATA_001" + b'\x9c\x47' # UART数据+CRC16校验位
|
||
test_data2 = b"CAN_TEST_DATA_001" + b'\x1a\x2b' # CAN数据+CRC16校验位
|
||
test_data3 = b"1553B_TEST_DATA_001" + b'\x0f\x3d' # 1553B数据+CRC16校验位
|
||
test_data4 = b"AD_TEST_DATA_001" + b'\x5f' # AD数据+累加和校验位
|
||
test_data5 = b"OC_TEST_DATA_001" + b'\x6a' # OC数据+累加和校验位
|
||
|
||
try:
|
||
# 1. 单端口发送测试（例：向Uart0(8880)发送数据）
|
||
print("===== 单端口发送测试 =====")
|
||
# client.send_hex_data(8880, test_data1)
|
||
client.send_hex_data(4000, test_data1)
|
||
time.sleep(1)
|
||
|
||
# 2. 多端口并行发送测试（验证端口互不干扰）
|
||
print("\n===== 多端口并行发送测试 =====")
|
||
port_data = [
|
||
(4000, test_data1), (4001, test_data2), (4002, test_data3),
|
||
(4003, test_data4), (4004, test_data1)
|
||
]
|
||
multi_port_send(client, port_data)
|
||
time.sleep(3)
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n\n用户终止测试")
|
||
finally:
|
||
# 关闭客户端
|
||
client.close()
|
||
time.sleep(1) |