tests/scripts/thread-cert/command.py - third_party/openthread - Git at Google

 import ipv6

 import network_layer
 import config
 import mle

 from enum import IntEnum

 class CheckType(IntEnum):
     CONTAIN = 0
     NOT_CONTAIN = 1
     OPTIONAL = 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.
     """
     command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid)

     source_rloc = source_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
     assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, \
             "Error: The IPv6 source address is not the RLOC of the originator. The source node's rloc is: " \
             + str(ipv6.ip_address(source_rloc)) + ", but the source_address in command msg is: " \
             + str(command_msg.ipv6_packet.ipv6_header.source_address)

     assert ipv6.ip_address(destination_address.decode('utf-8')) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The IPv6 destination address is not expected."

 def check_address_notification(command_msg, source_node, destination_node):
     """Verify source_node sent a properly formatted Address Notification command message to destination_node.
     """
     command_msg.assertCoapMessageRequestUriPath('/a/an')
     command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid)
     command_msg.assertCoapMessageContainsTlv(network_layer.Rloc16)
     command_msg.assertCoapMessageContainsTlv(network_layer.MlEid)

     source_rloc = source_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
     assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, "Error: The IPv6 source address is not the RLOC of the originator."

     destination_rloc = destination_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
     assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The IPv6 destination address is not the RLOC of the destination."

 def check_address_error_notification(command_msg, source_node, destination_address):
     """Verify source_node sent a properly formatted Address Error Notification command message to destination_address.
     """
     command_msg.assertCoapMessageRequestUriPath('/a/ae')
     command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid)
     command_msg.assertCoapMessageContainsTlv(network_layer.MlEid)

     source_rloc = source_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
     assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, \
             "Error: The IPv6 source address is not the RLOC of the originator. The source node's rloc is: " \
             + str(ipv6.ip_address(source_rloc)) + ", but the source_address in command msg is: " \
             + str(command_msg.ipv6_packet.ipv6_header.source_address)

     assert ipv6.ip_address(destination_address.decode('utf-8')) == command_msg.ipv6_packet.ipv6_header.destination_address, \
             "Error: The IPv6 destination address is not expected. The destination node's rloc is: " \
             + str(ipv6.ip_address(destination_address.decode('utf-8'))) + ", but the destination_address in command msg is: " \
             + str(command_msg.ipv6_packet.ipv6_header.destination_address)

 def check_address_release(command_msg, destination_node):
     """Verify the message is a properly formatted address release destined to the given node.
     """
     command_msg.assertCoapMessageRequestUriPath('/a/ar')
     command_msg.assertCoapMessageContainsTlv(network_layer.Rloc16)
     command_msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress)

     destination_rloc = destination_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
     assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The destination is not RLOC address"

 def check_tlv_request_tlv(command_msg, check_type, tlv_id):
     """Verify if TLV Request TLV contains specified TLV ID
     """
     tlv_request_tlv = command_msg.get_mle_message_tlv(mle.TlvRequest)

     if check_type == CheckType.CONTAIN:
         assert tlv_request_tlv is not None, "Error: The msg doesn't contain TLV Request TLV"
         assert any(tlv_id == tlv for tlv in tlv_request_tlv.tlvs), "Error: The msg doesn't contain TLV Request TLV ID: {}".format(tlv_id)

     elif check_type == CheckType.NOT_CONTAIN:
         if tlv_request_tlv is not None:
             assert any(tlv_id == tlv for tlv in tlv_request_tlv.tlvs) is False, "Error: The msg contains TLV Request TLV ID: {}".format(tlv_id)

     elif check_type == CheckType.OPTIONAL:
         if tlv_request_tlv is not None:
             if any(tlv_id == tlv for tlv in tlv_request_tlv.tlvs):
                 print("TLV Request TLV contains TLV ID: {}".format(tlv_id))
             else:
                 print("TLV Request TLV doesn't contain TLV ID: {}".format(tlv_id))
         else:
             print("The msg doesn't contain TLV Request TLV")

     else:
         raise ValueError("Invalid check type")

 def check_link_request(command_msg, source_address = CheckType.OPTIONAL, leader_data = CheckType.OPTIONAL, \
     tlv_request_address16 = CheckType.OPTIONAL, tlv_request_route64 = CheckType.OPTIONAL, \
     tlv_request_link_margin = CheckType.OPTIONAL):

     """Verify a properly formatted Link Request command message.
     """
     command_msg.assertMleMessageContainsTlv(mle.Challenge)
     command_msg.assertMleMessageContainsTlv(mle.Version)

     check_mle_optional_tlv(command_msg, source_address, mle.SourceAddress)
     check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData)

     check_tlv_request_tlv(command_msg, tlv_request_address16, mle.TlvType.ADDRESS16)
     check_tlv_request_tlv(command_msg, tlv_request_route64, mle.TlvType.ROUTE64)
     check_tlv_request_tlv(command_msg, tlv_request_link_margin, mle.TlvType.LINK_MARGIN)

 def check_link_accept(command_msg, destination_node, \
     leader_data = CheckType.OPTIONAL, link_margin = CheckType.OPTIONAL, mle_frame_counter = CheckType.OPTIONAL, \
     challenge = CheckType.OPTIONAL, address16 = CheckType.OPTIONAL, route64 = CheckType.OPTIONAL, \
     tlv_request_link_margin = CheckType.OPTIONAL):
     """verify a properly formatted link accept command message.
     """
     command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
     command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
     command_msg.assertMleMessageContainsTlv(mle.Response)
     command_msg.assertMleMessageContainsTlv(mle.Version)

     check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData)
     check_mle_optional_tlv(command_msg, link_margin, mle.LinkMargin)
     check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter)
     check_mle_optional_tlv(command_msg, challenge, mle.Challenge)
     check_mle_optional_tlv(command_msg, address16, mle.Address16)
     check_mle_optional_tlv(command_msg, route64, mle.Route64)

     check_tlv_request_tlv(command_msg, tlv_request_link_margin, mle.TlvType.LINK_MARGIN)

     destination_link_local = destination_node.get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL)
     assert ipv6.ip_address(destination_link_local) == command_msg.ipv6_packet.ipv6_header.destination_address, \
         "Error: The destination is unexpected"

 def check_icmp_path(sniffer, path, nodes, icmp_type = ipv6.ICMP_ECHO_REQUEST):
     """Verify icmp message is forwarded along the path.
     """
     len_path = len(path)

     # Verify icmp message is forwarded to the next node of the path.
     for i in range(0, len_path):
         node_msg = sniffer.get_messages_sent_by(path[i])
         node_icmp_msg = node_msg.get_icmp_message(icmp_type)

         if i < len_path - 1:
             next_node = nodes[path[i + 1]]
             next_node_rloc16 = next_node.get_addr16()
             assert next_node_rloc16 == node_icmp_msg.mac_header.dest_address.rloc, "Error: The path is unexpected."
         else:
             return True

     return False

 def check_id_set(command_msg, router_id):
     """Check the command_msg's Route64 tlv to verify router_id is an active router.
     """
     tlv = command_msg.assertMleMessageContainsTlv(mle.Route64)
     return ((tlv.router_id_mask >> (63 - router_id)) & 1)

 def get_routing_cost(command_msg, router_id):
     """Check the command_msg's Route64 tlv to get the routing cost to router.
     """
     tlv = command_msg.assertMleMessageContainsTlv(mle.Route64)

     # Get router's mask pos
     # Turn the number into binary string. Need to consider the preceding 0 omitted during conversion.
     router_id_mask_str = bin(tlv.router_id_mask).replace('0b','')
     prefix_len = 64 - len(router_id_mask_str)
     routing_entry_pos = 0

     for i in range(0, router_id - prefix_len):
         if router_id_mask_str[i] == '1':
             routing_entry_pos += 1

     assert router_id_mask_str[router_id - prefix_len] == '1', "Error: The router isn't in the topology. \n" \
             + "route64 tlv is: %s. \nrouter_id is: %s. \nrouting_entry_pos is: %s. \nrouter_id_mask_str is: %s." \
             %(tlv, router_id, routing_entry_pos, router_id_mask_str)

     return tlv.link_quality_and_route_data[routing_entry_pos].route

 def check_mle_optional_tlv(command_msg, type, tlv):
     if (type == CheckType.CONTAIN):
         command_msg.assertMleMessageContainsTlv(tlv)
     elif (type == CheckType.NOT_CONTAIN):
         command_msg.assertMleMessageDoesNotContainTlv(tlv)
     elif (type == CheckType.OPTIONAL):
         command_msg.assertMleMessageContainsOptionalTlv(tlv)
     else:
         raise ValueError("Invalid check type")

 def check_mle_advertisement(command_msg):
     command_msg.assertSentWithHopLimit(255)
     command_msg.assertSentToDestinationAddress(config.LINK_LOCAL_ALL_NODES_ADDRESS)
     command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
     command_msg.assertMleMessageContainsTlv(mle.LeaderData)
     command_msg.assertMleMessageContainsTlv(mle.Route64)

 def check_parent_request(command_msg):
     """Verify a properly formatted Parent Request command message.
     """
     command_msg.assertSentWithHopLimit(255)
     command_msg.assertSentToDestinationAddress(config.LINK_LOCAL_ALL_ROUTERS_ADDRESS)
     command_msg.assertMleMessageContainsTlv(mle.Mode)
     command_msg.assertMleMessageContainsTlv(mle.Challenge)
     command_msg.assertMleMessageContainsTlv(mle.ScanMask)
     command_msg.assertMleMessageContainsTlv(mle.Version)

 def check_parent_response(command_msg, mle_frame_counter = CheckType.OPTIONAL):
     """Verify a properly formatted Parent Response command message.
     """
     command_msg.assertMleMessageContainsTlv(mle.Challenge)
     command_msg.assertMleMessageContainsTlv(mle.Connectivity)
     command_msg.assertMleMessageContainsTlv(mle.LeaderData)
     command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
     command_msg.assertMleMessageContainsTlv(mle.LinkMargin)
     command_msg.assertMleMessageContainsTlv(mle.Response)
     command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
     command_msg.assertMleMessageContainsTlv(mle.Version)

     check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter)

 def check_child_id_request(command_msg, tlv_request = CheckType.OPTIONAL, \
     mle_frame_counter = CheckType.OPTIONAL, address_registration = CheckType.OPTIONAL, \
     active_timestamp = CheckType.OPTIONAL, pending_timestamp = CheckType.OPTIONAL):
     """Verify a properly formatted Child Id Request command message.
     """
     command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
     command_msg.assertMleMessageContainsTlv(mle.Mode)
     command_msg.assertMleMessageContainsTlv(mle.Response)
     command_msg.assertMleMessageContainsTlv(mle.Timeout)
     command_msg.assertMleMessageContainsTlv(mle.Version)

     check_mle_optional_tlv(command_msg, tlv_request, mle.TlvRequest)
     check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter)
     check_mle_optional_tlv(command_msg, address_registration, mle.AddressRegistration)
     check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp)
     check_mle_optional_tlv(command_msg, pending_timestamp, mle.PendingTimestamp)

 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):
     """Verify a properly formatted Child Id Response command message.
     """
     command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
     command_msg.assertMleMessageContainsTlv(mle.LeaderData)
     command_msg.assertMleMessageContainsTlv(mle.Address16)

     check_mle_optional_tlv(command_msg, route64, mle.Route64)
     check_mle_optional_tlv(command_msg, network_data, mle.NetworkData)
     check_mle_optional_tlv(command_msg, address_registration, mle.AddressRegistration)
     check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp)
     check_mle_optional_tlv(command_msg, pending_timestamp, mle.PendingTimestamp)
     check_mle_optional_tlv(command_msg, active_operational_dataset, mle.ActiveOperationalDataset)
     check_mle_optional_tlv(command_msg, pending_operational_dataset, mle.PendingOperationalDataset)

 def check_coap_optional_tlv(coap_msg, type, tlv):
     if (type == CheckType.CONTAIN):
         coap_msg.assertCoapMessageContainsTlv(tlv)
     elif (type == CheckType.NOT_CONTAIN):
         coap_msg.assertCoapMessageDoesNotContainTlv(tlv)
     elif (type == CheckType.OPTIONAL):
         coap_msg.assertCoapMessageContainsOptionalTlv(tlv)
     else:
         raise ValueError("Invalid check type")

 def check_router_id_cached(node, router_id, cached = True):
     """Verify if the node has cached any entries based on the router ID
     """
     eidcaches = node.get_eidcaches()
     if cached:
         assert any(router_id == (int(rloc, 16) >> 10) for (_, rloc) in eidcaches)
     else:
         assert any(router_id == (int(rloc, 16) >> 10) for (_, rloc) in eidcaches) is False
	import ipv6

	import network_layer
	import config
	import mle

	from enum import IntEnum

	class CheckType(IntEnum):
	CONTAIN = 0
	NOT_CONTAIN = 1
	OPTIONAL = 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.
	"""
	command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid)

	source_rloc = source_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
	assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, \
	"Error: The IPv6 source address is not the RLOC of the originator. The source node's rloc is: " \
	+ str(ipv6.ip_address(source_rloc)) + ", but the source_address in command msg is: " \
	+ str(command_msg.ipv6_packet.ipv6_header.source_address)

	assert ipv6.ip_address(destination_address.decode('utf-8')) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The IPv6 destination address is not expected."

	def check_address_notification(command_msg, source_node, destination_node):
	"""Verify source_node sent a properly formatted Address Notification command message to destination_node.
	"""
	command_msg.assertCoapMessageRequestUriPath('/a/an')
	command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid)
	command_msg.assertCoapMessageContainsTlv(network_layer.Rloc16)
	command_msg.assertCoapMessageContainsTlv(network_layer.MlEid)

	source_rloc = source_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
	assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, "Error: The IPv6 source address is not the RLOC of the originator."

	destination_rloc = destination_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
	assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The IPv6 destination address is not the RLOC of the destination."

	def check_address_error_notification(command_msg, source_node, destination_address):
	"""Verify source_node sent a properly formatted Address Error Notification command message to destination_address.
	"""
	command_msg.assertCoapMessageRequestUriPath('/a/ae')
	command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid)
	command_msg.assertCoapMessageContainsTlv(network_layer.MlEid)

	source_rloc = source_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
	assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, \
	"Error: The IPv6 source address is not the RLOC of the originator. The source node's rloc is: " \
	+ str(ipv6.ip_address(source_rloc)) + ", but the source_address in command msg is: " \
	+ str(command_msg.ipv6_packet.ipv6_header.source_address)

	assert ipv6.ip_address(destination_address.decode('utf-8')) == command_msg.ipv6_packet.ipv6_header.destination_address, \
	"Error: The IPv6 destination address is not expected. The destination node's rloc is: " \
	+ str(ipv6.ip_address(destination_address.decode('utf-8'))) + ", but the destination_address in command msg is: " \
	+ str(command_msg.ipv6_packet.ipv6_header.destination_address)

	def check_address_release(command_msg, destination_node):
	"""Verify the message is a properly formatted address release destined to the given node.
	"""
	command_msg.assertCoapMessageRequestUriPath('/a/ar')
	command_msg.assertCoapMessageContainsTlv(network_layer.Rloc16)
	command_msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress)

	destination_rloc = destination_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
	assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The destination is not RLOC address"

	def check_tlv_request_tlv(command_msg, check_type, tlv_id):
	"""Verify if TLV Request TLV contains specified TLV ID
	"""
	tlv_request_tlv = command_msg.get_mle_message_tlv(mle.TlvRequest)

	if check_type == CheckType.CONTAIN:
	assert tlv_request_tlv is not None, "Error: The msg doesn't contain TLV Request TLV"
	assert any(tlv_id == tlv for tlv in tlv_request_tlv.tlvs), "Error: The msg doesn't contain TLV Request TLV ID: {}".format(tlv_id)

	elif check_type == CheckType.NOT_CONTAIN:
	if tlv_request_tlv is not None:
	assert any(tlv_id == tlv for tlv in tlv_request_tlv.tlvs) is False, "Error: The msg contains TLV Request TLV ID: {}".format(tlv_id)

	elif check_type == CheckType.OPTIONAL:
	if tlv_request_tlv is not None:
	if any(tlv_id == tlv for tlv in tlv_request_tlv.tlvs):
	print("TLV Request TLV contains TLV ID: {}".format(tlv_id))
	else:
	print("TLV Request TLV doesn't contain TLV ID: {}".format(tlv_id))
	else:
	print("The msg doesn't contain TLV Request TLV")

	else:
	raise ValueError("Invalid check type")

	def check_link_request(command_msg, source_address = CheckType.OPTIONAL, leader_data = CheckType.OPTIONAL, \
	tlv_request_address16 = CheckType.OPTIONAL, tlv_request_route64 = CheckType.OPTIONAL, \
	tlv_request_link_margin = CheckType.OPTIONAL):

	"""Verify a properly formatted Link Request command message.
	"""
	command_msg.assertMleMessageContainsTlv(mle.Challenge)
	command_msg.assertMleMessageContainsTlv(mle.Version)

	check_mle_optional_tlv(command_msg, source_address, mle.SourceAddress)
	check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData)

	check_tlv_request_tlv(command_msg, tlv_request_address16, mle.TlvType.ADDRESS16)
	check_tlv_request_tlv(command_msg, tlv_request_route64, mle.TlvType.ROUTE64)
	check_tlv_request_tlv(command_msg, tlv_request_link_margin, mle.TlvType.LINK_MARGIN)

	def check_link_accept(command_msg, destination_node, \
	leader_data = CheckType.OPTIONAL, link_margin = CheckType.OPTIONAL, mle_frame_counter = CheckType.OPTIONAL, \
	challenge = CheckType.OPTIONAL, address16 = CheckType.OPTIONAL, route64 = CheckType.OPTIONAL, \
	tlv_request_link_margin = CheckType.OPTIONAL):
	"""verify a properly formatted link accept command message.
	"""
	command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
	command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
	command_msg.assertMleMessageContainsTlv(mle.Response)
	command_msg.assertMleMessageContainsTlv(mle.Version)

	check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData)
	check_mle_optional_tlv(command_msg, link_margin, mle.LinkMargin)
	check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter)
	check_mle_optional_tlv(command_msg, challenge, mle.Challenge)
	check_mle_optional_tlv(command_msg, address16, mle.Address16)
	check_mle_optional_tlv(command_msg, route64, mle.Route64)

	check_tlv_request_tlv(command_msg, tlv_request_link_margin, mle.TlvType.LINK_MARGIN)

	destination_link_local = destination_node.get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL)
	assert ipv6.ip_address(destination_link_local) == command_msg.ipv6_packet.ipv6_header.destination_address, \
	"Error: The destination is unexpected"

	def check_icmp_path(sniffer, path, nodes, icmp_type = ipv6.ICMP_ECHO_REQUEST):
	"""Verify icmp message is forwarded along the path.
	"""
	len_path = len(path)

	# Verify icmp message is forwarded to the next node of the path.
	for i in range(0, len_path):
	node_msg = sniffer.get_messages_sent_by(path[i])
	node_icmp_msg = node_msg.get_icmp_message(icmp_type)

	if i < len_path - 1:
	next_node = nodes[path[i + 1]]
	next_node_rloc16 = next_node.get_addr16()
	assert next_node_rloc16 == node_icmp_msg.mac_header.dest_address.rloc, "Error: The path is unexpected."
	else:
	return True

	return False

	def check_id_set(command_msg, router_id):
	"""Check the command_msg's Route64 tlv to verify router_id is an active router.
	"""
	tlv = command_msg.assertMleMessageContainsTlv(mle.Route64)
	return ((tlv.router_id_mask >> (63 - router_id)) & 1)

	def get_routing_cost(command_msg, router_id):
	"""Check the command_msg's Route64 tlv to get the routing cost to router.
	"""
	tlv = command_msg.assertMleMessageContainsTlv(mle.Route64)

	# Get router's mask pos
	# Turn the number into binary string. Need to consider the preceding 0 omitted during conversion.
	router_id_mask_str = bin(tlv.router_id_mask).replace('0b','')
	prefix_len = 64 - len(router_id_mask_str)
	routing_entry_pos = 0

	for i in range(0, router_id - prefix_len):
	if router_id_mask_str[i] == '1':
	routing_entry_pos += 1

	assert router_id_mask_str[router_id - prefix_len] == '1', "Error: The router isn't in the topology. \n" \
	+ "route64 tlv is: %s. \nrouter_id is: %s. \nrouting_entry_pos is: %s. \nrouter_id_mask_str is: %s." \
	%(tlv, router_id, routing_entry_pos, router_id_mask_str)

	return tlv.link_quality_and_route_data[routing_entry_pos].route

	def check_mle_optional_tlv(command_msg, type, tlv):
	if (type == CheckType.CONTAIN):
	command_msg.assertMleMessageContainsTlv(tlv)
	elif (type == CheckType.NOT_CONTAIN):
	command_msg.assertMleMessageDoesNotContainTlv(tlv)
	elif (type == CheckType.OPTIONAL):
	command_msg.assertMleMessageContainsOptionalTlv(tlv)
	else:
	raise ValueError("Invalid check type")

	def check_mle_advertisement(command_msg):
	command_msg.assertSentWithHopLimit(255)
	command_msg.assertSentToDestinationAddress(config.LINK_LOCAL_ALL_NODES_ADDRESS)
	command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
	command_msg.assertMleMessageContainsTlv(mle.LeaderData)
	command_msg.assertMleMessageContainsTlv(mle.Route64)

	def check_parent_request(command_msg):
	"""Verify a properly formatted Parent Request command message.
	"""
	command_msg.assertSentWithHopLimit(255)
	command_msg.assertSentToDestinationAddress(config.LINK_LOCAL_ALL_ROUTERS_ADDRESS)
	command_msg.assertMleMessageContainsTlv(mle.Mode)
	command_msg.assertMleMessageContainsTlv(mle.Challenge)
	command_msg.assertMleMessageContainsTlv(mle.ScanMask)
	command_msg.assertMleMessageContainsTlv(mle.Version)

	def check_parent_response(command_msg, mle_frame_counter = CheckType.OPTIONAL):
	"""Verify a properly formatted Parent Response command message.
	"""
	command_msg.assertMleMessageContainsTlv(mle.Challenge)
	command_msg.assertMleMessageContainsTlv(mle.Connectivity)
	command_msg.assertMleMessageContainsTlv(mle.LeaderData)
	command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
	command_msg.assertMleMessageContainsTlv(mle.LinkMargin)
	command_msg.assertMleMessageContainsTlv(mle.Response)
	command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
	command_msg.assertMleMessageContainsTlv(mle.Version)

	check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter)

	def check_child_id_request(command_msg, tlv_request = CheckType.OPTIONAL, \
	mle_frame_counter = CheckType.OPTIONAL, address_registration = CheckType.OPTIONAL, \
	active_timestamp = CheckType.OPTIONAL, pending_timestamp = CheckType.OPTIONAL):
	"""Verify a properly formatted Child Id Request command message.
	"""
	command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
	command_msg.assertMleMessageContainsTlv(mle.Mode)
	command_msg.assertMleMessageContainsTlv(mle.Response)
	command_msg.assertMleMessageContainsTlv(mle.Timeout)
	command_msg.assertMleMessageContainsTlv(mle.Version)

	check_mle_optional_tlv(command_msg, tlv_request, mle.TlvRequest)
	check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter)
	check_mle_optional_tlv(command_msg, address_registration, mle.AddressRegistration)
	check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp)
	check_mle_optional_tlv(command_msg, pending_timestamp, mle.PendingTimestamp)

	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):
	"""Verify a properly formatted Child Id Response command message.
	"""
	command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
	command_msg.assertMleMessageContainsTlv(mle.LeaderData)
	command_msg.assertMleMessageContainsTlv(mle.Address16)

	check_mle_optional_tlv(command_msg, route64, mle.Route64)
	check_mle_optional_tlv(command_msg, network_data, mle.NetworkData)
	check_mle_optional_tlv(command_msg, address_registration, mle.AddressRegistration)
	check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp)
	check_mle_optional_tlv(command_msg, pending_timestamp, mle.PendingTimestamp)
	check_mle_optional_tlv(command_msg, active_operational_dataset, mle.ActiveOperationalDataset)
	check_mle_optional_tlv(command_msg, pending_operational_dataset, mle.PendingOperationalDataset)

	def check_coap_optional_tlv(coap_msg, type, tlv):
	if (type == CheckType.CONTAIN):
	coap_msg.assertCoapMessageContainsTlv(tlv)
	elif (type == CheckType.NOT_CONTAIN):
	coap_msg.assertCoapMessageDoesNotContainTlv(tlv)
	elif (type == CheckType.OPTIONAL):
	coap_msg.assertCoapMessageContainsOptionalTlv(tlv)
	else:
	raise ValueError("Invalid check type")

	def check_router_id_cached(node, router_id, cached = True):
	"""Verify if the node has cached any entries based on the router ID
	"""
	eidcaches = node.get_eidcaches()
	if cached:
	assert any(router_id == (int(rloc, 16) >> 10) for (_, rloc) in eidcaches)
	else:
	assert any(router_id == (int(rloc, 16) >> 10) for (_, rloc) in eidcaches) is False