[thread-cert] add test to verify MED informs previous parent on reattach (#7005)

This commit verifies that MEDs can inform previous parent on reattach
with OPENTHREAD_CONFIG_MLE_INFORM_PREVIOUS_PARENT_ON_REATTACH enabled.

tests/scripts/thread-cert/Makefile.am

@@ -163,6 +163,7 @@ EXTRA_DIST                                                         = \
     test_dns_client_config_auto_start.py                             \
     test_dnssd.py                                                    \
     test_history_tracker.py                                          \
+    test_inform_previous_parent_on_reattach.py                       \
     test_ipv6.py                                                     \
     test_ipv6_fragmentation.py                                       \
     test_ipv6_source_selection.py                                    \
@@ -232,6 +233,7 @@ check_SCRIPTS                                                      = \
     test_dns_client_config_auto_start.py                             \
     test_dnssd.py                                                    \
     test_history_tracker.py                                          \
+    test_inform_previous_parent_on_reattach.py                       \
     test_ipv6.py                                                     \
     test_ipv6_fragmentation.py                                       \
     test_ipv6_source_selection.py                                    \

tests/scripts/thread-cert/node.py

@@ -39,7 +39,7 @@ import time
 import traceback
 import unittest
 from ipaddress import IPv6Address, IPv6Network
-from typing import Union, Dict, Optional, List
+from typing import Union, Dict, Optional, List, Any
 
 import pexpect
 import pexpect.popen_spawn
@@ -1579,6 +1579,63 @@ class NodeImpl:
         self.send_command(cmd)
         self._expect_done()
 
+    def get_child_table(self) -> Dict[int, Dict[str, Any]]:
+        """Get the table of attached children."""
+        cmd = 'child table'
+        self.send_command(cmd)
+        output = self._expect_command_output(cmd)
+
+        #
+        # Example output:
+        # | ID  | RLOC16 | Timeout    | Age        | LQ In | C_VN |R|D|N|Ver|CSL|QMsgCnt| Extended MAC     |
+        # +-----+--------+------------+------------+-------+------+-+-+-+---+---+-------+------------------+
+        # |   1 | 0xc801 |        240 |         24 |     3 |  131 |1|0|0|  3| 0 |     0 | 4ecede68435358ac |
+        # |   2 | 0xc802 |        240 |          2 |     3 |  131 |0|0|0|  3| 1 |     0 | a672a601d2ce37d8 |
+        # Done
+        #
+
+        headers = self.__split_table_row(output[0])
+
+        table = {}
+        for line in output[2:]:
+            line = line.strip()
+            if not line:
+                continue
+
+            fields = self.__split_table_row(line)
+            col = lambda colname: self.__get_table_col(colname, headers, fields)
+
+            id = int(col("ID"))
+            r, d, n = int(col("R")), int(col("D")), int(col("N"))
+            mode = f'{"r" if r else ""}{"d" if d else ""}{"n" if n else ""}'
+
+            table[int(id)] = {
+                'id': int(id),
+                'rloc16': int(col('RLOC16'), 16),
+                'timeout': int(col('Timeout')),
+                'age': int(col('Age')),
+                'lq_in': int(col('LQ In')),
+                'c_vn': int(col('C_VN')),
+                'mode': mode,
+                'extaddr': col('Extended MAC'),
+                'ver': int(col('Ver')),
+                'csl': bool(int(col('CSL'))),
+                'qmsgcnt': int(col('QMsgCnt')),
+            }
+
+        return table
+
+    def __split_table_row(self, row: str) -> List[str]:
+        if not (row.startswith('|') and row.endswith('|')):
+            raise ValueError(row)
+
+        fields = row.split('|')
+        fields = [x.strip() for x in fields[1:-1]]
+        return fields
+
+    def __get_table_col(self, colname: str, headers: List[str], fields: List[str]) -> str:
+        return fields[headers.index(colname)]
+
     def __getOmrAddress(self):
         prefixes = [prefix.split('::')[0] for prefix in self.get_prefixes()]
         omr_addrs = []

tests/scripts/thread-cert/test_inform_previous_parent_on_reattach.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python3
+#
+#  Copyright (c) 2021, 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 os
+import unittest
+
+import config
+import thread_cert
+
+from pktverify.packet_filter import PacketFilter
+
+# Test description:
+# The purpose of this test is to verify a MED will inform its previous parent when re-attaches to another parent.
+#
+# Initial Topology:
+#
+#  LEADER ----- ROUTER
+#    |
+#   MED
+#
+
+LEADER = 1
+ROUTER = 2
+MED = 3
+
+LONG_CHILD_TIMEOUT = 120
+
+
+class TestReset(thread_cert.TestCase):
+    SUPPORT_NCP = False
+    USE_MESSAGE_FACTORY = False
+
+    TOPOLOGY = {
+        LEADER: {
+            'name': 'LEADER',
+            'mode': 'rdn',
+            'allowlist': [ROUTER, MED]
+        },
+        ROUTER: {
+            'name': 'ROUTER',
+            'mode': 'rdn',
+            'allowlist': [LEADER]
+        },
+        MED: {
+            'name': 'MED',
+            'is_mtd': True,
+            'mode': 'rn',
+            'allowlist': [LEADER],
+            'timeout': LONG_CHILD_TIMEOUT,
+        },
+    }
+
+    def test(self):
+        if 'posix' in os.getenv('OT_CLI_PATH', ''):
+            self.skipTest("skip for posix tests")
+
+        self.nodes[LEADER].start()
+        self.simulator.go(5)
+        self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
+
+        self.nodes[ROUTER].start()
+        self.simulator.go(7)
+        self.assertEqual(self.nodes[ROUTER].get_state(), 'router')
+
+        self.nodes[MED].start()
+        self.simulator.go(7)
+        self.assertEqual(self.nodes[MED].get_state(), 'child')
+
+        self.assertIsChildOf(MED, LEADER)
+
+        self.nodes[LEADER].remove_allowlist(self.nodes[MED].get_addr64())
+        self.nodes[MED].remove_allowlist(self.nodes[LEADER].get_addr64())
+
+        self.nodes[ROUTER].add_allowlist(self.nodes[MED].get_addr64())
+        self.nodes[MED].add_allowlist(self.nodes[ROUTER].get_addr64())
+
+        self.nodes[MED].set_timeout(config.DEFAULT_CHILD_TIMEOUT)
+
+        self.simulator.go(config.DEFAULT_CHILD_TIMEOUT * 2)
+        self.assertIsChildOf(MED, ROUTER)
+
+        # Verify MED is not in the LEADER's Child Table.
+        med_extaddr = self.nodes[MED].get_addr64()
+        self.assertFalse(any(info['extaddr'] == med_extaddr for info in self.nodes[LEADER].get_child_table().values()))
+
+        self.collect_ipaddrs()
+        self.collect_rlocs()
+
+    def verify(self, pv):
+        pkts: PacketFilter = pv.pkts
+        pv.summary.show()
+
+        MED = pv.vars['MED']
+        LEADER_RLOC = pv.vars['LEADER_RLOC']
+
+        # MED should attach to LEADER first.
+        pv.verify_attached('MED', 'LEADER', child_type='MTD')
+        # MED should re-attach to ROUTER.
+        pv.verify_attached('MED', 'ROUTER', child_type='MTD')
+        # MED should send empty IPv6 message to inform previous parent (LEADER).
+        pkts.filter_wpan_src64(MED).filter('lowpan.dst == {LEADER_RLOC} and lowpan.next == 0x3b',
+                                           LEADER_RLOC=LEADER_RLOC).must_next()
+
+    def assertIsChildOf(self, childid, parentid):
+        childRloc16 = self.nodes[childid].get_addr16()
+        parentRloc16 = self.nodes[parentid].get_addr16()
+        self.assertEqual(parentRloc16 & 0xfc00, parentRloc16)
+        self.assertEqual(childRloc16 & 0xfc00, parentRloc16)
+
+        child_extaddr = self.nodes[childid].get_addr64()
+        self.assertTrue(
+            any(info['extaddr'] == child_extaddr for info in self.nodes[parentid].get_child_table().values()))
+
+
+if __name__ == '__main__':
+    unittest.main()