espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-07-26 22:09:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

[otci] fix some style issues (#8996)

Browse Source 
This commit fixes some style issues of otci. Also includes some minor
changes:
* added pytype check in otci
* exported `OTCommandHandler`
This commit is contained in:
Yakun Xu
2023-04-27 20:05:02 -07:00
committed by GitHub
parent 5657335de4
commit 674cbfaa5c
7 changed files with 89 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/otci.yml
+4
View File
@@ -68,6 +68,10 @@ jobs:
sudo rm /etc/apt/sources.list.d/* && sudo apt-get update
sudo apt-get --no-install-recommends install -y g++-multilib python3-setuptools python3-wheel
python3 -m pip install -r tests/scripts/thread-cert/requirements.txt
python3 -m pip install pytype adb-shell
- name: Style check
run: |
PYTHONPATH=./tests/scripts/thread-cert pytype tools/otci
- name: Build
run: |
./bootstrap
tools/otci/otci/__init__.py
+11 -2
View File
@@ -29,6 +29,7 @@
from . import errors
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, \
@@ -49,5 +50,13 @@ _connectors = [
'connect_cmd_handler',
]
__all__ = ['OTCI', 'errors', 'Rloc16', 'ChildId', 'NetifIdentifer', 'THREAD_VERSION_1_1', 'THREAD_VERSION_1_2'
] + _connectors
__all__ = [
'OTCI',
'OTCommandHandler',
'errors',
'Rloc16',
'ChildId',
'NetifIdentifier',
'THREAD_VERSION_1_1',
'THREAD_VERSION_1_2',
] + _connectors
tools/otci/otci/command_handlers.py
+6 -7
View File
@@ -31,7 +31,7 @@ import queue
import re
import threading
import time
from abc import abstractmethod
from abc import abstractmethod, ABC
from typing import Any, Callable, Optional, Union, List, Pattern
from .connectors import OtCliHandler
@@ -39,7 +39,7 @@ from .errors import ExpectLineTimeoutError, CommandError
from .utils import match_line
class OTCommandHandler:
class OTCommandHandler(ABC):
"""This abstract class defines interfaces of a OT Command Handler."""
@abstractmethod
@@ -50,12 +50,10 @@ class OTCommandHandler:
Note: each line SHOULD NOT contain '\r\n' at the end. The last line of output should be 'Done' or
'Error <code>: <msg>' following OT CLI conventions.
"""
pass
@abstractmethod
def close(self):
"""Method close should close the OT Command Handler."""
pass
@abstractmethod
def wait(self, duration: float) -> List[str]:
@@ -64,7 +62,6 @@ class OTCommandHandler:
Normally, OT CLI does not output when it's not executing any command. But OT CLI can also output
asynchronously in some cases (e.g. `Join Success` when Joiner joins successfully).
"""
pass
@abstractmethod
def set_line_read_callback(self, callback: Optional[Callable[[str], Any]]):
@@ -85,7 +82,7 @@ class OtCliCommandRunner(OTCommandHandler):
r'(Done|Error|Error \d+:.*|.*: command not found)$') # "Error" for spinel-cli.py
__PATTERN_LOG_LINE = re.compile(r'((\[(NONE|CRIT|WARN|NOTE|INFO|DEBG)\])'
r'|(-.*-+: )' # e.g. -CLI-----:
r'|(-.*-+: )' # e.g. -CLI-----:
r'|(\[[DINWC\-]\] (?=[\w\-]{14}:)\w+-*:)' # e.g. [I] Mac-----------:
r')')
"""regex used to filter logs"""
@@ -240,7 +237,9 @@ class OtbrSshCommandRunner(OTCommandHandler):
look_for_keys=False)
except paramiko.ssh_exception.AuthenticationException:
if not password:
self.__ssh.get_transport().auth_none(username)
transport = self.__ssh.get_transport()
assert transport is not None
transport.auth_none(username)
else:
raise
tools/otci/otci/connectors.py
+10 -10
View File
@@ -29,17 +29,16 @@
import logging
import subprocess
import time
from abc import abstractmethod
from abc import abstractmethod, ABC
from typing import Optional
class OtCliHandler:
class OtCliHandler(ABC):
"""This abstract class defines interfaces for a OT CLI Handler."""
@abstractmethod
def readline(self) -> str:
def readline(self) -> Optional[str]:
"""Method readline should return the next line read from OT CLI."""
pass
@abstractmethod
def writeline(self, s: str) -> None:
@@ -47,7 +46,6 @@ class OtCliHandler:
It should block until all characters are written to OT CLI.
"""
pass
@abstractmethod
def wait(self, duration: float) -> None:
@@ -56,15 +54,13 @@ class OtCliHandler:
A normal implementation should just call `time.sleep(duration)`. This is intended for proceeding Virtual Time
Simulation instances.
"""
pass
@abstractmethod
def close(self) -> None:
"""Method close should close the OT CLI Handler."""
pass
class Simulator:
class Simulator(ABC):
"""This abstract class defines interfaces for a Virtual Time Simulator."""
@abstractmethod
@@ -84,10 +80,12 @@ class OtCliPopen(OtCliHandler):
def __repr__(self):
return 'OTCli<%d>' % self.__nodeid
def readline(self) -> str:
def readline(self) -> Optional[str]:
assert self.__otcli_proc.stdout is not None
return self.__otcli_proc.stdout.readline().rstrip('\r\n')
def writeline(self, s: str):
assert self.__otcli_proc.stdin is not None
self.__otcli_proc.stdin.write(s + '\n')
self.__otcli_proc.stdin.flush()
@@ -100,6 +98,8 @@ class OtCliPopen(OtCliHandler):
time.sleep(duration)
def close(self):
assert self.__otcli_proc.stdin is not None
assert self.__otcli_proc.stdout is not None
self.__otcli_proc.stdin.close()
self.__otcli_proc.stdout.close()
self.__otcli_proc.wait()
@@ -120,7 +120,7 @@ class OtCliSim(OtCliPopen):
super().__init__(proc, nodeid, simulator)
class OtNcpSim(OtCliHandler):
class OtNcpSim(OtCliPopen):
"""Connector for OT NCP Simulation instances."""
def __init__(self, executable: str, nodeid: int, simulator: Simulator):
tools/otci/otci/otci.py
+55 -36
View File
@@ -64,7 +64,7 @@ class OTCI(object):
"""Gets the string representation of the OTCI instance."""
return repr(self.__otcmd)
def wait(self, duration: float, expect_line: Union[str, Pattern, Collection[Any]] = None):
def wait(self, duration: float, expect_line: Optional[Union[str, Pattern, Collection[Any]]] = None):
"""Wait for a given duration.
:param duration: The duration (in seconds) wait for.
@@ -104,6 +104,7 @@ class OTCI(object):
self.wait(2)
if i == self.__exec_command_retry:
raise
assert False
def __execute_command(self,
cmd: str,
@@ -219,7 +220,7 @@ class OTCI(object):
)
def ping(self,
ip: str,
ip: Union[str, Ip6Addr],
size: int = 8,
count: int = 1,
interval: float = 1,
@@ -261,15 +262,15 @@ class OTCI(object):
"""Stop sending ICMPv6 Echo Requests."""
self.execute_command('ping stop')
def discover(self, channel: int = None) -> List[Dict[str, Any]]:
def discover(self, channel: Optional[int] = None) -> List[Dict[str, Any]]:
"""Perform an MLE Discovery operation."""
return self.__scan_networks('discover', channel)
def scan(self, channel: int = None) -> List[Dict[str, Any]]:
def scan(self, channel: Optional[int] = None) -> List[Dict[str, Any]]:
"""Perform an IEEE 802.15.4 Active Scan."""
return self.__scan_networks('scan', channel)
def __scan_networks(self, cmd: str, channel: int = None) -> List[Dict[str, Any]]:
def __scan_networks(self, cmd: str, channel: Optional[int] = None) -> List[Dict[str, Any]]:
if channel is not None:
cmd += f' {channel}'
@@ -300,7 +301,7 @@ class OTCI(object):
return networks
def scan_energy(self, duration: float = None, channel: int = None) -> Dict[int, int]:
def scan_energy(self, duration: Optional[float] = None, channel: Optional[int] = None) -> Dict[int, int]:
"""Perform an IEEE 802.15.4 Energy Scan."""
cmd = 'scan energy'
if duration is not None:
@@ -496,9 +497,9 @@ class OTCI(object):
"""Try to switch to state detached, child, router or leader."""
self.execute_command(f'state {state}')
def get_rloc16(self) -> int:
def get_rloc16(self) -> Rloc16:
"""Get the Thread RLOC16 value."""
return self.__parse_int(self.execute_command('rloc16'), 16)
return Rloc16(self.__parse_int(self.execute_command('rloc16'), 16))
def get_router_id(self) -> int:
"""Get the Thread Router ID value."""
@@ -784,7 +785,9 @@ class OTCI(object):
for line in output:
k, v = line.split(': ')
if k == 'Server':
ip, port = re.match(OTCI._IPV6_SERVER_PORT_PATTERN, v).groups()
matched = re.match(OTCI._IPV6_SERVER_PORT_PATTERN, v)
assert matched is not None
ip, port = matched.groups()
config['server'] = (Ip6Addr(ip), int(port))
elif k == 'ResponseTimeout':
config['response_timeout'] = int(v[:-3])
@@ -799,9 +802,9 @@ class OTCI(object):
def dns_set_config(self,
server: Tuple[Union[str, ipaddress.IPv6Address], int],
response_timeout: int = None,
max_tx_attempts: int = None,
recursion_desired: bool = None):
response_timeout: Optional[int] = None,
max_tx_attempts: Optional[int] = None,
recursion_desired: Optional[bool] = None):
"""Set DNS client query config."""
cmd = f'dns config {str(server[0])} {server[1]}'
if response_timeout is not None:
@@ -936,6 +939,7 @@ class OTCI(object):
info = {'host': line}
result.append(info)
else:
assert info is not None
k, v = line.strip().split(': ')
if k == 'deleted':
if v not in ('true', 'false'):
@@ -962,6 +966,7 @@ class OTCI(object):
info = {'instance': line}
result.append(info)
else:
assert info is not None
k, v = line.strip().split(': ')
if k == 'deleted':
if v not in ('true', 'false'):
@@ -1130,7 +1135,7 @@ class OTCI(object):
port: int,
priority: int = 0,
weight: int = 0,
txt: Dict[str, Union[str, bytes, bool]] = None):
txt: Optional[Dict[str, Union[str, bytes, bool]]] = None):
instance = self.__escape_escapable(instance)
cmd = f'srp client service add {instance} {service} {port} {priority} {weight}'
if txt:
@@ -1164,7 +1169,9 @@ class OTCI(object):
def srp_client_get_server(self) -> Tuple[Ip6Addr, int]:
"""Get the SRP server (IP, port)."""
result = self.__parse_str(self.execute_command('srp client server'))
ip, port = re.match(OTCI._IPV6_SERVER_PORT_PATTERN, result).groups()
matched = re.match(OTCI._IPV6_SERVER_PORT_PATTERN, result)
assert matched
ip, port = matched.groups()
return Ip6Addr(ip), int(port)
def srp_client_get_service_key(self) -> bool:
@@ -1375,15 +1382,19 @@ class OTCI(object):
if k == 'Channel':
cfg['channel'] = int(v)
elif k == 'Timeout':
cfg['timeout'] = int(OTCI._CSL_TIMEOUT_PATTERN.match(v).group(1))
matched = OTCI._CSL_TIMEOUT_PATTERN.match(v)
assert matched is not None
cfg['timeout'] = int(matched.group(1))
elif k == 'Period':
cfg['period'] = int(OTCI._CSL_PERIOD_PATTERN.match(v).group(1))
matched = OTCI._CSL_PERIOD_PATTERN.match(v)
assert matched is not None
cfg['period'] = int(matched.group(1))
else:
logging.warning("Ignore unknown CSL parameter: %s: %s", k, v)
return cfg
def config_csl(self, channel: int = None, period: int = None, timeout: int = None):
def config_csl(self, channel: Optional[int] = None, period: Optional[int] = None, timeout: Optional[int] = None):
"""Configure CSL parameters.
:param channel: Set CSL channel.
@@ -1487,7 +1498,7 @@ class OTCI(object):
#
# Joiner operations
#
def joiner_start(self, psk: str, provisioning_url: str = None):
def joiner_start(self, psk: str, provisioning_url: Optional[str] = None):
"""Start the Joiner."""
cmd = f'joiner start {psk}'
if provisioning_url is not None:
@@ -1780,17 +1791,17 @@ class OTCI(object):
self.execute_command(cmd)
def dataset_set_buffer(self,
active_timestamp: int = None,
channel: int = None,
channel_mask: int = None,
extpanid: str = None,
mesh_local_prefix: str = None,
network_key: str = None,
network_name: str = None,
panid: int = None,
pskc: str = None,
security_policy: tuple = None,
pending_timestamp: int = None):
active_timestamp: Optional[int] = None,
channel: Optional[int] = None,
channel_mask: Optional[int] = None,
extpanid: Optional[str] = None,
mesh_local_prefix: Optional[str] = None,
network_key: Optional[str] = None,
network_name: Optional[str] = None,
panid: Optional[int] = None,
pskc: Optional[str] = None,
security_policy: Optional[tuple] = None,
pending_timestamp: Optional[int] = None):
if active_timestamp is not None:
self.execute_command(f'dataset activetimestamp {active_timestamp}')
@@ -1840,7 +1851,7 @@ class OTCI(object):
def disable_allowlist(self):
self.execute_command('macfilter addr disable')
def add_allowlist(self, addr: str, rssi: int = None):
def add_allowlist(self, addr: str, rssi: Optional[int] = None):
cmd = f'macfilter addr add {addr}'
if rssi is not None:
@@ -2053,7 +2064,10 @@ class OTCI(object):
return config
def set_backbone_router_config(self, seqno: int = None, delay: int = None, timeout: int = None):
def set_backbone_router_config(self,
seqno: Optional[int] = None,
delay: Optional[int] = None,
timeout: Optional[int] = None):
"""Configure local Backbone Router configuration for Thread 1.2 FTD.
Call register_backbone_router_dataset() to explicitly register Backbone Router service to Leader for Secondary Backbone Router.
@@ -2217,7 +2231,12 @@ class OTCI(object):
"""
self.execute_command(f'udp connect {ip} {port}')
def udp_send(self, ip: str = None, port: int = None, text: str = None, random_bytes: int = None, hex: str = None):
def udp_send(self,
ip: Optional[Union[str, Ip6Addr]] = None,
port: Optional[int] = None,
text: Optional[str] = None,
random_bytes: Optional[int] = None,
hex: Optional[str] = None):
"""Send a few bytes over UDP.
ip: the IPv6 destination address.
@@ -2292,11 +2311,11 @@ class OTCI(object):
"""Stops the application coap service."""
self.execute_command('coap stop')
def coap_get(self, addr: str, uri_path: str, type: str = "con"):
def coap_get(self, addr: Union[str, Ip6Addr], uri_path: str, type: str = "con"):
cmd = f'coap get {addr} {uri_path} {type}'
self.execute_command(cmd)
def coap_put(self, addr: str, uri_path: str, type: str = "con", payload: str = None):
def coap_put(self, addr: Union[str, Ip6Addr], uri_path: str, type: str = "con", payload: Optional[str] = None):
cmd = f'coap put {addr} {uri_path} {type}'
if payload is not None:
@@ -2304,7 +2323,7 @@ class OTCI(object):
self.execute_command(cmd)
def coap_post(self, addr: str, uri_path: str, type: str = "con", payload: str = None):
def coap_post(self, addr: Union[str, Ip6Addr], uri_path: str, type: str = "con", payload: Optional[str] = None):
cmd = f'coap post {addr} {uri_path} {type}'
if payload is not None:
@@ -2312,7 +2331,7 @@ class OTCI(object):
self.execute_command(cmd)
def coap_delete(self, addr: str, uri_path: str, type: str = "con", payload: str = None):
def coap_delete(self, addr: Union[str, Ip6Addr], uri_path: str, type: str = "con", payload: Optional[str] = None):
cmd = f'coap delete {addr} {uri_path} {type}'
if payload is not None:
tools/otci/otci/types.py
+2 -1
View File
@@ -28,6 +28,7 @@
#
import ipaddress
from collections import namedtuple
from enum import IntEnum
class ChildId(int):
@@ -52,7 +53,7 @@ class PartitionId(int):
pass
class NetifIdentifier(int):
class NetifIdentifier(IntEnum):
"""Represents a network interface identifier."""
UNSPECIFIED = 0
THERAD = 1
tools/otci/otci/utils.py
+1 -1
View File
@@ -33,7 +33,7 @@ from typing import Union, Collection, Any, Pattern
def match_line(line: str, expect_line: Union[str, Pattern, Collection[Any]]) -> bool:
"""Checks if a line is expected (matched by one of the given patterns)."""
if isinstance(expect_line, Pattern):
match = expect_line.match(line)
match = expect_line.match(line) is not None
elif isinstance(expect_line, str):
match = (line == expect_line)
else: