Skip to content

Commit

Permalink
Simplify framer test setup (#2290)
Browse files Browse the repository at this point in the history
  • Loading branch information
alexrudd2 authored Aug 9, 2024
1 parent 2b2dd06 commit db2e71f
Showing 1 changed file with 7 additions and 38 deletions.
45 changes: 7 additions & 38 deletions test/framers/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,35 +7,7 @@

from pymodbus.factory import ClientDecoder, ServerDecoder
from pymodbus.framer import Framer, FramerType
from pymodbus.transport import CommParams, ModbusProtocol


class DummyFramer(Framer):
"""Implement use of ModbusProtocol."""

def __init__(self,
framer_type: FramerType,
params: CommParams,
is_server: bool,
device_ids: list[int] | None,
):
"""Initialize a frame instance."""
super().__init__(framer_type, params, is_server, device_ids)
self.send = mock.Mock()
self.framer_type = framer_type

def callback_new_connection(self) -> ModbusProtocol:
"""Call when listener receive new connection request."""
return DummyFramer(self.framer_type, self.comm_params, self.is_server, self.device_ids) # pragma: no cover

def callback_connected(self) -> None:
"""Call when connection is succcesfull."""

def callback_disconnected(self, exc: Exception | None) -> None:
"""Call when connection is lost."""

def callback_request_response(self, data: bytes, device_id: int, tid: int) -> None:
"""Handle received modbus request/response."""
from pymodbus.transport import CommParams


@pytest.fixture(name="entry")
Expand All @@ -48,19 +20,16 @@ def prepare_is_server():
"""Return client/server."""
return False

@mock.patch.multiple(Framer, __abstractmethods__=set()) # eliminate abstract methods (callbacks)
@pytest.fixture(name="dummy_framer")
async def prepare_test_framer(entry, is_server):
"""Return framer object."""
framer = DummyFramer(
entry,
CommParams(),
is_server,
[0, 1],
)
framer = Framer(entry, CommParams(), is_server, [0, 1]) # type: ignore[abstract]
framer.send = mock.Mock() # type: ignore[method-assign]
if entry == FramerType.RTU:
func_table = (ServerDecoder if is_server else ClientDecoder)().lookup
func_table = (ServerDecoder if is_server else ClientDecoder)().lookup # type: ignore[attr-defined]
for key, ent in func_table.items():
fix_len = ent._rtu_frame_size if hasattr(ent, "_rtu_frame_size") else 0 # pylint: disable=protected-access
cnt_pos = ent. _rtu_byte_count_pos if hasattr(ent, "_rtu_byte_count_pos") else 0 # pylint: disable=protected-access
fix_len = getattr(ent, "_rtu_frame_size", 0)
cnt_pos = getattr(ent, "_rtu_byte_count_pos", 0)
framer.handle.set_fc_calc(key, fix_len, cnt_pos)
return framer

0 comments on commit db2e71f

Please sign in to comment.