espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-08-18 00:19:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

[tests] add traffic check in test case Cert_7_1_01 (#3405)

Browse Source
This commit is contained in:
Irving-cl
2019-02-18 13:41:30 -08:00
committed by Jonathan Hui
parent 968f9b6898
commit cf7bc35bbe
4 changed files with 142 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/command.py
+79 -13
View File
@@ -27,14 +27,16 @@
# POSSIBILITY OF SUCH DAMAGE.
#
import binascii
import sys
import ipv6
from network_data import Prefix, BorderRouter, LowpanId
import network_data
import network_layer
import config
import mle
from collections import Counter
from enum import IntEnum
class CheckType(IntEnum):
@@ -42,6 +44,10 @@ class CheckType(IntEnum):
NOT_CONTAIN = 1
OPTIONAL = 2
class NetworkDataCheckType:
PREFIX_CNT = 1
PREFIX_CONTENT = 2
def check_address_query(command_msg, source_node, destination_address):
"""Verify source_node sent a properly formatted Address Query Request message to the destination_address.
"""
@@ -303,10 +309,46 @@ def check_child_id_request(command_msg, tlv_request = CheckType.OPTIONAL, \
check_tlv_request_tlv(command_msg, CheckType.CONTAIN, mle.TlvType.ADDRESS16)
check_tlv_request_tlv(command_msg, CheckType.CONTAIN, mle.TlvType.NETWORK_DATA)
def find_prefix_tlv(tlvs, cond_map):
"""Find a prefix tlv in tlvs which matchs some conditions specified by cond_map
"""
for tlv in tlvs:
if network_data.TlvType.PREFIX in cond_map:
if binascii.hexlify(tlv.prefix) != cond_map[network_data.TlvType.PREFIX]:
continue
if network_data.TlvType.BORDER_ROUTER in cond_map:
border_router_tlv = get_sub_tlv(tlv.sub_tlvs, network_data.BorderRouter)
if border_router_tlv.border_router_16 != cond_map[network_data.TlvType.BORDER_ROUTER]:
continue
return tlv
return None
def check_network_data(data, check_detail):
check_type = check_detail[0]
prefixes = [tlv for tlv in data.tlvs if isinstance(tlv, network_data.Prefix)]
if check_type == NetworkDataCheckType.PREFIX_CNT:
# check_detail[1] should be a integer number representing the minimum count of prefixes should be
min_cnt = check_detail[1]
assert len(prefixes) >= min_cnt, 'Network data should contain at least {} prefixes'.format(mn_cnt)
for prefix in prefixes:
check_prefix(prefix)
elif check_type == NetworkDataCheckType.PREFIX_CONTENT:
# check_detail[1] should be a list of dictionary(like
# [{network_data.TlvType.PREFIX='...', network_data.TlvType.BORDER_ROUTER='...'}])
# each entry of the dictionary represents one thing to check of the prefix tlv
assert len(prefixes) >= len(check_detail[1]), 'Network data seems to have less prefixes than expected'
# basic check of prefixes
for prefix in prefixes:
check_prefix(prefix)
for cond_map in check_detail[1]:
tlv = find_prefix_tlv(prefixes, cond_map)
assert tlv is not None, 'Some prefix sub-tlv is not found:{}'.format(cond_map)
def check_child_id_response(command_msg, route64 = CheckType.OPTIONAL, network_data = CheckType.OPTIONAL, \
address_registration = CheckType.OPTIONAL, active_timestamp = CheckType.OPTIONAL, \
pending_timestamp = CheckType.OPTIONAL, active_operational_dataset = CheckType.OPTIONAL, \
pending_operational_dataset = CheckType.OPTIONAL):
pending_operational_dataset = CheckType.OPTIONAL,
network_data_check = None):
"""Verify a properly formatted Child Id Response command message.
"""
command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
@@ -321,6 +363,16 @@ def check_child_id_response(command_msg, route64 = CheckType.OPTIONAL, network_d
check_mle_optional_tlv(command_msg, active_operational_dataset, mle.ActiveOperationalDataset)
check_mle_optional_tlv(command_msg, pending_operational_dataset, mle.PendingOperationalDataset)
if network_data_check != None:
network_data_tlv = command_msg.assertMleMessageContainsTlv(mle.NetworkData)
check_network_data(network_data_tlv, network_data_check)
def check_prefix(prefix):
"""Verify if a prefix contains 6loWPAN sub-TLV and border router sub-TLV
"""
assert contains_tlv(prefix.sub_tlvs, network_data.BorderRouter), 'Prefix doesn\'t contain a border router sub-TLV!'
assert contains_tlv(prefix.sub_tlvs, network_data.LowpanId), 'Prefix doesn\'t contain a LowpanId sub-TLV!'
def check_child_update_request_by_child(command_msg):
command_msg.assertMleMessageContainsTlv(mle.LeaderData)
command_msg.assertMleMessageContainsTlv(mle.Mode)
@@ -356,24 +408,21 @@ def check_secure_mle_key_id_mode(command_msg, key_id_mode):
assert isinstance(command_msg.mle, mle.MleMessageSecured)
assert command_msg.mle.aux_sec_hdr.key_id_mode == key_id_mode
def check_data_response(command_msg, network_data=CheckType.OPTIONAL, prefixes=[],
active_timestamp=CheckType.OPTIONAL):
def check_data_response(command_msg, network_data_opt=CheckType.OPTIONAL,
active_timestamp=CheckType.OPTIONAL,
network_data_check=None):
"""Verify a properly formatted Data Response command message.
"""
check_secure_mle_key_id_mode(command_msg, 0x02)
command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
command_msg.assertMleMessageContainsTlv(mle.LeaderData)
check_mle_optional_tlv(command_msg, network_data, mle.NetworkData)
check_mle_optional_tlv(command_msg, network_data_opt, mle.NetworkData)
check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp)
if network_data == CheckType.CONTAIN and len(prefixes) > 0:
network_data_tlv = command_msg.get_mle_message_tlv(mle.NetworkData)
prefix_tlvs = [tlv for tlv in network_data_tlv.tlvs if isinstance(tlv, Prefix)]
assert len(prefix_tlvs) >= len(prefixes)
for prefix in prefix_tlvs:
assert contains_tlv(prefix.sub_tlvs, BorderRouter)
assert contains_tlv(prefix.sub_tlvs, LowpanId)
if network_data_check != None:
network_data_tlv = command_msg.assertMleMessageContainsTlv(mle.NetworkData)
check_network_data(network_data_tlv, network_data_check)
def check_child_update_request_from_parent(command_msg, leader_data=CheckType.OPTIONAL,
network_data=CheckType.OPTIONAL, challenge=CheckType.OPTIONAL,
@@ -389,7 +438,7 @@ def check_child_update_request_from_parent(command_msg, leader_data=CheckType.OP
check_mle_optional_tlv(command_msg, tlv_request, mle.TlvRequest)
check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp)
def check_child_update_response_from_parent(command_msg, timeout=CheckType.OPTIONAL,
def check_child_update_response(command_msg, timeout=CheckType.OPTIONAL,
address_registration=CheckType.OPTIONAL, address16=CheckType.OPTIONAL,
leader_data=CheckType.OPTIONAL, network_data=CheckType.OPTIONAL, response=CheckType.OPTIONAL,
link_layer_frame_counter=CheckType.OPTIONAL, mle_frame_counter=CheckType.OPTIONAL):
@@ -408,7 +457,24 @@ def check_child_update_response_from_parent(command_msg, timeout=CheckType.OPTIO
check_mle_optional_tlv(command_msg, link_layer_frame_counter, mle.LinkLayerFrameCounter)
check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter)
def unhashable_items_lists_equals(list1, list2):
return all(item in list1 for item in list2) and all(item in list2 for item in list1)
def check_message_address_registration_addr_set_equals(command_msg1, command_msg2):
"""Verify that all addresses in the address set of AddressRegistration tlv in msg2
are contained in that address set of AddressRegistration tlv in msg1
"""
addr_reg_tlv1 = command_msg1.assertMleMessageContainsTlv(mle.AddressRegistration)
addr_reg_tlv2 = command_msg2.assertMleMessageContainsTlv(mle.AddressRegistration)
# unordered lists comparison
assert unhashable_items_lists_equals(addr_reg_tlv1.addresses, addr_reg_tlv2.addresses), 'Some addresses are not included in AddressRegistration TLV'
def get_sub_tlv(tlvs, tlv_type):
for sub_tlv in tlvs:
if isinstance(sub_tlv, tlv_type):
return sub_tlv
def check_address_registration_tlv(addr_reg_tlv, address_set):
"""Verify all addresses contained in address_set are contained in add_reg_tlv
"""
assert all(addr in addr_reg_tlv.addresses for addr in address_set), 'Some addresses are not included in AddressRegistration TLV'