Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Simplify framer test setup #2290

Merged
merged 3 commits into from
Aug 9, 2024
Merged

Simplify framer test setup #2290

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9bd3ca7
eliminate DummyFramer
alexrudd2 Aug 9, 2024
5caf0a1
mypy
alexrudd2 Aug 9, 2024
77f31a3
use getattr()
alexrudd2 Aug 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 7 additions & 38 deletions test/framers/conftest.py
View file
Open in desktop
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