espressif/esp-modbus
1
0
Fork 1
mirror of https://github.com/espressif/esp-modbus.git synced 2026-08-03 20:24:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: add host slave tester for master based on robot framework

Browse Source
This commit is contained in:
Alex Lisitsyn
2026-05-22 11:58:15 +01:00
parent febaa774b6
commit c2f36916fc
16 changed files with 3213 additions and 438 deletions
Download Patch File Download Diff File Expand all files Collapse all files
conftest.py
+15 -3
View File
@@ -210,12 +210,13 @@ class MbObject:
class ModbusTestDut(IdfDut):
TEST_START_PROMPT = r"I\s\(([0-9]+)\) mb_console: (Start modbus instances)"
TEST_IP_PROMPT = r"Waiting IP\(([0-9]{1,2})\) from stdin:"
TEST_IP_ADDRESS_REGEXP = r"I \([0-9]+\) [a-z_]+: [A-Za-z\-]* IPv4 [A-Za-z\"_:\s]*address: ([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})"
TEST_APP_NAME = r"I \([0-9]+\) [a-z_]+: Project name:\s+([_a-z]*)"
TEST_EXPECT_STR_TIMEOUT = 120
TEST_IP_PROMPT_TOUT = 10
TEST_PROMPT_TOUT = 10
TEST_ACK_TIMEOUT = 60
TEST_MAX_CIDS = 8
@@ -427,6 +428,18 @@ class ModbusTestDut(IdfDut):
self.logger.info("Master Response time list couldn't be properly retrieved")
return 0
def send_message_start_dut(self) -> bool:
"""The function sends message to start DUT sequence"""
result: bool = True
self.logger.info(f"Sending start message to {self.dut_get_name()}")
self.write("mb start instances\n")
try:
self.expect(self.TEST_START_PROMPT, timeout=2)
except pexpect.TIMEOUT:
self.logger.error("Timeout waiting for start prompt confirmation.")
result = False
return result
def send_message_destroy_dut(self) -> None:
"""The function sends message to end caller DUT"""
self.logger.info("Sending destroy message to this DUT")
@@ -482,7 +495,6 @@ class ModbusTestDut(IdfDut):
)
if isinstance(expect_name, Match):
self.app_name = expect_name.group(1).decode("ascii")
self.logger.info(f"Project name registered: {self.app_name}")
return self.app_name
def dut_send_ip(
@@ -491,7 +503,7 @@ class ModbusTestDut(IdfDut):
"""The function sends the slave IP address defined as a parameter to master"""
addr_num: int = 0
try:
self.expect(self.TEST_IP_PROMPT, timeout=self.TEST_IP_PROMPT_TOUT)
self.expect(self.TEST_IP_PROMPT, timeout=self.TEST_PROMPT_TOUT)
except pexpect.TIMEOUT:
# Workaround for unreliable parsing of the master prompt.
# The expect() sometime does not catch it in spite it appears in the log.
examples/tcp/mb_tcp_master/main/tcp_master.c
+3 -3
View File
@@ -196,17 +196,17 @@ const mb_parameter_descriptor_t device_parameters[] = {
{
CID_RELAY_P1, STR("RelayP1"), STR("on/off"), MB_DEVICE_ADDR1, MB_PARAM_COIL, 2, 6,
COIL_OFFSET(coils_port0), PARAM_TYPE_U8, 1,
OPTS( 0xAA, 0x15, 0 ), PAR_PERMS_READ_WRITE_TRIGGER
OPTS( 0xAA, 0x2A, 0 ), PAR_PERMS_READ_WRITE_TRIGGER
},
{
CID_RELAY_P2, STR("RelayP2"), STR("on/off"), MB_DEVICE_ADDR1, MB_PARAM_COIL, 10, 6,
COIL_OFFSET(coils_port1), PARAM_TYPE_U8, 1,
OPTS( 0x55, 0x2A, 0 ), PAR_PERMS_READ_WRITE_TRIGGER
OPTS( 0x55, 0x15, 0 ), PAR_PERMS_READ_WRITE_TRIGGER
},
{
CID_DISCR_P1, STR("DiscreteInpP1"), STR("on/off"), MB_DEVICE_ADDR1, MB_PARAM_DISCRETE, 2, 7,
DISCR_OFFSET(discrete_input_port1), PARAM_TYPE_U8, 1,
OPTS( 0xAA, 0x15, 0 ), PAR_PERMS_READ_WRITE_TRIGGER
OPTS( 0xAA, 0x2A, 0 ), PAR_PERMS_READ_WRITE_TRIGGER
},
#if CONFIG_FMB_EXT_TYPE_SUPPORT
{
examples/tcp/mb_tcp_master/sdkconfig.ci.defaults
+1
View File
@@ -13,6 +13,7 @@ CONFIG_FMB_MASTER_DELAY_MS_CONVERT=300
CONFIG_FMB_TIMER_USE_ISR_DISPATCH_METHOD=y
CONFIG_FMB_EXT_TYPE_SUPPORT=y
CONFIG_FMB_TCP_UID_ENABLED=n
CONFIG_FMB_MDNS_INTEGRATION_ENABLE=y
CONFIG_EXAMPLE_CONNECT_IPV6=n
CONFIG_MB_SLAVE_IP_FROM_STDIN=y
CONFIG_LOG_DEFAULT_LEVEL_DEBUG=n
examples/tcp/mb_tcp_slave/main/tcp_slave.c
+1 -1
View File
@@ -1,5 +1,5 @@
/*
* SPDX-FileCopyrightText: 2016-2026 Espressif Systems (Shanghai) CO LTD
* SPDX-FileCopyrightText: 2020-2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
examples/tcp/mb_tcp_slave/sdkconfig.ci.defaults
+1
View File
@@ -13,6 +13,7 @@ CONFIG_FMB_MASTER_DELAY_MS_CONVERT=300
CONFIG_FMB_TIMER_USE_ISR_DISPATCH_METHOD=y
CONFIG_FMB_EXT_TYPE_SUPPORT=y
CONFIG_FMB_TCP_UID_ENABLED=n
CONFIG_FMB_MDNS_INTEGRATION_ENABLE=y
CONFIG_EXAMPLE_CONNECT_IPV6=n
CONFIG_MB_SLAVE_ADDR=1
CONFIG_LOG_DEFAULT_LEVEL_DEBUG=n
examples/tcp/pytest_mb_tcp_host_test_master.py
+250
View File
@@ -0,0 +1,250 @@
# SPDX-FileCopyrightText: 2016-2026 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
import sys
from typing import Any, Dict, List, Optional, Tuple
import re
# pytest required libraries
import pytest
from robot import run
from pathlib import Path
from robot.libraries.BuiltIn import BuiltIn
TEST_DIR = Path(__file__).resolve().parent
TEST_ROBOT_DIR = (TEST_DIR / "../../tools/robot").resolve()
TEST_CONF_DIR = (TEST_DIR / "../..").resolve()
if str(TEST_ROBOT_DIR) not in sys.path:
sys.path.insert(0, str(TEST_ROBOT_DIR))
if str(TEST_CONF_DIR) not in sys.path:
sys.path.insert(0, str(TEST_CONF_DIR))
from conftest import ModbusTestDut, MbParameter, PARAM_SUCCESS, Stages # noqa: E402
LOG_LEVEL = logging.DEBUG
LOGGER_NAME = "modbus_test"
ROBOT_SUITE_NAME = "ModbusTestSuiteSlave"
logger = logging.getLogger(LOGGER_NAME)
pattern_dict_master = {
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]+)\) [A-Z_]*: [0xa-f0-9,]*\s*Modbus master stack initialized"
),
Stages.STACK_CONNECT: (
r"I\s\(([0-9]+)\) mb_port.tcp.master: 0x[a-f0-9]+, Connected: [0-9], [0-9], start polling."
),
Stages.STACK_START: (r"I \(([0-9]+)\) [A-Z_]+: Master TCP is started"),
Stages.STACK_PAR_OK: (
r"[I|E] \(([0-9]+)\) [A-Z_]+: ([a-z0-9]+) Characteristic #([0-9]+) ([a-zA-Z0-9_]+) \([\%a-zA-Z_\/]*\) value = ([0-9a-zA-Z.]*)\s*([a-zA-Z0-9()]*)( read successful|, unexpected value)"
),
Stages.STACK_PAR_FAIL: (
r"E \(([0-9]+)\) [A-Z_]+: ([a-z0-9]+) Characteristic #([0-9]+) \(([a-zA-Z0-9_]+)\) read fail, err = [x0-9]+ \([_A-Z]+\)"
),
Stages.STACK_DESTROY: (r"I \(([0-9]+)\) [A-Z_]+: (Master TCP is completed)"),
Stages.STACK_OBJECT_CREATE: (
r"D \(([0-9]+)\) [a-z]+_[a-z]+\.([a-z]+)\: created object mb[a-z]\_tcp[#@](0x[0-9a-f]+)"
),
Stages.STACK_BAD_CONNECTION: (
r"I \([0-9]+\) example_connect: WiFi Connect failed [0-9]* times, stop reconnect."
),
Stages.STACK_CID_RESPONSE_TIME: (
r"D \(([0-9]+)\) mbc_[a-z]+.master: mbc_[a-z]+_master_get_parameter: ([a-zA-Z0-9]+) Good response for get cid\(([0-9]+)\) = ESP_OK"
),
}
class ModbusDataBridge:
ROBOT_LISTENER_API_VERSION = 2
def __init__(
self,
trace_list: Optional[List[Dict[str, Any]]] = None,
dut: Optional[ModbusTestDut] = None,
) -> None:
"""The constructor of the bridge interface that inherits the native robot listener API"""
self.ROBOT_LIBRARY_LISTENER = self
self.trace_list = trace_list
self.dut = dut
self.server_address: Optional[Tuple[str, int]] = None
self.logger = logging.getLogger("RobotFramework")
def setup_master_address(self, address: Tuple[str, int]) -> None:
"""The callback to setup master when server (slave) is ready."""
print(f"Setting up master address: {address}")
if self.dut:
dut_stdin_en = self.dut.app.sdkconfig.get("MB_SLAVE_IP_FROM_STDIN")
if dut_stdin_en:
self.dut.dut_send_ip(slave_ip=address[0], port=str(address[1]))
logger.info(
f"Initialized slave IP address for DUT {self.dut.dut_get_name()} = {address[0]}:{address[1]}."
)
if self.dut.send_message_start_dut():
logger.info(
f"The DUT:{self.dut.dut_get_name()} is ready for connection."
)
def end_keyword(self, name: str, attributes: Dict[str, Any]) -> None:
"""This method is a listener interface wrapper for each keyword."""
if ".Test Report Server Address" in name:
self.logger.info(f"Server is ready: {name}, {attributes}")
server_addr = BuiltIn().get_variable_value(
"${MODBUS_SERVER_ADDRESS}", "127.0.0.1"
)
server_port = BuiltIn().get_variable_value("${MODBUS_DEF_PORT}", 1502)
self.logger.info(
f"Setup slave address in master: {server_addr}:{server_port}"
)
self.setup_master_address((server_addr, server_port))
def end_test(self, name: str, attributes: Dict[str, Any]) -> None:
"""This method is a listener interface wrapper for each test."""
if (
"Test Modbus" in name
and "Test Async Run Modbus Case" in attributes["template"]
):
self.logger.info(
f"End test: {name}, {attributes}, Message: {attributes.get('message')}"
)
# This will track all tags after test completion using the test message as the container for parameter names
if self.trace_list is not None:
self.trace_list.append(
{
"timestamp": attributes.get("starttime"),
"action": name,
"args": attributes.get("args"),
"tags": attributes.get("message"),
"status": attributes.get("status"),
}
)
def trace_test_params(self) -> None:
"""This method will track the parameter keywords for each test (server side).
It gets the parameter data from log of master using ModbusTestDut class
and checks the status of characteristic tag of server side.
"""
if self.trace_list is None or self.dut is None:
return
self.logger.info(f"Test info List: {self.trace_list}")
for test_info in self.trace_list:
tags = test_info["tags"]
tags_str = (
tags.decode("ascii", errors="replace")
if isinstance(tags, bytes)
else str(tags)
if tags is not None
else ""
)
self.logger.info(f"Tags: {tags_str}")
params: List[MbParameter] = []
if tags_str and len(tags_str):
for tag in re.findall(r'"(\s*[^"\s][^"]*)"', tags_str):
self.logger.info(
f'Synchronize the "{tag}" with parameters in dut: {self.dut.dut_get_name()}'
)
params = self.dut.get_params_by_name(str(tag)) or []
if params:
for param in params:
self.logger.info(f"Found param: {param!r}")
if param.status == PARAM_SUCCESS:
val = param.get_value()
val_str = (
val.decode("ascii", errors="replace")
if isinstance(val, bytes)
else str(val)
if val is not None
else ""
)
self.logger.info(
f"Param: {param!r}, Value: {val_str}, Status: SUCCESS"
)
else:
# Master did not receive response. Keep information in the log and fail the test.
self.logger.error(f"Param: {param!r}, Status: FAIL.")
raise RuntimeError(
f"Param: {param!r}, Status: FAIL (Master did not get response from DUT)."
)
else:
# Continue test if the tag is not found
self.logger.error(
f"Could not find param with tag: {tag} in dut: {self.dut.dut_get_name()}"
)
@pytest.mark.parametrize("target", ["esp32"], indirect=True)
@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_master')}")],
indirect=True,
)
def test_modbus_tcp_master_to_host_slave_communication(
app_path: str, dut: ModbusTestDut
) -> None:
logger.info("DUT: %s start.", dut.dut_get_name())
dut_master_name = dut.dut_get_name()
dut_master_port = dut.app.sdkconfig.get("FMB_TCP_PORT_DEFAULT")
dut_master_ip_address = dut.dut_get_ip()
assert dut_master_ip_address is not None, "The DUT could not get IP address. Abort."
logger.info(f"DUT: {dut_master_name}, ip address: {dut_master_ip_address}.")
assert dut_master_port is not None, f"DUT port is not correct: {dut_master_port}"
logger.info(
f"Start test for the master application: {app_path}, {dut_master_ip_address}:{dut_master_port}"
)
try:
trace_data: List[Dict[str, Any]] = []
mb_data_bridge = ModbusDataBridge(trace_data, dut)
# The robot suite will run the test cases for Modbus Master node,
# and the data bridge will track the test parameters and their status after each test case.
# The test parameters are tagged in robot suite and synchronized with the data in DUT using the bridge interface.
return_code = run(
f"{TEST_ROBOT_DIR}/{ROBOT_SUITE_NAME}.robot",
variable=[
f"MODBUS_DEF_IP:{dut_master_ip_address}",
f"MODBUS_DEF_PORT:{dut_master_port}",
],
outputdir=f"{ROBOT_SUITE_NAME}_logs",
listener=mb_data_bridge,
output="trace_data.xml",
exitonfailure=False,
loglevel="DEBUG",
)
dut.dut_test_start(dictionary=pattern_dict_master)
mb_data_bridge.trace_test_params()
dut.write("mb stop instances\n") # Intentionally destroy DUT after test
dut.dut_check_errors()
if return_code != 0:
raise RuntimeError(
f"The robot suite {ROBOT_SUITE_NAME} returns an exception: {return_code}."
)
logger.info(
f"Test for the Modbus Master node: {dut_master_name} is completed, return code: {return_code}."
)
except Exception as e:
raise e
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-vv"]))
examples/tcp/pytest_mb_tcp_host_test_slave.py
+46 -32
View File
@@ -1,4 +1,4 @@
# SPDX-FileCopyrightText: 2016-2025 Espressif Systems (Shanghai) CO LTD
# SPDX-FileCopyrightText: 2016-2026 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
@@ -6,12 +6,11 @@
import logging
import os
import subprocess
# pytest required libraries
import pytest
from conftest import ModbusTestDut, Stages
from robot import run
TEST_DIR = os.path.abspath(os.path.dirname(__file__))
TEST_ROBOT_DIR = os.path.abspath(
@@ -19,14 +18,9 @@ TEST_ROBOT_DIR = os.path.abspath(
)
LOG_LEVEL = logging.DEBUG
LOGGER_NAME = "modbus_test"
ROBOT_SUITE_NAME = "ModbusTestSuiteMaster"
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]+\) [a-z_]+: [A-Za-z\-]* IPv4 [A-Za-z\"_:\s]*address: ([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})"
@@ -38,19 +32,19 @@ pattern_dict_slave = {
Stages.STACK_CONNECT: (
r"I\s\(([0-9]+)\) port.utils: Socket \(#[0-9]+\), accept client connection from address\[port\]: ([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})\[[0-9]+\]"
),
Stages.STACK_START: (r"I\s\(([0-9]+)\) [A-Z_]+: Slave TCP [#0-9]*\s*is started"),
Stages.STACK_START: (r"I\s\(([0-9]+)\) [A-Z_]+: (Start modbus test...)"),
Stages.STACK_PAR_OK: (
r"I\s\(([0-9]+)\) [A-Z_]+: OBJ (0x[a-fA-Z0-9]+),() ([A-Za-z\s]+) \([0-9]+ us\),\s*[A-Z:]*\s*[0-9,]*\s*[A-Z:]*[0-9]*, TYPE:[0-9]+, INST_ADDR:0x[a-fA-Z0-9]+[()0-9a-z]+, 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]+)\) [A-Z_]+: Destroy slave"),
Stages.STACK_DESTROY: (r"I\s\(([0-9]+)\) [A-Z_]+: (Destroy slave)"),
Stages.STACK_OBJECT_CREATE: (
r"D \(([0-9]+)\) [a-z]+_[a-z]+\.([a-z]+)\: created object mb[a-z]\_tcp[#@](0x[0-9a-f]+)"
r"D \(([0-9]+)\) [a-z]+_[a-z]+\.([a-z]+)\: created object mbs_tcp[#@](0x[0-9a-f]+)"
),
Stages.STACK_BAD_CONNECTION: (
r"I \([0-9]+\) example_connect: WiFi Connect failed [0-9]* times, stop reconnect."
r"E \([0-9]+\) mb_port[\.a-z]+: (0x[a-zA-Z0-9]+), node #[0-9]+, socket\(\#([0-9]+)\)\(([\.0-9a-f]+)\), communication fail, err= -([0-9]+)"
),
Stages.STACK_CID_RESPONSE_TIME: (
r"D \(([0-9]+)\) mbc_[a-z]+.slave: mbc_[a-z]+_slave_get_parameter: Good response for get cid\(([0-9]+)\) = ESP_OK"
@@ -66,7 +60,7 @@ pattern_dict_slave = {
[(1, f"{os.path.join(os.path.dirname(__file__), 'mb_tcp_slave')}")],
indirect=True,
)
@pytest.mark.flaky(reruns=1, reruns_delay=1)
@pytest.mark.flaky(reruns=1, reruns_delay=2)
def test_modbus_tcp_host_to_slave_communication(
app_path: str, dut: ModbusTestDut
) -> None:
@@ -77,28 +71,48 @@ def test_modbus_tcp_host_to_slave_communication(
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"
# Ensure the Slave test is started correctly and ready for requests
if dut.send_message_start_dut():
logger.info(
f"The DUT is ready for connection with the name: {dut.dut_get_name()}"
)
logger.info(
f"Start test for the slave: {app_path}, {dut_slave_ip_address}:{dut_slave_ip_port}"
)
p = subprocess.Popen(
cmd, stdin=None, stdout=None, stderr=None, shell=True, close_fds=CLOSE_FDS
return_code = run(
f"{TEST_ROBOT_DIR}/{ROBOT_SUITE_NAME}.robot",
variable=[
f"MODBUS_DEF_IP:{dut_slave_ip_address}",
f"MODBUS_DEF_PORT:{dut_slave_ip_port}",
],
outputdir=f"{ROBOT_SUITE_NAME}_logs",
# log=None, # Prevents stdout clutter
report="master_host_report.xml",
exitonfailure=False,
loglevel="DEBUG",
)
# Start and check DUT test sequence
dut.dut_test_start(dictionary=pattern_dict_slave)
p.wait()
logger.info(f"Test for the node: {dut_slave_ip_address} is completed.")
logging.info(f"Explicitly destroy slave: {dut.dut_get_name()}.")
dut.write("mb stop instances\n") # Intentionally destroy DUT after test
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.")
if return_code != 0:
raise RuntimeError(
f"The robot suite {ROBOT_SUITE_NAME}, returns an exception: {return_code}."
)
logger.info(
f"Suite {ROBOT_SUITE_NAME} for the Modbus Slave node: {dut_slave_ip_address} is completed, return code: {return_code}."
)
except Exception as e:
logging.error(f"Robot suite {ROBOT_SUITE_NAME} for {dut.dut_get_name()} fail.")
raise e
if __name__ == "__main__":
pytest.main(["pytest_mb_tcp_host_test_slave.py"])
examples/tcp/pytest_mb_tcp_master_slave.py
+12 -7
View File
@@ -28,19 +28,19 @@ pattern_dict_slave = {
Stages.STACK_CONNECT: (
r"I\s\(([0-9]+)\) port.utils: Socket \(#[0-9]+\), accept client connection from address\[port\]: ([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})\[[0-9]+\]"
),
Stages.STACK_START: (r"I\s\(([0-9]+)\) [A-Z_]+: Slave TCP [#0-9]*\s*is started"),
Stages.STACK_START: (r"I\s\(([0-9]+)\) [A-Z_]+: Start modbus test"),
Stages.STACK_PAR_OK: (
r"I\s\(([0-9]+)\) [A-Z_]+: OBJ (0x[a-fA-Z0-9]+),() ([A-Za-z\s]+) \([0-9]+ us\),\s*[A-Z:]*\s*[0-9,]*\s*[A-Z:]*[0-9]*, TYPE:[0-9]+, INST_ADDR:0x[a-fA-Z0-9]+[()0-9a-z]+, 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]+)\) [A-Z_]+: Destroy slave"),
Stages.STACK_DESTROY: (r"I\s\(([0-9]+)\) [A-Z_]+: (Destroy slave.)"),
Stages.STACK_OBJECT_CREATE: (
r"D \(([0-9]+)\) [a-z]+_[a-z]+\.([a-z]+)\: created object mb[a-z]\_tcp[#@](0x[0-9a-f]+)"
),
Stages.STACK_BAD_CONNECTION: (
r"I \([0-9]+\) example_connect: WiFi Connect failed [0-9]* times, stop reconnect."
r"[EWI] \([0-9]+\) mb_port[\.a-z]+: (0x[a-zA-Z0-9]+), node #[0-9]+, socket\(\#([0-9]+)\)\(([\.0-9a-f]+)\), communication fail, err= -([0-9]+)"
),
Stages.STACK_CID_RESPONSE_TIME: (
r"D \(([0-9]+)\) mbc_[a-z]+.slave: mbc_[a-z]+_slave_get_parameter: Good response for get cid\(([0-9]+)\) = ESP_OK"
@@ -118,18 +118,23 @@ def test_modbus_tcp_communication(
dut_master_name = dut_master.dut_get_name()
dut_slave_port = dut_slave.app.sdkconfig.get("FMB_TCP_PORT_DEFAULT")
dut_master_port = dut_master.app.sdkconfig.get("FMB_TCP_PORT_DEFAULT")
dut_stdin_en = dut_master.app.sdkconfig.get("MB_SLAVE_IP_FROM_STDIN")
dut_slave_ip_address = dut_slave.dut_get_ip()
logger.info(f"DUT: {dut_slave_name}, ip address: {dut_slave_ip_address}.")
dut_master_ip_address = dut_master.dut_get_ip()
logger.info(
f"DUT Slave: {dut_slave_name}, ip address[:port]: {dut_slave_ip_address}:{dut_slave_port}."
)
logger.info(
f"DUT Master: {dut_master_name}, ip address[:port]: {dut_master_ip_address}:{dut_master_port}."
)
dut_master_ip_address = dut_master.dut_get_ip()
logger.info(f"DUT: {dut_master_name}, ip address: {dut_master_ip_address}.")
if dut_stdin_en:
dut_master.dut_send_ip(slave_ip=dut_slave_ip_address)
print(f"Start testing for {dut_slave_name}:{dut_slave_port}, {dut_stdin_en}")
dut_master.dut_send_ip(slave_ip=dut_slave_ip_address, port=dut_slave_port)
dut_slave.dut_test_start(dictionary=pattern_dict_slave)
dut_master.dut_test_start(dictionary=pattern_dict_master)
tools/robot/ModbusTestLib.py → tools/robot/ModbusMasterLib.py
+236 -127
View File
@@ -1,5 +1,5 @@
#!/usr/bin/python
# SPDX-FileCopyrightText: 2024-2025 Espressif Systems (Shanghai) CO LTD
# SPDX-FileCopyrightText: 2024-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
import socket
@@ -10,17 +10,19 @@ from typing import Optional, Any, Tuple, List
from scapy.utils import wrpcap
from scapy.packet import Packet, Raw
from scapy.fields import XByteField
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 scapy.fields import XByteField
from robot.api.deco import keyword, library
from robot.api.logger import info
from ModbusSupport import (
modbus_exceptions,
Exceptions,
MB_EXCEPTION_MASK,
MB_EXCEPTION_FUNC_MASK,
ModbusADU_Request,
ModbusADU_Response,
ModbusPDUXX_Custom_Request, # noqa
@@ -30,6 +32,7 @@ from ModbusSupport import (
ModbusPDU04_Read_Input_Registers, # noqa
ModbusPDU01_Read_Coils, # noqa
ModbusPDU0F_Write_Multiple_Coils, # noqa
ModbusPDU05_Write_Single_Coil, # noqa
ModbusPDU02_Read_Discrete_Inputs, # noqa
ModbusPDU06_Write_Single_Register, # noqa
)
@@ -40,50 +43,54 @@ 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_SERVER_IP = "192.168.88.251" # "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_FUNC_REPORT_SLAVE_ID = 0x11
MB_DEF_QUANTITY = 1
MB_DEF_START_OFFS = 0x0001
MB_DEF_QUANTITY = 2
MB_DEF_START_OFFS = 0x0000
MB_DEF_REQ_TOUT = 5.0
MB_LOGGING_PATH = "."
# The constructed packets for self testing
TEST_PACKET_REPORT_CUSTOM_0X41 = "ModbusADU_Request(transId=MB_DEF_TRANS_ID, unitId=0x01, protoId=0)/\
ModbusPDUXX_Custom_Request(customBytes=[0x41])" # fmt: skip
TEST_PACKET_REPORT_SLAVE_ID_CUSTOM = "ModbusADU_Request(transId=MB_DEF_TRANS_ID, unitId=0x01, protoId=0)/\
ModbusPDUXX_Custom_Request(customBytes=[0x11])" # fmt: skip
TEST_PACKET_REPORT_SLAVE_ID = "ModbusADU_Request(transId=MB_DEF_TRANS_ID, unitId=0x01, protoId=0)/\
ModbusPDU11_Report_Slave_Id(funcCode=MB_DEF_FUNC_REPORT_SLAVE_ID)" # fmt: skip
TEST_PACKET_REPORT_CUSTOM_0X41 = (
"ModbusADU_Request(transId=MB_DEF_TRANS_ID, unitId=0x01, protoId=0)/\
ModbusPDUXX_Custom_Request(customBytes=[0x41])"
)
TEST_PACKET_REPORT_SLAVE_ID_CUSTOM = (
"ModbusADU_Request(transId=MB_DEF_TRANS_ID, unitId=0x01, protoId=0)/\
ModbusPDUXX_Custom_Request(customBytes=[0x11])"
)
TEST_PACKET_REPORT_SLAVE_ID = (
"ModbusADU_Request(transId=MB_DEF_TRANS_ID, unitId=0x01, protoId=0)/\
ModbusPDU11_Report_Slave_Id()"
)
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)" # fmt: skip
ModbusPDU03_Read_Holding_Registers(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])" # fmt: skip
ModbusPDU10_Write_Multiple_Registers(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)" # fmt: skip
ModbusPDU04_Read_Input_Registers(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)" # fmt: skip
ModbusPDU01_Read_Coils(startAddr=MB_DEF_START_OFFS, quantity=8)"
TEST_PACKET_DISCRETE_READ = "ModbusADU_Request(unitId=0x01, protoId=0)/\
ModbusPDU02_Read_Discrete_Inputs(startAddr=MB_DEF_START_OFFS, quantity=8)"
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])" # fmt: skip
ModbusPDU0F_Write_Multiple_Coils(startAddr=MB_DEF_START_OFFS, quantityOutput=MB_DEF_QUANTITY, outputsValue=[0xFF])"
TEST_PACKET_SINGLE_COIL_WRITE = "ModbusADU_Request(unitId=0x01, protoId=0)/\
ModbusPDU05_Write_Single_Coil(outputAddr=MB_DEF_START_OFFS, outputValue=0xFF00)"
TEST_PACKET_HOLDING_REG_WRITE = "ModbusADU_Request(unitId=0x01, protoId=0)/\
ModbusPDU06_Write_Single_Register(registerAddr=MB_DEF_START_OFFS, registerValue=[0x1234])"
# The simplified version of custom Modbus Library to check robot framework
@library(scope="GLOBAL", version="2.0.0") # fmt: skip
class ModbusTestLib:
# The simplified version of custom Modbus Master Library for robot framework
@library(scope="GLOBAL", version="2.1.0") # fmt: skip
class ModbusMasterLib:
"""
ModbusTestLib class is the custom Modbus library for robot framework.
ModbusMasterLib class is the custom Modbus Master 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: str = MB_DEF_SERVER_IP,
@@ -103,7 +110,7 @@ class ModbusTestLib:
self.in_pdu: Optional[Any] = 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)
path=MB_LOGGING_PATH, file="mbm_frames", ext="pcap", id=str(self.class_id)
)
if os.path.isfile(self.pcap_file_name):
os.remove(self.pcap_file_name)
@@ -177,7 +184,7 @@ class ModbusTestLib:
except Exception as exception:
raise Scapy_Exception(f"Send fail: {exception}")
@keyword('Get Class Id') # fmt: skip
@keyword("Get Class Id")
def get_class_id(self) -> int:
"""
Return unique class ID for robot suit debugging.
@@ -198,7 +205,7 @@ class ModbusTestLib:
if not packet[ModbusADU_Request].transId:
packet.transId = self.get_trans_id()
@keyword('Create Request') # fmt: skip
@keyword("Create Request") # fmt: skip
def create_request(self, packet_str: str) -> Packet:
"""
Create a Modbus packet based on the given string representation.
@@ -219,7 +226,7 @@ class ModbusTestLib:
raise ValueError(f"Failed to create packet: {str(exception)}")
# Connects to a target via TCP socket
@keyword('Connect') # fmt: skip
@keyword("Connect") # fmt: skip
def connect(
self, ip_addr: str = MB_DEF_SERVER_IP, port: int = MB_DEF_PORT
) -> StreamSocket:
@@ -239,6 +246,7 @@ class ModbusTestLib:
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print(f"Connect to server: {ip_addr}:{port}")
s.settimeout(8)
s.connect((ip_addr, port))
self._connection = StreamSocket(s, basecls=ModbusADU_Response)
self.host_ip = self.get_host_ip()
@@ -250,10 +258,12 @@ class ModbusTestLib:
self.node_address = ""
self.node_port = 0
self.socket = None
raise Scapy_Exception(f"Could not connect to socket: {exception}")
raise Scapy_Exception(
f"Could not connect to socket {ip_addr}:{port}: {exception}"
)
return self._connection
@keyword('Disconnect') # fmt: skip
@keyword("Disconnect") # fmt: skip
def disconnect(self) -> None:
"""
The disconnect from socket method to work as robot framework keyword.
@@ -275,7 +285,7 @@ class ModbusTestLib:
f"Connection close fail, exception occurred: ({exception})"
)
@keyword('Send Packet') # fmt: skip
@keyword("Send Packet") # fmt: skip
def send_packet_and_get_response(
self, pkt: Packet, timeout: int = 2, verbose: bool = True
) -> Optional[bytes]:
@@ -304,7 +314,6 @@ class ModbusTestLib:
response: Optional[Packet] = self._req_send_recv(
mb_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
@@ -336,7 +345,7 @@ class ModbusTestLib:
except Exception as exception:
raise Scapy_Exception(f"Send data fail, exception occurred: ({exception})")
@keyword('Translate Response') # fmt: skip
@keyword("Translate Response") # fmt: skip
def translate_response(self, pkt: bytes) -> Optional[Any]:
"""
Translates response received from server. Does dissection of the received packet.
@@ -351,45 +360,60 @@ class ModbusTestLib:
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)}")
print(f"Packet received: {packet.summary()}") # show(dump=True)
# Default extraction (may be Raw or a PDU class)
self.in_pdu, __ = packet.extract_padding(packet)
print(
f"Test received: pdu: {type(self.in_pdu)} {self.in_pdu!r}, {bytes(self.in_pdu)!r}"
)
# workaround to use dissected packets under robot framework
# issue with scapy dissector on incomplete packets
# Workaround: if Raw present try to identify correct PDU class using payload's guess_payload_class
if packet.haslayer(Raw):
raw_load = packet[Raw].load
print(f"Test workaround PDU: {raw_load!r}")
# We need to call guess_payload_class on the ADU layer to dissect the PDU payload
# The PDU class (e.g., ModbusPDU03_Read_Holding_Registers_Answer) is returned
pdu_class = packet.guess_payload_class(raw_load)
# Now instantiate the PDU class with the raw load
# Use the ADU layer's payload's guess_payload_class for correct PDU detection
try:
pdu_class = packet.payload.guess_payload_class(raw_load)
except Exception:
pdu_class = None
if pdu_class and pdu_class is not Raw:
self.in_pdu = pdu_class(raw_load)
try:
self.in_pdu = pdu_class(raw_load)
except Exception:
self.in_pdu = Raw(raw_load)
else:
self.in_pdu = Raw(raw_load)
# Check for exception
# Accessing fields requires type checking or hasattr
if hasattr(packet, "funcCode"):
func_code = packet.getfieldval("funcCode")
if (func_code & self.MB_EXCEPTION_MASK) and hasattr(
self.in_pdu, "exceptCode"
):
except_code = self.in_pdu.getfieldval("exceptCode")
if except_code in modbus_exceptions:
self.exception = except_code # Fixed: assign int
self.exception_message = modbus_exceptions[
self.exception
] # Fixed: index with int
# Check for exception: map exceptCode safely to Exceptions enum
try:
func_code = (
packet.getfieldval("funcCode")
if hasattr(packet, "getfieldval")
else None
)
except Exception:
func_code = None
if func_code is not None and (func_code & MB_EXCEPTION_MASK):
try:
if hasattr(self.in_pdu, "exceptCode"):
except_code = self.in_pdu.getfieldval("exceptCode")
# assign numeric exception and try to map to Exceptions enum
self.exception = int(except_code)
try:
self.exception_message = Exceptions(self.exception).name
except Exception:
self.exception_message = str(self.exception)
except Exception:
# best-effort: ignore mapping errors
pass
self.in_adu = packet[ModbusADU_Response]
print(f"PDU: {self.in_pdu}")
return self.in_pdu
except Exception as exception:
except Exception as e:
self.in_adu = None
self.in_pdu = None
raise Scapy_Exception(f"Parsing of response : ({exception})")
raise Scapy_Exception(f"Parsing of response : ({str(e)})")
def get_int(self, val: Any) -> int:
if isinstance(val, str):
@@ -408,7 +432,7 @@ class ModbusTestLib:
Check PDU frame from response. Check exception code
Args:
pdu: A Modbus PDU frame.
expected_func: timeout to send the data
expected_func: the expected function code
Returns:
exception: The exception code from Modbus frame
exception_message: exception message
@@ -430,13 +454,13 @@ class ModbusTestLib:
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")
func_code = pdu.getfieldval("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)}")
print(f"func code: {type(func_code)} {func_code}")
if (
(func_code & self.MB_EXCEPTION_MASK)
(func_code & MB_EXCEPTION_MASK)
and hasattr(pdu, "exceptCode")
and pdu.exceptCode
):
@@ -448,14 +472,14 @@ class ModbusTestLib:
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}"
assert (func_code & MB_EXCEPTION_FUNC_MASK) == exp_func, (
f"Unexpected function code: {func_code & MB_EXCEPTION_FUNC_MASK}, {exp_func}"
)
self.exception_message = modbus_exceptions[self.exception]
self.exception_message = Exceptions(self.exception).name
print(f"MB exception: {self.exception}, {self.exception_message}")
return self.exception, self.exception_message
@keyword('Check ADU') # fmt: skip
@keyword("Check ADU") # fmt: skip
def check_adu(self, adu_out: Packet, adu_in: Packet) -> Optional[int]:
"""
Check ADU frame fields.
@@ -482,7 +506,7 @@ class ModbusTestLib:
except Exception as exception:
raise Scapy_Exception(f"ADU check failed: ({exception})")
@keyword('Get Bits From PDU') # fmt: skip
@keyword("Get Bits From PDU") # fmt: skip
def get_bits_from_pdu(self, pdu: Packet) -> List[bool]:
"""
Check PDU frame, extract bits (coils or discrete) from PDU.
@@ -525,51 +549,54 @@ class ModbusTestLib:
print(f"Test: 0x41 <Custom command> packet: {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: {response!r}")
print(f"Test received: {response!r}")
pdu = self.translate_response(response)
if pdu is not None:
print(f"Received: {pdu}")
# print(f"PDU Exception: {self.check_response(pdu, packet.customBytes[0])}")
exception, msg = self.check_response(pdu, packet.customBytes[0])
if exception == Exceptions.UNDEFINED:
print(f"Received: {pdu}")
# print(f"PDU Exception: {self.check_response(pdu, packet.customBytes[0])}")
else:
print(f"PDU Exception: {exception}, {msg}")
# Test 0x11 Report Slave ID
packet = self.create_request(TEST_PACKET_REPORT_SLAVE_ID_CUSTOM)
print(f"Test: 0x11 <Report Slave ID> packet: {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: {response!r}")
print(f"Test received: {response!r}")
pdu = self.translate_response(response)
if pdu is not None:
# Assuming customBytes[0] holds the funcCode 0x11
exception, msg = self.check_response(pdu, packet.customBytes[0])
print(f"PDU Exception: {exception}, {msg}")
# Use getfieldval safely
slave_uid = (
pdu.getfieldval("slaveUid") if hasattr(pdu, "slaveUid") else "N/A"
)
run_status = (
pdu.getfieldval("runIdicatorStatus")
if hasattr(pdu, "runIdicatorStatus")
else "N/A"
)
ident_struct = (
pdu.getfieldval("slaveIdent") if hasattr(pdu, "slaveIdent") else "N/A"
)
print(
f"slaveUID: {slave_uid}, runIdicatorStatus: {run_status}, IdStruct: {ident_struct}"
)
if exception == Exceptions.UNDEFINED:
print(f"PDU Exception: {exception}, {msg}")
ident_struct = (
pdu.getfieldval("slaveIdent")
if hasattr(pdu, "slaveIdent")
else "N/A"
)
print(
f"slaveUID: {ident_struct}" if hasattr(pdu, "slaveIdent") else "N/A"
)
# Test 0x03 Read Holding Registers
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: {response!r}")
print(f"Test received: {response!r}")
pdu = self.translate_response(response)
if pdu is not None:
reg_val = (
pdu.getfieldval("registerVal") if hasattr(pdu, "registerVal") else []
)
print(f"Register values: {pdu}, len:{len(reg_val)}")
exception, msg = self.check_response(pdu, packet.getfieldval("funcCode"))
print(f"PDU Exception: {exception}, {msg}")
if pdu is not None and exception == Exceptions.UNDEFINED:
exception, msg = self.check_response(pdu, packet.funcCode)
if exception == Exceptions.UNDEFINED:
print(f"PDU Exception: {exception}")
reg_val = (
pdu.getfieldval("registerVal")
if hasattr(pdu, "registerVal")
else []
)
print(f"Register values: {pdu}, len:{len(reg_val)}")
else:
print(f"PDU Exception: {exception}, {msg}")
# Test 0x10 Write Multiple Registers
packet = self.create_request(TEST_PACKET_HOLDING_WRITE)
response = self.send_packet_and_get_response(packet, timeout=1, verbose=0)
@@ -577,50 +604,132 @@ class ModbusTestLib:
print(f"Write response ack: {response!r} ")
pdu = self.translate_response(response)
exception, msg = self.check_response(pdu, packet.getfieldval("funcCode"))
if pdu is not None: # check for pdu is sufficient
print(f"PDU Exception: {exception}, {msg}")
if pdu is not None:
exception, msg = self.check_response(pdu, packet.funcCode)
if exception == Exceptions.UNDEFINED:
print(f"PDU: StartAddr: {pdu.startAddr}, len:{pdu.quantityRegisters}")
else:
print(f"PDU Exception: {exception}, {msg}")
# Test 0x04 Read Input Registers
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: {response!r}")
assert response and len(response) > 1, "Incorrect response"
print(f"Test received: {response!r}")
pdu = self.translate_response(response)
exception, msg = self.check_response(pdu, packet.getfieldval("funcCode"))
if pdu is not None:
reg_val = (
pdu.getfieldval("registerVal") if hasattr(pdu, "registerVal") else []
)
print(f"Register values: {pdu}, len:{len(reg_val)}")
exception, msg = self.check_response(pdu, packet.funcCode)
if exception == Exceptions.UNDEFINED:
reg_val = (
pdu.getfieldval("registerVal")
if hasattr(pdu, "registerVal")
else []
)
print(f"Register values: {reg_val}, len:{len(reg_val)}")
else:
print(f"PDU Exception: {exception}, {msg}")
# Test 0x02 Read Discrete Registers
packet = self.create_request(TEST_PACKET_DISCRETE_READ)
print(f"Test: read discrete inputs request: {packet}")
response = self.send_packet_and_get_response(packet, timeout=1, verbose=0)
assert response and len(response) > 1, "Incorrect discrete input read response"
print(f"Test received: {response!r}")
pdu = self.translate_response(response)
if pdu is not None:
exception, msg = self.check_response(pdu, packet.funcCode)
if exception == Exceptions.UNDEFINED:
coil_status = (
pdu.getfieldval("inputStatus")
if hasattr(pdu, "inputStatus")
else []
)
print(f"Register values: {coil_status}, len:{len(coil_status)}")
else:
print(f"PDU Exception: {exception}, {msg}")
# Test 0x01 Read Coils
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: {response!r}")
print(f"Test received: {response!r}")
pdu = self.translate_response(response)
exception, msg = self.check_response(pdu, packet.getfieldval("funcCode"))
if pdu is not None:
coil_status = (
pdu.getfieldval("coilStatus") if hasattr(pdu, "coilStatus") else []
)
print(f"Register values: {pdu}, len:{len(coil_status)}")
exception, msg = self.check_response(pdu, packet.funcCode)
if exception == Exceptions.UNDEFINED:
coil_status = (
pdu.getfieldval("coilStatus") if hasattr(pdu, "coilStatus") else []
)
print(f"Register values: {coil_status}, len:{len(coil_status)}")
else:
print(f"PDU Exception: {exception}, {msg}")
# Test 0x0F Write Multiple Coils
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: {response!r}")
assert response and len(response) > 1, "Incorrect coil write response"
print(f"Test received: {response!r}")
pdu = self.translate_response(response)
exception, msg = self.check_response(pdu, packet.getfieldval("funcCode"))
if pdu is not None:
qty_output = (
pdu.getfieldval("quantityOutput")
if hasattr(pdu, "quantityOutput")
else 0
)
print(f"Register values: {pdu}, len: {qty_output}")
exception, msg = self.check_response(pdu, packet.funcCode)
if exception == Exceptions.UNDEFINED:
qty_output = (
pdu.getfieldval("quantityOutput")
if hasattr(pdu, "quantityOutput")
else 0
)
print(f"Register values: {pdu}, len: {qty_output}")
else:
print(f"PDU Exception: {exception}, {msg}")
# Test 0x05 Single Coil Write
packet = self.create_request(TEST_PACKET_SINGLE_COIL_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 write response"
print(f"Test received: {binascii.hexlify(bytes(response)).decode('ascii')}")
pdu = self.translate_response(response)
if pdu is not None:
exception, msg = self.check_response(pdu, packet.funcCode)
if exception == Exceptions.UNDEFINED:
qty_output = (
pdu.getfieldval("quantityOutput")
if hasattr(pdu, "quantityOutput")
else 0
)
reg_val = (
pdu.getfieldval("outputValue")
if hasattr(pdu, "outputValue")
else []
)
print(f"PDU: {pdu}, len: {qty_output}, values: {reg_val}")
else:
print(f"PDU Exception: {exception}, {msg}")
# Test 0x06 Write Single Register
packet = self.create_request(TEST_PACKET_HOLDING_REG_WRITE)
print(f"Test: write holding register request: {packet}")
response = self.send_packet_and_get_response(packet, timeout=1, verbose=0)
assert response and len(response) > 1, (
"Incorrect write holding register response"
)
print(f"Test received: {binascii.hexlify(bytes(response)).decode('ascii')}")
pdu = self.translate_response(response)
if pdu is not None:
exception, msg = self.check_response(pdu, packet.funcCode)
if exception == Exceptions.UNDEFINED:
reg_addr = (
pdu.getfieldval("registerAddr")
if hasattr(pdu, "registerAddr")
else 0
)
reg_val = (
pdu.getfieldval("registerValue")
if hasattr(pdu, "registerValue")
else 0
)
print(
f"Info: {pdu}, Register addr: {reg_addr}, registerValue: {reg_val}"
)
else:
print(f"PDU Exception: {exception}, {msg}")
self.disconnect()
####################################################################
@@ -629,6 +738,6 @@ class ModbusTestLib:
if __name__ == "__main__":
# interact(mydict=globals(), mybanner=banner)
test_lib = ModbusTestLib()
test_lib = ModbusMasterLib()
test_lib.self_test()
print("Test completed.")
tools/robot/ModbusSlaveLib.py
+1959
View File
File diff suppressed because it is too large Load Diff
tools/robot/ModbusSupport.py
+374 -239
View File
@@ -1,8 +1,12 @@
# SPDX-FileCopyrightText: 2024 Espressif Systems (Shanghai) CO LTD
# SPDX-FileCopyrightText: 2024-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
import struct
from typing import Any
import time
from datetime import datetime
from typing import Type
from scapy.packet import Packet
from enum import IntEnum
from scapy.fields import (
ShortField,
XShortField,
@@ -15,21 +19,52 @@ from scapy.fields import (
)
# 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
# to workaround some dissection issues with modbus packets and do explicit dissection of PDU
# 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",
}
MB_EXCEPTION_MASK = 0x80
MB_EXCEPTION_FUNC_MASK = 0x7F
# Supported default commands
class Commands(IntEnum):
UNDEFINED = 0x00
READ_COILS = 0x01
READ_DISCRETE_INPUTS = 0x02
READ_HOLDING_REGISTERS = 0x03
READ_INPUT_REGISTERS = 0x04
WRITE_SINGLE_COIL = 0x05
WRITE_SINGLE_HOLDING_REGISTER = 0x06
READ_EXCEPTION_STATE = 0x07
WRITE_MULTIPLE_COILS = 0x0F
WRITE_MULTIPLE_HOLDING_REGISTERS = 0x10
CUSTOM_COMMAND_41 = 0x41
REPORT_SLAVE_ID = 0x11
class Exceptions(IntEnum):
UNDEFINED = 0x00
ILLEGAL_FUNCTION = 0x01
ILLEGAL_DATA_ADDRESS = 0x02
ILLEGAL_DATA_VALUE = 0x03
SLAVE_DEVICE_FAILURE = 0x04
ACKNOWLEDGE = 0x05
SLAVE_DEVICE_BUSY = 0x06
NEGATIVE_ACKNOWLEDGE = 0x07
MEMORY_PARITY_ERROR = 0x08
GATEWAY_PATH_UNAVAILABLE = 0x10
GATEWAY_TARGET_DEVICE_FAILED_TO_RESPOND = 0x11
DEFAULT = 0xFE
MAX_EXCEPTION = 0xFF
class HandlingStateEnum(IntEnum):
DEFAULT = 0x00
RESPONDED = 0x01
IGNORED = 0x02
SKIPPED = 0x03
RANDOMIZED = 0x04
EXCEPTION = 0x05
# The common CRC16 checksum calculation method for Modbus Serial RTU frames
@@ -56,22 +91,42 @@ class ModbusMBAP(Packet):
XByteField("unitId", 0),
]
def get_time_stamp(self) -> float:
print(f"Get time_stamp of TID#{self.transId}: {self.time}")
return self.time
def get_time_stamp_str(self) -> str:
return datetime.fromtimestamp(self.time).strftime("%Y-%m-%d %H:%M:%S.%f")
def set_time_stamp(self) -> float:
self.time = time.time()
# print(f"Set time stamp of TID#{self.transId:#x}: {self.time}")
return self.time
# Can be used to replace all Modbus read
class ModbusPDU_Read_Generic(Packet):
name = "Read Generic"
fields_desc = [
XByteField("funcCode", 0x01),
XByteField("funcCode", Commands.READ_COILS),
XShortField("startAddr", 0x0000),
XShortField("quantity", 0x0001),
]
class ModbusPDU_Exception(Packet):
name = "Modbus Exception class"
fields_desc = [
XByteField("funcCode", 0x80),
ByteEnumField("exceptCode", 1, Exceptions),
]
# 0x01 - Read Coils
class ModbusPDU01_Read_Coils(Packet):
name = "Read Coils Request"
fields_desc = [
XByteField("funcCode", 0x01),
XByteField("funcCode", Commands.READ_COILS),
# 0x0000 to 0xFFFF
XShortField("startAddr", 0x0000),
XShortField("quantity", 0x0001),
@@ -81,7 +136,7 @@ class ModbusPDU01_Read_Coils(Packet):
class ModbusPDU01_Read_Coils_Answer(Packet):
name = "Read Coils Answer"
fields_desc = [
XByteField("funcCode", 0x01),
XByteField("funcCode", Commands.READ_COILS),
BitFieldLenField("byteCount", None, 8, count_of="coilStatus"),
FieldListField(
"coilStatus",
@@ -92,11 +147,11 @@ class ModbusPDU01_Read_Coils_Answer(Packet):
]
class ModbusPDU01_Read_Coils_Exception(Packet):
class ModbusPDU01_Read_Coils_Exception(ModbusPDU_Exception):
name = "Read Coils Exception"
fields_desc = [
XByteField("funcCode", 0x81),
ByteEnumField("exceptCode", 1, modbus_exceptions),
XByteField("funcCode", (Commands.READ_COILS | MB_EXCEPTION_MASK)),
ByteEnumField("exceptCode", 1, Exceptions),
]
@@ -104,7 +159,7 @@ class ModbusPDU01_Read_Coils_Exception(Packet):
class ModbusPDU02_Read_Discrete_Inputs(Packet):
name = "Read Discrete Inputs"
fields_desc = [
XByteField("funcCode", 0x02),
XByteField("funcCode", Commands.READ_DISCRETE_INPUTS),
XShortField("startAddr", 0x0000),
XShortField("quantity", 0x0001),
]
@@ -113,7 +168,7 @@ class ModbusPDU02_Read_Discrete_Inputs(Packet):
class ModbusPDU02_Read_Discrete_Inputs_Answer(Packet):
name = "Read Discrete Inputs Answer"
fields_desc = [
XByteField("funcCode", 0x02),
XByteField("funcCode", Commands.READ_DISCRETE_INPUTS),
BitFieldLenField("byteCount", None, 8, count_of="inputStatus"),
FieldListField(
"inputStatus",
@@ -124,11 +179,11 @@ class ModbusPDU02_Read_Discrete_Inputs_Answer(Packet):
]
class ModbusPDU02_Read_Discrete_Inputs_Exception(Packet):
class ModbusPDU02_Read_Discrete_Inputs_Exception(ModbusPDU_Exception):
name = "Read Discrete Inputs Exception"
fields_desc = [
XByteField("funcCode", 0x82),
ByteEnumField("exceptCode", 1, modbus_exceptions),
XByteField("funcCode", (Commands.READ_DISCRETE_INPUTS | MB_EXCEPTION_MASK)),
ByteEnumField("exceptCode", 1, Exceptions),
]
@@ -136,7 +191,7 @@ class ModbusPDU02_Read_Discrete_Inputs_Exception(Packet):
class ModbusPDU03_Read_Holding_Registers(Packet):
name = "Read Holding Registers"
fields_desc = [
XByteField("funcCode", 0x03),
XByteField("funcCode", Commands.READ_HOLDING_REGISTERS),
XShortField("startAddr", 0x0001),
XShortField("quantity", 0x0002),
]
@@ -145,22 +200,22 @@ class ModbusPDU03_Read_Holding_Registers(Packet):
class ModbusPDU03_Read_Holding_Registers_Answer(Packet):
name = "Read Holding Registers Answer"
fields_desc = [
XByteField("funcCode", 0x03),
XByteField("funcCode", Commands.READ_HOLDING_REGISTERS),
BitFieldLenField("byteCount", None, 8, count_of="registerVal"),
FieldListField(
"registerVal",
[0x0000],
ShortField("", 0x0000),
XShortField("", 0x0000),
count_from=lambda pkt: pkt.byteCount,
),
]
class ModbusPDU03_Read_Holding_Registers_Exception(Packet):
class ModbusPDU03_Read_Holding_Registers_Exception(ModbusPDU_Exception):
name = "Read Holding Registers Exception"
fields_desc = [
XByteField("funcCode", 0x83),
ByteEnumField("exceptCode", 1, modbus_exceptions),
XByteField("funcCode", (Commands.READ_HOLDING_REGISTERS | MB_EXCEPTION_MASK)),
ByteEnumField("exceptCode", 1, Exceptions),
]
@@ -168,7 +223,7 @@ class ModbusPDU03_Read_Holding_Registers_Exception(Packet):
class ModbusPDU04_Read_Input_Registers(Packet):
name = "Read Input Registers"
fields_desc = [
XByteField("funcCode", 0x04),
XByteField("funcCode", Commands.READ_INPUT_REGISTERS),
XShortField("startAddr", 0x0000),
XShortField("quantity", 0x0001),
]
@@ -177,24 +232,24 @@ class ModbusPDU04_Read_Input_Registers(Packet):
class ModbusPDU04_Read_Input_Registers_Answer(Packet):
name = "Read Input Registers Response"
fields_desc = [
XByteField("funcCode", 0x04),
XByteField("funcCode", Commands.READ_INPUT_REGISTERS),
BitFieldLenField(
"byteCount", None, 8, count_of="registerVal", adjust=lambda pkt, x: x * 2
),
FieldListField(
"registerVal",
[0x0000],
ShortField("", 0x0000),
XShortField("", 0x0000),
count_from=lambda pkt: pkt.byteCount,
),
]
class ModbusPDU04_Read_Input_Registers_Exception(Packet):
class ModbusPDU04_Read_Input_Registers_Exception(ModbusPDU_Exception):
name = "Read Input Registers Exception"
fields_desc = [
XByteField("funcCode", 0x84),
ByteEnumField("exceptCode", 1, modbus_exceptions),
XByteField("funcCode", (Commands.READ_INPUT_REGISTERS | MB_EXCEPTION_MASK)),
ByteEnumField("exceptCode", 1, Exceptions),
]
@@ -202,7 +257,7 @@ class ModbusPDU04_Read_Input_Registers_Exception(Packet):
class ModbusPDU05_Write_Single_Coil(Packet):
name = "Write Single Coil"
fields_desc = [
XByteField("funcCode", 0x05),
XByteField("funcCode", Commands.WRITE_SINGLE_COIL),
XShortField("outputAddr", 0x0000),
XShortField("outputValue", 0x0000),
]
@@ -211,17 +266,17 @@ class ModbusPDU05_Write_Single_Coil(Packet):
class ModbusPDU05_Write_Single_Coil_Answer(Packet):
name = "Write Single Coil"
fields_desc = [
XByteField("funcCode", 0x05),
XByteField("funcCode", Commands.WRITE_SINGLE_COIL),
XShortField("outputAddr", 0x0000),
XShortField("outputValue", 0x0000),
]
class ModbusPDU05_Write_Single_Coil_Exception(Packet):
class ModbusPDU05_Write_Single_Coil_Exception(ModbusPDU_Exception):
name = "Write Single Coil Exception"
fields_desc = [
XByteField("funcCode", 0x85),
ByteEnumField("exceptCode", 1, modbus_exceptions),
XByteField("funcCode", (Commands.WRITE_SINGLE_COIL | MB_EXCEPTION_MASK)),
ByteEnumField("exceptCode", 1, Exceptions),
]
@@ -229,7 +284,7 @@ class ModbusPDU05_Write_Single_Coil_Exception(Packet):
class ModbusPDU06_Write_Single_Register(Packet):
name = "Write Single Register"
fields_desc = [
XByteField("funcCode", 0x06),
XByteField("funcCode", Commands.WRITE_SINGLE_HOLDING_REGISTER),
XShortField("registerAddr", 0x0000),
XShortField("registerValue", 0x0000),
]
@@ -238,36 +293,41 @@ class ModbusPDU06_Write_Single_Register(Packet):
class ModbusPDU06_Write_Single_Register_Answer(Packet):
name = "Write Single Register Answer"
fields_desc = [
XByteField("funcCode", 0x06),
XByteField("funcCode", Commands.WRITE_SINGLE_HOLDING_REGISTER),
XShortField("registerAddr", 0x0000),
XShortField("registerValue", 0x0000),
]
class ModbusPDU06_Write_Single_Register_Exception(Packet):
class ModbusPDU06_Write_Single_Register_Exception(ModbusPDU_Exception):
name = "Write Single Register Exception"
fields_desc = [
XByteField("funcCode", 0x86),
ByteEnumField("exceptCode", 1, modbus_exceptions),
XByteField(
"funcCode", (Commands.WRITE_SINGLE_HOLDING_REGISTER | MB_EXCEPTION_MASK)
),
ByteEnumField("exceptCode", 1, Exceptions),
]
# 0x07 - Read Exception Status (Serial Line Only)
class ModbusPDU07_Read_Exception_Status(Packet):
name = "Read Exception Status"
fields_desc = [XByteField("funcCode", 0x07)]
fields_desc = [XByteField("funcCode", Commands.READ_EXCEPTION_STATE)]
class ModbusPDU07_Read_Exception_Status_Answer(Packet):
name = "Read Exception Status Answer"
fields_desc = [XByteField("funcCode", 0x07), XByteField("startAddr", 0x00)]
fields_desc = [
XByteField("funcCode", Commands.READ_EXCEPTION_STATE),
XByteField("startAddr", 0x00),
]
class ModbusPDU07_Read_Exception_Status_Exception(Packet):
class ModbusPDU07_Read_Exception_Status_Exception(ModbusPDU_Exception):
name = "Read Exception Status Exception"
fields_desc = [
XByteField("funcCode", 0x87),
ByteEnumField("exceptCode", 1, modbus_exceptions),
XByteField("funcCode", (Commands.READ_EXCEPTION_STATE | MB_EXCEPTION_MASK)),
ByteEnumField("exceptCode", 1, Exceptions),
]
@@ -275,7 +335,7 @@ class ModbusPDU07_Read_Exception_Status_Exception(Packet):
class ModbusPDU0F_Write_Multiple_Coils(Packet):
name = "Write Multiple Coils"
fields_desc = [
XByteField("funcCode", 0x0F),
XByteField("funcCode", Commands.WRITE_MULTIPLE_COILS),
XShortField("startAddr", 0x0000),
XShortField("quantityOutput", 0x0001),
BitFieldLenField(
@@ -293,24 +353,24 @@ class ModbusPDU0F_Write_Multiple_Coils(Packet):
class ModbusPDU0F_Write_Multiple_Coils_Answer(Packet):
name = "Write Multiple Coils Answer"
fields_desc = [
XByteField("funcCode", 0x0F),
XByteField("funcCode", Commands.WRITE_MULTIPLE_COILS),
XShortField("startAddr", 0x0000),
XShortField("quantityOutput", 0x0001),
]
class ModbusPDU0F_Write_Multiple_Coils_Exception(Packet):
class ModbusPDU0F_Write_Multiple_Coils_Exception(ModbusPDU_Exception):
name = "Write Multiple Coils Exception"
fields_desc = [
XByteField("funcCode", 0x8F),
ByteEnumField("exceptCode", 1, modbus_exceptions),
XByteField("funcCode", (Commands.WRITE_MULTIPLE_COILS | MB_EXCEPTION_MASK)),
ByteEnumField("exceptCode", 1, Exceptions),
]
class ModbusPDU10_Write_Multiple_Registers(Packet):
name = "Write Multiple Registers"
fields_desc = [
XByteField("funcCode", 0x10),
XByteField("funcCode", Commands.WRITE_MULTIPLE_HOLDING_REGISTERS),
XShortField("startAddr", 0x0000),
BitFieldLenField("quantityRegisters", None, 16, count_of="outputsValue"),
BitFieldLenField(
@@ -325,47 +385,49 @@ class ModbusPDU10_Write_Multiple_Registers(Packet):
]
class ModbusPDU10_Write_Multiple_Registers_Serial(ModbusPDU10_Write_Multiple_Registers):
name = "Write Multiple Registers Serial"
_crc: int = 0
# 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 get_crc(self) -> int:
# return self._crc
def post_build(self, p: bytes, pay: bytes) -> bytes:
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!r}, checksum = {self._crc!r}")
return p
# def post_build(self, p: bytes, pay: bytes) -> bytes:
# 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!r}, checksum = {self._crc!r}")
# return p
def guess_payload_class(self, payload: bytes) -> Any:
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)
# def guess_payload_class(self, payload: bytes) -> Any:
# 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),
XByteField("funcCode", Commands.WRITE_MULTIPLE_HOLDING_REGISTERS),
XShortField("startAddr", 0x0000),
XShortField("quantityRegisters", 0x0001),
]
class ModbusPDU10_Write_Multiple_Registers_Exception(Packet):
class ModbusPDU10_Write_Multiple_Registers_Exception(ModbusPDU_Exception):
name = "Write Multiple Registers Exception"
fields_desc = [
XByteField("funcCode", 0x90),
ByteEnumField("exceptCode", 1, modbus_exceptions),
XByteField(
"funcCode", (Commands.WRITE_MULTIPLE_HOLDING_REGISTERS | MB_EXCEPTION_MASK)
),
ByteEnumField("exceptCode", 1, Exceptions),
]
@@ -375,11 +437,11 @@ class ModbusPDUXX_Custom_Request(Packet):
fields_desc = [FieldListField("customBytes", [0x00], XByteField("", 0x00))]
class ModbusPDUXX_Custom_Exception(Packet):
class ModbusPDUXX_Custom_Exception(ModbusPDU_Exception):
name = "Custom Command Exception"
fields_desc = [
XByteField("funcCode", 0x00),
ByteEnumField("exceptCode", 1, modbus_exceptions),
XByteField("funcCode", MB_EXCEPTION_MASK),
ByteEnumField("exceptCode", 1, Exceptions),
]
@@ -389,16 +451,16 @@ class ModbusPDUXX_Custom_Answer(Packet):
fields_desc = [
ConditionalField(
XByteField("funcCode", 0x00),
lambda pkt: (type(pkt.underlayer) is ModbusADU_Response),
lambda pkt: type(pkt.underlayer) is ModbusADU_Response,
),
ConditionalField(
FieldListField(
"customBytes",
[0x00],
XByteField("", 0x00),
count_from=lambda pkt: pkt.underlayer.len
if pkt.underlayer is not None
else 0,
length_from=lambda pkt: (
pkt.underlayer.len if pkt.underlayer is not None else 0
),
),
lambda pkt: type(pkt.underlayer) is ModbusADU_Response,
),
@@ -408,13 +470,13 @@ class ModbusPDUXX_Custom_Answer(Packet):
# 0x11 - Report Slave Id
class ModbusPDU11_Report_Slave_Id(Packet):
name = "Report Slave Id"
fields_desc = [XByteField("funcCode", 0x11)]
fields_desc = [XByteField("funcCode", Commands.REPORT_SLAVE_ID)]
class ModbusPDU11_Report_Slave_Id_Answer(Packet):
name = "Report Slave Id Answer"
fields_desc = [
XByteField("funcCode", 0x11),
XByteField("funcCode", Commands.REPORT_SLAVE_ID),
BitFieldLenField("byteCount", None, 8, length_of="slaveUId"),
ConditionalField(XByteField("slaveUid", 0x00), lambda pkt: pkt.byteCount > 0),
ConditionalField(
@@ -429,20 +491,174 @@ class ModbusPDU11_Report_Slave_Id_Answer(Packet):
),
lambda pkt: pkt.byteCount > 0,
),
# ConditionalField(
# XByteField("slaveUid", 0x00),
# lambda pkt: pkt.byteCount>0
# ),
# ConditionalField(
# XByteField("runIdicatorStatus", 0x00),
# lambda pkt: pkt.byteCount>0
# ),
# ConditionalField(
# FieldListField("slaveIdent", [0x00],
# XByteField("", 0x00),
# count_from = lambda pkt: pkt.byteCount
# ),
# 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),
XByteField("funcCode", (Commands.REPORT_SLAVE_ID | MB_EXCEPTION_MASK)),
ByteEnumField("exceptCode", 1, Exceptions),
]
class ModbusADU_Response(ModbusMBAP):
name = "ModbusADU Response"
# static fields of the class used for dissection
_mb_exception: Exceptions = Exceptions.UNDEFINED
_last_packet: Packet = None
_modbus_pdu: Packet = None
_seq_num: int = 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
]
@classmethod
def get_sequence_num(self) -> int:
print(f"Get sequence number: {self._seq_num}")
return self._seq_num
@classmethod
def get_last_exception(self) -> Exceptions:
return self._mb_exception
# def get_hexcap(self):
# packet = self.from_hexcap()
# print(f"Hexcap dump: {packet}")
# return packet
# def extract_padding(self, s):
# print(f'Extract padding: {self, s, self.len, self.underlayer}')
# return self.guess_payload_class( s) #, s #self.extract_pedding(self, s)
def pre_dissect(self, s: bytes) -> bytes:
# print(f'Pre desect class: {self, s, self.len, self.underlayer}, seq_num: {self.__class__._seq_num}')
_last_packet = self
self.__class__._seq_num += 1
return s
# Dissects packets
def guess_payload_class(self, payload: bytes) -> Type[Packet]:
funcCode = payload[0]
self._mb_exception = Exceptions.UNDEFINED
if funcCode == Commands.READ_COILS:
self._modbus_pdu = ModbusPDU01_Read_Coils_Answer
return ModbusPDU01_Read_Coils_Answer
elif funcCode == (Commands.READ_COILS | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDU01_Read_Coils_Exception
return ModbusPDU01_Read_Coils_Exception
elif funcCode == Commands.READ_DISCRETE_INPUTS:
self._modbus_pdu = ModbusPDU02_Read_Discrete_Inputs_Answer
return ModbusPDU02_Read_Discrete_Inputs_Answer
elif funcCode == (Commands.READ_DISCRETE_INPUTS | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDU02_Read_Discrete_Inputs_Exception
return ModbusPDU02_Read_Discrete_Inputs_Exception
elif funcCode == Commands.READ_HOLDING_REGISTERS:
self._modbus_pdu = ModbusPDU03_Read_Holding_Registers_Answer
return ModbusPDU03_Read_Holding_Registers_Answer
elif funcCode == (Commands.READ_HOLDING_REGISTERS | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDU03_Read_Holding_Registers_Exception
return ModbusPDU03_Read_Holding_Registers_Exception
elif funcCode == Commands.READ_INPUT_REGISTERS:
self._modbus_pdu = ModbusPDU04_Read_Input_Registers_Answer
return ModbusPDU04_Read_Input_Registers_Answer
elif funcCode == (Commands.READ_INPUT_REGISTERS | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDU04_Read_Input_Registers_Exception
return ModbusPDU04_Read_Input_Registers_Exception
elif funcCode == Commands.WRITE_SINGLE_COIL:
self._modbus_pdu = ModbusPDU05_Write_Single_Coil_Answer
return ModbusPDU05_Write_Single_Coil_Answer
elif funcCode == (Commands.WRITE_SINGLE_COIL | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDU05_Write_Single_Coil_Exception
return ModbusPDU05_Write_Single_Coil_Exception
elif funcCode == Commands.WRITE_SINGLE_HOLDING_REGISTER:
self._modbus_pdu = ModbusPDU06_Write_Single_Register_Answer
return ModbusPDU06_Write_Single_Register_Answer
elif funcCode == (Commands.WRITE_SINGLE_HOLDING_REGISTER | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDU06_Write_Single_Register_Exception
return ModbusPDU06_Write_Single_Register_Exception
elif funcCode == Commands.READ_EXCEPTION_STATE:
self._modbus_pdu = ModbusPDU07_Read_Exception_Status_Answer
return ModbusPDU07_Read_Exception_Status_Answer
elif funcCode == (Commands.READ_EXCEPTION_STATE | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDU07_Read_Exception_Status_Exception
return ModbusPDU07_Read_Exception_Status_Exception
elif funcCode == Commands.WRITE_MULTIPLE_COILS:
self._modbus_pdu = ModbusPDU0F_Write_Multiple_Coils_Answer
return ModbusPDU0F_Write_Multiple_Coils_Answer
elif funcCode == (Commands.WRITE_MULTIPLE_COILS | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDU0F_Write_Multiple_Coils_Exception
return ModbusPDU0F_Write_Multiple_Coils_Exception
elif funcCode == Commands.WRITE_MULTIPLE_HOLDING_REGISTERS:
self._modbus_pdu = ModbusPDU10_Write_Multiple_Registers_Answer
return ModbusPDU10_Write_Multiple_Registers_Answer
elif funcCode == (
Commands.WRITE_MULTIPLE_HOLDING_REGISTERS | MB_EXCEPTION_MASK
):
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDU10_Write_Multiple_Registers_Exception
return ModbusPDU10_Write_Multiple_Registers_Exception
elif funcCode == Commands.REPORT_SLAVE_ID:
self._modbus_pdu = ModbusPDU11_Report_Slave_Id_Answer
return ModbusPDU11_Report_Slave_Id_Answer
elif funcCode == (Commands.REPORT_SLAVE_ID | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDU11_Report_Slave_Id_Exception
return ModbusPDU11_Report_Slave_Id_Exception
else:
if funcCode & MB_EXCEPTION_MASK:
self._mb_exception = Exceptions(payload[1])
self._modbus_pdu = ModbusPDUXX_Custom_Exception
return ModbusPDUXX_Custom_Exception
self._modbus_pdu = ModbusPDUXX_Custom_Answer
return ModbusPDUXX_Custom_Answer
# return Packet.guess_payload_class(self, payload)
class ModbusADU_Request(ModbusMBAP):
name = "ModbusADU Request"
_mb_exception: int = 0
_mb_exception: Exceptions = Exceptions.UNDEFINED
_seq_num: int = 0
_last_packet: Packet = None
fields_desc = [
XShortField("transId", 0x0000), # needs to be unique
XShortField("protoId", 0x0000), # needs to be zero (Modbus)
@@ -452,85 +668,104 @@ class ModbusADU_Request(ModbusMBAP):
), # 0xFF or 0x00 should be used for Modbus over TCP/IP
]
def mb_get_last_exception(self) -> int:
@classmethod
def get_sequence_num(self) -> int:
print(f"Get sequence number: {self._seq_num}")
return self._seq_num
def pre_dissect(self, s: bytes) -> bytes:
# print(f'Pre desect class: {self, s, self.len, self.underlayer}, seq_num: {self.__class__._seq_num}, time: {self.time}')
_last_packet = self
self.__class__._seq_num += 1
return s
def mb_get_last_exception(self) -> Exceptions:
return self._mb_exception
# Dissects packets
def guess_payload_class(self, payload: bytes) -> Packet:
funcCode = int(payload[0])
# print(f'Request guess payload class func: {funcCode}')
self._mb_exception = 0
# try:
# print(f'Request guess payload class func: {funcCode}:{Commands(funcCode).name}')
# except Exception as e:
# print(f'Request guess payload class func: {funcCode}:Unsupported')
self._mb_exception = Exceptions.UNDEFINED
if funcCode == 0x01:
if funcCode == Commands.READ_COILS:
return ModbusPDU01_Read_Coils
elif funcCode == 0x81:
self._mb_exception = int(payload[1])
elif funcCode == (Commands.READ_COILS | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
return ModbusPDU01_Read_Coils_Exception
elif funcCode == 0x02:
elif funcCode == Commands.READ_DISCRETE_INPUTS:
return ModbusPDU02_Read_Discrete_Inputs
elif funcCode == 0x82:
self._mb_exception = int(payload[1])
elif funcCode == (Commands.READ_DISCRETE_INPUTS | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
return ModbusPDU02_Read_Discrete_Inputs_Exception
elif funcCode == 0x03:
elif funcCode == Commands.READ_HOLDING_REGISTERS:
return ModbusPDU03_Read_Holding_Registers
elif funcCode == 0x83:
self._mb_exception = int(payload[1])
elif funcCode == (Commands.READ_HOLDING_REGISTERS | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
return ModbusPDU03_Read_Holding_Registers_Exception
elif funcCode == 0x04:
elif funcCode == Commands.READ_INPUT_REGISTERS:
return ModbusPDU04_Read_Input_Registers
elif funcCode == 0x84:
self._mb_exception = int(payload[1])
elif funcCode == (Commands.READ_INPUT_REGISTERS | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
return ModbusPDU04_Read_Input_Registers_Exception
elif funcCode == 0x05:
elif funcCode == Commands.WRITE_SINGLE_COIL:
return ModbusPDU05_Write_Single_Coil
elif funcCode == 0x85:
self._mb_exception = int(payload[1])
elif funcCode == (Commands.WRITE_SINGLE_COIL | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
return ModbusPDU05_Write_Single_Coil_Exception
elif funcCode == 0x06:
elif funcCode == Commands.WRITE_SINGLE_HOLDING_REGISTER:
return ModbusPDU06_Write_Single_Register
elif funcCode == 0x86:
self._mb_exception = int(payload[1])
elif funcCode == (Commands.WRITE_SINGLE_HOLDING_REGISTER | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
return ModbusPDU06_Write_Single_Register_Exception
elif funcCode == 0x07:
elif funcCode == Commands.READ_EXCEPTION_STATE:
return ModbusPDU07_Read_Exception_Status
elif funcCode == 0x87:
self._mb_exception = int(payload[1])
elif funcCode == (Commands.READ_EXCEPTION_STATE | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
return ModbusPDU07_Read_Exception_Status_Exception
elif funcCode == 0x0F:
elif funcCode == Commands.WRITE_MULTIPLE_COILS:
return ModbusPDU0F_Write_Multiple_Coils
elif funcCode == 0x8F:
self._mb_exception = int(payload[1])
elif funcCode == (Commands.WRITE_MULTIPLE_COILS | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(payload[1])
return ModbusPDU0F_Write_Multiple_Coils_Exception
elif funcCode == 0x11:
elif funcCode == Commands.REPORT_SLAVE_ID:
return ModbusPDU11_Report_Slave_Id
elif funcCode == 0x91:
self._mb_exception = int(payload[1])
elif funcCode == (Commands.REPORT_SLAVE_ID | MB_EXCEPTION_MASK):
self._mb_exception = Exceptions(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])
elif funcCode == (
Commands.WRITE_MULTIPLE_HOLDING_REGISTERS | MB_EXCEPTION_MASK
):
return ModbusPDU10_Write_Multiple_Registers
elif funcCode == (
Commands.WRITE_MULTIPLE_HOLDING_REGISTERS | MB_EXCEPTION_MASK
):
self._mb_exception = Exceptions(payload[1])
return ModbusPDU10_Write_Multiple_Registers_Exception
else:
if funcCode < 0x80 and len(payload) > 0: # Check for non-exception packets
if (
funcCode < MB_EXCEPTION_MASK and len(payload) > 0
): # Check for non-exception packets
# Assume custom request if it's not a known function code
return ModbusPDUXX_Custom_Request
elif funcCode & 0x80:
elif funcCode & MB_EXCEPTION_MASK:
# Assume custom exception if it's an exception but unknown
self._mb_exception = int(payload[1]) if len(payload) > 1 else 0
self._mb_exception = (
Exceptions(payload[1]) if len(payload) > 1 else Exceptions.UNDEFINED
)
return ModbusPDUXX_Custom_Exception
return Packet.guess_payload_class(self, payload)
@@ -545,105 +780,5 @@ class ModbusADU_Request(ModbusMBAP):
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: int = 0
_current_main_packet: Any = None
_modbus_pdu: Any = None
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) -> int:
return self._mb_exception
# def extract_padding(self, s):
# print(f'Extract pedding: {self, s, self.len, self.underlayer}')
# return self.guess_payload_class( s) #, s #self.extract_pedding(self, s)
def pre_dissect(self, s: bytes) -> bytes:
print(f"Pre desect: {self, s, self.len, self.underlayer}")
_current_main_packet = self
return s
# Dissects packets
def guess_payload_class(self, payload: bytes) -> Packet:
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:
if funcCode & 0x80:
self._mb_exception = int(payload[1]) if len(payload) > 1 else 0
return ModbusPDUXX_Custom_Exception
return ModbusPDUXX_Custom_Answer
# return Packet.guess_payload_class(self, payload)
return "%s = %s" % (f.name, reprval)
return ""
tools/robot/ModbusTestSuite.resource → tools/robot/ModbusTestSuiteMaster.resource
+9 -20
View File
@@ -1,21 +1,12 @@
*** Variables ***
${MODBUS_DEF_SERVER_IP} 127.0.0.1
${MODBUS_DEF_IP} 127.0.0.1
${MODBUS_DEF_PORT} 1502
${PAR1} 1
${PAR2} 2
${FUNC_REPORT_SLAVE_ID} ${0x11}
${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
Library ModbusMasterLib.py WITH NAME ModbusMasterLib
*** Keywords ***
Create Custom Command Request
@@ -25,47 +16,46 @@ Create Custom Command Request
Create Report Slave Id Request
[Arguments] ${uid} ${customData}
#${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU11_Report_Slave_Id(funcCode=${FUNC_REPORT_SLAVE_ID})
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDUXX_Custom_Request(customBytes=${customData})
RETURN ${packet}
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})
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0, len=6)/ModbusPDU04_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})
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0, len=6)/ModbusPDU03_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})
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU10_Write_Multiple_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})
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU06_Write_Single_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})
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU01_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})
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU0F_Write_Multiple_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})
${packet} = Create Request ModbusADU_Request(unitId=${uid}, protoId=0)/ModbusPDU02_Read_Discrete_Inputs(startAddr=${startAddr}, quantity=${quantity})
Log Packet: ${packet}
RETURN ${packet}
@@ -153,7 +143,6 @@ Read Input Registers
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}
tools/robot/ModbusTestSuite.robot → tools/robot/ModbusTestSuiteMaster.robot
+5 -4
View File
@@ -1,10 +1,11 @@
*** Settings ***
Documentation A test suite for Modbus commands.
...
Documentation A test suite for Modbus Slave testing.
... Implements the Modbus Master (Client) to send requests
... for Modbus Slave side (the DUT) and verifies the answers.
... Keywords are imported from the resource file
Resource ModbusTestSuite.resource
Resource ModbusTestSuiteMaster.resource
Default Tags multi_dut_modbus_generic
Suite Setup Create Connection ${MODBUS_DEF_SERVER_IP} ${MODBUS_DEF_PORT}
Suite Setup Create Connection ${MODBUS_DEF_IP} ${MODBUS_DEF_PORT}
Suite Teardown Disconnect
*** Variables ***
tools/robot/ModbusTestSuiteSlave.resource
+176
View File
@@ -0,0 +1,176 @@
*** Settings ***
Library Collections
Library ModbusSlaveLib.py WITH NAME ModbusSlaveLib
Library AsyncLibrary
*** Variables ***
${MODBUS_SERVER_ADDRESS} 127.0.0.1
${MODBUS_DEF_IP} 127.0.0.1
${MODBUS_DEF_PORT} ${1502}
${MODBUS_TIMEOUT} ${30}
@{EXP_TRANSACTIONS}
@{REGISTERED_HANDLES}
*** Keywords ***
Test Ensure Server Started
[Arguments] ${port}=${MODBUS_DEF_PORT} ${timeout}=${MODBUS_TIMEOUT} ${async}=False
[Documentation] Start the Modbus server and begin capturing packets
${class_uid} = ModbusSlaveLib.Get Class Id
Log "Try to start server: ${class_uid}:${port}"
${status} = ModbusSlaveLib.Start Server ${port} ${timeout} ${async}
RETURN ${status}
Test Stop Server
[Documentation] Stop the Modbus server and packet capture
ModbusSlaveLib.Stop Server
Test Report Server Address
[Arguments] ${address}=${None}
[Documentation] Test wrapper to get active server address and port tuple (handled by listener)
Log "Server reported address: ${address}"
Set Suite Variable $MODBUS_SERVER_ADDRESS ${address} children=True
Test Register Transaction
[Arguments] ${uid} ${func_code} ${start_addr}=0 ${quantity}=1 ${exception}=0 ${data}=${None} ${expected_response}=${None}
[Documentation] Register expected request and response
${class_uid} = ModbusSlaveLib.Get Class Id
Log "Library class: ${class_uid}"
${exp_transaction} = ModbusSlaveLib.Add Expected Transaction
... ${{eval($uid)}} ${{eval($func_code)}} ${{eval($start_addr)}} ${{eval($quantity)}} ${{eval($exception)}} ${{eval('${data}')}} ${{eval('${expected_response}')}}
Append To List ${EXP_TRANSACTIONS} ${exp_transaction}
Log "Transaction registered: ${exp_transaction}"
RETURN ${exp_transaction}
Test Register Transaction Wait Confirmation
[Arguments] ${uid} ${func_code} ${start_addr}=0 ${quantity}=1 ${exception}=0 ${data}=${None} ${expected_response}=${None} ${timeout}=${20}
[Documentation] Append expected request and response to the list then wait for transaction to be handled
Test Ensure Server Started ${MODBUS_DEF_PORT} ${timeout} ${True}
${transaction} = Test Register Transaction ${uid} ${func_code} ${start_addr} ${quantity} ${exception} ${data} ${expected_response}
Log "Transaction created: ${transaction}"
${result} = ModbusSlaveLib.Wait Transaction Confirmation transaction=${transaction} timeout=${timeout}
# Should Not Be Equal ${result} ${None}
Run Keyword If ${result} == ${None} Fail "No confirmation received for transaction: ${transaction.request.summary()}, ${transaction.response.summary()}"
RETURN ${result}
Test Async Run Modbus Case
[Arguments] ${name} ${uid} ${func_code} ${start_addr}=0 ${quantity}=1 ${exception}=0 ${data}=${None} ${expected_response}=${None} ${timeout}=${20}
[Documentation] Register expected transaction asynchronously (register+wait wrapper) and append handle
# Log "Parameters: ${uid} ${func_code} ${start_addr} ${quantity} ${exception} ${data} ${expected_response}"
${handle} = Async Run Test Register Transaction Wait Confirmation ${uid} ${func_code} ${start_addr} ${quantity} ${exception} ${data} ${expected_response} ${timeout}
Append To List ${REGISTERED_HANDLES} ${handle}
Log "Handle for function: ${func_code}, ${handle}, List: ${REGISTERED_HANDLES}"
# Register parameter name in test message attribute for tracing
Run Keyword If ${name} and ${${name.__len__()}} Set Test Message ${name} append=yes
RETURN ${handle}
Test Async Run Modbus Case Register Only
[Arguments] ${name} ${uid} ${func_code} ${start_addr}=0 ${quantity}=1 ${exception}=0 ${data}=${None} ${expected_response}=${None} ${timeout}=${2}
[Documentation] Register expected transaction asynchronously (register-only) and append async handle
${handle} = Async Run Test Register Transaction ${uid} ${func_code} ${start_addr} ${quantity} ${exception} ${data} ${expected_response}
Test Ensure Server Started ${MODBUS_DEF_PORT} ${timeout} ${True}
Append To List ${REGISTERED_HANDLES} ${handle}
Log "Register-only handle for function: ${func_code}, ${handle}, List: ${REGISTERED_HANDLES}"
Run Keyword If ${name} and ${${name.__len__()}} Set Test Message ${name} append=yes
RETURN ${handle}
Test Async Complete Modbus Cases
[Arguments] ${timeout}=${2}
[Documentation] Wait for all previously registered async test-case handles to complete and verify confirmations
# Allow small time for all async registration tasks to finish (acts like an "await" window)
Test Ensure Server Started ${MODBUS_DEF_PORT} ${timeout} ${True}
${address} = Get Server Address
Log "Server reported address: ${address}"
Test Report Server Address ${address[0]}
${n_keywords} = Get Length ${REGISTERED_HANDLES}
Log "Completing ${n_keywords} async tasks: ${REGISTERED_HANDLES}"
FOR ${i} IN RANGE ${n_keywords}
${handle} = Get From List ${REGISTERED_HANDLES} ${i}
Log "Completing transaction handle ${i} = ${handle}"
${return_value} Async Get ${handle}
${status} = Set Variable ${return_value}
IF "${return_value} is not ${None}"
${is_transaction} = Evaluate isinstance(${return_value}, ModbusSlaveLib.Transaction)
IF ${is_transaction}
${status} = ModbusSlaveLib.Wait Transaction Confirmation ${return_value} ${timeout}
Log "Keyword ${i}, transaction: ${return_value}, confirmation: ${status}"
Should Not Be Equal ${status} ${None}
ELSE
${is_int} = Evaluate isinstance(${status}, int)
IF ${is_int}
Log "Keyword ${i}, confirmed with TID: ${status}"
END
END
ELSE
Log "Keyword ${i}, is not confirmed, result: ${status}"
Fail "Keyword ${i} is not confirmed."
END
END
ModbusSlaveLib.Stop Server
Log "All async registered transactions completed."
BuiltIn.Sleep ${3}
# Clear registered handles list for next run
Set Test Variable @{REGISTERED_HANDLES} @{EMPTY}
RETURN
Test Start Verify Synchronous
[Arguments] ${uid} ${func_code} ${start_addr}=0 ${quantity}=1 ${exception}=0 ${data}=${None} ${expected_response}=${None} ${timeout}=${2}
[Documentation] Test a Modbus request and response
# Add the expected request and response
${class_uid} = ModbusSlaveLib.Get Class Id
Log "Library class: ${class_uid}"
Log "Parameters: ${uid} ${func_code} ${start_addr} ${quantity} ${exception} ${data} ${expected_response}"
ModbusSlaveLib.Add Expected Transaction ${{eval($uid)}} ${{eval($func_code)}} ${{eval($start_addr)}} ${{eval($quantity)}} ${{eval($exception)}} ${{eval('${data}')}} ${{eval('${expected_response}')}}
${server_state} = ModbusSlaveLib.Is Server Active
IF not ${server_state}
Test Ensure Server Started ${MODBUS_DEF_PORT} ${timeout} ${True}
END
# Get last processed transaction
${transaction} = ModbusSlaveLib.Wait Transaction Data timeout=${timeout}
Should Not Be Empty ${transaction.request} No request received
Should Not Be Empty ${transaction.response} No response sent
# Verify the request
${ex_code} = ModbusSlaveLib.Verify Expected Request ${transaction.request} ${uid} ${func_code} ${start_addr} ${quantity} ${exception} ${data}
Should Be Equal As Integers ${exception} ${ex_code} Request verification failed
Test Start Verify Asynchronous
[Arguments] ${timeout}=${2}
[Documentation] Start Modbus answering machine with timeout and then check registered expectations
${expectations} = Copy List ${EXP_TRANSACTIONS}
Test Ensure Server Started ${MODBUS_DEF_PORT} ${MODBUS_TIMEOUT} ${True}
${result} = ModbusSlaveLib.Verify Expectations ${expectations} ${timeout}
IF ${result} == ${0}
Log "All expectations met exp_length=${expectations.__len__()}"
ELSE
${exp_length} = Get length ${expectations}
IF ${exp_length} == ${0}
Log "All expectations met, exp_length=${exp_length}"
ELSE
Log "Status for following expectations (${exp_length}):"
${counter} = Set Variable ${0}
FOR ${i} IN RANGE ${exp_length}
${exp_transaction} = Get From List ${expectations} ${i}
Log "Transaction state: ${exp_transaction.state}, exception: ${exp_transaction.exception}"
IF "${exp_transaction}" != "${None}" and "${exp_transaction.response}" != "${None}"
IF "${exp_transaction.exception} == ${0}" and "${exp_transaction.state} == ${1}"
Log "Expectation ${i} - PASS: Request: ${exp_transaction.request} | Response: ${exp_transaction.response}"
ELSE
Log "Expectation ${i} - FAIL with ${exp_transaction.exception}: Request: ${exp_transaction.request} | Response: ${exp_transaction.response}"
${counter} = Set Variable ${counter + 1}
END
END
END
Run Keyword If ${counter} > 0 Fail "Found ${counter} failed expectations."
END
END
tools/robot/ModbusTestSuiteSlave.robot
+121
View File
@@ -0,0 +1,121 @@
*** Settings ***
Documentation A test suite for Modbus Master testing.
... This suite tests the Modbus slave implementation by waiting for
... requests from the master and sending appropriate responses.
... Keywords are imported from the resource file
Resource ModbusTestSuiteSlave.resource
Default Tags modbus_slave
Suite Setup Set Log Level DEBUG
# Suite Teardown Test Start Verify Asynchronous ${20}
Suite Teardown Test Async Complete Modbus Cases ${30}
*** Variables ***
${BINARY_REGS_AA} "[0xAAAA, 0xAAAA,
... 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA,
... 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA,
... 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA,
... 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA,
... 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA,
... 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA,
... 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA]"
${BINARY_REGS_00} "[0x0000, 0x0000,
... 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000,
... 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000,
... 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000,
... 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000,
... 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000,
... 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000,
... 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000, 0x0000]"
*** Variables ***
${SKIP} robot:skip
*** Test Cases ***
# Custom command send and report
# Test Register Transaction
Test Modbus Custom Command
[Documentation] Test custom command
[Template] Test Async Run Modbus Case
[Tags] Modbus Custom command
# Unit ID, Function Code, Unused, Unused, Exception, Custom Data, Expected Response
"" 0x01 0x41 0x0000 0 0 [0x4d, 0x61, 0x73, 0x74, 0x65, 0x72] [0x4d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x3A, 0x53, 0x6C, 0x61, 0x76, 0x65]
# The TCP master example does not implement this request for now, keep for future test
# Test Modbus Report Slave ID
# [Documentation] Test reporting slave ID
# [Template] Test Async Run Modbus Case
# [Tags] ${SKIP}
# # Unit ID, Function Code, Expected Response
# "" 0x01 0x11 0x0000 1 0 ${None} [0x11, 0x05, 0x01, 0x0f, 0x33, 0x22, 0x11]
# Read Input Registers observed offsets/quantities
Test Modbus Read Input Registers
[Documentation] Test reading input registers with different addresses and quantities
[Template] Test Async Run Modbus Case
"Data_channel_0" 0x01 0x04 0x0000 2 0 ${None} [0x5c29, 0x3f8f]
"Temperature_1" 0x01 0x04 0x0002 2 0 ${None} [0xc28f, 0x4015]
"Temperature_2" 0x01 0x04 0x0004 2 0 ${None} [0xd70a, 0x4063]
# "" 0x01 0x04 0x0004 2 [0xd70a, 0x4063] 04 04 d7 0a 40 63
Test Modbus Read Holding Registers
[Documentation] Test reading holding registers with different addresses and quantities
[Template] Test Async Run Modbus Case
# Name, Unit ID, Function Code, Start Address, Quantity, Exception, Data, Expected Response
# "Param1" 0x01 0x03 0x0000 2 0 ${None} [0x0055, 0x0055]
# "Param2" 0x01 0x03 0x0001 2 0 ${None} [0x0055, 0x5500]
# "" 0x01 0x03 0x0002 2 0 ${None} [0xc28f, 0x4015]
# "" 0x01 0x03 0x0004 2 0 ${None} [0x3039, 0x3039]
# "" 0x01 0x03 0x0006 2 0 ${None} [0x3930, 0x3930]
# "" 0x01 0x03 0x0008 4 0 ${None} [0x3039, 0x0000, 0x3039, 0x0000]
# "" 0x01 0x03 0x000c 4 0 ${None} [0x0000, 0x3039, 0x0000, 0x3039]
# "" 0x01 0x03 0x0010 4 0 ${None} [0x3930, 0x0000, 0x3930, 0x0000]
# "" 0x01 0x03 0x0014 4 0 ${None} [0x0000, 0x3039, 0x0000, 0x3039]
# "" 0x01 0x03 0x0018 4 0 ${None} [0xe400, 0x4640, 0xe400, 0x4640]
# "" 0x01 0x03 0x001c 4 0 ${None} [0x4640, 0xe400, 0x4640, 0xe400]
# "" 0x01 0x03 0x0020 4 0 ${None} [0x00e4, 0x4046, 0x00e4, 0x4046]
# "" 0x01 0x03 0x0024 4 0 ${None} [0x4046, 0x00e4, 0x4046, 0x00e4]
# "" 0x01 0x03 0x0028 8 0 ${None} [0x0000, 0x0000, 0x1c80, 0x40c8, 0x0000, 0x0000, 0x1c80, 0x40c8]
# "" 0x01 0x03 0x0030 8 0 ${None} [0xc840, 0x801c, 0x0000, 0x0000, 0xc840, 0x801c, 0x0000, 0x0000]
# "" 0x01 0x03 0x0038 8 0 ${None} [0x40c8, 0x1c80, 0x0000, 0x0000, 0x40c8, 0x1c80, 0x0000, 0x0000]
# "" 0x01 0x03 0x0040 8 0 ${None} [0x0000, 0x0000, 0x801c, 0xc840, 0x0000, 0x0000, 0x801c, 0xc840]
"Humidity_1" 0x02 0x03 0x004a 2 0 ${None} [0x0000, 0x0000]
"Humidity_2" 0x01 0x03 0x004c 2 0 ${None} [0x0000, 0x0000]
"Humidity_3" 0x01 0x03 0x004e 2 0 ${None} [0x0000, 0x0000]
"CustomHoldReg" 0x01 0x03 0x00f0 1 0 ${None} [0x0000]
"Test_regs" 0x01 0x03 0x0052 58 0 ${None} ${BINARY_REGS_00}
# "" 0x01 0x03 0x1234 1 0 ${None} [0x0000]
Test Modbus Write Multiple Registers
[Documentation] Test writing multiple holding registers
[Template] Test Async Run Modbus Case
# Name, Unit ID, Function Code, Start Address, Quantity, Byte Count, Register Values, Expected Response
#"" 0x01 0x10 0x0000 2 0 [0x1122,0x3344] [0x0000, 0x0002]
"" 0x01 0x10 0x0052 58 0 ${BINARY_REGS_AA} [0x0052, 0x003A]
Test Modbus Read Coils
[Documentation] Test reading coils with different addresses and quantities
[Template] Test Async Run Modbus Case
# Name, Unit ID, Function Code, Start Address, Quantity, Exception, Data, Expected Response
# "" 0x01 0x01 0x0002 6 0 ${None} [0x15]
"RelayP1" 0x01 0x01 0x0002 6 0 ${None} [0x2A]
# "RelayP2" 0x01 0x01 0x000A 6 0 ${None} [0x15]
# Test Modbus Write Single Coil
# [Documentation] Test writing a single coil
# [Template] Test Async Run Modbus Case
# "" 0x01 0x05 0x0000 1 0 [0xFF00] [0x00, 0x01]
# # Write Multiple Coils
# Test Modbus Write Multiple Coils
# [Documentation] Test writing multiple coils
# [Template] Test Async Run Modbus Case
# # Name, Unit ID, Function Code, Start Address, Quantity, Byte Count, Output Values, Expected Response
# "" 0x01 0x0F 0x0000 1 1 [0xFF] [0x0000,0x0001]
# Test Modbus Read Discrete Inputs
# [Documentation] Test reading discrete inputs with different addresses and quantities
# [Template] Test Async Run Modbus Case
# # Name, Unit ID, Function Code, Start Address, Quantity, Exception, Data, Expected Response
# "" 0x01 0x02 0x0002 7 0 ${None} [0x55]
tools/test_requirements.txt
+4 -2
View File
@@ -1,5 +1,7 @@
robotframework
# Pin RobotFramework for AsyncLibrary (robotframework-async-keyword) compatibility
robotframework>=6.0,<=7.3.2
dataclasses
scapy>=2.6.1
scapy==2.5.0 # intentionally use this version to workaround send issues
numpy>=1.21.6
matplotlib>=3.5.3
robotframework-async-keyword==1.1.15 # provides Python package AsyncLibrary (keyword "Async Run")