3、Python基础回顾：面向对象编程、socket编程基础、异常处理机制、日志模块配置
好，咱们进入第三讲。说实话，很多做DoIP测试的朋友，Python基础是有的，但一到写框架就卡壳。为啥？因为面向对象、socket、异常、日志这几个东西，单独拎出来都会，合在一起就懵了。
今天我就带你把这几块串起来。记住，咱们的目标不是学Python，而是用Python搞定DoIP测试。
3.1 面向对象编程：把DoIP报文封装成“活物”
面向对象，说白了就是“把数据和操作数据的方法打包在一起”。我在做DoIP测试时，最烦的就是到处散落着报文解析的代码。后来我学乖了——把每个DoIP报文都封装成一个类。
核心思想：DoIP报文有固定的结构——协议版本、负载类型、负载长度、实际负载。这些字段天然适合用对象来管理。
class DoIPMessage:
def __init__(self, protocol_version=0x02, payload_type=0x0001, payload=b''):
self.protocol_version = protocol_version
self.payload_type = payload_type
self.payload_length = len(payload)
self.payload = payload
def encode(self):
"""把对象变成字节流，方便发送"""
header = struct.pack('!BH', self.protocol_version, self.payload_type)
length_field = struct.pack('!I', self.payload_length)
return header + length_field + self.payload
@classmethod
def decode(cls, raw_data):
"""从字节流还原出对象"""
protocol_version = raw_data[0]
payload_type = struct.unpack('!H', raw_data[1:3])[0]
payload_length = struct.unpack('!I', raw_data[3:7])[0]
payload = raw_data[7:7+payload_length]
return cls(protocol_version, payload_type, payload)你看，这样一封装，发送和接收都变得特别清爽。我个人习惯是，每个DoIP报文类型都对应一个子类，比如路由激活报文、诊断报文，都继承自这个基类。
我的经验：刚开始写框架时，别急着搞继承。先把一个类写稳了，再考虑抽象。我见过太多人一上来就设计三层继承，结果改一个地方崩一片。
3.2 Socket编程基础：DoIP的“血管”
DoIP跑在TCP/IP上，所以socket编程是基本功。但注意，DoIP不是简单的“发-收”模式，它有连接建立、保活、超时重连这些讲究。
为什么会这样？因为车上的ECU可能随时休眠，也可能突然重启。你的测试脚本必须能处理这些“不靠谱”。
import socket
import time
class DoIPSocket:
def __init__(self, target_ip, target_port=13400):
self.target_ip = target_ip
self.target_port = target_port
self.sock = None
self.timeout = 5 # 秒
def connect(self):
"""建立TCP连接，带重试机制"""
retry_count = 3
for i in range(retry_count):
try:
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.settimeout(self.timeout)
self.sock.connect((self.target_ip, self.target_port))
print(f"连接成功: {self.target_ip}:{self.target_port}")
return True
except socket.timeout:
print(f"第{i+1}次连接超时，重试中...")
time.sleep(1)
except Exception as e:
print(f"连接异常: {e}")
break
return False
def send_message(self, doip_msg):
"""发送DoIP报文"""
if not self.sock:
raise ConnectionError("socket未连接")
raw_data = doip_msg.encode()
self.sock.sendall(raw_data)
def receive_message(self):
"""接收DoIP报文，处理粘包问题"""
# 先读7字节头部
header = self.sock.recv(7)
if len(header) < 7:
return None
payload_length = struct.unpack('!I', header[3:7])[0]
# 再读负载
payload = b''
while len(payload) < payload_length:
chunk = self.sock.recv(payload_length - len(payload))
if not chunk:
break
payload += chunk
return DoIPMessage.decode(header + payload)避坑指南：我曾经在项目里吃过“粘包”的亏。DoIP报文是流式的，recv一次不一定能拿到完整报文。所以一定要先读固定长度的头部，再根据头部里的长度字段去读负载。这个顺序不能乱。
3.3 异常处理机制：别让脚本“裸奔”
测试脚本跑着跑着突然崩了，这种事你遇到过吧？我早期写脚本时，异常处理就是print("出错了")，然后exit。后来被坑惨了——有一次半夜跑回归测试，脚本在凌晨3点崩了，第二天早上才发现，一晚上白跑了。
所以，异常处理要分层、要精细。
class DoIPError(Exception):
"""DoIP基础异常类"""
pass
class ConnectionError(DoIPError):
"""连接相关异常"""
pass
class TimeoutError(DoIPError):
"""超时异常"""
pass
class ProtocolError(DoIPError):
"""协议解析异常"""
pass
def safe_send_receive(socket_obj, doip_msg, max_retries=3):
"""带异常处理的发送接收函数"""
for attempt in range(max_retries):
try:
socket_obj.send_message(doip_msg)
response = socket_obj.receive_message()
if response is None:
raise TimeoutError("接收超时，未收到响应")
return response
except socket.timeout:
print(f"第{attempt+1}次尝试超时")
if attempt == max_retries - 1:
raise TimeoutError("重试耗尽，放弃")
time.sleep(2)
except ProtocolError as e:
print(f"协议错误: {e}")
# 记录错误报文，方便事后分析
log_error(f"协议错误: {e}, 报文: {doip_msg}")
raise
except Exception as e:
print(f"未知错误: {e}")
# 尝试重连
socket_obj.connect()
continue我的习惯：自定义异常类一定要继承自一个基础异常。这样你在最外层可以统一捕获DoIPError，而不用管具体是哪种错误。另外，异常信息里一定要带上上下文——哪个报文、哪个步骤、什么时间。不然查问题就像大海捞针。
3.4 日志模块配置：给测试过程“录像”
日志这东西，平时觉得没用，出问题时就真香了。我见过有人用print来记录测试过程，结果脚本一多，满屏的print，根本分不清是哪条消息。
Python自带的logging模块，配置好了非常好用。我一般这样配：
import logging
import logging.handlers
import os
from datetime import datetime
def setup_logger(log_dir="logs", log_level=logging.DEBUG):
"""配置日志系统"""
# 创建日志目录
os.makedirs(log_dir, exist_ok=True)
# 日志文件名带时间戳
log_file = os.path.join(log_dir, f"doip_test_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log")
# 创建logger
logger = logging.getLogger("DoIPTest")
logger.setLevel(log_level)
# 文件处理器：记录所有级别
file_handler = logging.handlers.RotatingFileHandler(
log_file, maxBytes=10*1024*1024, backupCount=5, encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)
# 控制台处理器：只记录INFO及以上
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 定义格式
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
# 添加处理器
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
# 使用示例
logger = setup_logger()
logger.info("DoIP测试开始")
logger.debug(f"发送报文: {doip_msg.encode().hex()}")
logger.warning("ECU响应超时，准备重试")
logger.error("连接断开，测试终止")关键配置说明：
| 配置项 | 推荐值 | 原因 |
|---|---|---|
| 日志级别 | DEBUG（文件）/ INFO（控制台） | 文件记录细节，控制台只看关键信息 |
| 文件轮转 | 10MB，保留5个备份 | 防止日志文件撑爆磁盘 |
| 格式 | 含时间、文件名、行号 | 方便定位问题来源 |
| 编码 | utf-8 | 避免中文乱码 |
避坑指南：我曾经犯过一个低级错误——在循环里反复调用setup_logger，结果每次调用都新增一个handler，日志重复打印。正确的做法是：在模块级别只调用一次，或者用logger.hasHandlers()判断一下。
3.5 把它们串起来：一个完整的DoIP测试片段
好了，前面四个知识点都讲完了。咱们把它们组合起来，看看实际怎么用。
def test_doip_routing_activation():
"""测试DoIP路由激活功能"""
logger = setup_logger()
socket_obj = DoIPSocket("192.168.1.100")
try:
# 连接
if not socket_obj.connect():
logger.error("连接失败，测试终止")
return False
# 构造路由激活请求
activation_req = DoIPMessage(
protocol_version=0x02,
payload_type=0x0005, # 路由激活
payload=bytes([0x01, 0x00, 0x00, 0x00]) # 激活类型
)
# 发送并接收
logger.info("发送路由激活请求")
response = safe_send_receive(socket_obj, activation_req)
# 验证响应
if response.payload_type == 0x0006: # 路由激活响应
logger.info("路由激活成功")
return True
else:
logger.warning(f"意外的响应类型: 0x{response.payload_type:04x}")
return False
except DoIPError as e:
logger.error(f"DoIP测试失败: {e}")
return False
except Exception as e:
logger.critical(f"未预期的异常: {e}", exc_info=True)
return False
finally:
socket_obj.sock.close()
logger.info("连接已关闭")你看，这样写出来的代码，可读性强、健壮性好、出了问题也好查。这就是面向对象 + socket + 异常 + 日志组合起来的效果。
嗯，这一讲就到这里。下一讲咱们开始真正搭建DoIP自动化测试框架的骨架。到时候你会发现，今天打的基础，全都能用上。