Compare commits

..
4 changed files with 340 additions and 610 deletions
@@ -17,10 +17,10 @@ uart:
baud_rate: 115200
port: /dev/null
# Shared 3-bus mesh (see the shared_yaml markers): addr 1 = typed read-only
# registers, addr 5 = the read/write 0x17 target, addr 2/3 on the second
# server hub. auto_start everywhere: the controller polls at boot, so the
# forwarding must already be live or early requests generate warnings.
# Shared 3-bus mesh (see the shared_yaml markers): addr 1 = typed registers
# backed by writable globals, addr 5 = the read/write 0x17 target, addr 2/3/6
# on the second server hub. auto_start everywhere: the controller polls at
# boot, so the forwarding must already be live or early requests generate warnings.
# Every test presses Start Scenario, so all merged actions fire in every test.
uart_mock:
- id: virtual_uart_server
@@ -64,6 +64,54 @@ globals:
- id: stored_1
type: uint16_t
initial_value: "0"
- id: stored_u_word
type: uint16_t
initial_value: "99"
- id: stored_u_word_s
type: uint16_t
initial_value: "4660"
- id: stored_s_word
type: int16_t
initial_value: "-99"
- id: stored_s_word_s
type: int16_t
initial_value: "-2"
- id: stored_u_dword
type: uint32_t
initial_value: "16909060"
- id: stored_s_dword
type: int32_t
initial_value: "-16909060"
- id: stored_u_dword_r
type: uint32_t
initial_value: "67305985"
- id: stored_s_dword_r
type: int32_t
initial_value: "-67305985"
- id: stored_u_qword
type: uint64_t
initial_value: "72623859790382856"
- id: stored_s_qword
type: int64_t
initial_value: "-72623859790382856"
- id: stored_u_qword_r
type: uint64_t
initial_value: "578437695752307201"
- id: stored_s_qword_r
type: int64_t
initial_value: "-578437695752307201"
- id: stored_fp32
type: float
initial_value: "3.14"
- id: stored_fp32_r
type: float
initial_value: "2.5"
- id: stored_bit_2
type: bool
initial_value: "false"
- id: stored_bit_3
type: bool
initial_value: "true"
modbus:
- uart_id: virtual_uart_server
@@ -90,6 +138,10 @@ modbus_controller:
modbus_id: virtual_modbus_client
id: modbus_controller_3
update_interval: 1s
- address: 6
modbus_id: virtual_modbus_client
id: modbus_controller_6
update_interval: 1s
modbus_server:
- address: 1
@@ -97,46 +149,60 @@ modbus_server:
registers:
- address: 0x01
value_type: U_WORD
read_lambda: return 99;
read_lambda: return id(stored_u_word);
write_lambda: id(stored_u_word) = x; return true;
- address: 0x02
value_type: U_WORD_S
read_lambda: return 4660;
read_lambda: return id(stored_u_word_s);
write_lambda: id(stored_u_word_s) = x; return true;
- address: 0x03
value_type: S_WORD
read_lambda: return -99;
read_lambda: return id(stored_s_word);
write_lambda: id(stored_s_word) = x; return true;
- address: 0x04
value_type: S_WORD_S
read_lambda: return -2;
read_lambda: return id(stored_s_word_s);
write_lambda: id(stored_s_word_s) = x; return true;
- address: 0x05
value_type: U_DWORD
read_lambda: return 16909060;
read_lambda: return id(stored_u_dword);
write_lambda: id(stored_u_dword) = x; return true;
- address: 0x08
value_type: S_DWORD
read_lambda: return -16909060;
read_lambda: return id(stored_s_dword);
write_lambda: id(stored_s_dword) = x; return true;
- address: 0x0B
value_type: U_DWORD_R
read_lambda: return 67305985;
read_lambda: return id(stored_u_dword_r);
write_lambda: id(stored_u_dword_r) = x; return true;
- address: 0x0E
value_type: S_DWORD_R
read_lambda: return -67305985;
read_lambda: return id(stored_s_dword_r);
write_lambda: id(stored_s_dword_r) = x; return true;
- address: 0x11
value_type: U_QWORD
read_lambda: return 72623859790382856;
read_lambda: return id(stored_u_qword);
write_lambda: id(stored_u_qword) = x; return true;
- address: 0x16
value_type: S_QWORD
read_lambda: return -72623859790382856;
read_lambda: return id(stored_s_qword);
write_lambda: id(stored_s_qword) = x; return true;
- address: 0x1B
value_type: U_QWORD_R
read_lambda: return 578437695752307201;
read_lambda: return id(stored_u_qword_r);
write_lambda: id(stored_u_qword_r) = x; return true;
- address: 0x20
value_type: S_QWORD_R
read_lambda: return -578437695752307201;
read_lambda: return id(stored_s_qword_r);
write_lambda: id(stored_s_qword_r) = x; return true;
- address: 0x25
value_type: FP32
read_lambda: return 3.14;
read_lambda: return id(stored_fp32);
write_lambda: id(stored_fp32) = x; return true;
- address: 0x28
value_type: FP32_R
read_lambda: return 3.14;
read_lambda: return id(stored_fp32_r);
write_lambda: id(stored_fp32_r) = x; return true;
- address: 5
modbus_id: virtual_modbus_server
registers:
@@ -165,6 +231,19 @@ modbus_server:
- address: 0x01
value_type: U_WORD
read_lambda: return 929;
- address: 6
modbus_id: virtual_modbus_server_2
bits:
- address: 0x00
read_lambda: return true;
- address: 0x01
read_lambda: return false;
- address: 0x02
read_lambda: return id(stored_bit_2);
write_lambda: id(stored_bit_2) = x; return true;
- address: 0x03
read_lambda: return id(stored_bit_3);
write_lambda: id(stored_bit_3) = x; return true;
sensor:
- platform: modbus_controller
@@ -280,6 +359,183 @@ sensor:
name: "client_read_1"
id: client_read_1
# The number schema caps min/max at 16777215 (float32 integer precision), so
# the large dword/qword baselines cannot be written back through these numbers.
number:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_word"
address: 0x01
register_type: holding
value_type: U_WORD
min_value: 0
max_value: 65535
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_word_s"
address: 0x02
register_type: holding
value_type: U_WORD_S
min_value: 0
max_value: 65535
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_word"
address: 0x03
register_type: holding
value_type: S_WORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_word_s"
address: 0x04
register_type: holding
value_type: S_WORD_S
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_dword"
address: 0x05
register_type: holding
value_type: U_DWORD
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_dword"
address: 0x08
register_type: holding
value_type: S_DWORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_dword_r"
address: 0x0B
register_type: holding
value_type: U_DWORD_R
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_dword_r"
address: 0x0E
register_type: holding
value_type: S_DWORD_R
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_qword"
address: 0x11
register_type: holding
value_type: U_QWORD
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_qword"
address: 0x16
register_type: holding
value_type: S_QWORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_qword_r"
address: 0x1B
register_type: holding
value_type: U_QWORD_R
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_qword_r"
address: 0x20
register_type: holding
value_type: S_QWORD_R
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_fp32"
address: 0x25
register_type: holding
value_type: FP32
min_value: -16777215
max_value: 16777215
step: 0.01
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_fp32_r"
address: 0x28
register_type: holding
value_type: FP32_R
min_value: -16777215
max_value: 16777215
step: 0.01
# The four bits are read both as coils (FC 0x01) and discrete inputs (FC 0x02);
# the server serves both from one shared table, so the two views must agree.
binary_sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_coil_0"
address: 0x00
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_coil_1"
address: 0x01
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_coil_2"
address: 0x02
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_coil_3"
address: 0x03
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_di_0"
address: 0x00
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_di_1"
address: 0x01
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_di_2"
address: 0x02
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_di_3"
address: 0x03
register_type: discrete_input
# write_bit_2 uses the single-coil write (FC 0x05); write_bit_3 opts into the
# multiple-coils write (FC 0x0F) so both server write paths are exercised.
switch:
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "write_bit_2"
address: 0x02
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "write_bit_3"
address: 0x03
register_type: coil
use_write_multiple: true
button:
- platform: template
name: "Start Scenario"
@@ -1,147 +0,0 @@
esphome:
name: uart-mock-modbus-srv-bits
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
# auto_start must be true for loopback fixtures: the modbus controller
# polls on its update_interval immediately at boot, so the uart_mock
# forwarding must already be active or early requests are lost and
# generate modbus warnings.
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true # See comment on virtual_uart_server above
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: stored_bit_2
type: bool
initial_value: "false"
- id: stored_bit_3
type: bool
initial_value: "true"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
update_interval: 1s
id: modbus_controller_1
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
bits:
- address: 0x00
read_lambda: return true;
- address: 0x01
read_lambda: return false;
- address: 0x02
read_lambda: return id(stored_bit_2);
write_lambda: id(stored_bit_2) = x; return true;
- address: 0x03
read_lambda: return id(stored_bit_3);
write_lambda: id(stored_bit_3) = x; return true;
# The same four bits are read both as coils (FC 0x01) and as discrete inputs
# (FC 0x02): the server serves both from one shared bit table, so the two
# views must always agree.
binary_sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_coil_0"
address: 0x00
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_coil_1"
address: 0x01
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_coil_2"
address: 0x02
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_coil_3"
address: 0x03
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_di_0"
address: 0x00
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_di_1"
address: 0x01
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_di_2"
address: 0x02
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_di_3"
address: 0x03
register_type: discrete_input
# write_bit_2 uses the single-coil write (FC 0x05); write_bit_3 opts into the
# multiple-coils write (FC 0x0F) so both server write paths are exercised.
switch:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_bit_2"
address: 0x02
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_bit_3"
address: 0x03
register_type: coil
use_write_multiple: true
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
@@ -1,371 +0,0 @@
esphome:
name: uart-mock-modbus-srv-write
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
# auto_start must be true for loopback fixtures: the modbus controller
# polls on its update_interval immediately at boot, so the uart_mock
# forwarding must already be active or early requests are lost and
# generate modbus warnings.
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true # See comment on virtual_uart_server above
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: stored_u_word
type: uint16_t
initial_value: "11"
- id: stored_u_word_s
type: uint16_t
initial_value: "4660"
- id: stored_s_word
type: int16_t
initial_value: "-11"
- id: stored_s_word_s
type: int16_t
initial_value: "-2"
- id: stored_u_dword
type: uint32_t
initial_value: "1001"
- id: stored_s_dword
type: int32_t
initial_value: "-1001"
- id: stored_u_dword_r
type: uint32_t
initial_value: "3003"
- id: stored_s_dword_r
type: int32_t
initial_value: "-3003"
- id: stored_u_qword
type: uint64_t
initial_value: "5005"
- id: stored_s_qword
type: int64_t
initial_value: "-5005"
- id: stored_u_qword_r
type: uint64_t
initial_value: "7007"
- id: stored_s_qword_r
type: int64_t
initial_value: "-7007"
- id: stored_fp32
type: float
initial_value: "1.5"
- id: stored_fp32_r
type: float
initial_value: "2.5"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
update_interval: 2s
id: modbus_controller_1
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
registers:
- address: 0x01
value_type: U_WORD
read_lambda: return id(stored_u_word);
write_lambda: id(stored_u_word) = x; return true;
- address: 0x02
value_type: U_WORD_S
read_lambda: return id(stored_u_word_s);
write_lambda: id(stored_u_word_s) = x; return true;
- address: 0x03
value_type: S_WORD
read_lambda: return id(stored_s_word);
write_lambda: id(stored_s_word) = x; return true;
- address: 0x04
value_type: S_WORD_S
read_lambda: return id(stored_s_word_s);
write_lambda: id(stored_s_word_s) = x; return true;
- address: 0x05
value_type: U_DWORD
read_lambda: return id(stored_u_dword);
write_lambda: id(stored_u_dword) = x; return true;
- address: 0x08
value_type: S_DWORD
read_lambda: return id(stored_s_dword);
write_lambda: id(stored_s_dword) = x; return true;
- address: 0x0B
value_type: U_DWORD_R
read_lambda: return id(stored_u_dword_r);
write_lambda: id(stored_u_dword_r) = x; return true;
- address: 0x0E
value_type: S_DWORD_R
read_lambda: return id(stored_s_dword_r);
write_lambda: id(stored_s_dword_r) = x; return true;
- address: 0x11
value_type: U_QWORD
read_lambda: return id(stored_u_qword);
write_lambda: id(stored_u_qword) = x; return true;
- address: 0x16
value_type: S_QWORD
read_lambda: return id(stored_s_qword);
write_lambda: id(stored_s_qword) = x; return true;
- address: 0x1B
value_type: U_QWORD_R
read_lambda: return id(stored_u_qword_r);
write_lambda: id(stored_u_qword_r) = x; return true;
- address: 0x20
value_type: S_QWORD_R
read_lambda: return id(stored_s_qword_r);
write_lambda: id(stored_s_qword_r) = x; return true;
- address: 0x25
value_type: FP32
read_lambda: return id(stored_fp32);
write_lambda: id(stored_fp32) = x; return true;
- address: 0x28
value_type: FP32_R
read_lambda: return id(stored_fp32_r);
write_lambda: id(stored_fp32_r) = x; return true;
sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_word"
address: 0x01
register_type: holding
value_type: U_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_word_s"
address: 0x02
register_type: holding
value_type: U_WORD_S
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_word"
address: 0x03
register_type: holding
value_type: S_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_word_s"
address: 0x04
register_type: holding
value_type: S_WORD_S
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_dword"
address: 0x05
register_type: holding
value_type: U_DWORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_dword"
address: 0x08
register_type: holding
value_type: S_DWORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_dword_r"
address: 0x0B
register_type: holding
value_type: U_DWORD_R
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_dword_r"
address: 0x0E
register_type: holding
value_type: S_DWORD_R
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_qword"
address: 0x11
register_type: holding
value_type: U_QWORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_qword"
address: 0x16
register_type: holding
value_type: S_QWORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_qword_r"
address: 0x1B
register_type: holding
value_type: U_QWORD_R
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_qword_r"
address: 0x20
register_type: holding
value_type: S_QWORD_R
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_fp32"
address: 0x25
register_type: holding
value_type: FP32
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_fp32_r"
address: 0x28
register_type: holding
value_type: FP32_R
number:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_word"
address: 0x01
register_type: holding
value_type: U_WORD
min_value: 0
max_value: 65535
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_word_s"
address: 0x02
register_type: holding
value_type: U_WORD_S
min_value: 0
max_value: 65535
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_word"
address: 0x03
register_type: holding
value_type: S_WORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_word_s"
address: 0x04
register_type: holding
value_type: S_WORD_S
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_dword"
address: 0x05
register_type: holding
value_type: U_DWORD
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_dword"
address: 0x08
register_type: holding
value_type: S_DWORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_dword_r"
address: 0x0B
register_type: holding
value_type: U_DWORD_R
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_dword_r"
address: 0x0E
register_type: holding
value_type: S_DWORD_R
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_qword"
address: 0x11
register_type: holding
value_type: U_QWORD
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_qword"
address: 0x16
register_type: holding
value_type: S_QWORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_qword_r"
address: 0x1B
register_type: holding
value_type: U_QWORD_R
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_qword_r"
address: 0x20
register_type: holding
value_type: S_QWORD_R
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_fp32"
address: 0x25
register_type: holding
value_type: FP32
min_value: -16777215
max_value: 16777215
step: 0.01
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_fp32_r"
address: 0x28
register_type: holding
value_type: FP32_R
min_value: -16777215
max_value: 16777215
step: 0.01
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
+66 -74
View File
@@ -19,23 +19,40 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
from aioesphomeapi import ButtonInfo, NumberInfo, SwitchInfo, TextSensorState
import pytest
from .state_utils import SensorTracker, find_entity, wait_for_state
from .state_utils import SensorTracker, find_entity, require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
@dataclass
class RegisterTestCase:
"""Test parameters for a single modbus register write/read round-trip."""
def _swap16(value: int) -> int:
"""Byte-swapped view of a 16-bit register as the raw U_WORD wire value."""
return ((value & 0xFF) << 8) | (value >> 8)
initial_value: object
write_number_name: str
write_value: float
post_write_value: object
# Raw U_WORD view of reg_u_word_s's initial 0x1234
MESH_RAW_U_WORD_S = _swap16(4660)
# Initial values of the mesh fixture's address 1 registers; the
# server_controller test reads them and the write test uses them as baseline.
MESH_INITIAL_VALUES: dict[str, object] = {
"reg_u_word": 99,
"reg_u_word_s": 4660,
"reg_s_word": -99,
"reg_s_word_s": -2,
"reg_u_dword": 16909060,
"reg_s_dword": -16909060,
"reg_u_dword_r": pytest.approx(67305985),
"reg_s_dword_r": pytest.approx(-67305985),
"reg_u_qword": pytest.approx(72623859790382856),
"reg_s_qword": pytest.approx(-72623859790382856),
"reg_u_qword_r": pytest.approx(578437695752307201),
"reg_s_qword_r": pytest.approx(-578437695752307201),
"reg_fp32": pytest.approx(3.14),
"reg_fp32_r": pytest.approx(2.5),
}
# ---------------------------------------------------------------------------
@@ -310,23 +327,7 @@ async def test_uart_mock_modbus_server_controller(
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
expected_values = {
"reg_u_word": 99,
"reg_u_word_s": 4660,
"reg_u_word_s_raw": 13330,
"reg_s_word": -99,
"reg_s_word_s": -2,
"reg_u_dword": 16909060,
"reg_s_dword": -16909060,
"reg_u_dword_r": pytest.approx(67305985),
"reg_s_dword_r": pytest.approx(-67305985),
"reg_u_qword": pytest.approx(72623859790382856),
"reg_s_qword": pytest.approx(-72623859790382856),
"reg_u_qword_r": pytest.approx(578437695752307201),
"reg_s_qword_r": pytest.approx(-578437695752307201),
"reg_fp32": pytest.approx(3.14),
"reg_fp32_r": pytest.approx(3.14),
}
expected_values = MESH_INITIAL_VALUES | {"reg_u_word_s_raw": MESH_RAW_U_WORD_S}
tracker = SensorTracker(list(expected_values.keys()))
futures = tracker.expect_all(expected_values)
@@ -334,14 +335,12 @@ async def test_uart_mock_modbus_server_controller(
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot, so the first values can already be in
# the states the device sends on connect; matching them there saves
# waiting for the next poll
await tracker.setup_and_start_scenario(client, match_initial_states=True)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller_write(
yaml_config: str,
@@ -357,51 +356,47 @@ async def test_uart_mock_modbus_server_controller_write(
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
register_test_cases: dict[str, RegisterTestCase] = {
"reg_u_word": RegisterTestCase(11, "write_u_word", 42, 42),
"reg_u_word_s": RegisterTestCase(4660, "write_u_word_s", 17185, 17185),
"reg_s_word": RegisterTestCase(-11, "write_s_word", -42, -42),
"reg_s_word_s": RegisterTestCase(-2, "write_s_word_s", -257, -257),
"reg_u_dword": RegisterTestCase(1001, "write_u_dword", 2002, 2002),
"reg_s_dword": RegisterTestCase(-1001, "write_s_dword", -2002, -2002),
"reg_u_dword_r": RegisterTestCase(3003, "write_u_dword_r", 4004, 4004),
"reg_s_dword_r": RegisterTestCase(-3003, "write_s_dword_r", -4004, -4004),
"reg_u_qword": RegisterTestCase(5005, "write_u_qword", 6006, 6006),
"reg_s_qword": RegisterTestCase(-5005, "write_s_qword", -6006, -6006),
"reg_u_qword_r": RegisterTestCase(7007, "write_u_qword_r", 8008, 8008),
"reg_s_qword_r": RegisterTestCase(-7007, "write_s_qword_r", -8008, -8008),
"reg_fp32": RegisterTestCase(
pytest.approx(1.5, abs=0.01),
"write_fp32",
3.14,
pytest.approx(3.14, abs=0.01),
),
"reg_fp32_r": RegisterTestCase(
pytest.approx(2.5, abs=0.01),
"write_fp32_r",
6.28,
pytest.approx(6.28, abs=0.01),
),
# Per read-back sensor: the number entity to write through and the value;
# floats read back within tolerance, everything else exactly
register_writes: dict[str, tuple[str, int | float]] = {
"reg_u_word": ("write_u_word", 42),
"reg_u_word_s": ("write_u_word_s", 17185),
"reg_s_word": ("write_s_word", -42),
"reg_s_word_s": ("write_s_word_s", -257),
"reg_u_dword": ("write_u_dword", 2002),
"reg_s_dword": ("write_s_dword", -2002),
"reg_u_dword_r": ("write_u_dword_r", 4004),
"reg_s_dword_r": ("write_s_dword_r", -4004),
"reg_u_qword": ("write_u_qword", 6006),
"reg_s_qword": ("write_s_qword", -6006),
"reg_u_qword_r": ("write_u_qword_r", 8008),
"reg_s_qword_r": ("write_s_qword_r", -8008),
"reg_fp32": ("write_fp32", 6.28),
"reg_fp32_r": ("write_fp32_r", 9.42),
}
tracker = SensorTracker(list(register_test_cases.keys()))
tracker = SensorTracker([*register_writes, "reg_u_word_s_raw"])
# The raw U_WORD view of 0x02 pins the byte swap on the write path: the
# round trip through write_u_word_s applies the swap an even number of
# times, so only the raw sensor can catch a symmetrically dropped swap.
# Phase 1: expect initial baseline values
initial_futures = tracker.expect_all(
{name: case.initial_value for name, case in register_test_cases.items()}
MESH_INITIAL_VALUES | {"reg_u_word_s_raw": MESH_RAW_U_WORD_S}
)
# Phase 2: expect post-write values (registered now so on_state can match them)
written_futures = tracker.expect_all(
{name: case.post_write_value for name, case in register_test_cases.items()}
{
name: pytest.approx(value, abs=0.01) if isinstance(value, float) else value
for name, (_, value) in register_writes.items()
}
| {"reg_u_word_s_raw": _swap16(register_writes["reg_u_word_s"][1])}
)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot, so the baseline can already be in the
# states the device sends on connect; matching it there saves waiting for
# the next poll
entities = await tracker.setup_and_start_scenario(
client, match_initial_states=True
)
@@ -410,19 +405,22 @@ async def test_uart_mock_modbus_server_controller_write(
# connection is working before issuing writes
await tracker.await_all(initial_futures, timeout=4.0)
# Issue write commands for all register types
for case in register_test_cases.values():
entity = find_entity(entities, case.write_number_name, NumberInfo)
assert entity is not None, (
f"{case.write_number_name} number entity not found"
)
client.number_command(entity.key, case.write_value)
# Issue write commands for all register types; exact object_id match,
# since several write_* names are prefixes of a sibling
numbers = {
e.object_id.lower(): e for e in entities if isinstance(e, NumberInfo)
}
for number_name, value in register_writes.values():
entity = numbers.get(number_name)
assert entity is not None, f"{number_name} number entity not found"
client.number_command(entity.key, value)
# Wait for sensors to reflect the written values (round-trip write+read)
await tracker.await_all(written_futures, timeout=4.0)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller_bits(
yaml_config: str,
@@ -468,8 +466,6 @@ async def test_uart_mock_modbus_server_controller_bits(
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot and binary sensors drop repeats, so the
# baseline can arrive only in the states the device sends on connect
entities = await tracker.setup_and_start_scenario(
client, match_initial_states=True
)
@@ -480,8 +476,7 @@ async def test_uart_mock_modbus_server_controller_bits(
# Flip both writable bits: 0x02 false -> true, 0x03 true -> false
for switch_name, value in (("write_bit_2", True), ("write_bit_3", False)):
entity = find_entity(entities, switch_name, SwitchInfo)
assert entity is not None, f"{switch_name} switch entity not found"
entity = require_entity(entities, switch_name, SwitchInfo)
client.switch_command(entity.key, value)
# Wait for both read views to reflect the written values
@@ -508,9 +503,6 @@ async def test_uart_mock_modbus_server_controller_multiple(
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot, so the first values can already be in
# the states the device sends on connect; matching them there saves
# waiting for the next poll
await tracker.setup_and_start_scenario(client, match_initial_states=True)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)