mirror of
https://github.com/espressif/openthread.git
synced 2026-07-27 06:17:47 +00:00
[tests] enhance Cert_5_1_06_RemoveRouterId (#3379)
This commit is contained in:
@@ -30,6 +30,8 @@
|
||||
import time
|
||||
import unittest
|
||||
|
||||
import command
|
||||
from command import CheckType
|
||||
import config
|
||||
import mle
|
||||
import network_layer
|
||||
@@ -82,9 +84,6 @@ class Cert_5_1_06_RemoveRouterId(unittest.TestCase):
|
||||
self.simulator.go(5)
|
||||
self.assertEqual(self.nodes[ROUTER1].get_state(), 'router')
|
||||
|
||||
for addr in self.nodes[ROUTER1].get_addrs():
|
||||
self.assertTrue(self.nodes[LEADER].ping(addr))
|
||||
|
||||
leader_messages = self.simulator.get_messages_sent_by(LEADER)
|
||||
router1_messages = self.simulator.get_messages_sent_by(ROUTER1)
|
||||
|
||||
@@ -100,26 +99,26 @@ class Cert_5_1_06_RemoveRouterId(unittest.TestCase):
|
||||
msg = router1_messages.next_coap_message("0.02")
|
||||
msg.assertCoapMessageRequestUriPath("/a/as")
|
||||
|
||||
msg = leader_messages.next_coap_message("2.04")
|
||||
leader_messages.next_coap_message("2.04")
|
||||
|
||||
# 2 - N/A
|
||||
|
||||
# 3 - Router1
|
||||
router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST)
|
||||
leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE)
|
||||
msg = router1_messages.next_mle_message(mle.CommandType.PARENT_REQUEST)
|
||||
command.check_parent_request(msg, is_first_request=True)
|
||||
|
||||
router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST)
|
||||
msg = leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE)
|
||||
msg.assertSentToNode(self.nodes[ROUTER1])
|
||||
|
||||
msg = router1_messages.next_coap_message(code="0.02", uri_path="/a/as")
|
||||
msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress)
|
||||
msg.assertCoapMessageContainsTlv(network_layer.Status)
|
||||
msg = router1_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST)
|
||||
msg.assertSentToNode(self.nodes[LEADER])
|
||||
command.check_child_id_request(msg, tlv_request=CheckType.CONTAIN,
|
||||
mle_frame_counter=CheckType.OPTIONAL, address_registration=CheckType.NOT_CONTAIN,
|
||||
active_timestamp=CheckType.OPTIONAL, pending_timestamp=CheckType.OPTIONAL)
|
||||
|
||||
msg = leader_messages.next_coap_message("2.04")
|
||||
msg.assertCoapMessageContainsTlv(network_layer.Status)
|
||||
msg.assertCoapMessageContainsOptionalTlv(network_layer.RouterMask)
|
||||
msg = router1_messages.next_coap_message(code="0.02")
|
||||
command.check_address_solicit(msg, was_router=True)
|
||||
|
||||
status_tlv = msg.get_coap_message_tlv(network_layer.Status)
|
||||
self.assertEqual(network_layer.StatusValues.SUCCESS, status_tlv.status)
|
||||
# 4 - Router1
|
||||
for addr in self.nodes[ROUTER1].get_addrs():
|
||||
self.assertTrue(self.nodes[LEADER].ping(addr))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@@ -96,6 +96,15 @@ def check_address_error_notification(command_msg, source_node, destination_addre
|
||||
+ str(ipv6.ip_address(destination_address)) + ", but the destination_address in command msg is: " \
|
||||
+ str(command_msg.ipv6_packet.ipv6_header.destination_address)
|
||||
|
||||
def check_address_solicit(command_msg, was_router):
|
||||
command_msg.assertCoapMessageRequestUriPath('/a/as')
|
||||
command_msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress)
|
||||
command_msg.assertCoapMessageContainsTlv(network_layer.Status)
|
||||
if was_router:
|
||||
command_msg.assertCoapMessageContainsTlv(network_layer.Rloc16)
|
||||
else:
|
||||
command_msg.assertMleMessageDoesNotContainTlv(network_layer.Rloc16)
|
||||
|
||||
def check_address_release(command_msg, destination_node):
|
||||
"""Verify the message is a properly formatted address release destined to the given node.
|
||||
"""
|
||||
@@ -234,14 +243,22 @@ def check_mle_advertisement(command_msg):
|
||||
command_msg.assertMleMessageContainsTlv(mle.LeaderData)
|
||||
command_msg.assertMleMessageContainsTlv(mle.Route64)
|
||||
|
||||
def check_parent_request(command_msg):
|
||||
def check_parent_request(command_msg, is_first_request):
|
||||
"""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)
|
||||
scan_mask = command_msg.assertMleMessageContainsTlv(mle.ScanMask)
|
||||
if not scan_mask.router:
|
||||
raise ValueError("Parent request without R bit set")
|
||||
if is_first_request:
|
||||
if scan_mask.end_device:
|
||||
raise ValueError("First parent request with E bit set")
|
||||
elif not scan_mask.end_device:
|
||||
raise ValueError("Second parent request without E bit set")
|
||||
|
||||
command_msg.assertMleMessageContainsTlv(mle.Version)
|
||||
|
||||
def check_parent_response(command_msg, mle_frame_counter = CheckType.OPTIONAL):
|
||||
|
||||
Reference in New Issue
Block a user