diff --git a/staticlibs/tests/unit/host/python/apf_utils_test.py b/staticlibs/tests/unit/host/python/apf_utils_test.py index caaf959523..b9a270fc68 100644 --- a/staticlibs/tests/unit/host/python/apf_utils_test.py +++ b/staticlibs/tests/unit/host/python/apf_utils_test.py @@ -26,11 +26,15 @@ from net_tests_utils.host.python.apf_utils import ( get_apf_counter, get_apf_counters_from_dumpsys, get_hardware_address, + is_send_raw_packet_downstream_supported, send_broadcast_empty_ethercat_packet, send_raw_packet_downstream, ) from net_tests_utils.host.python.assert_utils import UnexpectedBehaviorError +TEST_IFACE_NAME = "eth0" +TEST_PACKET_IN_HEX = "AABBCCDDEEFF" + class TestApfUtils(base_test.BaseTestClass, parameterized.TestCase): @@ -125,13 +129,13 @@ IpClient.wlan1 self, mock_adb_shell: MagicMock ) -> None: mock_adb_shell.return_value = "" # Successful command output - iface_name = "eth0" - packet_in_hex = "AABBCCDDEEFF" - send_raw_packet_downstream(self.mock_ad, iface_name, packet_in_hex) + send_raw_packet_downstream( + self.mock_ad, TEST_IFACE_NAME, TEST_PACKET_IN_HEX + ) mock_adb_shell.assert_called_once_with( self.mock_ad, "cmd network_stack send-raw-packet-downstream" - f" {iface_name} {packet_in_hex}", + f" {TEST_IFACE_NAME} {TEST_PACKET_IN_HEX}", ) @patch("net_tests_utils.host.python.adb_utils.adb_shell") @@ -142,7 +146,13 @@ IpClient.wlan1 "Any Unexpected Output" ) with asserts.assert_raises(UnexpectedBehaviorError): - send_raw_packet_downstream(self.mock_ad, "eth0", "AABBCCDDEEFF") + send_raw_packet_downstream( + self.mock_ad, TEST_IFACE_NAME, TEST_PACKET_IN_HEX + ) + asserts.assert_true( + is_send_raw_packet_downstream_supported(self.mock_ad), + "Send raw packet should be supported.", + ) @patch("net_tests_utils.host.python.adb_utils.adb_shell") def test_send_raw_packet_downstream_unsupported( @@ -152,7 +162,13 @@ IpClient.wlan1 cmd="", stdout="Unknown command", stderr="", ret_code=3 ) with asserts.assert_raises(UnsupportedOperationException): - send_raw_packet_downstream(self.mock_ad, "eth0", "AABBCCDDEEFF") + send_raw_packet_downstream( + self.mock_ad, TEST_IFACE_NAME, TEST_PACKET_IN_HEX + ) + asserts.assert_false( + is_send_raw_packet_downstream_supported(self.mock_ad), + "Send raw packet should not be supported.", + ) @parameterized.parameters( ("2,2048,1", ApfCapabilities(2, 2048, 1)), # Valid input diff --git a/staticlibs/testutils/host/python/apf_test_base.py b/staticlibs/testutils/host/python/apf_test_base.py new file mode 100644 index 0000000000..86e7a26ba6 --- /dev/null +++ b/staticlibs/testutils/host/python/apf_test_base.py @@ -0,0 +1,51 @@ +# Copyright (C) 2024 The Android Open Source Project +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from mobly import asserts +from net_tests_utils.host.python import adb_utils, apf_utils, multi_devices_test_base, tether_utils +from net_tests_utils.host.python.tether_utils import UpstreamType + + +class ApfTestBase(multi_devices_test_base.MultiDevicesTestBase): + + def setup_class(self): + super().setup_class() + tether_utils.assume_hotspot_test_preconditions( + self.serverDevice, self.clientDevice, UpstreamType.NONE + ) + asserts.abort_class_if( + not apf_utils.is_send_raw_packet_downstream_supported( + self.serverDevice + ), + "NetworkStack is too old to support send raw packet, skip test.", + ) + + client = self.clientDevice.connectivity_multi_devices_snippet + + self.server_iface_name, client_network = ( + tether_utils.setup_hotspot_and_client_for_upstream_type( + self.serverDevice, self.clientDevice, UpstreamType.NONE + ) + ) + self.client_iface_name = client.getInterfaceNameFromNetworkHandle( + client_network + ) + + adb_utils.set_doze_mode(self.clientDevice, True) + + def teardown_class(self): + adb_utils.set_doze_mode(self.clientDevice, False) + tether_utils.cleanup_tethering_for_upstream_type( + self.serverDevice, UpstreamType.NONE + ) diff --git a/staticlibs/testutils/host/python/apf_utils.py b/staticlibs/testutils/host/python/apf_utils.py index 415799c2f6..7ebf792b5c 100644 --- a/staticlibs/testutils/host/python/apf_utils.py +++ b/staticlibs/testutils/host/python/apf_utils.py @@ -142,6 +142,19 @@ def send_broadcast_empty_ethercat_packet( send_raw_packet_downstream(ad, iface_name, packet) +def is_send_raw_packet_downstream_supported( + ad: android_device.AndroidDevice, +) -> bool: + try: + # Invoke the shell command with empty argument and see how NetworkStack respond. + # If supported, an IllegalArgumentException with help page will be printed. + send_raw_packet_downstream(ad, "", "") + except assert_utils.UnexpectedBehaviorError: + return True + except UnsupportedOperationException: + return False + + def send_raw_packet_downstream( ad: android_device.AndroidDevice, iface_name: str, diff --git a/tests/cts/multidevices/apfv4_test.py b/tests/cts/multidevices/apfv4_test.py index 5844e493b5..7652c651ed 100644 --- a/tests/cts/multidevices/apfv4_test.py +++ b/tests/cts/multidevices/apfv4_test.py @@ -12,57 +12,32 @@ # See the License for the specific language governing permissions and # limitations under the License. -from net_tests_utils.host.python import adb_utils, apf_utils, assert_utils, mdns_utils, multi_devices_test_base, tether_utils +from mobly import asserts +from net_tests_utils.host.python import adb_utils, apf_test_base, apf_utils, assert_utils, tether_utils from net_tests_utils.host.python.tether_utils import UpstreamType COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED = "DROPPED_ETHERTYPE_NOT_ALLOWED" -class ApfV4Test(multi_devices_test_base.MultiDevicesTestBase): +class ApfV4Test(apf_test_base.ApfTestBase): def test_apf_drop_ethercat(self): - tether_utils.assume_hotspot_test_preconditions( - self.serverDevice, self.clientDevice, UpstreamType.NONE + count_before_test = apf_utils.get_apf_counter( + self.clientDevice, + self.client_iface_name, + COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED, + ) + apf_utils.send_broadcast_empty_ethercat_packet( + self.serverDevice, self.server_iface_name ) - client = self.clientDevice.connectivity_multi_devices_snippet - try: - server_iface_name, client_network = ( - tether_utils.setup_hotspot_and_client_for_upstream_type( - self.serverDevice, self.clientDevice, UpstreamType.NONE - ) - ) - client_iface_name = client.getInterfaceNameFromNetworkHandle( - client_network - ) - adb_utils.set_doze_mode(self.clientDevice, True) - - count_before_test = apf_utils.get_apf_counter( - self.clientDevice, - client_iface_name, - COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED, - ) - try: - apf_utils.send_broadcast_empty_ethercat_packet( - self.serverDevice, server_iface_name - ) - except apf_utils.UnsupportedOperationException: - asserts.skip( - "NetworkStack is too old to support send raw packet, skip test." + assert_utils.expect_with_retry( + lambda: apf_utils.get_apf_counter( + self.clientDevice, + self.client_iface_name, + COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED, ) + > count_before_test + ) - assert_utils.expect_with_retry( - lambda: apf_utils.get_apf_counter( - self.clientDevice, - client_iface_name, - COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED, - ) - > count_before_test - ) - - # TODO: Verify the packet is not actually received. - finally: - adb_utils.set_doze_mode(self.clientDevice, False) - tether_utils.cleanup_tethering_for_upstream_type( - self.serverDevice, UpstreamType.NONE - ) + # TODO: Verify the packet is not actually received.