[OIS] Integration test suite improvements (#27516)

Fix handling none arguments commands.
Add common functions to parsing help response
and log messages from response.
Increasing the default timeout value of the
wait_for_output function to 30s.
Update shell test.

Signed-off-by: ATmobica <artur.tynecki@arm.com>

Author: ATmobica

src/test_driver/openiotsdk/integration-tests/common/device.py

@@ -38,7 +38,7 @@ def __init__(self, name: Optional[str] = None):
         else:
             self.name = name
 
-    def send(self, command, expected_output=None, wait_before_read=None, wait_for_response=10, assert_output=True):
+    def send(self, command, expected_output=None, wait_before_read=None, wait_for_response=30, assert_output=True):
         """
         Send command for client
         :param command: Command
@@ -73,7 +73,7 @@ def flush(self, timeout: float = 0) -> [str]:
             except queue.Empty:
                 return lines
 
-    def wait_for_output(self, search: str, timeout: float = 10, assert_timeout: bool = True, verbose: bool = False) -> [str]:
+    def wait_for_output(self, search: str, timeout: float = 30, assert_timeout: bool = True, verbose: bool = False) -> [str]:
         """
         Wait for expected output response
         :param search: Expected response string

src/test_driver/openiotsdk/integration-tests/common/utils.py

@@ -37,7 +37,7 @@ def get_setup_payload(device):
     :param device: serial device instance
     :return: setup payload or None
     """
-    ret = device.wait_for_output("SetupQRCode", timeout=30)
+    ret = device.wait_for_output("SetupQRCode")
     if ret is None or len(ret) < 2:
         return None
 
@@ -139,7 +139,10 @@ def send_zcl_command(devCtrl, cluster: str, command: str, nodeId: int, endpoint:
 
         clusterObj = getattr(GeneratedObjects, cluster)
         commandObj = getattr(clusterObj.Commands, command)
-        req = commandObj(**args)
+        if args is not None:
+            req = commandObj(**args)
+        else:
+            req = commandObj()
 
         res = asyncio.run(devCtrl.SendCommand(int(nodeId), int(endpoint), req, timedRequestTimeoutMs=requestTimeoutMs))
 
@@ -232,3 +235,21 @@ def read_zcl_attribute(devCtrl, cluster: str, attribute: str, nodeId: int, endpo
         err = -1
 
     return (err, res)
+
+
+def get_shell_commands_from_help_response(response):
+    """
+    Parse shell help command response to get the list of supported commands
+    :param response: help command response
+    :return: list of supported commands
+    """
+    return [line.split()[0].strip() for line in response]
+
+
+def get_log_messages_from_response(response):
+    """
+    Parse shell help command response to get the list of supported commands
+    :param response: device response
+    :return: raw log messages
+    """
+    return [' '.join(line.split()[2:]) for line in response]

src/test_driver/openiotsdk/integration-tests/lock-app/test_app.py

@@ -48,7 +48,7 @@ def controllerConfig(request):
 def test_smoke_test(device):
     ret = device.wait_for_output("Open IoT SDK lock-app example application start")
     assert ret is not None and len(ret) > 0
-    ret = device.wait_for_output("Open IoT SDK lock-app example application run", timeout=30)
+    ret = device.wait_for_output("Open IoT SDK lock-app example application run")
     assert ret is not None and len(ret) > 0
 
 
@@ -74,7 +74,7 @@ def test_commissioning(device, controller):
     assert nodeId is not None
     log.info("Device {} connected".format(commissionable_device.addresses[0]))
 
-    ret = device.wait_for_output("Commissioning completed successfully", timeout=30)
+    ret = device.wait_for_output("Commissioning completed successfully")
     assert ret is not None and len(ret) > 0
 
     assert disconnect_device(devCtrl, nodeId)
@@ -104,7 +104,7 @@ def test_lock_ctrl(device, controller):
     nodeId = connect_device(devCtrl, setupPayload, commissionable_device)
     assert nodeId is not None
 
-    ret = device.wait_for_output("Commissioning completed successfully", timeout=30)
+    ret = device.wait_for_output("Commissioning completed successfully")
     assert ret is not None and len(ret) > 0
 
     err, res = send_zcl_command(devCtrl, "DoorLock", "SetUser", nodeId, LOCK_CTRL_TEST_ENDPOINT_ID,

src/test_driver/openiotsdk/integration-tests/shell/test_app.py

@@ -22,6 +22,7 @@
 import pytest
 from chip import exceptions
 from chip.setup_payload import SetupPayload
+from common.utils import get_shell_commands_from_help_response
 from packaging import version
 
 log = logging.getLogger(__name__)
@@ -40,10 +41,6 @@ def binaryPath(request, rootDir):
                       "echo", "log", "rand"]
 
 
-def get_shell_command(response):
-    return [line.split()[0].strip() for line in response]
-
-
 def parse_config_response(response):
     config = {}
     for param in response:
@@ -100,7 +97,7 @@ def test_command_check(device):
     # Help
     ret = device.send(command="help", expected_output="Done")
     assert ret is not None and len(ret) > 1
-    shell_commands = get_shell_command(ret[1:-1])
+    shell_commands = get_shell_commands_from_help_response(ret[1:-1])
     assert set(SHELL_COMMAND_NAME) == set(shell_commands)
 
     # Echo

src/test_driver/openiotsdk/integration-tests/unit-tests/test_app.py

@@ -32,9 +32,9 @@ def binaryPath(request, rootDir):
 
 
 def test_unit_tests(device):
-    ret = device.wait_for_output("Open IoT SDK unit-tests start", timeout=30)
+    ret = device.wait_for_output("Open IoT SDK unit-tests start")
     assert ret is not None and len(ret) > 0
-    ret = device.wait_for_output("Open IoT SDK unit-tests run", timeout=30)
+    ret = device.wait_for_output("Open IoT SDK unit-tests run")
     assert ret is not None and len(ret) > 0
 
     ret = device.wait_for_output("Test status:", 1200)