Skip to content

Commit

Permalink
tests for asyncio modbusclients created in coroutine
Browse files Browse the repository at this point in the history
  • Loading branch information
Emilv2 committed Feb 12, 2021
1 parent 58e53a5 commit a8f3850
Showing 1 changed file with 39 additions and 0 deletions.
39 changes: 39 additions & 0 deletions test/test_client_async_asyncio.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,11 @@
ReconnectingAsyncioModbusTcpClient,
ModbusClientProtocol, ModbusUdpClientProtocol)
from test.asyncio_test_helper import return_as_coroutine, run_coroutine
from pymodbus.client.asynchronous import schedulers
from pymodbus.client.asynchronous.serial import AsyncModbusSerialClient
from pymodbus.client.asynchronous.tcp import AsyncModbusTCPClient
from pymodbus.client.asynchronous.tls import AsyncModbusTLSClient
from pymodbus.client.asynchronous.udp import AsyncModbusUDPClient
from pymodbus.factory import ClientDecoder
from pymodbus.exceptions import ConnectionException
from pymodbus.transaction import ModbusSocketFramer
Expand Down Expand Up @@ -68,6 +73,40 @@ def test_factory_initialization_state(self):
assert client.loop is mock_loop
assert client.protocol_class is mock_protocol_class

@pytest.mark.asyncio
async def test_initialization_tcp_in_loop(self):
_, client = AsyncModbusTCPClient(schedulers.ASYNC_IO, port=5020)
client = await client

assert not client.connected
assert client.port == 5020
assert client.delay_ms < client.DELAY_MAX_MS

@pytest.mark.asyncio
async def test_initialization_udp_in_loop(self):
_, client = AsyncModbusUDPClient(schedulers.ASYNC_IO, port=5020)
client = await client

assert client.connected
assert client.port == 5020
assert client.delay_ms < client.DELAY_MAX_MS

@pytest.mark.asyncio
async def test_initialization_tls_in_loop(self):
_, client = AsyncModbusTLSClient(schedulers.ASYNC_IO, port=5020)
client = await client

assert not client.connected
assert client.port == 5020
assert client.delay_ms < client.DELAY_MAX_MS

@pytest.mark.asyncio
async def test_initialization_serial_in_loop(self):
_, client = AsyncModbusSerialClient(schedulers.ASYNC_IO, port='/tmp/ptyp0', baudrate=9600, method='rtu')

assert client.port == '/tmp/ptyp0'
assert client.baudrate == 9600

def test_factory_reset_wait_before_reconnect(self):
mock_protocol_class = mock.MagicMock()
mock_loop = mock.MagicMock()
Expand Down

0 comments on commit a8f3850

Please sign in to comment.