diff --git a/clickplc/__init__.py b/clickplc/__init__.py index 4ce2732..25391cf 100644 --- a/clickplc/__init__.py +++ b/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()) diff --git a/clickplc/driver.py b/clickplc/driver.py index d2b0c50..df07399 100644 --- a/clickplc/driver.py +++ b/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. diff --git a/clickplc/tests/test_driver.py b/clickplc/tests/test_driver.py index fd83ede..3d11650 100644 --- a/clickplc/tests/test_driver.py +++ b/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') diff --git a/clickplc/util.py b/clickplc/util.py index 22c07fe..0dae3cf 100644 --- a/clickplc/util.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 2a9a3c4..aced58a 100644 --- a/pyproject.toml +++ b/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]