openthread/tests/scripts/thread-cert/network_diag.py

#!/usr/bin/env python3
#
#  Copyright (c) 2020, The OpenThread Authors.
#  All rights reserved.
#
#  Redistribution and use in source and binary forms, with or without
#  modification, are permitted provided that the following conditions are met:
#  1. Redistributions of source code must retain the above copyright
#     notice, this list of conditions and the following disclaimer.
#  2. Redistributions in binary form must reproduce the above copyright
#     notice, this list of conditions and the following disclaimer in the
#     documentation and/or other materials provided with the distribution.
#  3. Neither the name of the copyright holder nor the
#     names of its contributors may be used to endorse or promote products
#     derived from this software without specific prior written permission.
#
#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
#  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
#  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
#  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
#  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
#  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
#  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
#  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
#  POSSIBILITY OF SUCH DAMAGE.
#

import struct

from enum import IntEnum
from typing import List

import common
import ipaddress
import mle


class TlvType(IntEnum):
    EXT_ADDRESS = 0
    ADDRESS16 = 1
    MODE = 2
    POLLING_PERIOD = 3
    CONNECTIVITY = 4
    ROUTE64 = 5
    LEADER_DATA = 6
    NETWORK_DATA = 7
    IPV6_ADDRESS_LIST = 8
    MAC_COUNTERS = 9
    BATTERY_LEVEL = 14
    SUPPLY_VOLTAGE = 15
    CHILD_TABLE = 16
    CHANNEL_PAGES = 17
    TYPE_LIST = 18
    MAX_CHILD_TIMEOUT = 19


class Ipv6AddressList:

    def __init__(self, addresses: List[ipaddress.IPv6Address]):
        self._addresses = addresses

    @property
    def addresses(self):
        return self._addresses

    def __eq__(self, other):
        common.expect_the_same_class(self, other)
        return self.addresses == other.addresses

    def __repr__(self):
        return f'Ipv6AddressList({self.addresses})'


class Ipv6AddressListFactory:

    def parse(self, data, message_info):
        addresses = []
        while data.tell() < message_info.length:
            addresses.append(ipaddress.IPv6Address(data.read(16)))
        return Ipv6AddressList(addresses)


class MacCounters:

    def __init__(self, counters: List[int]):
        self._counters = counters

    @property
    def if_in_unknown_protos(self):
        return self._counters[0]

    @property
    def if_in_errors(self):
        return self._counters[1]

    @property
    def if_out_errors(self):
        return self._counters[2]

    @property
    def if_in_ucast_pkts(self):
        return self._counters[3]

    @property
    def if_in_broadcast_pkts(self):
        return self._counters[4]

    @property
    def if_in_discards(self):
        return self._counters[5]

    @property
    def if_out_ucast_pkts(self):
        return self._counters[6]

    @property
    def if_out_broadcast_pkts(self):
        return self._counters[7]

    @property
    def if_out_discards(self):
        return self._counters[8]

    @property
    def counters(self):
        return self._counters

    def __eq__(self, other):
        common.expect_the_same_class(self, other)

        return self.counters == other.counters

    def __repr__(self):
        return ('MacCounters(' + f'if_in_unknown_protos={self.if_in_unknown_protos}, ' +
                f'if_in_errors={self.if_in_errors}, ' + f'if_out_errors={self.if_out_errors}, ' +
                f'if_in_ucast_pkts={self.if_in_ucast_pkts}, ' + f'if_in_broadcast_pkts={self.if_in_broadcast_pkts}, ' +
                f'if_in_discards={self.if_in_discards}, ' + f'if_out_ucast_pkts={self.if_out_ucast_pkts}, ' +
                f'if_out_broadcast_pkts={self.if_out_broadcast_pkts}, ' + f'if_out_discards={self.if_out_discards})')


class MacCountersFactory:

    def parse(self, data, message_info):
        return MacCounters(struct.unpack('>9I', data.read(4 * 9)))


class BatteryLevel:

    def __init__(self, battery_level: int):
        self._battery_level = battery_level

    @property
    def battery_level(self):
        return self._battery_level

    def __eq__(self, other):
        common.expect_the_same_class(self, other)

        return self.battery_level == other.battery_level

    def __repr__(self):
        return f'BatteryLevel(battery_level={self.battery_level})'


class BatteryLevelFactory:

    def parse(self, data, message_info):
        return BatteryLevel(struct.unpack('>B', data.read(1))[0])


class SupplyVoltage:

    def __init__(self, supply_voltage: int):
        self._supply_voltage = supply_voltage

    @property
    def supply_voltage(self):
        return self._supply_voltage

    def __eq__(self, other):
        common.expect_the_same_class(self, other)

        return self.supply_voltage == other.supply_voltage

    def __repr__(self):
        return f'SupplyVoltage(supply_voltage={self.supply_voltage})'


class SupplyVoltageFactory:

    def parse(self, data, message_info):
        return SupplyVoltage(struct.unpack('>H', data.read(2))[0])


class ChildTableEntry:

    def __init__(self, timeout: int, child_id: int, mode: mle.Mode):
        self._timeout = timeout
        self._child_id = child_id
        self._mode = mode

    @property
    def timeout(self):
        return self._timeout

    @property
    def child_id(self):
        return self._child_id

    @property
    def mode(self):
        return self._mode

    def __eq__(self, other):
        common.expect_the_same_class(self, other)

        return (self.timeout == other.timeout and self.child_id == other.child_id and self.mode == other.mode)

    def __repr__(self):
        return f'ChildTableEntry(timeout={self.timeout}, child_id={self.child_id}, mode={self.mode})'


class ChildTable:

    def __init__(self, children: List[ChildTableEntry]):
        self._children = sorted(children, key=lambda child: child.child_id)

    @property
    def children(self):
        return self._children

    def __eq__(self, other):
        common.expect_the_same_class(self, other)

        return self.children == other.children

    def __repr__(self):
        return f'ChildTable({self.children})'


class ChildTableFactory:

    def parse(self, data, message_info):
        children = []
        while message_info.length > 0:
            timeout_and_id = struct.unpack('>H', data.read(2))[0]
            message_info.length -= 2

            timeout = (timeout_and_id & 0xf800) >> 11
            child_id = timeout_and_id & 0x1fff

            mode = mle.ModeFactory().parse(data, message_info)
            message_info.length -= 1

            children.append(ChildTableEntry(timeout, child_id, mode))
        return ChildTable(children)


class ChannelPages:

    def __init__(self, channel_pages: bytes):
        self._channel_pages = channel_pages

    @property
    def channel_pages(self):
        return self._channel_pages

    def __eq__(self, other):
        common.expect_the_same_class(self, other)

        return self.channel_pages == other.channel_pages

    def __repr__(self):
        return f'ChannelPages(channel_pages={self.channel_pages})'


class ChannelPagesFactory:

    def parse(self, data, message_info):
        return ChannelPages(data.getvalue())


class TypeList:

    def __init__(self, tlv_types: List[int]):
        self._tlv_types = tlv_types

    @property
    def tlv_types(self):
        return self._tlv_types

    def __eq__(self, other):
        common.expect_the_same_class(self, other)

        return self.tlv_types == other.tlv_types

    def __repr__(self):
        return f'TypeList(tlv_types={self.tlv_types})'


class TypeListFactory:

    def parse(self, data, message_info):
        return TypeList([ord(t) for t in data.getvalue()])


class MaxChildTimeout:

    def __init__(self, max_child_timeout: int):
        self._max_child_timeout = max_child_timeout

    @property
    def max_child_timeout(self):
        return self._max_child_timeout

    def __eq__(self, other):
        common.expect_the_same_class(self, other)

        return self.max_child_timeout == other.max_child_timeout

    def __repr__(self):
        return f'MaxChildTimeout(max_child_timeout={self.max_child_timeout})'


class MaxChildTimeoutFactory:

    def parse(self, data, message_info):
        return MaxChildTimeout(struct.unpack('>I', data.read(4))[0])