[mlr] add Commissioner API to register Multicast Listeners (#5346)

tests/scripts/thread-cert/node.py

@@ -281,6 +281,37 @@ class Node:
 
         return results
 
+    def _expect_command_output(self, cmd: str):
+        lines = []
+        cmd_output_started = False
+
+        while True:
+            self._expect(r"[^\n]+")
+            line = self.pexpect.match.group(0).decode('utf8').strip()
+
+            if line.startswith('> '):
+                line = line[2:]
+
+            if line == '':
+                continue
+
+            if line == cmd:
+                cmd_output_started = True
+                continue
+
+            if not cmd_output_started:
+                continue
+
+            if line == 'Done':
+                break
+            elif line.startswith('Error '):
+                raise Exception(line)
+            else:
+                lines.append(line)
+
+        print(f'_expect_command_output({cmd!r}) returns {lines!r}')
+        return lines
+
     def __init_soc(self, nodeid):
         """ Initialize a System-on-a-chip node connected via UART. """
         import fdpexpect
@@ -564,6 +595,24 @@ class Node:
         self.send_command(cmd)
         self._expect('Done')
 
+    def register_multicast_listener(self, *ipaddrs: Union[ipaddress.IPv6Address, str], timeout=None):
+        assert len(ipaddrs) > 0, ipaddrs
+
+        cmd = f'mlr reg {" ".join(ipaddrs)}'
+        if timeout is not None:
+            cmd += f' {int(timeout)}'
+        self.send_command(cmd)
+        self.simulator.go(3)
+        lines = self._expect_command_output(cmd)
+        m = re.match(r'status (\d+), (\d+) failed', lines[0])
+        assert m is not None, lines
+        status = int(m.group(1))
+        failed_num = int(m.group(2))
+        assert failed_num == len(lines) - 1
+        failed_ips = list(map(ipaddress.IPv6Address, lines[1:]))
+        print(f"register_multicast_listener {ipaddrs} => status: {status}, failed ips: {failed_ips}")
+        return status, failed_ips
+
     def set_link_quality(self, addr, lqi):
         cmd = 'macfilter rss add-lqi %s %s' % (addr, lqi)
         self.send_command(cmd)

tests/scripts/thread-cert/v1_2_test_multicast_listener_registration.py

@@ -339,6 +339,104 @@ class TestMulticastListenerRegistration(thread_cert.TestCase):
         self.simulator.go(WAIT_REDUNDANCE)
         self.flush_all()
 
+    def testCommissionerRegisterMulticastListeners(self):
+        self._bootstrap()
+
+        # Use ROUTER_1_2 as the Commissioner
+        commissioiner = self.nodes[ROUTER_1_2]
+
+        self.assertRaisesRegex(Exception, "InvalidState", lambda: commissioiner.register_multicast_listener("ff04::1"))
+
+        commissioiner.commissioner_start()
+        self.simulator.go(10)
+
+        # Now the Commissioner should be able to register MAs
+        for ip in ["ff04::1", "ff04::2"]:
+            status, failed_ips = commissioiner.register_multicast_listener(ip)
+            self.assertTrue(status == 0 and not failed_ips)
+            self.__check_multicast_listener(ip, expect_mlr_timeout_range=[290, 300])
+
+        # Register existing MA with a new timeout should be able to update the timeout
+        for ip in ["ff04::1", "ff04::2"]:
+            status, failed_ips = commissioiner.register_multicast_listener(ip, timeout=1000)
+            self.assertTrue(status == 0 and not failed_ips)
+            self.__check_multicast_listener(ip, expect_mlr_timeout_range=[990, 1000])
+
+        # Register MAs with given timeouts
+        for ip, timeout in [("ff05::1", 400), ("ff05::2", 500), ("ff05::3", 600)]:
+            status, failed_ips = commissioiner.register_multicast_listener(ip, timeout=timeout)
+            self.assertTrue(status == 0 and not failed_ips)
+            self.__check_multicast_listener(ip, expect_mlr_timeout_range=[timeout - 10, timeout])
+
+        # Register multiple MAs with one call
+        ips = ["ff05::4", "ff05::5", "ff05::6"]
+        status, failed_ips = commissioiner.register_multicast_listener(*ips, timeout=700)
+        self.assertTrue(status == 0 and not failed_ips)
+        for ip in ips:
+            self.__check_multicast_listener(ip, expect_mlr_timeout_range=[690, 700])
+
+        # Register multiple MAs with one call (without timeout)
+        ips = ["ff05::7", "ff05::8", "ff05::9"]
+        status, failed_ips = commissioiner.register_multicast_listener(*ips)
+        self.assertTrue(status == 0 and not failed_ips)
+        for ip in ips:
+            self.__check_multicast_listener(ip, expect_mlr_timeout_range=[290, 300])
+
+        # Unregister MAs using timeout=0
+        for ip in ["ff05::1", "ff05::2", "ff05::3"]:
+            status, failed_ips = commissioiner.register_multicast_listener(ip, timeout=0)
+            self.assertTrue(status == 0 and not failed_ips)
+            self.__check_multicast_listener(ip, expect_not_present=True)
+
+        # Unregister multiple MAs
+        ips = ["ff05::4", "ff05::5", "ff05::6"]
+        status, failed_ips = commissioiner.register_multicast_listener(*ips, timeout=0)
+        self.assertTrue(status == 0 and not failed_ips)
+        self.__check_multicast_listener(ip, expect_not_present=True)
+
+        # Remove MAs that are not subscribed should not fail
+        ips = ["ff06::1", "ff02::1", "2001::1"]
+        status, failed_ips = commissioiner.register_multicast_listener(*ips, timeout=0)
+        self.assertTrue(status == 0 and not failed_ips)
+        self.__check_multicast_listener(ip, expect_not_present=True)
+
+        # Register invalid MAs should fail
+        for ip in ["ff02::1", "ff03::1", "2001::1"]:
+            status, failed_ips = commissioiner.register_multicast_listener(ip)
+            self.assertEqual(status, 2)
+            self.assertEqual(set(failed_ips), {ipaddress.IPv6Address(ip)})
+            self.__check_multicast_listener(ip, expect_not_present=True)
+
+        # Register valid MAs with invalid MAs should succeed partially
+        ips = ["ff05::1", "ff05::2", "ff02::1", "2001::1"]
+        status, failed_ips = commissioiner.register_multicast_listener(*ips)
+        self.assertEqual(status, 2)
+        self.assertEqual(set(failed_ips), {ipaddress.IPv6Address("ff02::1"), ipaddress.IPv6Address("2001::1")})
+        self.__check_multicast_listener("ff05::1")
+        self.__check_multicast_listener("ff05::2")
+        self.__check_multicast_listener("ff02::1", expect_not_present=True)
+        self.__check_multicast_listener("2001::1", expect_not_present=True)
+
+        # Registering persistent MAs should fail for now
+        status, failed_ips = commissioiner.register_multicast_listener("ff06::1", timeout=0xffffffff)
+        self.assertEqual(status, 3)
+        # "ff06::1" should not be included in failed IPs because all IPs failed
+        self.assertTrue(not failed_ips == 0)
+
+    def __check_multicast_listener(self, *addrs, expect_mlr_timeout_range=None, expect_not_present=False):
+        addrs = map(ipaddress.IPv6Address, addrs)
+        listeners = self.nodes[BBR_1].multicast_listener_list()
+        logging.info("__check_multicast_listener get listeners: %s", listeners)
+
+        for addr in addrs:
+            if not expect_not_present:
+                self.assertIn(addr, listeners)
+                if expect_mlr_timeout_range is not None:
+                    self.assertGreaterEqual(listeners[addr], expect_mlr_timeout_range[0])
+                    self.assertLessEqual(listeners[addr], expect_mlr_timeout_range[1])
+            else:
+                self.assertNotIn(addr, listeners)
+
     def __test_multicast_listeners_table_api(self):
         self.assertTrue(self.nodes[BBR_1].multicast_listener_list() == {})