[otci] update to support more commands (openthread#11015)

* otci now supports all commands the Thread Test Harness requires
* improve typehinting across the board
* fixes some typos

Author: thcu-gp

tests/scripts/thread-cert/command.py

@@ -144,8 +144,8 @@ def check_tlv_request_tlv(command_msg, check_type, tlv_id):
 
     elif check_type == CheckType.NOT_CONTAIN:
         if tlv_request_tlv is not None:
-            assert (any(tlv_id == tlv for tlv in tlv_request_tlv.tlvs) is
-                    False), "Error: The msg contains TLV Request TLV ID: {}".format(tlv_id)
+            assert (not any(tlv_id == tlv
+                            for tlv in tlv_request_tlv.tlvs)), f"Error: The msg contains TLV Request TLV ID: {tlv_id}"
 
     elif check_type == CheckType.OPTIONAL:
         if tlv_request_tlv is not None:

tools/otci/otci/__init__.py

@@ -31,14 +31,15 @@
 from .constants import THREAD_VERSION_1_1, THREAD_VERSION_1_2
 from .command_handlers import OTCommandHandler
 from .otci import OTCI
-from .otci import \
-    connect_cli_sim, \
-    connect_cli_serial, \
-    connect_ncp_sim, \
-    connect_cmd_handler, \
-    connect_otbr_ssh, \
-    connect_otbr_adb_tcp, \
-    connect_otbr_adb_usb
+from .otci import (
+    connect_cli_sim,
+    connect_cli_serial,
+    connect_ncp_sim,
+    connect_cmd_handler,
+    connect_otbr_ssh,
+    connect_otbr_adb_tcp,
+    connect_otbr_adb_usb,
+)
 
 from .types import Rloc16, ChildId, NetifIdentifier
 
@@ -61,4 +62,6 @@
     'NetifIdentifier',
     'THREAD_VERSION_1_1',
     'THREAD_VERSION_1_2',
+    'THREAD_VERSION_1_3',
+    'THREAD_VERSION_1_4',
 ] + _connectors

tools/otci/otci/command_handlers.py

@@ -105,21 +105,21 @@ class OtCliCommandRunner(OTCommandHandler):
 
     __ASYNC_COMMANDS = {'scan', 'ping', 'discover'}
 
-    def __init__(self, otcli: OtCliHandler, is_spinel_cli=False):
+    def __init__(self, otcli: OtCliHandler, is_spinel_cli: bool = False):
         self.__otcli: OtCliHandler = otcli
         self.__is_spinel_cli = is_spinel_cli
         self.__expect_command_echoback = not self.__is_spinel_cli
         self.__line_read_callback = None
 
-        self.__pending_lines = queue.Queue()
+        self.__pending_lines: queue.Queue[str] = queue.Queue()
         self.__should_close = threading.Event()
         self.__otcli_reader = threading.Thread(target=self.__otcli_read_routine, daemon=True)
         self.__otcli_reader.start()
 
     def __repr__(self):
         return repr(self.__otcli)
 
-    def execute_command(self, cmd, timeout=10) -> List[str]:
+    def execute_command(self, cmd: str, timeout: float = 10) -> List[str]:
         assert not self.__should_close.is_set(), "OT CLI is already closed."
         self.__otcli.writeline(cmd)
 
@@ -137,13 +137,13 @@ def execute_command(self, cmd, timeout=10) -> List[str]:
                                     asynchronous=cmd.split()[0] in OtCliCommandRunner.__ASYNC_COMMANDS)
         return output
 
-    def execute_platform_command(self, cmd, timeout=10) -> List[str]:
+    def execute_platform_command(self, cmd: str, timeout: float = 10) -> List[str]:
         raise NotImplementedError(f'Platform command is not supported on {self.__class__.__name__}')
 
     def wait(self, duration: float) -> List[str]:
         self.__otcli.wait(duration)
 
-        output = []
+        output: List[str] = []
         try:
             while True:
                 line = self.__pending_lines.get_nowait()
@@ -166,8 +166,11 @@ def set_line_read_callback(self, callback: Optional[Callable[[str], Any]]):
     # Private methods
     #
 
-    def __expect_line(self, timeout: float, expect_line: Union[str, Pattern], asynchronous=False) -> List[str]:
-        output = []
+    def __expect_line(self,
+                      timeout: float,
+                      expect_line: Union[str, Pattern[str]],
+                      asynchronous: bool = False) -> List[str]:
+        output: List[str] = []
 
         if not asynchronous:
             while True:
@@ -222,12 +225,13 @@ def __otcli_read_routine(self):
             logging.debug('%s: %s', self.__otcli, line)
 
             if not OtCliCommandRunner.__PATTERN_LOG_LINE.match(line):
+                logging.info('%s: %s', self.__otcli, line)
                 self.__pending_lines.put(line)
 
 
 class OtbrSshCommandRunner(OTCommandHandler):
 
-    def __init__(self, host, port, username, password, sudo):
+    def __init__(self, host: str, port: int, username: str, password: str, sudo: bool):
         import paramiko
 
         self.__host = host
@@ -272,16 +276,16 @@ def execute_command(self, cmd: str, timeout: float) -> List[str]:
 
         return output
 
-    def execute_platform_command(self, cmd, timeout=10) -> List[str]:
+    def execute_platform_command(self, cmd: str, timeout: float = 10) -> List[str]:
         if self.__sudo:
             cmd = 'sudo ' + cmd
 
         return self.shell(cmd, timeout=timeout)
 
     def shell(self, cmd: str, timeout: float) -> List[str]:
-        cmd_in, cmd_out, cmd_err = self.__ssh.exec_command(cmd, timeout=int(timeout), bufsize=1024)
-        errput = [l.rstrip('\r\n') for l in cmd_err.readlines()]
-        output = [l.rstrip('\r\n') for l in cmd_out.readlines()]
+        _, cmd_out, cmd_err = self.__ssh.exec_command(cmd, timeout=int(timeout), bufsize=1024)
+        errput = [line.rstrip('\r\n') for line in cmd_err.readlines()]
+        output = [line.rstrip('\r\n') for line in cmd_out.readlines()]
 
         if errput:
             raise CommandError(cmd, errput)

tools/otci/otci/connectors.py

@@ -30,7 +30,7 @@
 import subprocess
 import time
 from abc import abstractmethod, ABC
-from typing import Optional
+from typing import Any, Optional
 
 
 class OtCliHandler(ABC):
@@ -72,7 +72,7 @@ def go(self, duration: float):
 class OtCliPopen(OtCliHandler):
     """Connector for OT CLI process (a Popen instance)."""
 
-    def __init__(self, proc: subprocess.Popen, nodeid: int, simulator: Simulator):
+    def __init__(self, proc: subprocess.Popen[Any], nodeid: int, simulator: Optional[Simulator]):
         self.__otcli_proc = proc
         self.__nodeid = nodeid
         self.__simulator = simulator
@@ -108,7 +108,7 @@ def close(self):
 class OtCliSim(OtCliPopen):
     """Connector for OT CLI Simulation instances."""
 
-    def __init__(self, executable: str, nodeid: int, simulator: Simulator):
+    def __init__(self, executable: str, nodeid: int, simulator: Optional[Simulator]):
         logging.info('%s: executable=%s', self.__class__.__name__, executable)
 
         proc = subprocess.Popen(args=[executable, str(nodeid)],
@@ -123,7 +123,7 @@ def __init__(self, executable: str, nodeid: int, simulator: Simulator):
 class OtNcpSim(OtCliPopen):
     """Connector for OT NCP Simulation instances."""
 
-    def __init__(self, executable: str, nodeid: int, simulator: Simulator):
+    def __init__(self, executable: str, nodeid: int, simulator: Optional[Simulator]):
         logging.info('%s: executable=%s', self.__class__.__name__, executable)
 
         proc = subprocess.Popen(args=f'spinel-cli.py -p "{executable}" -n {nodeid} 2>&1',
@@ -144,6 +144,7 @@ def __init__(self, dev: str, baudrate: int):
 
         import serial
         self.__serial = serial.Serial(self.__dev, self.__baudrate, timeout=0.1, exclusive=True)
+        self.writeline('\r\n')
         self.__linebuffer = b''
 
     def __repr__(self):
@@ -164,7 +165,7 @@ def readline(self) -> Optional[str]:
         return None
 
     def writeline(self, s: str):
-        self.__serial.write((s + '\n').encode('utf-8'))
+        self.__serial.write((s + '\r\n').encode('utf-8'))
 
     def wait(self, duration: float):
         time.sleep(duration)

tools/otci/otci/constants.py

@@ -30,3 +30,5 @@
 # Thread versions
 THREAD_VERSION_1_1 = 2
 THREAD_VERSION_1_2 = 3
+THREAD_VERSION_1_3 = 4
+THREAD_VERSION_1_4 = 5

tools/otci/otci/errors.py

@@ -26,7 +26,9 @@
 #  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 #  POSSIBILITY OF SUCH DAMAGE.
 #
-from typing import List
+
+import re
+from typing import Collection, List, Pattern, Union
 
 
 class OTCIError(Exception):
@@ -37,15 +39,26 @@ class OTCIError(Exception):
 class ExpectLineTimeoutError(OTCIError):
     """OTCI failed to find an expected line before timeout."""
 
-    def __init__(self, line):
+    def __init__(self, line: Union[str, Pattern[str], Collection[str]]):
         super(ExpectLineTimeoutError, self).__init__("Expected line %r, but timed out" % line)
 
 
 class CommandError(OTCIError):
     """OTCI failed to execute a command."""
 
+    __COMMAND_OUTPUT_ERROR_PATTERN = re.compile(r'Error (\d+): (.*)')
+
     def __init__(self, cmd: str, output: List[str]):
         self.__output = output
+
+        for line in output:
+            m = self.__COMMAND_OUTPUT_ERROR_PATTERN.match(line)
+            if not m:
+                continue
+            code, msg = m.groups()
+            self.code, self.msg = int(code), str(msg)
+            break
+
         super(CommandError, self).__init__("Command error while executing %r:\n%s\n" % (cmd, '\n'.join(output)))
 
     def error(self) -> str: