Sync changes from internal dev branch.

betocq/ap_utils.py

@@ -18,14 +18,14 @@
 from mobly.controllers.wifi import openwrt_device
 from mobly.controllers.wifi.lib import wifi_configs
 
-from betocq import nc_constants
+from betocq import constants
 
 
 def start_wifi(
     ap: openwrt_device.OpenWrtDevice,
     wifi_channel: int,
     country_code: str,
-    test_parameters: nc_constants.TestParameters,
+    test_parameters: constants.TestParameters,
 ):
   """Starts a WiFi network and sets SSID and password to the test parameters.
 
@@ -49,13 +49,13 @@ def start_wifi(
       'Started WiFi network with ssid "%s" password "%s"', ssid, password
   )
 
-  if wifi_channel == nc_constants.PROGRAMMABLE_AP_CHANNEL_2G:
+  if wifi_channel == constants.PROGRAMMABLE_AP_CHANNEL_2G:
     test_parameters.wifi_2g_ssid = ssid
     test_parameters.wifi_2g_password = password
-  elif wifi_channel == nc_constants.PROGRAMMABLE_AP_CHANNEL_5G:
+  elif wifi_channel == constants.PROGRAMMABLE_AP_CHANNEL_5G:
     test_parameters.wifi_5g_ssid = ssid
     test_parameters.wifi_5g_password = password
-  elif wifi_channel == nc_constants.PROGRAMMABLE_AP_CHANNEL_5G_DFS:
+  elif wifi_channel == constants.PROGRAMMABLE_AP_CHANNEL_5G_DFS:
     test_parameters.wifi_dfs_5g_ssid = ssid
     test_parameters.wifi_dfs_5g_password = password
   else:

betocq/app/res/xml/file_paths_up.xml

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="utf-8"?>
+<paths>
+  <!--
+    app-specific external storage directories of files
+    This is scoped storage path, can't access by adb shell on Android 11 release key devices
+    /sdcard/Android/data/com.google.android.nearby.mobly.snippet.betocore/files/
+  -->
+  <external-files-path
+      name="external_files_up"
+      path="."/>
+  <!--
+    Nearby test files: the folder to save test files for NS tests.
+    Please adb push the test files to
+    /sdcard/Download/up_test_files/
+  -->
+  <external-path
+      name="nearby_test_files_up"
+      path="Download/up_test_files/"/>
+</paths>

betocq/app/src/main/com/google/android/nearby/mobly/snippet/connection/MediumSettingsFactory.java

@@ -46,7 +46,7 @@ public final class MediumSettingsFactory {
   private static final int MEDIUM_UPGRADE_TO_WIFIDIRECT = 7;
   private static final int MEDIUM_BLE_L2CAP_ONLY = 8;
   private static final int MEDIUM_UPGRADE_TO_ALL_WIFI = 9;
-  // LINT.ThenChange(//depot/google3/wireless/android/platform/testing/bettertogether/betocq/nc_constants.py)
+  // LINT.ThenChange(//depot/google3/wireless/android/platform/testing/bettertogether/betocq/constants.py)
 
   private static final int MEDIUM_UPGRADE_TYPE_DISRUPTIVE = 1;
   private static final int MEDIUM_UPGRADE_TYPE_NON_DISRUPTIVE = 2;

betocq/app/src/main/com/google/android/nearby/mobly/snippet/utils/WifiManagerSnippet.java

@@ -47,6 +47,7 @@
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
 import java.lang.reflect.Field;
+import java.net.InetAddress;
 import java.security.cert.CertificateException;
 import java.security.cert.CertificateFactory;
 import java.security.cert.X509Certificate;
@@ -997,6 +998,23 @@ public boolean isWifiStandardSupported(int standard) {
     return wifiManager.isWifiStandardSupported(standard);
   }
 
+  /**
+   * Connects/Enables TDLS connection using remote IP address.
+   *
+   * @param remoteIpAddress Peer IP address
+   * @param enable {@code true} to setup a connection to the given peer IP address, {@code false} to
+   *     tear down TDLS
+   */
+  @Rpc(description = "set wifi TDLS enabled or disabled with remote ip address).")
+  public void wifiSetTdlsEnable(String remoteIpAddress, boolean enable) {
+    try {
+      InetAddress ipAddress = InetAddress.getByName(remoteIpAddress);
+      wifiManager.setTdlsEnabled(ipAddress, enable);
+    } catch (Exception e) {
+      Log.e("Failed to set wifi TDLS enabled or disabled with remote ip address: " + e);
+    }
+  }
+
   /**
    * Connects/Enables TDLS connection using peer MAC address. True to enable, False to disable
    *
@@ -1005,7 +1023,7 @@ public boolean isWifiStandardSupported(int standard) {
    *     false} to tear down TDLS
    */
   @Rpc(description = "Set TDLS connection enabled/disabled with peer MAC address.")
-  public void setTdlsEnabledWithMacAddress(String remoteMacAddress, boolean enable) {
+  public void wifiSetTdlsEnabledWithMacAddress(String remoteMacAddress, boolean enable) {
     wifiManager.setTdlsEnabledWithMacAddress(remoteMacAddress, enable);
   }
 

betocq/base_test.py

@@ -16,22 +16,23 @@
 
 import logging
 import sys
-
 import os
 import traceback
 
 from mobly import base_test
 from mobly import records
+from mobly import signals
 from mobly import utils
 from mobly.controllers import android_device
+from mobly.controllers.android_device_lib import adb
 from mobly.controllers.android_device_lib import errors
 from mobly.controllers.wifi import local_sniffer_device
 from mobly.controllers.wifi import openwrt_device
 from mobly.controllers.wifi.lib import wifi_configs
 import yaml
 
 from betocq import ap_utils
-from betocq import nc_constants
+from betocq import constants
 from betocq import setup_utils
 
 # TODO: Need to design external path for OEM.
@@ -69,8 +70,8 @@ def __init__(self, configs):
     self.ads: list[android_device.AndroidDevice] = []
     self.advertiser: android_device.AndroidDevice = None
     self.discoverer: android_device.AndroidDevice = None
-    self.test_parameters: nc_constants.TestParameters = (
-        nc_constants.TestParameters.from_user_params(self.user_params)
+    self.test_parameters: constants.TestParameters = (
+        constants.TestParameters.from_user_params(self.user_params)
     )
     logging.info('all test parameters: %s', self.test_parameters)
     self.num_bug_reports: int = 0
@@ -80,38 +81,26 @@ def __init__(self, configs):
 
   def setup_class(self) -> None:
     if sys.version_info < (3, 12):
-      setup_utils.abort_all_and_report_error_on_setup(
-          self, 'Python 3.12 or higher is required for this test.'
+      setup_utils.report_error_on_setup_class(
+          self,
+          'Python 3.12 or higher is required for this test.',
+          error_class=constants.PythonVersionError,
       )
-    if (
-        not self.test_parameters.use_programmable_ap
-        and self.test_parameters.abort_all_if_any_ap_not_ready
-    ):
-      error_messages = ''
-      if not self.test_parameters.wifi_2g_ssid:
-        error_messages += '2G AP is not ready for this test.\n'
-        logging.warning('2G AP is not ready for this test.')
-      if not self.test_parameters.wifi_5g_ssid:
-        error_messages += '5G AP is not ready for this test.\n'
-        logging.warning('5G AP is not ready for this test.')
-      if not self.test_parameters.wifi_dfs_5g_ssid:
-        error_messages += '5G DFS AP is not ready for this test.\n'
-        logging.warning('5G DFS AP is not ready for this test.')
-      if error_messages:
-        setup_utils.abort_all_and_report_error_on_setup(self, error_messages)
     try:
       self.ads = self.register_controller(android_device, min_number=2)
-    except errors.Error as e:
-      setup_utils.abort_all_and_report_error_on_setup(
+    except (errors.Error, adb.Error, signals.ControllerError) as e:
+      setup_utils.report_error_on_setup_class(
           self,
           'Failed to get Android devices with error: %s,'
           f' {traceback.format_exception(e)}',
+          abort_all=not self.test_parameters.run_all_tests_in_suite,
+          error_class=constants.DeviceRegistrationError,
       )
     for ad in self.ads:
       if hasattr(ad, 'dimensions') and 'role' in ad.dimensions:
         ad.role = ad.dimensions['role']
       if self.is_using_gms_api:
-        ad.gms_info = nc_constants.GmsInfo()
+        ad.gms_info = constants.GmsInfo()
     try:
       self.discoverer = android_device.get_device(
           self.ads, role='source_device'
@@ -161,8 +150,10 @@ def _assert_general_nc_test_conditions(self):
     if not self.test_parameters.allow_unrooted_device:
       if not self.advertiser.is_adb_root or not self.discoverer.is_adb_root:
         logging.warning('The test is aborted because the device is unrooted.')
-        setup_utils.abort_all_and_report_error_on_setup(
-            self, 'The test only can run on rooted device.'
+        setup_utils.report_error_on_setup_class(
+            self,
+            'The test only can run on rooted device.',
+            error_class=constants.UnrootedDeviceError,
         )
     else:
       logging.warning(
@@ -172,51 +163,17 @@ def _assert_general_nc_test_conditions(self):
     if self.test_parameters.is_wifi_chipset_model_mandatory:
       if not self.test_parameters.wifi_chipset_model:
         if not self.advertiser.wifi_chipset or not self.discoverer.wifi_chipset:
-          setup_utils.abort_all_and_report_error_on_setup(
-              self, 'wifi_chipset is empty in the config file'
+          setup_utils.report_error_on_setup_class(
+              self,
+              'wifi_chipset is empty in the config file',
+              error_class=constants.MissingConfigError,
           )
 
   def _assert_test_conditions(self) -> None:
     """Asserts the test conditions for all devices."""
 
-  def _get_snippet_apk_path(self, snippet_name: str) -> str | None:
-    """Gets the APK path for the given snippet name from user params.
-
-    Args:
-      snippet_name: The snippet name used to find the snippet
-      APK in user_params (e.g., 'nearby_snippet').
-
-    Returns:
-      The path to the snippet APK, or None if not provided.
-    """
-    file_tag = 'files' if 'files' in self.user_params else 'mh_files'
-    apk_paths = self.user_params.get(file_tag, {}).get(snippet_name, [''])
-    if not apk_paths or not apk_paths[0]:
-      # allow the apk_path to be empty as github release does not install
-      # the apk in the script.
-      return None
-    return apk_paths[0]
-
-  @property
-  def nearby_snippet_config(self) -> nc_constants.SnippetConfig:
-    """Snippet config for loading the first nearby snippet instance."""
-    return nc_constants.SnippetConfig(
-        snippet_name='nearby',
-        package_name=nc_constants.NEARBY_SNIPPET_PACKAGE_NAME,
-        apk_path=self._get_snippet_apk_path('nearby_snippet'),
-    )
-
-  @property
-  def nearby2_snippet_config(self) -> nc_constants.SnippetConfig:
-    """Snippet config for loading the second nearby snippet instance."""
-    return nc_constants.SnippetConfig(
-        snippet_name='nearby2',
-        package_name=nc_constants.NEARBY_SNIPPET_2_PACKAGE_NAME,
-        apk_path=self._get_snippet_apk_path('nearby_snippet_2'),
-    )
-
   def setup_wifi_env(
-      self, d2d_type: nc_constants.WifiD2DType, country_code: str
+      self, d2d_type: constants.WifiD2DType, country_code: str
   ):
     """Sets up the WiFi environment with given d2d type and country code.
 
@@ -229,7 +186,7 @@ def setup_wifi_env(
       d2d_type: The Wi-Fi D2D type.
       country_code: The country code of the test.
     """
-    if d2d_type == nc_constants.WifiD2DType.MCC_5G_AND_5G_DFS_STA and (
+    if d2d_type == constants.WifiD2DType.MCC_5G_AND_5G_DFS_STA and (
         self.test_parameters.use_programmable_ap
         or self.test_parameters.use_sniffer
     ):
@@ -240,7 +197,7 @@ def setup_wifi_env(
       )
       return
 
-    wifi_channel = nc_constants.get_wifi_channel_for_programmable_ap(d2d_type)
+    wifi_channel = constants.get_wifi_channel_for_programmable_ap(d2d_type)
     if self.test_parameters.use_programmable_ap:
       self._setup_programmable_ap(wifi_channel, country_code)
 
@@ -352,7 +309,7 @@ def on_fail(self, record: records.TestResultRecord) -> None:
         )
     ):
       error_message = (
-          f'Abort all test due to the following error happened during the'
+          f'The following error happened during the'
           ' test:\n'
           f'{record.stacktrace}\n'
           'it could be one of the following issues:\n'
@@ -361,12 +318,11 @@ def on_fail(self, record: records.TestResultRecord) -> None:
           ' killed from the logcat, disable the GMS auto update from the play'
           ' store (Settings -> Network perferences) and retry the test;\n'
           '3. The test snippet might be killed by a security app or service'
-          ' from the device, especially if this happens very frequently, check'
-          ' the logcat to verify if '
-          f' {nc_constants.NEARBY_SNIPPET_PACKAGE_NAME} or'
-          f' {nc_constants.NEARBY_SNIPPET_2_PACKAGE_NAME} or'
-          f' {nc_constants.DCT_SNIPPET_PACKAGE_NAME} or'
-          f' {nc_constants.DCT_SNIPPET_2_PACKAGE_NAME} was killed; you should'
+          ' from the device, especially if this happens very frequently,'
+          ' check the logcat to verify if '
+          f' {constants.NEARBY_SNIPPET_PACKAGE_NAME} or'
+          f' {constants.NEARBY_SNIPPET_2_PACKAGE_NAME} or'
+          ' was killed; you should'
           ' put them to the allowlist of the security app.\n'
           '4. The USB cable or port is not stable, change the USB cable or the'
           ' connection portal and try again;\n'
@@ -375,7 +331,12 @@ def on_fail(self, record: records.TestResultRecord) -> None:
       )
       logging.error(error_message)
       # show the error in setup_class clearly.
-      setup_utils.abort_all_and_report_error_on_setup(self, error_message)
+      setup_utils.report_error_on_setup_class(
+          self,
+          error_message,
+          abort_all=not self.test_parameters.run_all_tests_in_suite,
+          error_class=constants.SnippetDisconnectionError,
+      )
 
     # Reset the Nearby Connection state to ensure the testbed is in a good
     # state for the next test.
@@ -390,7 +351,7 @@ def on_fail(self, record: records.TestResultRecord) -> None:
       logging.info('skip bug report for failure')
     else:
       self.num_bug_reports = self.num_bug_reports + 1
-      if self.num_bug_reports <= nc_constants.MAX_NUM_BUG_REPORT:
+      if self.num_bug_reports <= constants.MAX_NUM_BUG_REPORT:
         logging.info('take bug report for failure')
         android_device.take_bug_reports(
             self.ads,
@@ -423,7 +384,7 @@ def _set_run_identifier(self) -> None:
     if BaseTestClass._run_identifier_is_set:
       return
     suite_name_items = [
-        nc_constants.BETOCQ_NAME,
+        constants.BETOCQ_NAME,
     ]
     if 'suite_name' in self.user_params:
       suite_name_items.append(self.user_params['suite_name'])