From 3cc069adea7f001525eec0a53001b7f2525d6bfc Mon Sep 17 00:00:00 2001 From: Jimi Chen Date: Wed, 19 Mar 2025 14:40:52 +0000 Subject: [PATCH] Add MLDv2 offload for APFv6 multi-devices tests Test: m connectivity_multi_devices_snippet && atest CtsConnectivityMultiDevicesTestCases Change-Id: If07deb3405b889382e938550c7b09f90030f9d31 --- tests/cts/multidevices/apfv6_test.py | 36 +++++++++++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/tests/cts/multidevices/apfv6_test.py b/tests/cts/multidevices/apfv6_test.py index fb45f4a36f..655ce14d86 100644 --- a/tests/cts/multidevices/apfv6_test.py +++ b/tests/cts/multidevices/apfv6_test.py @@ -14,7 +14,16 @@ from mobly import asserts from scapy.layers.inet import IP, ICMP, IPOption_Router_Alert -from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply +from scapy.layers.inet6 import ( + IPv6, + IPv6ExtHdrHopByHop, + ICMPv6EchoRequest, + ICMPv6EchoReply, + ICMPv6MLQuery2, + ICMPv6MLReport2, + ICMPv6MLDMultAddrRec, + RouterAlert +) from scapy.layers.l2 import Ether from scapy.contrib.igmpv3 import IGMPv3, IGMPv3mq, IGMPv3mr, IGMPv3gr from net_tests_utils.host.python import apf_test_base, apf_utils, adb_utils, assert_utils, packet_utils @@ -161,3 +170,28 @@ class ApfV6Test(apf_test_base.ApfTestBase): self.clientDevice, f'ip addr del {addr}/32 dev {self.client_iface_name}' ) + + @apf_utils.at_least_B() + @apf_utils.apf_ram_at_least(3000) + def test_mldv2_general_query_offload(self): + ether = Ether(src=self.server_mac_address, dst='33:33:00:00:00:01') + ip = IPv6(src=self.server_ipv6_addresses[0], dst='ff02::1', hlim=1) + hopOpts = IPv6ExtHdrHopByHop(options=[RouterAlert(otype=5)]) + mld = ICMPv6MLQuery2() + mldv2_general_query = bytes(ether/ip/hopOpts/mld).hex() + + ether = Ether(src=self.client_mac_address, dst='33:33:00:00:00:16') + ip = IPv6(src=self.client_ipv6_addresses[0], dst='ff02::16', hlim=1) + + mcast_addrs = apf_utils.get_exclude_all_host_ipv6_multicast_addresses( + self.clientDevice, self.client_iface_name + ) + + mld_records = [] + for addr in mcast_addrs: + mld_records.append(ICMPv6MLDMultAddrRec(dst=addr, rtype=2)) + mld = ICMPv6MLReport2(records=mld_records) + expected_mldv2_report = bytes(ether/ip/hopOpts/mld).hex() + self.send_packet_and_expect_reply_received( + mldv2_general_query, "DROPPED_IPV6_MLD_V2_GENERAL_QUERY_REPLIED", expected_mldv2_report + )