From db2e71f58f789dc27745a43a13e5a44452e90fbf Mon Sep 17 00:00:00 2001 From: Alex <52292902+alexrudd2@users.noreply.github.com> Date: Fri, 9 Aug 2024 00:46:08 -0500 Subject: [PATCH] Simplify framer test setup (#2290) --- test/framers/conftest.py | 45 +++++++--------------------------------- 1 file changed, 7 insertions(+), 38 deletions(-) diff --git a/test/framers/conftest.py b/test/framers/conftest.py index 078599c99..e5cb9304d 100644 --- a/test/framers/conftest.py +++ b/test/framers/conftest.py @@ -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") @@ -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