[otci] get router link information (#6105)

This commit adds link information to the router table
(i.e. get_router_table()).
This commit is contained in:
Simon Lin
2021-01-22 11:59:55 -08:00
committed by GitHub
parent a751b35748
commit de62590edb
3 changed files with 41 additions and 6 deletions
+28 -5
View File
@@ -37,7 +37,8 @@ from . import connectors
from .command_handlers import OTCommandHandler, OtCliCommandRunner, OtbrSshCommandRunner from .command_handlers import OTCommandHandler, OtCliCommandRunner, OtbrSshCommandRunner
from .connectors import Simulator from .connectors import Simulator
from .errors import UnexpectedCommandOutput, ExpectLineTimeoutError, CommandError, InvalidArgumentsError from .errors import UnexpectedCommandOutput, ExpectLineTimeoutError, CommandError, InvalidArgumentsError
from .types import ChildId, Rloc16, Ip6Addr, ThreadState, PartitionId, DeviceMode, RouterId, SecurityPolicy, Ip6Prefix from .types import ChildId, Rloc16, Ip6Addr, ThreadState, PartitionId, DeviceMode, RouterId, SecurityPolicy, Ip6Prefix, \
RouterTableEntry
from .utils import match_line, constant_property from .utils import match_line, constant_property
@@ -575,7 +576,7 @@ class OTCI(object):
line = self.__parse_str(self.execute_command('router list')) line = self.__parse_str(self.execute_command('router list'))
return list(map(RouterId, line.strip().split())) return list(map(RouterId, line.strip().split()))
def get_router_table(self) -> Dict[RouterId, Dict[str, Any]]: def get_router_table(self) -> Dict[RouterId, RouterTableEntry]:
"""table of routers.""" """table of routers."""
output = self.execute_command('router table') output = self.execute_command('router table')
if len(output) < 2: if len(output) < 2:
@@ -606,7 +607,7 @@ class OTCI(object):
col = lambda colname: self.__get_table_col(colname, headers, fields) col = lambda colname: self.__get_table_col(colname, headers, fields)
id = col('ID') id = col('ID')
table[RouterId(id)] = { table[RouterId(id)] = router = RouterTableEntry({
'id': RouterId(id), 'id': RouterId(id),
'rloc16': Rloc16(col('RLOC16'), 16), 'rloc16': Rloc16(col('RLOC16'), 16),
'next_hop': int(col('Next Hop')), 'next_hop': int(col('Next Hop')),
@@ -615,11 +616,33 @@ class OTCI(object):
'lq_out': int(col('LQ Out')), 'lq_out': int(col('LQ Out')),
'age': int(col('Age')), 'age': int(col('Age')),
'extaddr': col('Extended MAC'), 'extaddr': col('Extended MAC'),
} })
if 'Link' in headers:
router['link'] = int(col('Link'))
else:
# support older version of OT which does not output `Link` field
router['link'] = self.get_router_info(router['id'], silent=True)['link']
return table return table
# TODO: router <id> def get_router_info(self, id: int, silent: bool = False) -> RouterTableEntry:
cmd = f'router {id}'
info = {}
output = self.execute_command(cmd, silent=silent)
items = [line.strip().split(': ') for line in output]
headers = [h for h, _ in items]
fields = [f for _, f in items]
col = lambda colname: self.__get_table_col(colname, headers, fields)
return RouterTableEntry({
'id': RouterId(id),
'rloc16': Rloc16(col('Rloc'), 16),
'alloc': int(col('Alloc')),
'next_hop': int(col('Next Hop'), 16) >> 10, # convert RLOC16 to Router ID
'link': int(col('Link')),
})
# #
# Router utilities: Child management # Router utilities: Child management
+9 -1
View File
@@ -44,7 +44,7 @@ class Rloc16(int):
"""Represents a RLOC16.""" """Represents a RLOC16."""
def __repr__(self): def __repr__(self):
return hex(self) return '0x%04x' % self
class PartitionId(int): class PartitionId(int):
@@ -122,6 +122,14 @@ class Ip6Prefix(ipaddress.IPv6Network):
SecurityPolicy = namedtuple('SecurityPolicy', ['rotation_time', 'flags']) SecurityPolicy = namedtuple('SecurityPolicy', ['rotation_time', 'flags'])
"""Represents a Security Policy configuration.""" """Represents a Security Policy configuration."""
class RouterTableEntry(dict):
@property
def is_link_established(self):
return bool(self['link'])
if __name__ == '__main__': if __name__ == '__main__':
assert Ip6Addr('2001:0:0:0:0:0:0:1') == '2001::1' assert Ip6Addr('2001:0:0:0:0:0:0:1') == '2001::1'
assert repr(Ip6Addr('2001:0:0:0:0:0:0:1')) == '2001::1' assert repr(Ip6Addr('2001:0:0:0:0:0:0:1')) == '2001::1'
+4
View File
@@ -260,6 +260,7 @@ class TestOTCI(unittest.TestCase):
self.assertEqual(1, len(leader.get_router_list())) self.assertEqual(1, len(leader.get_router_list()))
self.assertEqual(1, len(leader.get_router_table())) self.assertEqual(1, len(leader.get_router_table()))
logging.info("Leader router table: %r", leader.get_router_table()) logging.info("Leader router table: %r", leader.get_router_table())
self.assertFalse(list(leader.get_router_table().values())[0].is_link_established)
logging.info('scan: %r', leader.scan()) logging.info('scan: %r', leader.scan())
logging.info('scan energy: %r', leader.scan_energy()) logging.info('scan energy: %r', leader.scan_energy())
@@ -499,6 +500,9 @@ class TestOTCI(unittest.TestCase):
self.assertEqual(2, len(leader.get_router_list())) self.assertEqual(2, len(leader.get_router_list()))
self.assertEqual(2, len(leader.get_router_table())) self.assertEqual(2, len(leader.get_router_table()))
logging.info('leader router table: %r', leader.get_router_table())
self.assertEqual({False, True},
set(router.is_link_established for router in leader.get_router_table().values()))
self.assertFalse(leader.is_singleton()) self.assertFalse(leader.is_singleton())