Use pymodbus simulator instead of python mock

clickplc/__init__.py

@@ -33,7 +33,10 @@ async def get():
                 d.update(await plc.get('ctd1-ctd250'))
             print(json.dumps(d, indent=4))
 
-    loop = asyncio.new_event_loop()
+    try:
+        loop = asyncio.get_event_loop()
+    except RuntimeError:
+        loop = asyncio.new_event_loop()
     loop.run_until_complete(get())
 
 

clickplc/driver.py

@@ -55,10 +55,10 @@ def __init__(self, address, tag_filepath='', timeout=1):
 
         """
         super().__init__(address, timeout)
-        self.bigendian = Endian.BIG if self.pymodbus35plus else Endian.Big  # type:ignore[attr-defined]
-        self.lilendian = Endian.LITTLE if self.pymodbus35plus else Endian.Little  # type:ignore[attr-defined]
         self.tags = self._load_tags(tag_filepath)
         self.active_addresses = self._get_address_ranges(self.tags)
+        self.bigendian = Endian.BIG if self.pymodbus35plus else Endian.Big  # type:ignore[attr-defined]
+        self.lilendian = Endian.LITTLE if self.pymodbus35plus else Endian.Little  # type:ignore[attr-defined]
 
     def get_tags(self) -> dict:
         """Return all tags and associated configuration information.

clickplc/tests/test_driver.py

@@ -1,15 +1,46 @@
 """Test the driver correctly parses a tags file and responds with correct data."""
 import asyncio
+import contextlib
 from unittest import mock
 
 import pytest
 
-from clickplc import command_line
-from clickplc.mock import ClickPLC
+try:
+    from pymodbus.server import ModbusTcpServer
+except ImportError:
+    from pymodbus.server.async_io import ModbusTcpServer  # type: ignore[no-redef]
 
-ADDRESS = 'fakeip'
-# from clickplc.driver import ClickPLC
+from clickplc import ClickPLC, command_line
+from clickplc.mock import ClickPLC as MockClickPLC
+
+# Test against pymodbus simulator
+ADDRESS = '127.0.0.1'
+autouse = True
+# Uncomment below to use a real PLC
 # ADDRESS = '172.16.0.168'
+# autouse = False
+
+@pytest.fixture(scope='session', autouse=autouse)
+async def _sim():
+    """Start a modbus server and datastore."""
+    from pymodbus.datastore import (
+        ModbusSequentialDataBlock,
+        ModbusServerContext,
+        ModbusSlaveContext,
+    )
+    store = ModbusSlaveContext(
+        di=ModbusSequentialDataBlock(0, [0] * 65536),  # Discrete Inputs
+        co=ModbusSequentialDataBlock(0, [0] * 65536),  # Coils
+        hr=ModbusSequentialDataBlock(0, [0] * 65536),  # Holding Registers
+        ir=ModbusSequentialDataBlock(0, [0] * 65536)   # Input Registers
+    )
+    context = ModbusServerContext(slaves=store, single=True)
+    server = ModbusTcpServer(context=context, address=("127.0.0.1", 5020))
+    asyncio.ensure_future(server.serve_forever())  # noqa: RUF006
+    await(asyncio.sleep(0))
+    yield
+    with contextlib.suppress(AttributeError):  # 2.x
+        await server.shutdown()  # type: ignore
 
 # ruff: noqa: E302
 @pytest.fixture(scope='session')
@@ -42,7 +73,7 @@ def expected_tags():
         'timer': {'address': {'start': 449153}, 'id': 'CTD1', 'type': 'int32'},
     }
 
-@mock.patch('clickplc.ClickPLC', ClickPLC)
+
 def test_driver_cli(capsys):
     """Confirm the commandline interface works without a tags file."""
     command_line([ADDRESS])
@@ -51,7 +82,19 @@ def test_driver_cli(capsys):
     assert 'c100' in captured.out
     assert 'df100' in captured.out
 
-@mock.patch('clickplc.ClickPLC', ClickPLC)
+
+@mock.patch('clickplc.ClickPLC', MockClickPLC)
+def test_driver_cli_tags_mock(capsys):
+    """Confirm the (mocked) commandline interface works with a tags file."""
+    command_line([ADDRESS, 'clickplc/tests/plc_tags.csv'])
+    captured = capsys.readouterr()
+    assert 'P_101' in captured.out
+    assert 'VAHH_101_OK' in captured.out
+    assert 'TI_101' in captured.out
+    with pytest.raises(SystemExit):
+        command_line([ADDRESS, 'tags', 'bogus'])
+
+
 def test_driver_cli_tags(capsys):
     """Confirm the commandline interface works with a tags file."""
     command_line([ADDRESS, 'clickplc/tests/plc_tags.csv'])
@@ -62,8 +105,8 @@ def test_driver_cli_tags(capsys):
     with pytest.raises(SystemExit):
         command_line([ADDRESS, 'tags', 'bogus'])
 
-@pytest.mark.asyncio(loop_scope='session')
-async def test_unsupported_tags():
+@mock.patch('clickplc.util.AsyncioModbusClient.__init__')
+def test_unsupported_tags(mock_init):
     """Confirm the driver detects an improper tags file."""
     with pytest.raises(TypeError, match='unsupported data type'):
         ClickPLC(ADDRESS, 'clickplc/tests/bad_tags.csv')

clickplc/util.py

@@ -26,10 +26,11 @@ class AsyncioModbusClient:
     def __init__(self, address, timeout=1):
         """Set up communication parameters."""
         self.ip = address
+        self.port = 5020 if address == '127.0.0.1' else 502  # pymodbus simulator is 127.0.0.1:5020
         self.timeout = timeout
         self._detect_pymodbus_version()
         if self.pymodbus30plus:
-            self.client = AsyncModbusTcpClient(address, timeout=timeout)  # pyright: ignore [reportPossiblyUnboundVariable]
+            self.client = AsyncModbusTcpClient(address, timeout=timeout, port=self.port)  # pyright: ignore [reportPossiblyUnboundVariable]
         else:  # 2.x
             self.client = ReconnectingAsyncioModbusTcpClient()  # pyright: ignore [reportPossiblyUnboundVariable]
         self.lock = asyncio.Lock()
@@ -56,7 +57,7 @@ async def _connect(self) -> None:
             if self.pymodbus30plus:
                 await asyncio.wait_for(self.client.connect(), timeout=self.timeout)  # 3.x
             else:  # 2.4.x - 2.5.x
-                await self.client.start(self.ip)  # type: ignore[attr-defined]
+                await self.client.start(host=self.ip, port=self.port)  # type: ignore[attr-defined]
         except Exception as e:
             raise OSError(f"Could not connect to '{self.ip}'.") from e
 

pyproject.toml

@@ -40,6 +40,7 @@ test = [
   'pytest-asyncio>=0.23.8',
   'ruff==0.9.2',
   'types-PyYAML',
+  'pyserial-asyncio>=0.4.0; python_version == "3.9"',  # pymodbus 2.x
 ]
 
 [project.scripts]