Merge "Baseline one more Lint FlaggedApi violation" into main
diff --git a/staticlibs/tests/unit/Android.bp b/staticlibs/tests/unit/Android.bp
index e9f8858..3186033 100644
--- a/staticlibs/tests/unit/Android.bp
+++ b/staticlibs/tests/unit/Android.bp
@@ -76,7 +76,14 @@
test_suites: [
"general-tests",
],
+ // MoblyBinaryHostTest doesn't support unit_test.
test_options: {
- unit_test: true,
+ unit_test: false,
+ },
+ // Needed for applying VirtualEnv.
+ version: {
+ py3: {
+ embedded_launcher: false,
+ },
},
}
diff --git a/staticlibs/tests/unit/host/python/adb_utils_test.py b/staticlibs/tests/unit/host/python/adb_utils_test.py
index b75d464..8fcca37 100644
--- a/staticlibs/tests/unit/host/python/adb_utils_test.py
+++ b/staticlibs/tests/unit/host/python/adb_utils_test.py
@@ -12,15 +12,21 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import unittest
from unittest.mock import MagicMock, patch
+from absl.testing import parameterized
+from mobly import asserts
+from mobly import base_test
+from mobly import config_parser
from net_tests_utils.host.python import adb_utils
from net_tests_utils.host.python.assert_utils import UnexpectedBehaviorError
-class TestAdbUtils(unittest.TestCase):
+class TestAdbUtils(base_test.BaseTestClass, parameterized.TestCase):
- def setUp(self):
+ def __init__(self, configs: config_parser.TestRunConfig):
+ super().__init__(configs)
+
+ def setup_test(self):
self.mock_ad = MagicMock() # Mock Android device object
self.mock_ad.log = MagicMock()
self.mock_ad.adb.shell.return_value = b"" # Default empty return for shell
@@ -49,25 +55,21 @@
@patch("net_tests_utils.host.python.adb_utils._get_screen_state")
def test_set_screen_state_failure(self, mock_get_screen_state):
mock_get_screen_state.return_value = False # State doesn't change
- with self.assertRaises(UnexpectedBehaviorError):
+ with asserts.assert_raises(UnexpectedBehaviorError):
adb_utils._set_screen_state(self.mock_ad, True)
+ @parameterized.parameters(
+ ("Awake", True),
+ ("Asleep", False),
+ ("Dozing", False),
+ ("SomeOtherState", False),
+ ) # Declare inputs for state_str and expected_result.
@patch("net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys")
- def test_get_screen_state(self, mock_get_value):
- # Test cases for different return values of get_value_of_key_from_dumpsys
- # TODO: Make it parameterized.
- test_cases = [
- ("Awake", True),
- ("Asleep", False),
- ("Dozing", False),
- ("SomeOtherState", False),
- ]
-
- for state_str, expected_result in test_cases:
- mock_get_value.return_value = state_str
- self.assertEqual(
- adb_utils._get_screen_state(self.mock_ad), expected_result
- )
+ def test_get_screen_state(self, state_str, expected_result, mock_get_value):
+ mock_get_value.return_value = state_str
+ asserts.assert_equal(
+ adb_utils._get_screen_state(self.mock_ad), expected_result
+ )
def test_get_value_of_key_from_dumpsys(self):
self.mock_ad.adb.shell.return_value = (
@@ -76,36 +78,34 @@
result = adb_utils.get_value_of_key_from_dumpsys(
self.mock_ad, "power", "mWakefulness"
)
- self.assertEqual(result, "Awake")
+ asserts.assert_equal(result, "Awake")
+ @parameterized.parameters(
+ (True, ["true"]),
+ (False, ["false"]),
+ (
+ True,
+ ["false", "true"],
+ ), # Expect True, get False which is unexpected, then get True
+ (
+ False,
+ ["true", "false"],
+ ), # Expect False, get True which is unexpected, then get False
+ ) # Declare inputs for expected_state and returned_value
@patch("net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys")
- def test_expect_dumpsys_state_with_retry_success(self, mock_get_value):
- # Test cases for different combinations of expected_state and get_value return value
- # TODO: Make it parameterized.
- test_cases = [
- (True, ["true"]), # Expect True, get True
- (False, ["false"]), # Expect False, get False
- (
- True,
- ["false", "true"],
- ), # Expect True, get False which is unexpected, then get True
- (
- False,
- ["true", "false"],
- ), # Expect False, get True which is unexpected, then get False
- ]
-
- for expected_state, returned_value in test_cases:
- mock_get_value.side_effect = returned_value
- # Verify the method returns and does not throw.
- adb_utils.expect_dumpsys_state_with_retry(
- self.mock_ad, "service", "key", expected_state, 0
- )
+ def test_expect_dumpsys_state_with_retry_success(
+ self, expected_state, returned_value, mock_get_value
+ ):
+ mock_get_value.side_effect = returned_value
+ # Verify the method returns and does not throw.
+ adb_utils.expect_dumpsys_state_with_retry(
+ self.mock_ad, "service", "key", expected_state, 0
+ )
@patch("net_tests_utils.host.python.adb_utils.get_value_of_key_from_dumpsys")
def test_expect_dumpsys_state_with_retry_failure(self, mock_get_value):
mock_get_value.return_value = "false"
- with self.assertRaises(UnexpectedBehaviorError):
+ with asserts.assert_raises(UnexpectedBehaviorError):
adb_utils.expect_dumpsys_state_with_retry(
self.mock_ad, "service", "key", True, 0
)
@@ -116,11 +116,7 @@
mock_get_value.return_value = None
# Expect the function to raise UnexpectedBehaviorError due to the exception
- with self.assertRaises(UnexpectedBehaviorError):
+ with asserts.assert_raises(UnexpectedBehaviorError):
adb_utils.expect_dumpsys_state_with_retry(
self.mock_ad, "service", "key", True
)
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/staticlibs/tests/unit/host/python/apf_utils_test.py b/staticlibs/tests/unit/host/python/apf_utils_test.py
index f7fb93b..50bc884 100644
--- a/staticlibs/tests/unit/host/python/apf_utils_test.py
+++ b/staticlibs/tests/unit/host/python/apf_utils_test.py
@@ -12,18 +12,29 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import unittest
from unittest.mock import MagicMock, patch
+from mobly import asserts
+from mobly import base_test
+from mobly import config_parser
+from mobly.controllers.android_device_lib.adb import AdbError
from net_tests_utils.host.python.apf_utils import (
PatternNotFoundException,
+ UnsupportedOperationException,
get_apf_counter,
get_apf_counters_from_dumpsys,
+ get_hardware_address,
+ send_broadcast_empty_ethercat_packet,
+ send_raw_packet_downstream,
)
+from net_tests_utils.host.python.assert_utils import UnexpectedBehaviorError
-class TestApfUtils(unittest.TestCase):
+class TestApfUtils(base_test.BaseTestClass):
- def setUp(self):
+ def __init__(self, configs: config_parser.TestRunConfig):
+ super().__init__(configs)
+
+ def setup_test(self):
self.mock_ad = MagicMock() # Mock Android device object
@patch("net_tests_utils.host.python.adb_utils.get_dumpsys_for_service")
@@ -36,9 +47,8 @@
COUNTER_NAME1: 123
COUNTER_NAME2: 456
"""
- iface_name = "wlan0"
- counters = get_apf_counters_from_dumpsys(self.mock_ad, iface_name)
- self.assertEqual(counters, {"COUNTER_NAME1": 123, "COUNTER_NAME2": 456})
+ counters = get_apf_counters_from_dumpsys(self.mock_ad, "wlan0")
+ asserts.assert_equal(counters, {"COUNTER_NAME1": 123, "COUNTER_NAME2": 456})
@patch("net_tests_utils.host.python.adb_utils.get_dumpsys_for_service")
def test_get_apf_counters_from_dumpsys_exceptions(
@@ -58,7 +68,7 @@
for dumpsys_output in test_cases:
mock_get_dumpsys.return_value = dumpsys_output
- with self.assertRaises(PatternNotFoundException):
+ with asserts.assert_raises(PatternNotFoundException):
get_apf_counters_from_dumpsys(self.mock_ad, "wlan0")
@patch("net_tests_utils.host.python.apf_utils.get_apf_counters_from_dumpsys")
@@ -68,10 +78,83 @@
"COUNTER_NAME1": 123,
"COUNTER_NAME2": 456,
}
- self.assertEqual(get_apf_counter(self.mock_ad, iface, "COUNTER_NAME1"), 123)
+ asserts.assert_equal(
+ get_apf_counter(self.mock_ad, iface, "COUNTER_NAME1"), 123
+ )
# Not found
- self.assertEqual(get_apf_counter(self.mock_ad, iface, "COUNTER_NAME3"), 0)
+ asserts.assert_equal(
+ get_apf_counter(self.mock_ad, iface, "COUNTER_NAME3"), 0
+ )
+ @patch("net_tests_utils.host.python.adb_utils.adb_shell")
+ def test_get_hardware_address_success(
+ self, mock_adb_shell: MagicMock
+ ) -> None:
+ mock_adb_shell.return_value = """
+46: wlan0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq ...
+ link/ether 72:05:77:82:21:e0 brd ff:ff:ff:ff:ff:ff
+"""
+ mac_address = get_hardware_address(self.mock_ad, "wlan0")
+ asserts.assert_equal(mac_address, "72:05:77:82:21:E0")
-if __name__ == "__main__":
- unittest.main()
+ @patch("net_tests_utils.host.python.adb_utils.adb_shell")
+ def test_get_hardware_address_not_found(
+ self, mock_adb_shell: MagicMock
+ ) -> None:
+ mock_adb_shell.return_value = "Some output without MAC address"
+ 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
+ ) -> None:
+ mock_adb_shell.return_value = "" # Successful command output
+ packet_type = "BEEF"
+ iface_name = "eth0"
+ dst_mac = "1234567890AB"
+ packet_in_hex = "AABBCCDDEEFF"
+ send_raw_packet_downstream(
+ self.mock_ad, packet_type, iface_name, dst_mac, packet_in_hex
+ )
+ mock_adb_shell.assert_called_once_with(
+ self.mock_ad,
+ "cmd network_stack send-raw-packet-downstream"
+ f" {packet_type} {iface_name} {dst_mac} {packet_in_hex}",
+ )
+
+ @patch("net_tests_utils.host.python.adb_utils.adb_shell")
+ def test_send_raw_packet_downstream_failure(
+ self, mock_adb_shell: MagicMock
+ ) -> None:
+ mock_adb_shell.return_value = ( # Unexpected command output
+ "Any Unexpected Output"
+ )
+ with asserts.assert_raises(UnexpectedBehaviorError):
+ send_raw_packet_downstream(
+ self.mock_ad, "BEEF", "eth0", "1234567890AB", "AABBCCDDEEFF"
+ )
+
+ @patch("net_tests_utils.host.python.adb_utils.adb_shell")
+ def test_send_raw_packet_downstream_unsupported(
+ self, mock_adb_shell: MagicMock
+ ) -> None:
+ mock_adb_shell.side_effect = AdbError(
+ cmd="", stdout="Unknown command", stderr="", ret_code=3
+ )
+ with asserts.assert_raises(UnsupportedOperationException):
+ send_raw_packet_downstream(
+ self.mock_ad, "BEEF", "eth0", "1234567890AB", "AABBCCDDEEFF"
+ )
diff --git a/staticlibs/tests/unit/host/python/assert_utils_test.py b/staticlibs/tests/unit/host/python/assert_utils_test.py
index b970d14..7a33373 100644
--- a/staticlibs/tests/unit/host/python/assert_utils_test.py
+++ b/staticlibs/tests/unit/host/python/assert_utils_test.py
@@ -12,11 +12,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import unittest
+from mobly import asserts
+from mobly import base_test
from net_tests_utils.host.python.assert_utils import UnexpectedBehaviorError, expect_with_retry
-class TestAssertUtils(unittest.TestCase):
+class TestAssertUtils(base_test.BaseTestClass):
def test_predicate_succeed(self):
"""Test when the predicate becomes True within retries."""
@@ -28,12 +29,12 @@
return call_count > 2 # True on the third call
expect_with_retry(predicate, max_retries=5, retry_interval_sec=0)
- self.assertEqual(call_count, 3) # Ensure it was called exactly 3 times
+ asserts.assert_equal(call_count, 3) # Ensure it was called exactly 3 times
def test_predicate_failed(self):
"""Test when the predicate never becomes True."""
- with self.assertRaises(UnexpectedBehaviorError):
+ with asserts.assert_raises(UnexpectedBehaviorError):
expect_with_retry(
predicate=lambda: False, max_retries=3, retry_interval_sec=0
)
@@ -52,7 +53,9 @@
max_retries=5,
retry_interval_sec=0,
)
- self.assertFalse(retry_action_called) # Assert retry_action was NOT called
+ asserts.assert_false(
+ retry_action_called, "retry_action called."
+ ) # Assert retry_action was NOT called
def test_retry_action_not_called_failed(self):
"""Test that the retry_action is not called if the max_retries is reached."""
@@ -62,14 +65,16 @@
nonlocal retry_action_called
retry_action_called = True
- with self.assertRaises(UnexpectedBehaviorError):
+ with asserts.assert_raises(UnexpectedBehaviorError):
expect_with_retry(
predicate=lambda: False,
retry_action=retry_action,
max_retries=1,
retry_interval_sec=0,
)
- self.assertFalse(retry_action_called) # Assert retry_action was NOT called
+ asserts.assert_false(
+ retry_action_called, "retry_action called."
+ ) # Assert retry_action was NOT called
def test_retry_action_called(self):
"""Test that the retry_action is executed when provided."""
@@ -79,11 +84,11 @@
nonlocal retry_action_called
retry_action_called = True
- with self.assertRaises(UnexpectedBehaviorError):
+ with asserts.assert_raises(UnexpectedBehaviorError):
expect_with_retry(
predicate=lambda: False,
retry_action=retry_action,
max_retries=2,
retry_interval_sec=0,
)
- self.assertTrue(retry_action_called)
+ asserts.assert_true(retry_action_called, "retry_action not called.")
diff --git a/staticlibs/tests/unit/host/python/run_tests.py b/staticlibs/tests/unit/host/python/run_tests.py
index c44f1a8..fa6a310 100644
--- a/staticlibs/tests/unit/host/python/run_tests.py
+++ b/staticlibs/tests/unit/host/python/run_tests.py
@@ -14,14 +14,22 @@
"""Main entrypoint for all of unittest."""
-import unittest
-
-# Import all unittest classes here, so it can be discovered by unittest module.
-# TODO: make the tests can be executed without manually import classes.
+import sys
from host.python.adb_utils_test import TestAdbUtils
from host.python.apf_utils_test import TestApfUtils
from host.python.assert_utils_test import TestAssertUtils
+from mobly import suite_runner
if __name__ == "__main__":
- unittest.main()
+ # For MoblyBinaryHostTest, this entry point will be called twice:
+ # 1. List tests.
+ # <mobly-par-file-name> -- --list_tests
+ # 2. Run tests.
+ # <mobly-par-file-name> -- --config=<yaml-path> --device_serial=<device-serial> --log_path=<log-path>
+ # Strip the "--" since suite runner doesn't recognize it.
+ sys.argv.pop(1)
+ # TODO: make the tests can be executed without manually list classes.
+ suite_runner.run_suite(
+ [TestAssertUtils, TestAdbUtils, TestApfUtils], sys.argv
+ )
diff --git a/staticlibs/tests/unit/host/python/test_config.xml b/staticlibs/tests/unit/host/python/test_config.xml
index 26ee9e2..d3b200a 100644
--- a/staticlibs/tests/unit/host/python/test_config.xml
+++ b/staticlibs/tests/unit/host/python/test_config.xml
@@ -14,9 +14,11 @@
limitations under the License.
-->
<configuration description="Config for NetworkStaticLibHostPythonTests">
- <target_preparer class="com.android.tradefed.targetprep.RootTargetPreparer" />
- <test class="com.android.tradefed.testtype.python.PythonBinaryHostTest">
- <option name="par-file-name" value="NetworkStaticLibHostPythonTests" />
- <option name="test-timeout" value="3m" />
+ <target_preparer class="com.android.tradefed.targetprep.PythonVirtualenvPreparer">
+ <option name="dep-module" value="absl-py" />
+ </target_preparer>
+ <test class="com.android.tradefed.testtype.mobly.MoblyBinaryHostTest" >
+ <option name="mobly-par-file-name" value="NetworkStaticLibHostPythonTests" />
+ <option name="mobly-test-timeout" value="3m" />
</test>
</configuration>
diff --git a/staticlibs/testutils/host/python/apf_utils.py b/staticlibs/testutils/host/python/apf_utils.py
index 331e970..eea10de 100644
--- a/staticlibs/testutils/host/python/apf_utils.py
+++ b/staticlibs/testutils/host/python/apf_utils.py
@@ -14,13 +14,23 @@
import re
from mobly.controllers import android_device
-from net_tests_utils.host.python import adb_utils
+from mobly.controllers.android_device_lib.adb import AdbError
+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."""
+class UnsupportedOperationException(Exception):
+ pass
+
+
def get_apf_counter(
ad: android_device.AndroidDevice, iface: str, counter_name: str
) -> int:
@@ -75,3 +85,115 @@
ad.log.debug("Getting apf counters: " + str(result))
return result
+
+
+def get_hardware_address(
+ ad: android_device.AndroidDevice, iface_name: str
+) -> str:
+ """Retrieves the hardware (MAC) address for a given network interface.
+
+ Returns:
+ The hex representative of the MAC address in uppercase.
+ E.g. 12:34:56:78:90:AB
+
+ Raises:
+ PatternNotFoundException: If the MAC address is not found in the command
+ output.
+ """
+
+ # Run the "ip link" command and get its output.
+ ip_link_output = adb_utils.adb_shell(ad, f"ip link show {iface_name}")
+
+ # Regular expression to extract the MAC address.
+ # Parse hardware address from ip link output like below:
+ # 46: wlan0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq ...
+ # link/ether 72:05:77:82:21:e0 brd ff:ff:ff:ff:ff:ff
+ pattern = r"link/ether (([0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2})"
+ match = re.search(pattern, ip_link_output)
+
+ if match:
+ return match.group(1).upper() # Extract the MAC address string.
+ else:
+ raise PatternNotFoundException(
+ "Cannot get hardware address for " + iface_name
+ )
+
+
+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, ETH_P_ETHERCAT, iface_name, ETHER_BROADCAST, packet
+ )
+
+
+def send_raw_packet_downstream(
+ ad: android_device.AndroidDevice,
+ packetType: str,
+ iface_name: str,
+ dst_mac: str,
+ packet_in_hex: str,
+) -> None:
+ """Sends a raw packet over the specified downstream interface.
+
+ This function constructs and sends a raw packet using the
+ `send-raw-packet-downstream`
+ command provided by NetworkStack process. It's primarily intended for testing
+ purposes.
+
+ Args:
+ ad: The AndroidDevice object representing the connected device.
+ packetType: The type of packet to send (e.g., "88A4" for EtherCAT).
+ iface_name: The name of the network interface to use (e.g., "wlan0",
+ "eth0").
+ dst_mac: The destination MAC address of the packet (e.g., "FFFFFFFFFFFF"
+ for broadcast).
+ packet_in_hex: The raw packet data starting from L2 header encoded in
+ hexadecimal string format.
+
+ Raises:
+ UnsupportedOperationException: If the NetworkStack doesn't support
+ the `send-raw-packet` command.
+ UnexpectedBehaviorException: If the command execution produces unexpected
+ output other than an empty response or "Unknown command".
+
+ Important Considerations:
+ Security: This method only works on tethering downstream interfaces due
+ to security restrictions.
+ Packet Format: The `packet_in_hex` must be a valid hexadecimal
+ representation of a packet starting from L2 header.
+ """
+
+ cmd = (
+ "cmd network_stack send-raw-packet-downstream"
+ f" {packetType} {iface_name} {dst_mac} {packet_in_hex}"
+ )
+
+ # Expect no output or Unknown command if NetworkStack is too old. Throw otherwise.
+ try:
+ output = adb_utils.adb_shell(ad, cmd)
+ except AdbError as e:
+ output = str(e.stdout)
+ if output:
+ if "Unknown command" in output:
+ raise UnsupportedOperationException(
+ "send-raw-packet-downstream command is not supported."
+ )
+ raise assert_utils.UnexpectedBehaviorError(
+ f"Got unexpected output: {output} for command: {cmd}."
+ )
diff --git a/tests/cts/net/src/android/net/cts/NsdManagerDownstreamTetheringTest.kt b/tests/cts/net/src/android/net/cts/NsdManagerDownstreamTetheringTest.kt
index 284fcae..f45f881 100644
--- a/tests/cts/net/src/android/net/cts/NsdManagerDownstreamTetheringTest.kt
+++ b/tests/cts/net/src/android/net/cts/NsdManagerDownstreamTetheringTest.kt
@@ -59,7 +59,6 @@
@After
override fun tearDown() {
super.tearDown()
- setIncludeTestInterfaces(false)
}
@Test
@@ -107,7 +106,6 @@
@Test
fun testMdnsDiscoveryWorkOnTetheringInterface() {
assumeFalse(isInterfaceForTetheringAvailable())
- setIncludeTestInterfaces(true)
var downstreamIface: TestNetworkInterface? = null
var tetheringEventCallback: MyTetheringEventCallback? = null