Refactor Drop EtherCat packet test
This is a no-op refactoring, which is preparation for follow-up
changes to add more apf tests.
This includes:
1. Add a method to send packet and expect counter increases.
2. Inline the packet construction code in the test method.
Test: atest NetworkStaticLibHostPythonTests
Test: m connectivity_multi_devices_snippet && \
atest CtsConnectivityMultiDevicesTestCases
Bug: 350880989
Change-Id: Iadcfd97740033306af1b6a474004cbf0699ddd1b
diff --git a/staticlibs/tests/unit/host/python/apf_utils_test.py b/staticlibs/tests/unit/host/python/apf_utils_test.py
index b9a270f..b5a941b 100644
--- a/staticlibs/tests/unit/host/python/apf_utils_test.py
+++ b/staticlibs/tests/unit/host/python/apf_utils_test.py
@@ -27,7 +27,6 @@
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
@@ -112,18 +111,6 @@
with asserts.assert_raises(PatternNotFoundException):
get_hardware_address(self.mock_ad, "wlan0")
- @patch("net_tests_utils.host.python.apf_utils.get_hardware_address")
- @patch("net_tests_utils.host.python.apf_utils.send_raw_packet_downstream")
- def test_send_broadcast_empty_ethercat_packet(
- self,
- mock_send_raw_packet_downstream: MagicMock,
- mock_get_hardware_address: MagicMock,
- ) -> None:
- mock_get_hardware_address.return_value = "12:34:56:78:90:AB"
- send_broadcast_empty_ethercat_packet(self.mock_ad, "eth0")
- # Assuming you'll mock the packet construction part, verify calls to send_raw_packet_downstream.
- mock_send_raw_packet_downstream.assert_called_once()
-
@patch("net_tests_utils.host.python.adb_utils.adb_shell")
def test_send_raw_packet_downstream_success(
self, mock_adb_shell: MagicMock
@@ -162,9 +149,9 @@
cmd="", stdout="Unknown command", stderr="", ret_code=3
)
with asserts.assert_raises(UnsupportedOperationException):
- send_raw_packet_downstream(
- self.mock_ad, TEST_IFACE_NAME, TEST_PACKET_IN_HEX
- )
+ 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.",
diff --git a/staticlibs/testutils/host/python/apf_test_base.py b/staticlibs/testutils/host/python/apf_test_base.py
index 86e7a26..7203265 100644
--- a/staticlibs/testutils/host/python/apf_test_base.py
+++ b/staticlibs/testutils/host/python/apf_test_base.py
@@ -13,7 +13,7 @@
# 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 import adb_utils, apf_utils, assert_utils, multi_devices_test_base, tether_utils
from net_tests_utils.host.python.tether_utils import UpstreamType
@@ -21,6 +21,8 @@
def setup_class(self):
super().setup_class()
+
+ # Check test preconditions.
tether_utils.assume_hotspot_test_preconditions(
self.serverDevice, self.clientDevice, UpstreamType.NONE
)
@@ -31,8 +33,8 @@
"NetworkStack is too old to support send raw packet, skip test.",
)
+ # Fetch device properties and storing them locally for later use.
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
@@ -41,7 +43,11 @@
self.client_iface_name = client.getInterfaceNameFromNetworkHandle(
client_network
)
+ self.server_mac_address = apf_utils.get_hardware_address(
+ self.serverDevice, self.server_iface_name
+ )
+ # Enable doze mode to activate APF.
adb_utils.set_doze_mode(self.clientDevice, True)
def teardown_class(self):
@@ -49,3 +55,26 @@
tether_utils.cleanup_tethering_for_upstream_type(
self.serverDevice, UpstreamType.NONE
)
+
+ def send_packet_and_expect_counter_increased(
+ self, packet: str, counter_name: str
+ ) -> None:
+ count_before_test = apf_utils.get_apf_counter(
+ self.clientDevice,
+ self.client_iface_name,
+ counter_name,
+ )
+ apf_utils.send_raw_packet_downstream(
+ self.serverDevice, self.server_iface_name, packet
+ )
+
+ assert_utils.expect_with_retry(
+ lambda: apf_utils.get_apf_counter(
+ self.clientDevice,
+ self.client_iface_name,
+ counter_name,
+ )
+ > count_before_test
+ )
+
+ # TODO: Verify the packet is not actually received.
diff --git a/staticlibs/testutils/host/python/apf_utils.py b/staticlibs/testutils/host/python/apf_utils.py
index 7ebf792..a3ec6e9 100644
--- a/staticlibs/testutils/host/python/apf_utils.py
+++ b/staticlibs/testutils/host/python/apf_utils.py
@@ -20,11 +20,6 @@
from net_tests_utils.host.python import adb_utils, assert_utils
-# Constants.
-ETHER_BROADCAST = "FFFFFFFFFFFF"
-ETH_P_ETHERCAT = "88A4"
-
-
class PatternNotFoundException(Exception):
"""Raised when the given pattern cannot be found."""
@@ -121,27 +116,6 @@
)
-def send_broadcast_empty_ethercat_packet(
- ad: android_device.AndroidDevice, iface_name: str
-) -> None:
- """Transmits a broadcast empty EtherCat packet on the specified interface."""
-
- # Get the interface's MAC address.
- mac_address = get_hardware_address(ad, iface_name)
-
- # TODO: Build packet by using scapy library.
- # Ethernet header (14 bytes).
- packet = ETHER_BROADCAST # Destination MAC (broadcast)
- packet += mac_address.replace(":", "") # Source MAC
- packet += ETH_P_ETHERCAT # EtherType (EtherCAT)
-
- # EtherCAT header (2 bytes) + 44 bytes of zero padding.
- packet += "00" * 46
-
- # Send the packet using a raw socket.
- send_raw_packet_downstream(ad, iface_name, packet)
-
-
def is_send_raw_packet_downstream_supported(
ad: android_device.AndroidDevice,
) -> bool:
diff --git a/tests/cts/multidevices/apfv4_test.py b/tests/cts/multidevices/apfv4_test.py
index 7652c65..4633d37 100644
--- a/tests/cts/multidevices/apfv4_test.py
+++ b/tests/cts/multidevices/apfv4_test.py
@@ -12,32 +12,24 @@
# 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_test_base, apf_utils, assert_utils, tether_utils
-from net_tests_utils.host.python.tether_utils import UpstreamType
+from net_tests_utils.host.python import apf_test_base
+# Constants.
COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED = "DROPPED_ETHERTYPE_NOT_ALLOWED"
+ETHER_BROADCAST_ADDR = "FFFFFFFFFFFF"
+ETH_P_ETHERCAT = "88A4"
class ApfV4Test(apf_test_base.ApfTestBase):
def test_apf_drop_ethercat(self):
- 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
- )
+ # Ethernet header (14 bytes).
+ packet = ETHER_BROADCAST_ADDR # Destination MAC (broadcast)
+ packet += self.server_mac_address.replace(":", "") # Source MAC
+ packet += ETH_P_ETHERCAT # EtherType (EtherCAT)
- 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
+ # EtherCAT header (2 bytes) + 44 bytes of zero padding.
+ packet += "00" * 46
+ self.send_packet_and_expect_counter_increased(
+ packet, COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED
)
-
- # TODO: Verify the packet is not actually received.