From de62590edb5b67a58ad283936b4eea75dfd09cd3 Mon Sep 17 00:00:00 2001 From: Simon Lin Date: Sat, 23 Jan 2021 03:59:55 +0800 Subject: [PATCH] [otci] get router link information (#6105) This commit adds link information to the router table (i.e. get_router_table()). --- tools/otci/otci/otci.py | 33 ++++++++++++++++++++++++++++----- tools/otci/otci/types.py | 10 +++++++++- tools/otci/tests/test_otci.py | 4 ++++ 3 files changed, 41 insertions(+), 6 deletions(-) diff --git a/tools/otci/otci/otci.py b/tools/otci/otci/otci.py index 65686a3c6..7070bb775 100644 --- a/tools/otci/otci/otci.py +++ b/tools/otci/otci/otci.py @@ -37,7 +37,8 @@ from . import connectors from .command_handlers import OTCommandHandler, OtCliCommandRunner, OtbrSshCommandRunner from .connectors import Simulator 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 @@ -575,7 +576,7 @@ class OTCI(object): line = self.__parse_str(self.execute_command('router list')) 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.""" output = self.execute_command('router table') if len(output) < 2: @@ -606,7 +607,7 @@ class OTCI(object): col = lambda colname: self.__get_table_col(colname, headers, fields) id = col('ID') - table[RouterId(id)] = { + table[RouterId(id)] = router = RouterTableEntry({ 'id': RouterId(id), 'rloc16': Rloc16(col('RLOC16'), 16), 'next_hop': int(col('Next Hop')), @@ -615,11 +616,33 @@ class OTCI(object): 'lq_out': int(col('LQ Out')), 'age': int(col('Age')), '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 - # TODO: router + 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 diff --git a/tools/otci/otci/types.py b/tools/otci/otci/types.py index 0ed8a8110..f5d39a1de 100644 --- a/tools/otci/otci/types.py +++ b/tools/otci/otci/types.py @@ -44,7 +44,7 @@ class Rloc16(int): """Represents a RLOC16.""" def __repr__(self): - return hex(self) + return '0x%04x' % self class PartitionId(int): @@ -122,6 +122,14 @@ class Ip6Prefix(ipaddress.IPv6Network): SecurityPolicy = namedtuple('SecurityPolicy', ['rotation_time', 'flags']) """Represents a Security Policy configuration.""" + +class RouterTableEntry(dict): + + @property + def is_link_established(self): + return bool(self['link']) + + if __name__ == '__main__': 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' diff --git a/tools/otci/tests/test_otci.py b/tools/otci/tests/test_otci.py index ee7d48544..7c9eb4f2e 100644 --- a/tools/otci/tests/test_otci.py +++ b/tools/otci/tests/test_otci.py @@ -260,6 +260,7 @@ class TestOTCI(unittest.TestCase): self.assertEqual(1, len(leader.get_router_list())) self.assertEqual(1, len(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 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_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())