move test of examples to subdirectory. (#1592)

examples/client_async.py

@@ -23,12 +23,11 @@
 """
 import asyncio
 import logging
-import os
 
 # --------------------------------------------------------------------------- #
 # import the various client implementations
 # --------------------------------------------------------------------------- #
-from examples.helper import get_commandline
+from examples import helper
 from pymodbus.client import (
     AsyncModbusSerialClient,
     AsyncModbusTcpClient,
@@ -38,11 +37,14 @@
 
 
 _logger = logging.getLogger()
+_logger.setLevel("DEBUG")
 
 
 def setup_async_client(description=None, cmdline=None):
     """Run client setup."""
-    args = get_commandline(server=False, description=description, cmdline=cmdline)
+    args = helper.get_commandline(
+        server=False, description=description, cmdline=cmdline
+    )
     _logger.info("### Create client object")
     if args.comm == "tcp":
         client = AsyncModbusTcpClient(
@@ -90,13 +92,6 @@ def setup_async_client(description=None, cmdline=None):
             #    handle_local_echo=False,
         )
     elif args.comm == "tls":
-        cwd = os.getcwd().split("/")[-1]
-        if cwd == "examples":
-            path = "."
-        elif cwd == "test":
-            path = "../examples"
-        else:
-            path = "examples"
         client = AsyncModbusTlsClient(
             args.host,
             port=args.port,
@@ -109,8 +104,8 @@ def setup_async_client(description=None, cmdline=None):
             #    strict=True,
             # TLS setup parameters
             #    sslctx=sslctx,
-            certfile=f"{path}/certificates/pymodbus.crt",
-            keyfile=f"{path}/certificates/pymodbus.key",
+            certfile=helper.get_certificate("crt"),
+            keyfile=helper.get_certificate("key"),
             #    password="none",
             server_hostname="localhost",
         )
@@ -128,25 +123,15 @@ async def run_async_client(client, modbus_calls=None):
     _logger.info("### End of Program")
 
 
-async def helper():
-    """Combine the setup and run"""
-    args = [
-        "--comm",
-        "udp",
-        "--host",
-        "127.0.0.1",
-        "--port",
-        "5020",
-        "--framer",
-        "socket",
-        "--log",
-        "debug",
-    ]
-    testclient = setup_async_client(
-        description="Run asynchronous client.", cmdline=args
-    )
-    await run_async_client(testclient)
+async def run_a_few_calls(client):
+    """Test connection works."""
+    rr = await client.read_coils(32, 1, slave=1)
+    assert len(rr.bits) == 8
+    rr = await client.read_holding_registers(4, 2, slave=1)
+    assert rr.registers[0] == 17
+    assert rr.registers[1] == 17
 
 
 if __name__ == "__main__":
-    asyncio.run(helper(), debug=True)
+    testclient = setup_async_client(description="Run asynchronous client.")
+    asyncio.run(run_async_client(testclient, modbus_calls=run_a_few_calls), debug=True)

examples/client_calls.py

@@ -47,6 +47,7 @@
 
 
 _logger = logging.getLogger()
+_logger.setLevel("DEBUG")
 
 
 SLAVE = 0x01
@@ -84,7 +85,7 @@ async def async_template_call(client):
 def template_call(client):
     """Show complete modbus call, sync version."""
     try:
-        rr = client.read_coils(1, 1, slave=SLAVE)
+        rr = client.read_coils(32, 1, slave=SLAVE)
     except ModbusException as exc:
         txt = f"ERROR: exception in pymodbus {exc}"
         _logger.error(txt)
@@ -289,14 +290,14 @@ def run_sync_calls(client):
     template_call(client)
 
 
-async def helper():
+async def async_helper():
     """Combine the setup and run"""
     testclient = setup_async_client(description="Run asynchronous client.")
     await run_async_client(testclient, modbus_calls=run_async_calls)
 
 
 if __name__ == "__main__":
-    asyncio.run(helper())
+    asyncio.run(async_helper())
     testclient = setup_sync_client(
         description="Run modbus calls in synchronous client."
     )

examples/client_custom_msg.py

@@ -124,9 +124,9 @@ def __init__(self, address, **kwargs):
 # --------------------------------------------------------------------------- #
 
 
-def run_custom_client():
+def run_custom_client(host, port):
     """Run versions of read coil."""
-    with ModbusClient(host="localhost", port=5020) as client:
+    with ModbusClient(host=host, port=port) as client:
         # Standard call
         request = ReadCoilsRequest(32, 1, slave=1)
         result = client.execute(request)
@@ -145,4 +145,4 @@ def run_custom_client():
 
 
 if __name__ == "__main__":
-    run_custom_client()
+    run_custom_client("localhost", "5020")

examples/client_payload.py

@@ -136,11 +136,11 @@ async def run_payload_calls(client):
         print("\n")
 
 
-async def helper():
+async def async_helper():
     """Combine the setup and run"""
     testclient = setup_async_client(description="Run asynchronous client.")
     await run_async_client(testclient, modbus_calls=run_payload_calls)
 
 
 if __name__ == "__main__":
-    asyncio.run(helper())
+    asyncio.run(async_helper())

examples/client_sync.py

@@ -22,12 +22,11 @@
     python3 server_sync.py
 """
 import logging
-import os
 
 # --------------------------------------------------------------------------- #
 # import the various client implementations
 # --------------------------------------------------------------------------- #
-from examples.helper import get_commandline
+from examples import helper
 from pymodbus.client import (
     ModbusSerialClient,
     ModbusTcpClient,
@@ -37,11 +36,12 @@
 
 
 _logger = logging.getLogger()
+_logger.setLevel("DEBUG")
 
 
 def setup_sync_client(description=None, cmdline=None):
     """Run client setup."""
-    args = get_commandline(
+    args = helper.get_commandline(
         server=False,
         description=description,
         cmdline=cmdline,
@@ -93,13 +93,6 @@ def setup_sync_client(description=None, cmdline=None):
             #    handle_local_echo=False,
         )
     elif args.comm == "tls":
-        cwd = os.getcwd().split("/")[-1]
-        if cwd == "examples":
-            path = "."
-        elif cwd == "test":
-            path = "../examples"
-        else:
-            path = "examples"
         client = ModbusTlsClient(
             args.host,
             port=args.port,
@@ -112,8 +105,8 @@ def setup_sync_client(description=None, cmdline=None):
             #    strict=True,
             # TLS setup parameters
             #    sslctx=None,
-            certfile=f"{path}/certificates/pymodbus.crt",
-            keyfile=f"{path}/certificates/pymodbus.key",
+            certfile=helper.get_certificate("crt"),
+            keyfile=helper.get_certificate("key"),
             #    password=None,
             server_hostname="localhost",
         )
@@ -130,6 +123,15 @@ def run_sync_client(client, modbus_calls=None):
     _logger.info("### End of Program")
 
 
+def run_a_few_calls(client):
+    """Test connection works."""
+    rr = client.read_coils(32, 1, slave=1)
+    assert len(rr.bits) == 8
+    rr = client.read_holding_registers(4, 2, slave=1)
+    assert rr.registers[0] == 17
+    assert rr.registers[1] == 17
+
+
 if __name__ == "__main__":
     testclient = setup_sync_client(description="Run synchronous client.")
-    run_sync_client(testclient)
+    run_sync_client(testclient, modbus_calls=run_a_few_calls)

examples/client_sync_test.py

@@ -29,17 +29,18 @@
 # --------------------------------------------------------------------------- #
 # import the various client implementations
 # --------------------------------------------------------------------------- #
-from examples.helper import get_commandline
+from examples import helper
 from pymodbus.client import ModbusTcpClient
 from pymodbus.exceptions import ModbusException
 
 
 _logger = logging.getLogger()
+_logger.setLevel("DEBUG")
 
 
 def run_sync_client():
     """Run sync client."""
-    args = get_commandline()
+    args = helper.get_commandline(server=False)
     _logger.info("### Create client object")
     client = ModbusTcpClient(
         args.host,

examples/helper.py

@@ -6,6 +6,7 @@
 """
 import argparse
 import logging
+import os
 
 from pymodbus import pymodbus_apply_logging_config
 from pymodbus.transaction import (
@@ -116,3 +117,20 @@ def get_commandline(server=False, description=None, extras=None, cmdline=None):
     if args.comm != "serial" and args.port:
         args.port = int(args.port)
     return args
+
+
+def get_certificate(suffix: str):
+    """Get example certificate."""
+    delimiter = "\\" if os.name == "nt" else "/"
+    cwd = os.getcwd().split(delimiter)[-1]
+    if cwd == "examples":
+        path = "."
+    elif cwd == "sub_examples":
+        path = "../../examples"
+    elif cwd == "test":
+        path = "../examples"
+    elif cwd == "pymodbus":
+        path = "examples"
+    else:
+        raise RuntimeError(f"**Error** Cannot find certificate path={cwd}")
+    return f"{path}/certificates/pymodbus.{suffix}"