|
19 | 19 | from unittest.mock import patch
|
20 | 20 |
|
21 | 21 | import pytest
|
22 |
| -from pymodbus.pdu.register_read_message import ( |
| 22 | +from pymodbus.pdu.register_message import ( |
23 | 23 | ReadHoldingRegistersResponse,
|
24 | 24 | ReadInputRegistersResponse,
|
25 | 25 | )
|
@@ -72,45 +72,46 @@ def bypass_get_data_fixture():
|
72 | 72 | yield
|
73 | 73 |
|
74 | 74 |
|
75 |
| -def read_input_register(register, start, count) -> ReadInputRegistersResponse: |
| 75 | +def read_input_register( |
| 76 | + address: int, register, start, count |
| 77 | +) -> ReadInputRegistersResponse: |
76 | 78 | """Read a slice from the input register."""
|
77 |
| - return ReadInputRegistersResponse(register[start : start + count]) |
| 79 | + return ReadInputRegistersResponse( |
| 80 | + address=address, count=count, registers=register[start : start + count] |
| 81 | + ) |
78 | 82 |
|
79 | 83 |
|
80 | 84 | async def read_input_registers_wpm(
|
81 |
| - address: int, |
82 |
| - count: int = 1, |
83 |
| - slave: int = 0, |
84 |
| - **kwargs: Any, |
| 85 | + address: int, *, count: int = 1, slave: int = 0, no_response_expected: bool = False |
85 | 86 | ):
|
86 | 87 | """Simulate reads on the input registers on wpm models."""
|
87 | 88 | system_info = [2, 390]
|
88 | 89 | if address >= 5000:
|
89 |
| - return read_input_register(system_info, address - 5000, count) |
90 |
| - return ReadInputRegistersResponse(list(range(count))) |
| 90 | + return read_input_register(address, system_info, address - 5000, count) |
| 91 | + return ReadInputRegistersResponse( |
| 92 | + address=address, count=count, registers=list(range(count)) |
| 93 | + ) |
91 | 94 |
|
92 | 95 |
|
93 | 96 | async def read_input_registers_lwz(
|
94 |
| - address: int, |
95 |
| - count: int = 1, |
96 |
| - slave: int = 0, |
97 |
| - **kwargs: Any, |
| 97 | + address: int, *, count: int = 1, slave: int = 0, no_response_expected: bool = False |
98 | 98 | ):
|
99 | 99 | """Simulate reads on the input registers on lwz models ."""
|
100 | 100 | system_info = [2, 103]
|
101 | 101 | if address >= 5000:
|
102 |
| - return read_input_register(system_info, address - 5000, count) |
103 |
| - return ReadInputRegistersResponse(list(range(count))) |
| 102 | + return read_input_register(address, system_info, address - 5000, count) |
| 103 | + return ReadInputRegistersResponse( |
| 104 | + address=address, count=count, registers=list(range(count)) |
| 105 | + ) |
104 | 106 |
|
105 | 107 |
|
106 | 108 | async def read_holding_registers(
|
107 |
| - address: int, |
108 |
| - count: int = 1, |
109 |
| - slave: int = 0, |
110 |
| - **kwargs: Any, |
| 109 | + address: int, *, count: int = 1, slave: int = 0, no_response_expected: bool = False |
111 | 110 | ):
|
112 | 111 | """Simulate reads on the holding registers on lwz models ."""
|
113 |
| - return ReadHoldingRegistersResponse(list(range(count))) |
| 112 | + return ReadHoldingRegistersResponse( |
| 113 | + address=address, count=count, registers=list(range(count)) |
| 114 | + ) |
114 | 115 |
|
115 | 116 |
|
116 | 117 | # This fixture, when used, will result in calls to read_input_registers to return mock data. To have the call
|
|
0 commit comments