【Python搞定车载自动化测试】——Python基于Pytest框架实现UDS诊断自动化（含Python源码）

系列文章目录

【Python搞定车载自动化测试】系列文章目录汇总

文章目录

系列文章目录
💯💯💯 前言💯💯💯
- 一、环境搭建
- - 1.软件环境
  - 2.硬件环境
- 二、目录结构
- 三、源码展示
- - 1.诊断基础函数方法
  - 2.诊断业务函数方法
  - 3.27服务安全解锁
  - 4.自动测试用例
  - 5.配置参数
- 四、日志和报告
- - 1.测试日志
  - 2.测试报告
- 五、完整源码链接

💯💯💯 前言💯💯💯

在之前的专栏【如何学习CAN总线测试】中介绍了如何通过CAPL语言实现UDS诊断测试，但CAPL语言依赖于Vector CANoe设备，随着智能电动汽车行业的发展，目前使用Python语言实现自动化测试的需求越来越广，本章节主要介绍如何使用Python语言基于Pytest自动化框架来实现UDS诊断自动化测试。
首先回顾一下UDS诊断基础知识：

统一诊断服务（Unified Diagnostic Services），简称UDS。是ISO 15765和ISO 14229定义的一种汽车通用诊断协议，位于OSI模型中的应用层，它可在不同的汽车总线（例如CAN、LIN、Flexray、Internet、K-line）上实现，是当前汽车领域广泛使用的一种车载诊断协议标准。

UDS协议的应用层定义是ISO 14229-1，目前大部分汽车厂商均采用UDS on CAN的诊断协议。
根据UDS的诊断协议，汽车上的控制系统需要根据规则化的诊断协议进行故障记录和处理，最终体现为诊断故障代码（Diagnostic Trouble Code，DTC）的方式。例如，网络通信丢失的故障诊断机制：
自动变速箱控制单元（Transmission Control Unit，TCU）和制动防抱死系统（Antilock Brake System，ABS）是CAN车载网络上的两大电子控制单元，这2个ECU要通过CAN网络进行大量的信息交互。但是由于电磁干扰、串扰、静电等外界干扰或电子控制单元本身控制策略引起的通信停止等原因，2个电子控制单元之间可能会出现通信丢失的现象。
控制系统需要将故障信息（例如：通信丢失故障信息）诊断出来，以处理通信被破坏时出现丢失帧的故障现象，并记录为DTC。一旦某一控制系统，如TCU监测到一段规定的时间内并没有接收到ABS发来的通信数据，便将此DTC记录下来。外部诊断设备通过规则的诊断通信与控制系统建立诊断通信连接，并选择相应的诊断方式。例如：读取故障信息服务时，就会将此故障信息读出，并在诊断仪中显示出来。

UDS诊断服务共分为六大单元：
1、诊断和通信管理功能单元（Diagnostic and Communication Management）
$10 - 诊断会话控制（DiagnosticSessionControl）
$11 - 电控单元复位（ECUReset）
$27 - 安全访问（SecurityAccess）
$28 - 通讯控制（CommunicationControl）
$3E - 待机握手（TesterPresent）
$83 - 访问时间参数（AccessTimingParameter）
$84 - 安全数据传输（SecuredDataTransmission）
$85 - 诊断故障码设置控制（ControlDTCSetting）
$86 - 事件响应（ResponseOnEvent）
$87 - 链路控制（LinkControl）
2、数据传输功能单元（Data Transmission）
$22 - 通过ID读数据（ReadDataByldentifier）
$23 - 通过地址读内存（ReadMemoryByAddress）
$24 - 通过ID读比例数据（ReadScalingDataByidentifier）
$2A - 通过周期ID读取数据（ReadDataUyPeriodicidentifier）
$2C - 动态定义标识符（DynamicallyDefineDataldentifier）
$2E - 通过ID写数据（WriteDataByldentifier）
$3D - 通过地址写内存（WriteMemoryByAddress）
3、存储数据传输功能单元（Stored Data Transmission）
$14 - 清除诊断信息（ClearDiagnosticInformation）
$19 - 读取故障码信息（ReadDTCInformation）
4、输入输出控制功能单元（Input & Output Control）
$2F - 通过标识符控制输入输出（InputOutputControlByIdentifier）
5、例行程序功能单元（Remote Activation of Routine）
$31 - 例行程序控制（RoutineControl）
6 、传下载功能单元（Upload & Download）
$34 - 请求下载（RequestDownload）
$35 - 请求上传（RequestUpload）
$36 - 数据传输（TransferData）
$37 - 请求退出传输（RequestTransferExit）
$38 - 请求文件传输（RequestFileTransfer）

一、环境搭建

1.软件环境

Python版本：python3.9
第三方库：
pip install allure-pytest2.13.5
pip install can-isotp2.0.4
pip install python-can4.3.1
pip install udsoncan1.23.0
allure：安装allure工具并设置环境变量，https://github.com/allure-framework/allure2/releases

2.硬件环境

支持CAN设备硬件接口：
在这里插入图片描述
https://python-can.readthedocs.io/en/stable/configuration.html

二、目录结构

在这里插入图片描述
dll目录：存放27服务安全解锁DLL文件（同CANoe中使用的DLL文件）。
public_method目录：存放公共函数方法和27安全解锁工具。
test_case目录：存放python自动化测试用例。
update目录：存放升级包和效验文件。
config.py文件：CAN相关配置参数。
run.py文件：运行入口文件。

三、源码展示

1.诊断基础函数方法

base_can.py文件主要封装了CAN初始化、诊断配置、诊断请求、诊断响应基础方法，源码如下：

class CanBus:
    def __init__(self, interface: str = None, channel: int = None, bitrate: int = None, fd: bool = None,
                 data_bitrate: int = None, can_filters: CanFilters = None, *args, **kwargs):
        self.interface = interface
        self.channel = channel
        self.bitrate = bitrate
        self.fd = fd
        self.data_bitrate = data_bitrate
        self.can_filters = can_filters
        try:
            self.bus = can.interface.Bus(channel=self.channel, interface=self.interface, app_name="CANoe",
                                         bitrate=self.bitrate, fd=self.fd, data_bitrate=self.data_bitrate,
                                         can_filters=self.can_filters, *args, **kwargs)
        except Exception as e:
            raise Exception("初始化失败：%s" % e)
        else:
            print("初始化成功")

    def diag_congfig(self, tx: int, rx: int, addressingmode=isotp.AddressingMode.Normal_11bits):
        """
        诊断配置函数
        :param tx: 诊断请求ID，功能寻址、物理寻址
        :param rx: 诊断响应ID
        :return:
        """
        self.isotp_params = {
            'stmin': 20,  # 流控帧间隔时间
            'blocksize': 8,  # 流控帧单包大小，0表示不限制
            'tx_padding': 0,  # 当 notNone表示用于填充发送的消息的字节。
            'rx_flowcontrol_timeout': 1000,  # 在停止接收和触发之前等待流控制帧的毫秒数
            'rx_consecutive_frame_timeout': 1000,  # 在停止接收和触发 a 之前等待连续帧的毫秒数
        }
        try:
            self.tp_addr = isotp.Address(addressing_mode=addressingmode, txid=tx, rxid=rx)  # 网络层寻址方案
            tp_stack = isotp.CanStack(bus=self.bus, address=self.tp_addr, params=self.isotp_params)  # 网络/传输层（IsoTP 协议）
            self.conn = PythonIsoTpConnection(tp_stack)  # 应用层和传输层之间的接口
        except Exception as e:
            print("UDS配置失败:%s" % e)
        return self.conn

    def diag_request(self, request_command: str, request_data_log_flag=True):
        """
        诊断请求
        """
        requestPdu = binascii.a2b_hex(request_command.replace(' ', ''))
        if not self.conn.is_open():
            self.conn.open()
        try:
            self.conn.send(requestPdu)
        except Exception as e:
            print("诊断请求失败:%s" % e)
        else:
            req_info = ''
            request_command = request_command.replace(' ', '')
            for i in range(len(request_command)):
                if i >= len(request_command) / 2:
                    break
                req_info += request_command[2 * i:2 * i + 2] + ' '
            if request_data_log_flag:
                print("诊断请求：%s" % req_info)

    def diag_respond(self, timeout1=1):
        """
        诊断响应
        """
        try:
            respPdu = self.conn.wait_frame(timeout=timeout1)
        except Exception as e:
            print(e)
        else:
            if respPdu is None:
                return None
            resp1 = respPdu.hex().upper()
            resp2 = ''
            for i in range(len(resp1)):
                if i != 0 and i % 2 == 0:
                    resp2 += ' '
                resp2 += resp1[i]
            print("诊断响应：%s" % resp2)
            return resp2

2.诊断业务函数方法

fun_can.py主要二次封装UDS诊断函数，包括：27安全解锁，34服务、36服务、诊断78响应处理、UDS诊断测试、CRC效验等函数，源码如下：

import binascii
import os
import subprocess
import time

import pytest

from config import Parameter
from public_method.base_can import CanBus


class CanMethod(CanBus):
    def __init__(self, config):
        self.interface = config['can']['interface']
        self.channel = config['can']['channel']
        self.bitrate = config['can']['bitrate']
        self.fd = config['can']['canfd']
        self.data_bitrate = config['can']['data_bitrate']
        self.addressingmode = config['can']['addressing_mode']
        self.tx = config['can']['physics_id_default']
        self.rx = config['can']['response_id_default']
        CanBus.__init__(self, interface=self.interface, channel=self.channel, bitrate=self.bitrate, fd=self.fd,
                        data_bitrate=self.data_bitrate, )
        self.diag_congfig(addressingmode=self.addressingmode, tx=self.tx, rx=self.rx)
        self.sign_nrc78 = 0

    def __diag_get_seed(self, req_data="27 01"):
        """
        27服务获取种子并解析
        """
        self.diag_request(req_data)
        try:
            uds_res_data = self.conn.specific_wait_frame(timeout=2)
            while uds_res_data[0] == 0x7F and uds_res_data[2] == 0x78:
                print("已收到 %d bytes : [%s]" % (len(uds_res_data), binascii.hexlify(uds_res_data)))
                uds_res_data = self.conn.specific_wait_frame(timeout=3)
            resp1 = uds_res_data.hex().upper()
            resp2 = ''
            for i in range(len(resp1)):
                if i != 0 and i % 2 == 0:
                    resp2 += ' '
                resp2 += resp1[i]
            print("诊断响应：%s" % resp2)
        except:
            print("响应数据失败")
        else:
            seed = []
            res_seed = resp2.split(' ')[2:]
            for i in range(len(res_seed)):
                seed.append(eval('0x' + res_seed[i]))
            print("seed：%s" % seed)
            return seed

    def get_key_level(self, seed):
        """
        dll_security_unlock.exe工具解锁
        语法：
        dll_security_unlock.exe --dll_path dome.dll --seed [11,22,33,44] --seedsize 4 --level 1 --keysize 4
        --dll_path DLL路径
        --seed 请求种子
        --seedsize 种子长度
        --level 安全级别
        --keysize 秘钥长度
        """
        seed = str(seed).replace(' ', '')
        tool = os.path.join(os.path.dirname(__file__), 'dll_security_unlock.exe')
        cmd = '{} --dll_path {} --seed {}'.format(tool, os.path.join(
            os.path.dirname(os.path.dirname(__file__)), r'dll\dome.dll'), seed)
        key = subprocess.getoutput(cmd)
        return key

    def unlock_level(self):
        """27安全解锁"""
        seed = self.__diag_get_seed(req_data="27 01")
        if seed is not None:
            if seed != 0 and len(seed) > 1:
                key = self.get_key_level(seed)
                print("key= %s" % key)
                req_data = "27 02 %s" % key
                self.diag_request(req_data)
                self.uds_respond_0x78()
                time.sleep(0.1)
        else:
            print("seed响应不正确")

    def diag_send(self, req_data="3E 80"):
        """
        发送诊断请求，不断言诊断响应
        """
        self.diag_request(req_data)
        response = self.uds_respond_0x78()
        time.sleep(0.1)
        return response

    def diag_send_exp(self, req_data="3E 80", exp=None):
        """
        发送诊断请求，并断言诊断响应
        """
        self.diag_request(req_data)
        result = self.__diag_test_response_judge(exp=exp)
        time.sleep(0.1)
        return result

    def diag_send_0x34(self, req_data="34 00 44", address="00 00 00 00", size=0, exp=None):
        """
        刷写时使用，请求传输升级包
        """
        print("传输包大小= %s" % size)
        self.diag_request(req_data + address + "{:08X}".format(size))
        self.__diag_test_response_judge(exp=exp)
        time.sleep(0.1)

    def diag_send_0x36(self, req_data="36", trans_size=255, path="", exp=None):
        """
        36服务传包
        """
        total_size = os.path.getsize(path)
        print("size = %s" % total_size)
        with open(path, "rb") as f:
            file_read = f.read()
        print(
            "CRC= %s" % "{:02X}".format(binascii.crc32(file_read))
        )
        file_byte = []
        for i in range(len(file_read)):
            file_byte.append("{:02X}".format(file_read[i]))
        sequence = 1
        transmitted_size = 0
        try:
            status = True
            while status:
                trans_data = ""
                for i in range(trans_size):
                    if transmitted_size < total_size:
                        trans_data = trans_data + file_byte[transmitted_size]
                        transmitted_size = transmitted_size + 1
                    else:
                        status = False
                        break
                print("data_num=%s" % transmitted_size)
                self.diag_request(
                    request_command=req_data + "{:02X}".format(sequence) + trans_data,
                    request_data_log_flag=False,
                )
                print(req_data + "{:02X}".format(sequence) + "...")
                self.__diag_test_response_judge(exp=exp)
                sequence += 1
                if sequence == 256:
                    sequence = 0
        finally:
            print("36传输结束")

    def diag_crc32(self, req_data="31 01 02 02", path="", exp=None):
        """
        刷写时使用，CRC32校验
        """
        size = os.path.getsize(path)
        print("size = %s" % size)
        with open(path, "rb") as f:
            file_read = f.read()
        crc32 = "{:08X}".format(binascii.crc32(file_read))
        print("crc 32 = %s " % crc32)
        self.diag_send_exp(req_data=req_data + crc32, exp=exp)

    def __diag_session_mode(self, session):
        """
        诊断会话模式
        """
        if session == "01":
            self.diag_send(req_data="10 01")
        elif session == "03":
            self.diag_send(req_data="10 03")
        elif session == "02":
            self.diag_send(req_data="10 03")
            self.unlock_level()
            self.diag_send(req_data="10 02")

    def uds_respond_0x78(self, timeout1=2):
        """
        78响应处理
        """
        response = self.diag_respond(timeout1=timeout1)
        if response is not None:
            try:
                response2 = response.replace(' ', '')
                cyc = 0
                while response2[:2] == '7F' and response2[4:6] == '78':
                    self.sign_nrc78 = 1
                    response = self.diag_respond(timeout1=timeout1)
                    if response is not None:
                        response2 = response.replace(' ', '')
                    cyc += 1
                    if cyc > 20:
                        break
            except Exception as e:
                print("异常:%s" % e)
        return response

    def __diag_test_response_judge(self, exp=None):
        """
        断言响应结果与预期结果是否一致
        """
        response = self.uds_respond_0x78()
        response_return = response
        if (exp is not None) & (response is not None):
            exp = exp.replace(" ", "").upper()
            exp2 = ""
            for i in range(len(exp)):
                if i != 0 and i % 2 == 0:
                    exp2 += " "
                exp2 += exp[i]
            exp = exp2
            if len(exp) < len(response):
                response = response[0: len(exp)]
        if response == exp:
            return response_return
        else:
            print("诊断结果与预期结果不匹配")
            pytest.fail("诊断结果与预期结果不匹配")

    def diag_test(self, session="01", req_data=None, exp=None):
        """
        诊断测试
        :param session: 执行前会话模式，01默认会话，02编程会话，03扩展会话
        :param req_data:请求数据
        :param exp:预期结果
        :return:
        """
        self.__diag_session_mode(session=session)
        if req_data is not None:
            self.diag_request(req_data)
            self.__diag_test_response_judge(exp=exp)


uds = CanMethod(Parameter.config)

3.27服务安全解锁

dll_security_unlock.exe文件可实现DLL安全解锁，使用方法如下：
语法：
举例：dll_security_unlock.exe --dll_path dome.dll --seed [11,22,33,44] --seedsize 4 --level 1 --keysize 4
–dll_path DLL路径
–seed 请求种子
–seedsize 种子长度
–level 安全级别
–keysize 秘钥长度

4.自动测试用例

test_uds.py主要是自动化测试用例举例，包括10服务测试、11服务测试、14服务测试、19服务测试、22服务测试、28服务测试、31服务测试、85服务测试等，源码如下：

import allure

from public_method.fun_can import uds


class TestDiag:
    @allure.title("诊断测试10服务")
    def test_diag_0x10(self):
        print("#####诊断测试10服务#####")
        uds.diag_test(session="", req_data="10 01", exp="50 01")

    @allure.title("诊断测试11服务")
    def test_diag_0x11(self):
        print("#####诊断测试11服务#####")
        uds.diag_test(session="", req_data="11 01", exp="51 01")

    @allure.title("诊断测试14服务")
    def test_diag_0x14(self):
        print("#####诊断测试14服务#####")
        uds.diag_test(session="03", req_data="14 C0 00 00", exp="54 C0")

    @allure.title("诊断测试19服务")
    def test_diag_0x19(self):
        print("#####诊断测试59服务#####")
        uds.diag_test(session="", req_data="19 01 09", exp="59 01 09")

    @allure.title("诊断测试22服务")
    def test_diag_0x22(self):
        print("#####诊断测试22服务#####")
        uds.diag_test(session="03", req_data="22 60 D7", exp="62 60 D7")

    @allure.title("诊断测试28服务")
    def test_diag_0x28(self):
        print("#####诊断测试28服务#####")
        uds.diag_test(session="03", req_data="28 03 01", exp="68 03")

    @allure.title("诊断测试31服务")
    def test_diag_0x31(self):
        print("#####诊断测试31服务#####")
        uds.diag_test(session="03", req_data="31 01 E0 F5", exp="71 01 E0 F5")

    @allure.title("诊断测试85服务")
    def test_diag_0x85(self):
        print("#####诊断测试85服务#####")
        uds.diag_test(session="03", req_data="85 01", exp="C5 01")

5.配置参数

config主要配置CAN和诊断相关的参数：
interface：配置can设备类型（支持python-can三方库的设备）
channel：通道
bitrate：波特率
addressing_mode：数据比特率
physics_id_default：物理寻址
response_id_default：响应寻址

class Parameter():
    """
    CAN参数配置
    """
    config = {
        "can": {
            "interface": "vector",
            "channel": 0,
            "bitrate": 500000,
            "data_bitrate": 2000000,
            "canfd": False,  # 是否canfd
            "addressing_mode": 0,
            "physics_id_default": 0x56A,
            "response_id_default": 0x56B,
            "function_id_default": 0x56C,
        }
    }