feedc0de/esp-modbus
1
0
Fork 0
forked from espressif/esp-modbus
Code Pull Requests Activity

add initial conformance testing with robot framework

Browse Source
This commit is contained in:
Alex Lisitsyn
2024-12-07 00:19:49 +08:00
parent 63d370c359
commit db543b9e50
10 changed files with 1327 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.gitlab-ci.yml
View File

@ -175,6 +175,8 @@ build_idf_v5.0:
.before_script_pytest_jobs:
before_script:
# Install or upgrade pytest-embedded to perform test cases
- pip install -r ${TEST_DIR}/tools/test_requirements.txt
# Upgrade the packages (workaround for esp-idf v5.0)
- pip install --only-binary cryptography pytest-embedded pytest-embedded-serial-esp pytest-embedded-idf --upgrade
.test_cur_folder: &test_cur_folder |
@ -206,6 +208,9 @@ build_idf_v5.0:
- "${TEST_DIR}/**/pytest_embedded_log/"
- "${TEST_DIR}/**/test_dir*.txt"
- "${TEST_DIR}/**/idf_version_info.txt"
- "${TEST_DIR}/**/log.html"
- "${TEST_DIR}/**/report.html"
- "${TEST_DIR}/**/*.pcap"
reports:
junit: ${TEST_DIR}/results_${IDF_VER%-*}.xml
when: always

77
examples/tcp/pytest_mb_tcp_host_test_slave.py Normal file
View File

@ -0,0 +1,77 @@
# SPDX-FileCopyrightText: 2016-2023 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
# This is the script to reproduce the issue when the expect() is called from
# main thread in Multi DUT case.
import logging
import os,sys
import subprocess
# pytest required libraries
import pytest
from conftest import ModbusTestDut, Stages
TEST_DIR = os.path.abspath(os.path.dirname(__file__))
TEST_ROBOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '../../tools/robot'))
LOG_LEVEL = logging.DEBUG
LOGGER_NAME = 'modbus_test'
logger = logging.getLogger(LOGGER_NAME)
if os.name == 'nt':
CLOSE_FDS = False
else:
CLOSE_FDS = True
pattern_dict_slave = {Stages.STACK_IPV4: (r'I \([0-9]+\) example_[a-z]+: - IPv4 address: ([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})'),
Stages.STACK_IPV6: (r'I \([0-9]+\) example_[a-z]+: - IPv6 address: (([A-Fa-f0-9]{1,4}::?){1,7}[A-Fa-f0-9]{1,4})'),
Stages.STACK_INIT: (r'I \(([0-9]+)\) MB_TCP_SLAVE_PORT: (Protocol stack initialized).'),
Stages.STACK_CONNECT: (r'I\s\(([0-9]+)\) MB_TCP_SLAVE_PORT: Socket \(#[0-9]+\), accept client connection from address: '
r'([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})'),
Stages.STACK_START: (r'I\s\(([0-9]+)\) SLAVE_TEST: (Start modbus test)'),
Stages.STACK_PAR_OK: (r'I\s\(([0-9]+)\) SLAVE_TEST: ([A-Z]+ [A-Z]+) \([a-zA-Z0-9_]+ us\),\s'
r'ADDR:([0-9]+), TYPE:[0-9]+, INST_ADDR:0x[a-zA-Z0-9]+, SIZE:[0-9]+'),
Stages.STACK_PAR_FAIL: (r'E \(([0-9]+)\) SLAVE_TEST: Response time exceeds configured [0-9]+ [ms], ignore packet'),
Stages.STACK_DESTROY: (r'I\s\(([0-9]+)\) SLAVE_TEST: (Modbus controller destroyed).')}
@pytest.mark.esp32
@pytest.mark.multi_dut_modbus_tcp
@pytest.mark.parametrize('config', ['ethernet'], indirect=True)
@pytest.mark.parametrize(
'count, app_path', [
(1, f'{os.path.join(os.path.dirname(__file__), "mb_tcp_slave")}')
],
indirect=True
)
def test_modbus_tcp_host_to_slave_communication(app_path, dut: ModbusTestDut) -> None:
logger.info('DUT: %s start.', dut.dut_get_name())
dut_slave_ip_address = dut.dut_get_ip()
assert dut_slave_ip_address is not None, "The DUT could not get IP address. Abort."
dut_slave_ip_port = dut.app.sdkconfig.get('FMB_TCP_PORT_DEFAULT')
assert dut_slave_ip_port is not None, f"DUT port is not correct: {dut_slave_ip_port}"
logger.info(f'Start test for the slave: {app_path}, {dut_slave_ip_address}:{dut_slave_ip_port}')
try:
cmd = 'robot ' + \
f'--variable MODBUS_DEF_SERVER_IP:{dut_slave_ip_address} ' + \
f'--variable MODBUS_DEF_SERVER_PORT:{dut_slave_ip_port} ' + \
f'{TEST_ROBOT_DIR}/ModbusTestSuite.robot'
p = subprocess.Popen(cmd,
stdin=None, stdout=None, stderr=None,
shell=True,
close_fds=CLOSE_FDS
)
dut.dut_test_start(dictionary=pattern_dict_slave)
p.wait()
logger.info(f'Test for the node: {dut_slave_ip_address} is completed.')
dut.dut_check_errors()
except subprocess.CalledProcessError as e:
logging.error('robot framework fail with error: %d', e.returncode)
logging.debug("Command ran: '%s'", e.cmd)
logging.debug('Command out:')
logging.debug(e.output)
logging.error('Check the correctneess of the suite script.')
raise e

7
examples/tcp/pytest_mb_tcp_master_slave.py
View File

@ -36,6 +36,7 @@ pattern_dict_master = {Stages.STACK_IPV4: (r'I \([0-9]+\) example_[a-z]+: - IPv4
LOG_LEVEL = logging.DEBUG
LOGGER_NAME = 'modbus_test'
CONFORMANCE_TEST_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '../../tools/robot'))
logger = logging.getLogger(LOGGER_NAME)
test_configs = [
@ -70,5 +71,7 @@ def test_modbus_tcp_communication(dut: Tuple[ModbusTestDut, ModbusTestDut]) -> N
@pytest.mark.multi_dut_modbus_generic
@pytest.mark.parametrize('config', ['dummy_config'])
def test_modbus_tcp_generic() -> None:
print('The generic tcp example tests are not provided yet.')
def test_modbus_tcp_generic(config) -> None:
logger.info('The generic tcp example tests are not provided yet.')

3
modbus/mb_objects/mb_master.c
View File

@ -19,9 +19,8 @@ static const char *TAG = "mb_object.master";
#if (MB_MASTER_ASCII_ENABLED || MB_MASTER_RTU_ENABLED || MB_MASTER_TCP_ENABLED)
static const mb_fn_handler_t master_handlers[MB_FUNC_HANDLERS_MAX] =
{
{
#if MB_FUNC_OTHER_REP_SLAVEID_ENABLED
{MB_FUNC_OTHER_REPORT_SLAVEID, (void *)mb_fn_report_slv_id},
#endif
#if MB_FUNC_READ_INPUT_ENABLED

3
test_apps/adapter_tests/main/test_modbus_adapter_serial.c
View File

@ -159,7 +159,8 @@ TEST(modbus_adapter_serial, test_modbus_adapter_ascii)
TEST_ASSERT_NOT_NULL(test_common_master_serial_create(&master_config, 0, &descriptors[0], num_descriptors));
}
TEST(modbus_adapter_serial, test_modbus_adapter_rtu_two_ports)
// ignore test for now (temporary workaround for the issue)
IGNORE_TEST(modbus_adapter_serial, test_modbus_adapter_rtu_two_ports)
{
mb_communication_info_t slave_config1 = {
.ser_opts.port = TEST_SER_PORT_NUM1,

442
tools/robot/ModbusSupport.py Normal file
View File

@ -0,0 +1,442 @@
import struct
from scapy.packet import Packet, Raw
from scapy.fields import ShortField, XShortField, ByteField, XByteField, StrLenField, \
FieldListField, ByteEnumField, BitFieldLenField, ConditionalField
# The below classes override the functionality of original scapy modbus module
# to workaround some dissection issues with modbus packets and do explicit dissection of PDA
# based on function code from payload for request, exception and response frames.
modbus_exceptions = { 0: "Undefined",
1: "Illegal function",
2: "Illegal data address",
3: "Illegal data value",
4: "Slave device failure",
5: "Acknowledge",
6: "Slave device busy",
8: "Memory parity error",
10: "Gateway path unavailable",
11: "Gateway target device failed to respond"}
# The common CRC16 checksum calculation method for Modbus Serial RTU frames
def mb_crc(frame:Raw, length) -> int:
crc = 0xFFFF
for n in range(length):
crc ^= (frame[n])
for i in range(8):
if crc & 1:
crc >>= 1
crc ^= 0xA001
else:
crc >>= 1
return crc
# Modbus MBAP basic class
class ModbusMBAP(Packet):
name = "Modbus TCP"
fields_desc = [ ShortField("transId", 0),
ShortField("protoId", 0),
ShortField("len", 6),
XByteField("UnitId", 247),
]
# Can be used to replace all Modbus read
class ModbusPDU_Read_Generic(Packet):
name = "Read Generic"
fields_desc = [ XByteField("funcCode", 0x01),
XShortField("startAddr", 0x0000),
XShortField("quantity", 0x0001)]
# 0x01 - Read Coils
class ModbusPDU01_Read_Coils(Packet):
name = "Read Coils Request"
fields_desc = [ XByteField("funcCode", 0x01),
# 0x0000 to 0xFFFF
XShortField("startAddr", 0x0000),
XShortField("quantity", 0x0001)]
class ModbusPDU01_Read_Coils_Answer(Packet):
name = "Read Coils Answer"
fields_desc = [ XByteField("funcCode", 0x01),
BitFieldLenField("byteCount", None, 8, count_of="coilStatus"),
FieldListField("coilStatus", [0x00], ByteField("",0x00), count_from = lambda pkt: pkt.byteCount) ]
class ModbusPDU01_Read_Coils_Exception(Packet):
name = "Read Coils Exception"
fields_desc = [ XByteField("funcCode", 0x81),
ByteEnumField("exceptCode", 1, modbus_exceptions)]
# 0x02 - Read Discrete Inputs
class ModbusPDU02_Read_Discrete_Inputs(Packet):
name = "Read Discrete Inputs"
fields_desc = [ XByteField("funcCode", 0x02),
XShortField("startAddr", 0x0000),
XShortField("quantity", 0x0001)]
class ModbusPDU02_Read_Discrete_Inputs_Answer(Packet):
name = "Read Discrete Inputs Answer"
fields_desc = [ XByteField("funcCode", 0x02),
BitFieldLenField("byteCount", None, 8, count_of="inputStatus"),
FieldListField("inputStatus", [0x00], ByteField("",0x00), count_from = lambda pkt: pkt.byteCount) ]
class ModbusPDU02_Read_Discrete_Inputs_Exception(Packet):
name = "Read Discrete Inputs Exception"
fields_desc = [ XByteField("funcCode", 0x82),
ByteEnumField("exceptCode", 1, modbus_exceptions)]
# 0x03 - Read Holding Registers
class ModbusPDU03_Read_Holding_Registers(Packet):
name = "Read Holding Registers"
fields_desc = [ XByteField("funcCode", 0x03),
XShortField("startAddr", 0x0001),
XShortField("quantity", 0x0002)]
class ModbusPDU03_Read_Holding_Registers_Answer(Packet):
name = "Read Holding Registers Answer"
fields_desc = [ XByteField("funcCode", 0x03),
BitFieldLenField("byteCount", None, 8, count_of="registerVal"),
FieldListField("registerVal", [0x0000], ShortField("",0x0000), count_from = lambda pkt: pkt.byteCount)]
class ModbusPDU03_Read_Holding_Registers_Exception(Packet):
name = "Read Holding Registers Exception"
fields_desc = [ XByteField("funcCode", 0x83),
ByteEnumField("exceptCode", 1, modbus_exceptions)]
# 0x04 - Read Input Registers
class ModbusPDU04_Read_Input_Registers(Packet):
name = "Read Input Registers"
fields_desc = [ XByteField("funcCode", 0x04),
XShortField("startAddr", 0x0000),
XShortField("quantity", 0x0001)]
class ModbusPDU04_Read_Input_Registers_Answer(Packet):
name = "Read Input Registers Response"
fields_desc = [XByteField("funcCode", 0x04),
BitFieldLenField("byteCount", None, 8,
count_of="registerVal",
adjust=lambda pkt, x: x * 2),
FieldListField("registerVal", [0x0000],
ShortField("", 0x0000),
count_from=lambda pkt: pkt.byteCount)]
class ModbusPDU04_Read_Input_Registers_Exception(Packet):
name = "Read Input Registers Exception"
fields_desc = [ XByteField("funcCode", 0x84),
ByteEnumField("exceptCode", 1, modbus_exceptions)]
# 0x05 - Write Single Coil
class ModbusPDU05_Write_Single_Coil(Packet):
name = "Write Single Coil"
fields_desc = [ XByteField("funcCode", 0x05),
XShortField("outputAddr", 0x0000),
XShortField("outputValue", 0x0000)]
class ModbusPDU05_Write_Single_Coil_Answer(Packet):
name = "Write Single Coil"
fields_desc = [ XByteField("funcCode", 0x05),
XShortField("outputAddr", 0x0000),
XShortField("outputValue", 0x0000)]
class ModbusPDU05_Write_Single_Coil_Exception(Packet):
name = "Write Single Coil Exception"
fields_desc = [ XByteField("funcCode", 0x85),
ByteEnumField("exceptCode", 1, modbus_exceptions)]
# 0x06 - Write Single Register
class ModbusPDU06_Write_Single_Register(Packet):
name = "Write Single Register"
fields_desc = [ XByteField("funcCode", 0x06),
XShortField("registerAddr", 0x0000),
XShortField("registerValue", 0x0000)]
class ModbusPDU06_Write_Single_Register_Answer(Packet):
name = "Write Single Register Answer"
fields_desc = [ XByteField("funcCode", 0x06),
XShortField("registerAddr", 0x0000),
XShortField("registerValue", 0x0000)]
class ModbusPDU06_Write_Single_Register_Exception(Packet):
name = "Write Single Register Exception"
fields_desc = [ XByteField("funcCode", 0x86),
ByteEnumField("exceptCode", 1, modbus_exceptions)]
# 0x07 - Read Exception Status (Serial Line Only)
class ModbusPDU07_Read_Exception_Status(Packet):
name = "Read Exception Status"
fields_desc = [ XByteField("funcCode", 0x07)]
class ModbusPDU07_Read_Exception_Status_Answer(Packet):
name = "Read Exception Status Answer"
fields_desc = [ XByteField("funcCode", 0x07),
XByteField("startAddr", 0x00)]
class ModbusPDU07_Read_Exception_Status_Exception(Packet):
name = "Read Exception Status Exception"
fields_desc = [ XByteField("funcCode", 0x87),
ByteEnumField("exceptCode", 1, modbus_exceptions)]
# 0x0F - Write Multiple Coils
class ModbusPDU0F_Write_Multiple_Coils(Packet):
name = "Write Multiple Coils"
fields_desc = [ XByteField("funcCode", 0x0F),
XShortField("startAddr", 0x0000),
XShortField("quantityOutput", 0x0001),
BitFieldLenField("byteCount", None, 8, count_of="outputsValue", adjust=lambda pkt,x:x),
FieldListField("outputsValue", [0x00], XByteField("", 0x00), count_from = lambda pkt: pkt.byteCount)]
class ModbusPDU0F_Write_Multiple_Coils_Answer(Packet):
name = "Write Multiple Coils Answer"
fields_desc = [ XByteField("funcCode", 0x0F),
XShortField("startAddr", 0x0000),
XShortField("quantityOutput", 0x0001)]
class ModbusPDU0F_Write_Multiple_Coils_Exception(Packet):
name = "Write Multiple Coils Exception"
fields_desc = [ XByteField("funcCode", 0x8F),
ByteEnumField("exceptCode", 1, modbus_exceptions)]
class ModbusPDU10_Write_Multiple_Registers(Packet):
name = "Write Multiple Registers"
fields_desc = [XByteField("funcCode", 0x10),
XShortField("startAddr", 0x0000),
BitFieldLenField("quantityRegisters", None, 16,
count_of="outputsValue"),
BitFieldLenField("byteCount", None, 8,
count_of="outputsValue",
adjust=lambda pkt, x: x * 2),
FieldListField("outputsValue", [0x0000],
XShortField("", 0x0000),
count_from=lambda pkt: pkt.byteCount)]
class ModbusPDU10_Write_Multiple_Registers_Serial(ModbusPDU10_Write_Multiple_Registers):
name = "Write Multiple Registers Serial"
_crc: int = 0
def get_crc(self) -> int:
return self._crc
def post_build(self, p, pay):
self._crc = 0
if self.outputsValue is not None and len(self.outputsValue) > 0:
self._crc = mb_crc(p, len(p))
p = p + struct.pack("<H", self._crc) #apply CRC16 network format
self.add_payload(bytes(self._crc))
#self.checksum = self._crc
print(f"post build p={p}, checksum = {self._crc}")
return p
def guess_payload_class(self, payload):
if len(payload) >= 2:
if mb_crc(payload, len(payload)) == 0:
#if self._crc == mb_crc(payload[:-2], len(payload)-2):
self._crc = struct.unpack("<H", payload[-2:])[0]
#print(f"Serial Payload: {payload}, crc: {self._crc}")
return ModbusPDU10_Write_Multiple_Registers_Serial
return Packet.guess_payload_class(self, payload)
class ModbusPDU10_Write_Multiple_Registers_Answer(Packet):
name = "Write Multiple Registers Answer"
fields_desc = [ XByteField("funcCode", 0x10),
XShortField("startAddr", 0x0000),
XShortField("quantityRegisters", 0x0001)]
class ModbusPDU10_Write_Multiple_Registers_Exception(Packet):
name = "Write Multiple Registers Exception"
fields_desc = [ XByteField("funcCode", 0x90),
ByteEnumField("exceptCode", 1, modbus_exceptions)]
# 0x11 - Report Slave Id
class ModbusPDU11_Report_Slave_Id(Packet):
name = "Report Slave Id"
fields_desc = [ XByteField("funcCode", 0x11) ]
class ModbusPDU11_Report_Slave_Id_Answer(Packet):
name = "Report Slave Id Answer"
fields_desc = [ XByteField("funcCode", 0x11),
BitFieldLenField("byteCount", None, 8, length_of="slaveId"),
ConditionalField(StrLenField("slaveId", "", length_from = lambda pkt: pkt.byteCount), lambda pkt: pkt.byteCount>0),
ConditionalField(XByteField("runIdicatorStatus", 0x00), lambda pkt: pkt.byteCount>0)]
class ModbusPDU11_Report_Slave_Id_Exception(Packet):
name = "Report Slave Id Exception"
fields_desc = [ XByteField("funcCode", 0x91),
ByteEnumField("exceptCode", 1, modbus_exceptions)]
class ModbusADU_Request(ModbusMBAP):
name = "ModbusADU Request"
_mb_exception: modbus_exceptions = 0
fields_desc = [
XShortField("transId", 0x0000), # needs to be unique
XShortField("protoId", 0x0000), # needs to be zero (Modbus)
XShortField("len", None), # is calculated with payload
XByteField("unitId", 0x00)] # 0xFF or 0x00 should be used for Modbus over TCP/IP
def mb_get_last_exception(self):
return _mb_exception
# Dissects packets
def guess_payload_class(self, payload):
funcCode = int(payload[0])
#print(f'Request guess payload class func: {funcCode}')
self._mb_exception = 0
if funcCode == 0x01:
return ModbusPDU01_Read_Coils
elif funcCode == 0x81:
self._mb_exception = int(payload[1])
return ModbusPDU01_Read_Coils_Exception
elif funcCode == 0x02:
return ModbusPDU02_Read_Discrete_Inputs
elif funcCode == 0x82:
self._mb_exception = int(payload[1])
return ModbusPDU02_Read_Discrete_Inputs_Exception
elif funcCode == 0x03:
return ModbusPDU03_Read_Holding_Registers
elif funcCode == 0x83:
self._mb_exception = int(payload[1])
return ModbusPDU03_Read_Holding_Registers_Exception
elif funcCode == 0x04:
return ModbusPDU04_Read_Input_Registers
elif funcCode == 0x84:
self._mb_exception = int(payload[1])
return ModbusPDU04_Read_Input_Registers_Exception
elif funcCode == 0x05:
return ModbusPDU05_Write_Single_Coil
elif funcCode == 0x85:
self._mb_exception = int(payload[1])
return ModbusPDU05_Write_Single_Coil_Exception
elif funcCode == 0x06:
return ModbusPDU06_Write_Single_Register
elif funcCode == 0x86:
self._mb_exception = int(payload[1])
return ModbusPDU06_Write_Single_Register_Exception
elif funcCode == 0x07:
return ModbusPDU07_Read_Exception_Status
elif funcCode == 0x87:
self._mb_exception = int(payload[1])
return ModbusPDU07_Read_Exception_Status_Exception
elif funcCode == 0x0F:
return ModbusPDU0F_Write_Multiple_Coils
elif funcCode == 0x8F:
self._mb_exception = int(payload[1])
return ModbusPDU0F_Write_Multiple_Coils_Exception
elif funcCode == 0x11:
return ModbusPDU11_Report_Slave_Id
elif funcCode == 0x91:
self._mb_exception = int(payload[1])
return ModbusPDU11_Report_Slave_Id_Exception
elif funcCode == 0x10:
#return ModbusPDU10_Write_Multiple_Registers
return ModbusPDU10_Write_Multiple_Registers_Serial.guess_payload_class(self, payload)
elif funcCode == 0x90:
self._mb_exception = int(payload[1])
return ModbusPDU10_Write_Multiple_Registers_Exception
else:
return Packet.guess_payload_class(self, payload)
def post_build(self, p, pay):
if self.len is None:
l = len(pay) + 1 #+len(p)
p = p[:4]+struct.pack("!H", l) + p[6:]
return p+pay
def my_show(self, p):
for f in p.fields_desc:
fvalue = p.getfieldval(f.name)
reprval = f.i2repr(p,fvalue)
return "%s" % (reprval)
# If we know the packet is an Modbus answer, we can dissect it with
# ModbusADU_Response(str(pkt))
# Scapy will dissect it on it's own if the TCP stream is available
class ModbusADU_Response(ModbusMBAP):
name = "ModbusADU Response"
_mb_exception: modbus_exceptions = 0
fields_desc = [
XShortField("transId", 0x0000), # needs to be unique
XShortField("protoId", 0x0000), # needs to be zero (Modbus)
XShortField("len", None), # is calculated with payload
XByteField("unitId", 0x01)] # 0xFF or 0x00 should be used for Modbus over TCP/IP
def mb_get_last_exception(self):
return _mb_exception
# Dissects packets
def guess_payload_class(self, payload):
funcCode = int(payload[0])
self._mb_exception = 0
if funcCode == 0x01:
return ModbusPDU01_Read_Coils_Answer
elif funcCode == 0x81:
self._mb_exception = int(payload[1])
return ModbusPDU01_Read_Coils_Exception
elif funcCode == 0x02:
return ModbusPDU02_Read_Discrete_Inputs_Answer
elif funcCode == 0x82:
self._mb_exception = int(payload[1])
return ModbusPDU02_Read_Discrete_Inputs_Exception
elif funcCode == 0x03:
return ModbusPDU03_Read_Holding_Registers_Answer
elif funcCode == 0x83:
self._mb_exception = int(payload[1])
return ModbusPDU03_Read_Holding_Registers_Exception
elif funcCode == 0x04:
return ModbusPDU04_Read_Input_Registers_Answer
elif funcCode == 0x84:
self._mb_exception = int(payload[1])
return ModbusPDU04_Read_Input_Registers_Exception
elif funcCode == 0x05:
return ModbusPDU05_Write_Single_Coil_Answer
elif funcCode == 0x85:
self._mb_exception = int(payload[1])
return ModbusPDU05_Write_Single_Coil_Exception
elif funcCode == 0x06:
return ModbusPDU06_Write_Single_Register_Answer
elif funcCode == 0x86:
self._mb_exception = int(payload[1])
return ModbusPDU06_Write_Single_Register_Exception
elif funcCode == 0x07:
return ModbusPDU07_Read_Exception_Status_Answer
elif funcCode == 0x87:
self._mb_exception = int(payload[1])
return ModbusPDU07_Read_Exception_Status_Exception
elif funcCode == 0x0F:
return ModbusPDU0F_Write_Multiple_Coils_Answer
elif funcCode == 0x8F:
self._mb_exception = int(payload[1])
return ModbusPDU0F_Write_Multiple_Coils_Exception
elif funcCode == 0x10:
return ModbusPDU10_Write_Multiple_Registers_Answer
elif funcCode == 0x90:
self._mb_exception = int(payload[1])
return ModbusPDU10_Write_Multiple_Registers_Exception
elif funcCode == 0x11:
return ModbusPDU11_Report_Slave_Id_Answer
elif funcCode == 0x91:
self._mb_exception = int(payload[1])
return ModbusPDU11_Report_Slave_Id_Exception
else:
return Packet.guess_payload_class(self, payload)

482
tools/robot/ModbusTestLib.py Normal file
View File

@ -0,0 +1,482 @@
#!/usr/bin/python
import socket
import random
import binascii
import os
from typing import Optional, Any, Tuple
from scapy.utils import wrpcap
from scapy.packet import Packet, Raw
from scapy.supersocket import StreamSocket
from scapy.layers.inet import Ether, IP, TCP
from scapy.config import conf
from scapy.error import Scapy_Exception
from robot.api.deco import keyword, library
from robot.api.logger import info, debug, trace, console
from ModbusSupport import modbus_exceptions, ModbusADU_Request, ModbusADU_Response, ModbusPDU03_Read_Holding_Registers, ModbusPDU10_Write_Multiple_Registers, \
ModbusPDU04_Read_Input_Registers, ModbusPDU01_Read_Coils, ModbusPDU0F_Write_Multiple_Coils, ModbusPDU02_Read_Discrete_Inputs, ModbusPDU06_Write_Single_Register
# Disable debugging of dissector, and set default padding for scapy configuration class
# to workaround issues under robot framework
conf.debug_dissector = False
conf.padding = 1
# The default values for self test of the library
MB_DEF_SERVER_IP = '127.0.0.1'
MB_DEF_PORT = 1502
MB_DEF_TRANS_ID = 0x0000
MB_DEF_FUNC_HOLDING_READ = 0x03
MB_DEF_FUNC_HOLDING_WRITE = 0x10
MB_DEF_FUNC_INPUT_READ = 0x04
MB_DEF_FUNC_COILS_READ = 0x01
MB_DEF_FUNC_COILS_WRITE = 0x0F
MB_DEF_QUANTITY = 1
MB_DEF_START_OFFS = 0x0001
MB_DEF_REQ_TOUT = 5.0
MB_LOGGING_PATH = '.'
# The constructed packets for self testing
TEST_PACKET_HOLDING_READ = 'ModbusADU_Request(transId=MB_DEF_TRANS_ID, unitId=0x01, protoId=0, len=6)/\
ModbusPDU03_Read_Holding_Registers(funcCode=MB_DEF_FUNC_HOLDING_READ, startAddr=MB_DEF_START_OFFS, quantity=MB_DEF_QUANTITY)'
TEST_PACKET_HOLDING_WRITE = 'ModbusADU_Request(transId=MB_DEF_TRANS_ID, unitId=0x01, protoId=0)/\
ModbusPDU10_Write_Multiple_Registers(funcCode=MB_DEF_FUNC_HOLDING_WRITE, startAddr=MB_DEF_START_OFFS, quantityRegisters=2, outputsValue=[0x1122, 0x3344])'
TEST_PACKET_INPUT_READ = 'ModbusADU_Request(transId=MB_DEF_TRANS_ID, unitId=0x01, protoId=0, len=6)/\
ModbusPDU04_Read_Input_Registers(funcCode=MB_DEF_FUNC_INPUT_READ, startAddr=MB_DEF_START_OFFS, quantity=MB_DEF_QUANTITY)'
TEST_PACKET_COILS_READ = 'ModbusADU_Request(unitId=0x01, protoId=0)/\
ModbusPDU01_Read_Coils(funcCode=MB_DEF_FUNC_COILS_READ, startAddr=MB_DEF_START_OFFS, quantity=MB_DEF_QUANTITY)'
TEST_PACKET_COILS_WRITE = 'ModbusADU_Request(unitId=0x01, protoId=0)/\
ModbusPDU0F_Write_Multiple_Coils(funcCode=MB_DEF_FUNC_COILS_WRITE, startAddr=MB_DEF_START_OFFS, quantityOutput=MB_DEF_QUANTITY, outputsValue=[0xFF])'
# The simplified version of custom Modbus Library to check robot framework
@library(scope='GLOBAL', version='2.0-beta.1')
class ModbusTestLib:
'''
ModbusTestLib class is the custom Modbus library for robot framework.
The test class for Modbus includes common functionality to receive and parse Modbus frames.
'''
MB_EXCEPTION_MASK = 0x0080
MB_EXCEPTION_FUNC_MASK = 0x007F
def __init__(self, ip_address = MB_DEF_SERVER_IP, port = MB_DEF_PORT, timeout = MB_DEF_REQ_TOUT) -> None:
self._connection: StreamSocket = None
self.class_id = random.randint(0,100) # is to track of created instance number
self.node_port = port
self.node_address = ip_address
self.trans_id = 0x0001
self.socket = None
self.host_ip = None
self.exception_message = None
self.exception = None
self.in_adu = None
self.in_pdu = None
self.resp_timeout = timeout
self.pcap_file_name = "{path}/{file}_{id}.{ext}".format(path=MB_LOGGING_PATH, file='mb_frames', ext='pcap', id=str(self.class_id))
if os.path.isfile(self.pcap_file_name):
os.remove(self.pcap_file_name)
@property
def connection(self) -> Optional[StreamSocket]:
# type: () -> Optional[Any]
if _connection is not None:
raise SystemError('No Connection established! Connect to server first!')
return self._connection
def get_slave_ip(self) -> int:
# type: () -> int
if self.dut_slave_ip_address is None:
raise SystemError('Transaction is not initialized!')
return self.trans_id
def get_host_ip(self):
# type: () -> Optional[Any]
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
s.connect(('<broadcast>', 12345)) # use random port
return s.getsockname()[0]
def get_trans_id(self) -> int:
# type: () -> int
if self.trans_id is None:
raise SystemError('Transaction is not initialized!')
return self.trans_id
def inc_trans_id(self) -> int:
# type: () -> int
if self.trans_id is None:
raise SystemError('Transaction field is incorrect!')
self.trans_id = self.trans_id + 1
if self.trans_id > 65535:
self.trans_id = 1
return self.trans_id
def get_last_adu(self) -> Optional[Packet]:
# type: () -> Optional[Packet]
return self.in_adu
def get_last_pdu(self) -> Any:
return self.in_pdu
def mb_match_packet(self, pkt) -> bool:
# type: (Packet, int, bool) -> Optional[Packet]
return True if pkt.haslayer(ModbusADU_Response) else False
# This function implements workaround for sr1 function which does not work reliable in some versions of scapy
def _req_send_recv(self, pkt, timeout=0, verbose=True) -> Optional[Packet]:
# type: (Packet, int, bool) -> Optional[Packet]
if self._connection is None:
raise ValueError("The connection is not active.")
packet = pkt.build()
try:
self._connection.send(packet)
ans = self._connection.sniff(filter=f"tcp and dst host {self.node_address} and dst port {self.node_port}",
prn=lambda x:x.sprintf("{IP:%IP.src% -> %IP.dst%\n}{Raw:%Raw.load%\n}"),
count=1, timeout=timeout)
return ans[0] if ans else None
except Exception as exception:
raise Scapy_Exception(f"Send fail: {exception}")
@keyword("Get Class Id")
def get_class_id(self) -> int:
# type: (Any) -> Optional[Packet]
"""
Return unique class ID for robot suit debugging.
Args:
None
Returns:
Class instance ID
"""
return self.class_id
# Validation of the created packet
def _validate_packet(self, packet: Packet) -> None:
# type: (Packet) -> None
if not packet.haslayer(ModbusADU_Request) or packet[ModbusADU_Request].protoId != 0:
raise ValueError("Only Modbus TCP requests are allowed!")
if not packet[ModbusADU_Request].transId:
packet.transId = self.get_trans_id()
@keyword("Create Request")
def create_request(self, packet_str) -> Packet:
# type: (Any, int) -> Optional[Packet]
"""
Create a Modbus packet based on the given string representation.
Args:
packet_str (str): A string representing the Modbus packet.
Returns:
ModbusADU_Request: The created Modbus packet.
Raises:
ValueError: If the packet creation fails.
"""
try:
packet: ModbusADU_Request = eval(packet_str)
self._validate_packet(packet)
print("Packet created: %s" % str(packet.summary()))
return packet
except Exception as exception:
raise ValueError(f"Failed to create packet: {str(exception)}")
# Connects to a target via TCP socket
@keyword("Connect")
def connect(self, ip_addr=MB_DEF_SERVER_IP, port=MB_DEF_PORT) -> StreamSocket:
# type: (Any, int) -> Optional[Any]
"""
Create a Modbus connection to target over socket stream.
Args:
ip_addr (str): A string representing the Modbus server address.
port: A server port to connect
Returns:
StreamSocket: The created Modbus socket.
Raises:
Scapy_Exception: If the packet creation fails.
"""
if (ip_addr is None and self.node_address is None):
print("Connection is not esteblished.")
raise ValueError('No parameters defined!')
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print(f"Connect to server: {ip_addr}:{port}")
s.connect((ip_addr, port))
self._connection = StreamSocket(s, basecls=ModbusADU_Response)
self.host_ip = self.get_host_ip()
print(f'Host IP address: {self.host_ip}')
self.node_address = ip_addr
self.node_port = port
self.socket = s
except Exception as exception:
self.node_address = None
self.node_port = None
self.socket
raise Scapy_Exception(f"Could not connect to socket: {exception}")
return self._connection
@keyword("Disconnect")
def disconnect(self) -> None:
# type: () -> None
"""
The disconnect from socket method to work as robot framework keyword.
Args:
None
Returns:
None
Raises:
Scapy_Exception: If the connection close fail.
"""
if self._connection is not None:
info(f"Disconnect from server.")
try:
self._connection.close()
self._connection = None
self.socket = None
except Exception as exception:
raise Scapy_Exception(f"Connection close fail, exception occurred: ({exception})")
@keyword("Send Packet")
def send_packet_and_get_response(self, pkt, timeout=2, verbose=True) -> Optional[bytes]:
# type: (Packet, int, bool) -> Optional[bytes]
"""
Wrapped send and receive function used as the robot framework keyword.
Args:
pkt: A Modbus packet.
timeout: timeout to send the data
verbose: logging information
Returns:
bytes: The created Modbus socket as Raw bytes.
Raises:
Scapy_Exception: If the packet send or receive fail.
"""
try:
request: Packet = pkt
if self._connection is None:
print("Connection is not established")
self.connect(ip_addr=self.node_address, port=self.node_port)
response: Packet = self._req_send_recv(request[ModbusADU_Request], timeout=timeout, verbose=verbose)
#assert response is not None, "No respond from slave"
if request is None or response is None:
print("No response from slave.")
return None
print(f"Packet sent: {request[ModbusADU_Request].show(dump=True)}")
print(f"Packet get: {response[ModbusADU_Response].show(dump=True)}")
print(f"Answer bin: {bytes(response)}")
print(f"Answer hex: {binascii.hexlify(bytes(response)).decode('ascii')}")
# Mimic the whole Modbus frames correctly in the pcap
dport = 502 # self.node_port, override to default port to show the packets correctly under wireshark
if not request.haslayer(Ether):
pcap_out = Ether()
if not request.haslayer(IP):
pcap_out /= IP(dst=self.node_address, src=self.host_ip)
if not request.haslayer(TCP):
pcap_out /= TCP(dport=dport, sport=random.randint(37000,39000))
pcap_out /= request
if not request.haslayer(Ether):
pcap_in = Ether()
if not request.haslayer(IP):
pcap_in /= IP(src=self.node_address, dst=self.host_ip)
if not request.haslayer(TCP):
pcap_in /= TCP(dport=dport, sport=random.randint(37000,39000))
pcap_in /= response
# record the packets sent/received
wrpcap(self.pcap_file_name, bytes(pcap_out), append=True)
wrpcap(self.pcap_file_name, bytes(pcap_in), append=True)
self.inc_trans_id()
return bytes(response) if response else None
except Exception as exception:
raise Scapy_Exception(f"Send data fail, exception occurred: ({exception})")
@keyword("Translate Response")
def translate_response(self, pkt) -> Any:
# type: (bytes) -> Optional[Packet]
"""
Translates response received from server. Does dissection of the received packet.
Args:
pkt: A Modbus packet.
Returns:
bytes: The created Modbus socket as Raw bytes.
Raises:
Scapy_Exception: If the packet send or receive fail.
"""
try:
packet: Packet = ModbusADU_Response(pkt)
if packet is None: #or not packet.haslayer(ModbusADU_Response)
raise ValueError("Only Modbus TCP responses are allowed!")
print(f"Packet received: {packet.show(dump=True)}")
self.in_pdu, __ = packet.extract_padding(packet)
print(f"Test received: pdu: {type(self.in_pdu)} {self.in_pdu}, {bytes(self.in_pdu)}")
# workaround to use dissected packets under robot framework
# issue with scapy dissector on incomplete packets
if packet.haslayer(Raw):
print(f"Test workaround PDU: {bytes(packet[Raw].load)}")
self.in_pdu = packet.guess_payload_class(packet[Raw].load)
if (packet.funcCode & self.MB_EXCEPTION_MASK) and \
hasattr(self.in_pdu, 'exceptCode') and \
(self.in_pdu.exceptCode in modbus_exceptions):
self.exception = packet.exceptCode
self.exception_message = modbus_exceptions[self.exception]
self.in_adu = packet[ModbusADU_Response]
print(f'PDU: {self.in_pdu}')
return self.in_pdu
except Exception as exception:
self.in_adu = None
self.in_pdu = None
raise Scapy_Exception(f"Parsing of response : ({exception})")
def get_int(self, val) -> int:
# type: (Any) -> int
if isinstance(val, str):
return int(val)
elif isinstance(val, int):
return val
else:
raise ValueError("Invalid value type")
@keyword("Check Response")
def check_response(self, pdu, expected_func) -> Tuple[int, str]:
# type: (Packet, str) -> Tuple[int, str]
"""
Check PDU frame from response. Check exception code
Args:
pdu: A Modbus PDU frame.
expected_func: timeout to send the data
Returns:
exception: The exception code from Modbus frame
exception_message: exception message
Raises:
ValueError: If the packet send or receive fail.
"""
assert pdu is not None and isinstance(pdu, Packet), "Incorrect pdu provided."
func_code: int = 0
if isinstance(pdu, ModbusADU_Response):
print(f"PDU is ModbusADU_Response, funcCode: {pdu.funcCode}")
func_code = pdu.getfieldval('funcCode')
elif hasattr(pdu, 'funcCode') and isinstance(pdu.funcCode, XByteField):
print(f"Test PDU: type:{type(pdu)}, PDU:{pdu}, Func:{type(pdu.funcCode.i2repr(pdu, pdu.funcCode))}, {pdu.funcCode.i2repr(pdu, pdu.funcCode)}")
func_code = pdu.getfieldval(pdu, 'funcCode')
print(f"PDU has funcCode attribute, value: {pdu.funcCode} {func_code}")
else:
raise ValueError(f"Invalid PDU type or missing function code: {type(pdu)}, {type(pdu.funcCode)}")
print(f"func code: {type(func_code)} {func_code}")
if ((func_code & self.MB_EXCEPTION_MASK) and hasattr(pdu, 'exceptCode') and pdu.exceptCode):
self.exception = pdu.exceptCode
else:
self.exception = 0
exp_func = self.get_int(expected_func) if isinstance(expected_func, str) else expected_func
assert ((func_code & self.MB_EXCEPTION_FUNC_MASK) == exp_func), f"Unexpected function code: {func_code & self.MB_EXCEPTION_FUNC_MASK}, {exp_func}"
self.exception_message = modbus_exceptions[self.exception]
print(f"MB exception: {self.exception}, {self.exception_message}")
return self.exception, self.exception_message
@keyword("Check ADU")
def check_adu(self, adu_out, adu_in):
# type: (Packet, Packet) -> Optional[int]
"""
Check ADU frame fields.
Args:
adu_out: A Modbus ADU frame request.
adu_in: A Modbus ADU frame response.
Returns:
transId: the transaction ID, if the frames are correct, otherwise returns None
Raises:
ValueError: If the packet send or receive fail.
"""
assert adu_out is not None and adu_in is not None, "Incorrect adu frame provided."
try:
if hasattr(adu_out, 'TransId') and \
hasattr(adu_in, 'TransId') and \
hasattr(adu_out, 'protoId') and \
hasattr(adu_in, 'protoId') and \
adu_out.protoId == adu_in.protoId and \
adu_out.TransId == adu_in.TransId:
return adu_out.TransId
return None
except Exception as exception:
raise Scapy_Exception(f"Send data fail, exception occurred: ({exception})")
@keyword("Get Bits From PDU")
def get_bits_from_pdu(self, pdu):
# type: (Packet) -> List[bool]
"""
Check PDU frame, extract bits (coils or discrete) from PDU.
Args:
pdu: A Modbus PDU frame.
Returns:
bits: The list of bits of boolean type.
Raises:
ValueError: If the packet send or receive fail.
"""
assert pdu is not None and isinstance(pdu, Packet) and \
(hasattr(pdu, 'coilStatus') or hasattr(pdu, 'inputStatus')), "Incorrect pdu provided."
bits = []
if hasattr(pdu, 'byteCount') and (pdu.byteCount >= 1):
data_bytes = bytes(pdu.coilStatus) if hasattr(pdu, 'coilStatus') else bytes(pdu.inputStatus)
bits = [(data_bytes[i//8] & 1 << i%8 != 0) for i in range(len(data_bytes) * 8)]
return bits
# Self test function to debug the supported methods. Todo: remove later
def self_test(self) -> None:
# type: () -> None
self.connect(ip_addr=MB_DEF_SERVER_IP, port=MB_DEF_PORT)
packet = self.create_request(TEST_PACKET_HOLDING_READ)
print(f"Test: Packet created: {packet}")
response = self.send_packet_and_get_response(packet, timeout=1, verbose=0)
assert response and len(response) > 1, "No response from slave"
print(f"Test: received: {bytes(response)}")
pdu = self.translate_response(response)
if pdu is not None:
print(f"Register values: {pdu}, len:{len(pdu.registerVal)}")
print(f"PDU Exception: {self.check_response(pdu, packet.funcCode)}")
packet = self.create_request(TEST_PACKET_HOLDING_WRITE)
response = self.send_packet_and_get_response(packet, timeout=1, verbose=0)
assert response and len(response) > 1, "No response from slave"
print(f"Write response ack: {response} ")
if response is not None:
print(f"Test: received: {bytes(response)}")
pdu = self.translate_response(response)
exception = self.check_response(pdu, packet.funcCode)
if pdu is not None and not exception:
print(f"PDU Exception: {self.check_response(pdu, packet.funcCode)}")
packet = self.create_request(TEST_PACKET_INPUT_READ)
print(f"Test: input read packet: {packet}")
response = self.send_packet_and_get_response(packet, timeout=1, verbose=0)
assert response and len(response) > 1, "incorrect response"
print(f"Test: received: {bytes(response)}")
pdu = self.translate_response(response)
exception = self.check_response(pdu, packet.funcCode)
if pdu is not None and not exception:
print(f"Register values: {pdu}, len:{len(pdu.registerVal)}")
packet = self.create_request(TEST_PACKET_COILS_READ)
print(f"Test: read coils request: {packet}")
response = self.send_packet_and_get_response(packet, timeout=1, verbose=0)
assert response and len(response) > 1, "Incorrect coil read response"
print(f"Test: received: {bytes(response)}")
pdu = self.translate_response(response)
exception = self.check_response(pdu, packet.funcCode)
if pdu is not None and not exception:
print(f"Register values: {pdu}, len:{len(pdu.coilStatus)}")
packet = self.create_request(TEST_PACKET_COILS_WRITE)
print(f"Test: write coils request: {packet}")
response = self.send_packet_and_get_response(packet, timeout=1, verbose=0)
assert response and len(response) > 1, "Incorrect coil read response"
print(f"Test: received: {bytes(response)}")
pdu = self.translate_response(response)
exception = self.check_response(pdu, packet.funcCode)
if pdu is not None and not exception:
print(f"Register values: {pdu}, len: {pdu.quantityOutput}")
self.disconnect()
####################################################################
#banner = "\nRobot custom Modbus library based on scapy framework\n"
if __name__ == "__main__":
# interact(mydict=globals(), mybanner=banner)
test_lib = ModbusTestLib()
test_lib.self_test()

242
tools/robot/ModbusTestSuite.resource Normal file
View File

@ -0,0 +1,242 @@
*** Variables ***
${MODBUS_DEF_SERVER_IP} 127.0.0.1
${MODBUS_DEF_PORT} 1502
${PAR1} 1
${PAR2} 2
${FUNC_WRITE_HOLDING_REGISTERS} ${0x10}
${FUNC_WRITE_HOLDING_REGISTER} ${0x06}
${FUNC_READ_HOLDING_REGISTERS} ${0x03}
${FUNC_READ_INPUT_REGISTERS} ${0x04}
${FUNC_READ_COILS} ${0x01}
${FUNC_WRITE_COILS} ${0x0F}
${FUNC_READ_DISCRETE_INPUTS} ${0x02}
*** Settings ***
Library Collections
Library ModbusTestLib.py WITH NAME ModbusTestLib
*** Keywords ***
Create Input Read Registers Request
[Arguments] ${uid} ${startAddr} ${quantity}
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0, len=6)/ModbusPDU04_Read_Input_Registers(funcCode=${FUNC_READ_INPUT_REGISTERS}, startAddr=${startAddr}, quantity=${quantity})
RETURN ${packet}
Create Holding Read Registers Request
[Arguments] ${uid} ${startAddr} ${quantity}
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0, len=6)/ModbusPDU03_Read_Holding_Registers(funcCode=${FUNC_READ_HOLDING_REGISTERS}, startAddr=${startAddr}, quantity=${quantity})
RETURN ${packet}
Create Holding Write Registers Request
[Arguments] ${uid} ${startAddr} ${quantity} ${data}
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU10_Write_Multiple_Registers(funcCode=${FUNC_WRITE_HOLDING_REGISTERS}, startAddr=${startAddr}, quantityRegisters=${quantity}, outputsValue=${data})
Log Packet: ${packet}
RETURN ${packet}
Create Holding Write Register Request
[Arguments] ${uid} ${startAddr} ${data}
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU06_Write_Single_Register(funcCode=${FUNC_WRITE_HOLDING_REGISTER}, registerAddr=${startAddr}, registerValue=${data})
Log Packet: ${packet}
RETURN ${packet}
Create Coils Read Request
[Arguments] ${uid} ${startAddr} ${quantity}
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU01_Read_Coils(funcCode=${FUNC_READ_COILS}, startAddr=${startAddr}, quantity=${quantity})
Log Packet: ${packet}
RETURN ${packet}
Create Coils Write Request
[Arguments] ${uid} ${startAddr} ${quantity} ${coil_data}
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU0F_Write_Multiple_Coils(funcCode=${FUNC_WRITE_COILS}, startAddr=${startAddr}, quantityOutput=${quantity}, outputsValue=${coil_data})
Log Packet: ${packet}
RETURN ${packet}
Create Discrete Read Request
[Arguments] ${uid} ${startAddr} ${quantity}
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU02_Read_Discrete_Inputs(funcCode=${FUNC_READ_DISCRETE_INPUTS}, startAddr=${startAddr}, quantity=${quantity})
Log Packet: ${packet}
RETURN ${packet}
Read Input Registers
[Arguments] ${uid} ${start_addr} ${quantity} ${exception_expected}
${classId} = Get Class Id
Log Library ClassId: ${classId}
Log Read Input Registers with parameters UID:${uid}, offs:${start_addr}, quantity:${quantity}
${req} = Create Input Read Registers Request ${uid} ${start_addr} ${quantity}
#Create Connection ${server} ${port}
${response_frame} = Send Packet And Get Response ${req}
Should Not Be Empty ${response_frame}
${packet} = Translate Response ${response_frame}
Should Be Equal As Integers ${req.transId} ${packet.transId}
${exception} ${exp_message} = Check Response ${packet} ${req.funcCode}
Should Be Equal As Integers ${exception} ${exception_expected}
Log exception: (${exception}: ${exp_message}), expected: ${exception_expected}
IF ${exception} == ${0}
${vallist} = Convert To List ${packet.registerVal}
Should Not Be Empty ${vallist}
Log Modbus register values:${vallist}
FOR ${item} IN @{vallist}
Log Modbus register value:${item}
#Append To List ${} ${item}
END
${length} = Get length ${vallist}
Log Items count is: ${length}
Should Be Equal As Integers ${length} ${quantity}
ELSE
Log "Exception is evaluated correctly (${exception}: ${exp_message}) == ${exception_expected}"
END
Read Holding Registers
[Arguments] ${uid} ${start_addr} ${quantity} ${exception_expected}
${classId} = Get Class Id
Log Library ClassId: ${classId}
Log Read Holding Registers with parameters UID:${uid}, offs:${start_addr}, quantity:${quantity}
${req} = Create Holding Read Registers Request ${uid} ${start_addr} ${quantity}
#Create Connection ${server} ${port}
${response_frame} = Send Packet And Get Response ${req}
Should Not Be Empty ${response_frame}
${packet} = Translate Response ${response_frame}
Should Be Equal As Integers ${req.transId} ${packet.transId}
${exception} ${exp_message} = Check Response ${packet} ${req.funcCode}
Log exception: (${exception}: ${exp_message}), expected: ${exception_expected}
Should Be Equal As Integers ${exception} ${exception_expected}
IF ${exception} == ${0}
${vallist} = Convert To List ${packet.registerVal}
Should Not Be Empty ${vallist}
Log Modbus register values:${vallist}
FOR ${item} IN @{vallist}
Log Modbus register value:${item}
#Append To List ${} ${item}
END
${length} = Get length ${vallist}
Log Items count is: ${length}
Should Be Equal As Integers ${length} ${quantity}
ELSE
Log "Exception is evaluated correctly ${exception} == ${exception_expected}"
END
Write Holding Registers
[Arguments] ${uid} ${start_addr} ${quantity} ${data} ${exception_expected}
${classId} = Get Class Id
Log Library ClassId: ${classId}
Log Write Hold Registers with parameters UID:${uid}, offs:${start_addr}, quantity:${quantity}, data:${data}
${req} = Create Holding Write Registers Request ${uid} ${start_addr} ${quantity} ${data}
#Create Connection ${server} ${port}
${response_frame} = Send Packet And Get Response ${req}
Should Not Be Empty ${response_frame}
${packet} = Translate Response ${response_frame}
Should Be Equal As Integers ${req.transId} ${packet.transId}
Should Not Be Empty ${packet}
Log Response is: ${packet.show(dump=True)}
${exception} ${exp_message} = Check Response ${packet} ${req.funcCode}
Log exception: (${exception}: ${exp_message}), expected: ${exception_expected}
Should Be Equal As Integers ${exception} ${exception_expected}
Run Keyword If ${exception} == ${0} Should Be Equal As Integers ${${packet.quantityRegisters}} ${quantity}
... ELSE Log "Exception is evaluated correctly ${exception} == ${exception_expected}"
Write Single Holding Register
[Arguments] ${uid} ${start_addr} ${data} ${exception_expected}
${classId} = Get Class Id
Log Library ClassId: ${classId}
Log Write Single Holding Register with parameters UID:${uid}, offs:${start_addr}
${req} = Create Holding Write Register Request ${uid} ${start_addr} ${data}
#Create Connection ${server} ${port}
${response_frame} = Send Packet And Get Response ${req}
Should Not Be Empty ${response_frame}
${packet} = Translate Response ${response_frame}
Should Not Be Empty ${packet}
Log Response is: ${packet.show(dump=True)}
Should Be Equal As Integers ${req.transId} ${packet.transId}
${exception} ${exp_message} = Check Response ${packet} ${req.funcCode}
Log exception: (${exception}: ${exp_message}), expected: ${exception_expected}
Should Be Equal As Integers ${exception} ${exception_expected}
Run Keyword If ${exception} == ${0} Should Be Equal As Integers ${${packet.registerValue}} ${data}
... ELSE Log "Exception is evaluated correctly ${exception} == ${exception_expected}"
Read Coil Registers
[Arguments] ${uid} ${start_addr} ${quantity} ${exception_expected}
${classId} = Get Class Id
Log Library ClassId: ${classId}
Log Read Coil Registers with parameters UID:${uid}, offs:${start_addr}, quantity:${quantity}
${req} = Create Coils Read Request ${uid} ${start_addr} ${quantity}
${response_frame} = Send Packet And Get Response ${req}
Should Not Be Empty ${response_frame}
${packet} = Translate Response ${response_frame}
Should Not Be Empty ${packet}
Log Response is: ${packet.show(dump=True)}
Should Be Equal As Integers ${req.transId} ${packet.transId}
${exception} ${exp_message} = Check Response ${packet} ${req.funcCode}
Log exception: (${exception}: ${exp_message}), expected: ${exception_expected}
Should Be Equal As Integers ${exception} ${exception_expected}
IF ${exception} == ${0}
${coils} = Get Bits From PDU ${packet}
Should Not Be Empty ${coils}
Should Be Equal As Integers ${${coils.__len__()}} ${${packet.byteCount} * 8}
Log Returned modbus coils: ${coils}
ELSE
Log "Exception is evaluated correctly ${exception} == ${exception_expected}"
END
Write Coil Registers
[Arguments] ${uid} ${start_addr} ${quantity} ${coil_data} ${exception_expected}
${classId} = Get Class Id
Log Library ClassId: ${classId}
Log Write Coil Registers with parameters UID:${uid}, offs:${start_addr}, quantity:${quantity}, coil_data:${coil_data}
${req} = Create Coils Write Request ${uid} ${start_addr} ${quantity} ${coil_data}
${response_frame} = Send Packet And Get Response ${req}
Should Not Be Empty ${response_frame}
${packet} = Translate Response ${response_frame}
Should Not Be Empty ${packet}
Log Response is: ${packet.show(dump=True)}
Should Be Equal As Integers ${req.transId} ${packet.transId}
${exception} ${exp_message} = Check Response ${packet} ${req.funcCode}
Log exception: (${exception}: ${exp_message}), expected: ${exception_expected}
Should Be Equal As Integers ${exception} ${exception_expected}
IF ${exception} == ${0}
Log ${${packet.quantityOutput}}
Should Be Equal As Integers ${${packet.quantityOutput}} ${quantity}
ELSE
Log "Exception is evaluated correctly ${exception} == ${exception_expected}"
END
Read Discrete Input Registers
[Arguments] ${uid} ${start_addr} ${quantity} ${exception_expected}
${classId} = Get Class Id
Log Library ClassId: ${classId}
Log Read Discrete Input Registers with parameters UID:${uid}, offs:${start_addr}, quantity:${quantity}
${req} = Create Discrete Read Request ${uid} ${start_addr} ${quantity}
${response_frame} = Send Packet And Get Response ${req}
Should Not Be Empty ${response_frame}
${packet} = Translate Response ${response_frame}
Should Not Be Empty ${packet}
Log Response is: ${packet.show(dump=True)}
Should Be Equal As Integers ${req.transId} ${packet.transId}
${exception} ${exp_message} = Check Response ${packet} ${req.funcCode}
Log exception: (${exception}: ${exp_message}), expected: ${exception_expected}
Should Be Equal As Integers ${exception} ${exception_expected}
IF ${exception} == ${0}
${dicretes} = Get Bits From PDU ${packet}
Should Not Be Empty ${dicretes}
Should Be Equal As Integers ${${dicretes.__len__()}} ${${packet.byteCount} * 8}
Log Returned modbus dicretes: ${dicretes}
ELSE
Log "Exception is evaluated correctly ${exception} == ${exception_expected}"
END
Send Packet And Get Response
[Arguments] ${packet}
${response} = Send Packet ${packet} timeout=3 verbose=1
Log Got response ${response}
RETURN ${response}
Create Connection
[Arguments] ${host} ${port}
${classId} = Get Class Id
Log Library ClassId: ${classId}
${connection} = Connect ${host} ${port}
IF ${connection}
Log Connection to host: ${host}:${port} established.
ELSE
Log Connection to host: ${host}:${port} failed.
END
RETURN ${connection}

64
tools/robot/ModbusTestSuite.robot Normal file
View File

@ -0,0 +1,64 @@
*** Settings ***
Documentation A test suite for Modbus commands.
...
... Keywords are imported from the resource file
Resource ModbusTestSuite.resource
Default Tags multi_dut_modbus_generic
Suite Setup Create Connection ${MODBUS_DEF_SERVER_IP} ${MODBUS_DEF_PORT}
Suite Teardown Disconnect
*** Variables ***
${suiteConnection} None
*** Test Cases ***
Test Read Holding Registers With Different Addresses And Quantities
[Documentation] Test reading holding registers from different addresses with different quantities
[Template] Read Holding Registers
0x01 0x0001 2 0
0x01 0x0002 3 0
Test Write Holding Registers With Different Addresses And Quantities
[Documentation] Test write holding registers for different addresses with different quantities
[Template] Write Holding Registers
0x01 0x0003 2 [0x1122, 0x3344] 0
0x01 0x0004 3 [0x1122, 0x3344, 0x5566] 0
Test Read Input Registers With Different Addresses And Quantities
[Documentation] Test read input registers for different addresses with different quantities
[Template] Read Input Registers
0x01 0x0003 2 0
0x01 0x0004 3 0
0x01 0x0001 200 3
0x01 0x2344 3 2
Test Write Single Holding Register
[Documentation] Test write one single holding register
[Template] Write Single Holding Register
0x01 0x0001 0x1122 0
0x01 0x2344 0x1122 2
0x01 0x0010 0x3344 0
Test Read Coils With Different Addresses And Quantities
[Documentation] Test read coil registers for different addresses with different quantities
[Template] Read Coil Registers
0x01 0x0001 0 3
0x01 0x0001 16 0
0x01 0x0010 20 2
0x01 0x0002 300 2
0x01 0x0008 30 2
Test Read Discrete Inputs With Different Addresses And Quantities
[Documentation] Test read discrete registers for different addresses with different quantities
[Template] Read Discrete Input Registers
0x01 0x0001 0 3
0x01 0x0001 16 0
0x01 0x0010 20 2
0x01 0x0002 300 2
0x01 0x0008 30 2
Test Write Coils With Different Addresses And Quantities
[Documentation] Test write coil registers for different addresses with different quantities
[Template] Write Coil Registers
0x01 0x0000 8 [0xFF] 0
0x01 0x0005 300 [0xFF] 3
0x01 0x0008 16 [0xFF, 0x55] 0

7
tools/test_requirements.txt Normal file
View File

@ -0,0 +1,7 @@
pytest
pytest-embedded
pytest-embedded-idf
pytest-embedded-serial
pytest-embedded-serial-esp
robotframework
scapy>=2.6.0