Initial commit for BetoCQ v2

TODO:
- hardware capabilities file
- atest targets per test class
- script to generate executable test zip
- openwrt support

Bug: 322046897
Test: atest; python3 with executable zip
Change-Id: I120c348b867f7fbf8adb286d2843500401368173

44 files changed, 6147 insertions, 0 deletions

diff --git a/tests/bettertogether/betocq/Android.bp b/tests/bettertogether/betocq/Android.bp
new file mode 100644
index 000000000..11a9afa5f
--- /dev/null
+++ b/tests/bettertogether/betocq/Android.bp
@@ -0,0 +1,228 @@
+// 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.
+
+package {
+    default_applicable_licenses: ["Android-Apache-2.0"],
+    default_team: "trendy_team_fwk_uwb",
+}
+
+python_defaults {
+    name: "betocq_lib_defaults",
+    pkg_path: "betocq",
+}
+
+python_defaults {
+    name: "betocq_test_defaults",
+    version: {
+        py3: {
+            embedded_launcher: false,
+        },
+    },
+}
+
+// Libraries
+
+python_library_host {
+    name: "betocq_lib",
+    defaults: ["betocq_lib_defaults"],
+    srcs: [
+        "android_wifi_utils.py",
+        "gms_auto_updates_util.py",
+        "nc_constants.py",
+        "nearby_connection_wrapper.py",
+        "setup_utils.py",
+        "version.py",
+    ],
+    libs: [
+        "mobly",
+    ],
+}
+
+python_library_host {
+    name: "base_betocq_suite",
+    defaults: ["betocq_lib_defaults"],
+    srcs: [
+        "base_betocq_suite.py",
+    ],
+    libs: [
+        "mobly",
+        "pyyaml",
+    ],
+}
+
+python_library_host {
+    name: "d2d_performance_test_base",
+    defaults: ["betocq_lib_defaults"],
+    srcs: [
+        "d2d_performance_test_base.py",
+    ],
+    libs: [
+        "betocq_lib",
+        "mobly",
+        "betocq_nc_base_test",
+    ],
+}
+
+python_library_host {
+    name: "betocq_nc_base_test",
+    defaults: ["betocq_lib_defaults"],
+    srcs: [
+        "nc_base_test.py",
+    ],
+    libs: [
+        "betocq_lib",
+        "mobly",
+        "pyyaml",
+    ],
+}
+
+python_library_host {
+    name: "betocq_cuj_tests",
+    defaults: ["betocq_lib_defaults"],
+    srcs: ["cuj_tests/*.py"],
+    libs: [
+        "betocq_lib",
+        "d2d_performance_test_base",
+        "mobly",
+    ],
+}
+
+python_library_host {
+    name: "betocq_directed_tests",
+    defaults: ["betocq_lib_defaults"],
+    srcs: ["directed_tests/*.py"],
+    libs: [
+        "betocq_lib",
+        "d2d_performance_test_base",
+        "mobly",
+    ],
+}
+
+python_library_host {
+    name: "betocq_function_tests",
+    defaults: ["betocq_lib_defaults"],
+    srcs: ["function_tests/*.py"],
+    libs: [
+        "betocq_lib",
+        "mobly",
+    ],
+}
+
+// TODO: Add modules for individual test classes.
+
+// Test suites
+
+python_test_host {
+    name: "batch_all_performance_test_suite",
+    main: "batch_all_performance_test_suite.py",
+    srcs: ["batch_all_performance_test_suite.py"],
+    libs: [
+        "base_betocq_suite",
+        "betocq_cuj_tests",
+        "betocq_directed_tests",
+        "betocq_function_tests",
+        "betocq_lib",
+        "mobly",
+    ],
+    data: [
+        "local_dev_testbed.yml",
+        // package the snippes for atest
+        ":nearby_snippet",
+        ":nearby_snippet_2",
+    ],
+    test_suites: [],
+    test_options: {
+        unit_test: false, // as Mobly tests require device(s)
+        // This tag is used to enable the ATest Mobly runner
+        tags: ["mobly"],
+    },
+}
+
+python_test_host {
+    name: "batch_performance_testing_with_2g_ap_test_suite",
+    main: "batch_performance_testing_with_2g_ap_test_suite.py",
+    srcs: ["batch_performance_testing_with_2g_ap_test_suite.py"],
+    defaults: ["betocq_test_defaults"],
+    libs: [
+        "base_betocq_suite",
+        "betocq_cuj_tests",
+        "betocq_directed_tests",
+        "betocq_function_tests",
+        "betocq_lib",
+        "mobly",
+    ],
+    data: [
+        "local_dev_testbed.yml",
+        // package the snippes for atest
+        ":nearby_snippet",
+        ":nearby_snippet_2",
+    ],
+    test_suites: [],
+    test_options: {
+        unit_test: false, // as Mobly tests require device(s)
+        // This tag is used to enable the ATest Mobly runner
+        tags: ["mobly"],
+    },
+}
+
+python_test_host {
+    name: "batch_performance_testing_with_5g_ap_test_suite",
+    main: "batch_performance_testing_with_5g_ap_test_suite.py",
+    srcs: ["batch_performance_testing_with_5g_ap_test_suite.py"],
+    defaults: ["betocq_test_defaults"],
+    libs: [
+        "base_betocq_suite",
+        "betocq_directed_tests",
+        "betocq_lib",
+        "mobly",
+    ],
+    data: [
+        "local_dev_testbed.yml",
+        // package the snippes for atest
+        ":nearby_snippet",
+        ":nearby_snippet_2",
+    ],
+    test_suites: [],
+    test_options: {
+        unit_test: false, // as Mobly tests require device(s)
+        // This tag is used to enable the ATest Mobly runner
+        tags: ["mobly"],
+    },
+}
+
+python_test_host {
+    name: "batch_performance_testing_with_dfs_5g_ap_test_suite",
+    main: "batch_performance_testing_with_dfs_5g_ap_test_suite.py",
+    srcs: ["batch_performance_testing_with_dfs_5g_ap_test_suite.py"],
+    defaults: ["betocq_test_defaults"],
+    libs: [
+        "base_betocq_suite",
+        "betocq_cuj_tests",
+        "betocq_directed_tests",
+        "betocq_lib",
+        "mobly",
+    ],
+    data: [
+        "local_dev_testbed.yml",
+        // package the snippes for atest
+        ":nearby_snippet",
+        ":nearby_snippet_2",
+    ],
+    test_suites: [],
+    test_options: {
+        unit_test: false, // as Mobly tests require device(s)
+        // This tag is used to enable the ATest Mobly runner
+        tags: ["mobly"],
+    },
+}
diff --git a/tests/bettertogether/betocq/CHANGELOG.md b/tests/bettertogether/betocq/CHANGELOG.md
new file mode 100644
index 000000000..632c69acc
--- /dev/null
+++ b/tests/bettertogether/betocq/CHANGELOG.md
@@ -0,0 +1,29 @@
+# BetoCQ test suite release history
+
+## 2.0
+
+### New
+* BetoCQ test suite: more accurate connectivity quality test suite with better
+coverage.
+  * CUJ tests: Connectivity performance evaluation for the specific CUJs, such
+  as Quickstart, Quickshare, etc.
+  * Directed tests: Specific performance tests for mediums used by D2D
+  connection.
+  * Function tests: Tests for the basic functions used by D2D connection.
+
+## 1.6
+
+### New
+* `nearby_share_stress_test.py` for testing Nearby Share using Wifi only.
+
+### Fixes
+* Change discovery medium to BLE only.
+* Increase 1G file transfer timeout to 400s.
+* Disable GMS auto-updates for the duration of the test.
+
+## 1.5
+
+### New
+* `esim_transfer_stress_test.py` for testing eSIM transfer using Bluetooth only.
+* `quick_start_stress_test.py` for testing the Quickstart flow using both
+   Bluetooth and Wifi.
\ No newline at end of file
diff --git a/tests/bettertogether/betocq/__init__.py b/tests/bettertogether/betocq/__init__.py
new file mode 100644
index 000000000..f915584ae
--- /dev/null
+++ b/tests/bettertogether/betocq/__init__.py
@@ -0,0 +1,13 @@
+#  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.
diff --git a/tests/bettertogether/betocq/android_wifi_utils.py b/tests/bettertogether/betocq/android_wifi_utils.py
new file mode 100644
index 000000000..1fe84be6c
--- /dev/null
+++ b/tests/bettertogether/betocq/android_wifi_utils.py
@@ -0,0 +1,203 @@
+#  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.
+
+"""Utils for Android Wi-Fi operations."""
+
+import dataclasses
+import datetime
+import enum
+import re
+import time
+
+from mobly.controllers import android_device
+from mobly.controllers.android_device_lib import adb
+
+_DELAY_AFTER_CHANGE_WIFI_STATUS = datetime.timedelta(seconds=5)
+_WAIT_FOR_CONNECTION = datetime.timedelta(seconds=30)
+
+_SAVED_WIFI_LIST_PATTERN = re.compile(
+    r'(?P<id>\d+)\s+(?P<ssid>.*)\s+(?P<security>.*)'
+)
+_SSID_PATTERN = re.compile(rb'Wifi is connected to "(?P<ssid>.*?)"')
+
+
+@dataclasses.dataclass(frozen=True)
+class SavedWifiInfo:
+  """Information about a saved Wi-Fi network."""
+
+  id: str
+  ssid: str
+  security: str
+
+
+@enum.unique
+class WiFiSecurity(enum.StrEnum):
+  """Security type of the Wi-Fi network."""
+
+  OPEN = 'open'
+  OWE = 'owe'
+  WPA2 = 'wpa2'
+  WPA3 = 'wpa3'
+  WEP = 'wep'
+
+
+class AndroidWiFiError(Exception):
+  """Error when failed to operate Wi-Fi on Android device."""
+
+  def __init__(self, ad: android_device.AndroidDevice, message: str) -> None:
+    self._ad = ad
+    self._message = message
+
+  def __str__(self) -> str:
+    return self._message if self._ad is None else f'{self._ad} {self._message}'
+
+
+def connect_to_wifi(
+    ad: android_device.AndroidDevice,
+    ssid: str,
+    passphrase: str | None = None,
+    security: WiFiSecurity | None = None,
+) -> None:
+  """Connects to a W-Fi network and adds to saved networks list."""
+  enable_wifi(ad)
+  if get_current_wifi(ad) == ssid:
+    ad.log.info(f'Wi-Fi was already connected to {repr(ssid)}')
+    return
+  cmd = ['cmd', 'wifi', 'connect-network', f'"{ssid}"']
+  if passphrase is None:
+    cmd.append(f'"{security or WiFiSecurity.OPEN}"')
+  else:
+    cmd.extend([f'"{security or WiFiSecurity.WPA2}"', f'"{passphrase}"'])
+  ad.adb.shell(cmd)
+  if not _wait_for_data_connected(ad) or get_current_wifi(ad) != ssid:
+    raise AndroidWiFiError(ad, f'Fail to connect to Wi-Fi {repr(ssid)}')
+  ad.log.info(f'Wi-Fi connected to {repr(ssid)}')
+
+
+def disable_wifi(ad: android_device.AndroidDevice) -> None:
+  """Disables Wi-Fi."""
+  if not is_wifi_enabled(ad):
+    ad.log.info('Wi-Fi was already disabled.')
+    return
+  ad.log.info('Disabling Wi-Fi...')
+  ad.adb.shell(['cmd', 'wifi', 'set-wifi-enabled', 'disabled'])
+  start_time = time.monotonic()
+  timeout = start_time + _DELAY_AFTER_CHANGE_WIFI_STATUS.total_seconds()
+  while time.monotonic() < timeout:
+    if not is_wifi_enabled(ad):
+      ad.log.info('Wi-Fi is disabled.')
+      return
+  raise AndroidWiFiError(
+      ad,
+      'Fail to disable Wi-Fi after waiting for'
+      f' {_DELAY_AFTER_CHANGE_WIFI_STATUS.total_seconds()} seconds',
+  )
+
+
+def enable_wifi(ad: android_device.AndroidDevice) -> None:
+  """Enables Wi-Fi."""
+  if is_wifi_enabled(ad):
+    ad.log.info('Wi-Fi was already enabled.')
+    return
+  ad.log.info('Enabling Wi-Fi...')
+  ad.adb.shell(['cmd', 'wifi', 'set-wifi-enabled', 'enabled'])
+  start_time = time.monotonic()
+  timeout = start_time + _DELAY_AFTER_CHANGE_WIFI_STATUS.total_seconds()
+  while time.monotonic() < timeout:
+    if is_wifi_enabled(ad):
+      ad.log.info('Wi-Fi is enabled.')
+      return
+  raise AndroidWiFiError(
+      ad,
+      'Fail to enable Wi-Fi after waiting for'
+      f' {_DELAY_AFTER_CHANGE_WIFI_STATUS.total_seconds()} seconds',
+  )
+
+
+def forget_all_wifi(ad: android_device.AndroidDevice) -> None:
+  """Forgets all Wi-Fi from saved networks list."""
+  for saved_wifi in list_saved_wifi(ad):
+    ad.adb.shell(['cmd', 'wifi', 'forget-network', saved_wifi.id])
+  saved_wifis = list_saved_wifi(ad)
+  if saved_wifis:
+    raise AndroidWiFiError(
+        ad,
+        'Fail to forget all Wi-Fi networks, remaining in the list:'
+        f' {saved_wifis}',
+    )
+
+
+def forget_wifi(ad: android_device.AndroidDevice, ssid: str) -> None:
+  """Forgets Wi-Fi from saved networks list."""
+  saved_wifis = list_saved_wifi(ad)
+  for saved_wifi in saved_wifis:
+    if saved_wifi.ssid == ssid:
+      stdout = ad.adb.shell(['cmd', 'wifi', 'forget-network', saved_wifi.id])
+      if stdout == b'Forget successful\n':
+        ad.log.info(f'Wi-Fi {repr(ssid)} was forgotten from saved networks.')
+        return
+      raise AndroidWiFiError(ad, f'Fail to forget Wi-Fi {repr(ssid)}')
+  ad.log.info(f'Nothing was deleted since Wi-Fi {repr(ssid)} was not saved')
+  return
+
+
+def get_current_wifi(ad: android_device.AndroidDevice) -> str:
+  """Returns current Wi-Fi network."""
+  if match := _SSID_PATTERN.search(ad.adb.shell(['cmd', 'wifi', 'status'])):
+    return match.group('ssid').decode()
+  return ''
+
+
+def is_wifi_enabled(ad: android_device.AndroidDevice) -> bool:
+  """Returns True if Wi-Fi is enabled, False otherwise."""
+  return ad.adb.shell(['cmd', 'wifi', 'status']).startswith(b'Wifi is enabled')
+
+
+def list_saved_wifi(ad: android_device.AndroidDevice) -> list[SavedWifiInfo]:
+  """Returns list of saved Wi-Fi networks."""
+  saved_wifis = []
+  stdout = ad.adb.shell(['cmd', 'wifi', 'list-networks']).decode()
+  if stdout == 'No networks\n':
+    return saved_wifis
+  for line in stdout.splitlines()[1:]:
+    if match := _SAVED_WIFI_LIST_PATTERN.search(line):
+      saved_wifis.append(
+          SavedWifiInfo(
+              id=match.group('id'),
+              ssid=match.group('ssid').strip(),
+              security=match.group('security'),
+          )
+      )
+  return saved_wifis
+
+
+def _is_data_connected(ad: android_device.AndroidDevice) -> bool:
+  """Returns True if data is connected, False otherwise."""
+  try:
+    return b'5 received' in ad.adb.shell(['ping', '-c', '5', '8.8.8.8'])
+  except adb.AdbError:
+    return False
+
+
+def _wait_for_data_connected(
+    ad: android_device.AndroidDevice,
+    timeout: datetime.timedelta = _WAIT_FOR_CONNECTION,
+) -> bool:
+  """Returns True if data is connected before timeout, False otherwise."""
+  start_time = time.monotonic()
+  timeout = start_time + timeout.total_seconds()
+  while time.monotonic() < timeout:
+    if _is_data_connected(ad):
+      return True
+  return False
diff --git a/tests/bettertogether/betocq/base_betocq_suite.py b/tests/bettertogether/betocq/base_betocq_suite.py
new file mode 100644
index 000000000..1bc29668f
--- /dev/null
+++ b/tests/bettertogether/betocq/base_betocq_suite.py
@@ -0,0 +1,83 @@
+#  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.
+
+"""Base class for BetoCQ test suites."""
+
+import logging
+import os
+
+from mobly import base_suite
+from mobly import records
+import yaml
+
+
+class BaseBetocqSuite(base_suite.BaseSuite):
+  """Base class for BetoCQ test suites.
+
+  Contains methods for aggregating and exporting suite data.
+  """
+
+  def __init__(self, runner, config):
+    super().__init__(runner, config)
+    self._summary_path = None
+    self._summary_writer = None
+
+  @property
+  def summary_path(self):
+    """Returns the path to the summary file.
+
+    NOTE: This path is only correctly resolved if called within teardown_suite.
+    """
+    if self._summary_path is None:
+      # pylint: disable-next=protected-access
+      self._summary_path = self._runner._test_run_metadata.summary_file_path
+    return self._summary_path
+
+  def _retrieve_user_data_from_summary(self):
+    """Retrieves all user_data entries from the currently streamed summary.
+
+    Use this method to aggregate data written by record_data in test classes.
+
+    NOTE: This method can only be called within teardown_suite.
+
+    Returns:
+      A list of dictionaries, each corresponding to a USER_DATA entry.
+    """
+    if not os.path.isfile(self.summary_path):
+      logging.error(
+          'Cannot retrieve user data for the suite. '
+          'The summary file does not exist: %s',
+          self.summary_path,
+      )
+      return []
+
+    with open(self.summary_path, 'r') as f:
+      return [
+          entry
+          for entry in yaml.safe_load_all(f)
+          if entry['Type'] == records.TestSummaryEntryType.USER_DATA.value
+      ]
+
+  def _record_suite_properties(self, properties):
+    """Record suite properties to the test summary file.
+
+    NOTE: This method can only be called within teardown_suite.
+
+    Args:
+      properties: dict, the properties to add to the summary
+    """
+    if self._summary_writer is None:
+      self._summary_writer = records.TestSummaryWriter(self.summary_path)
+    content = {'properties': properties}
+    self._summary_writer.dump(content, records.TestSummaryEntryType.USER_DATA)
diff --git a/tests/bettertogether/betocq/batch_all_performance_test_suite.py b/tests/bettertogether/betocq/batch_all_performance_test_suite.py
new file mode 100644
index 000000000..844f2034a
--- /dev/null
+++ b/tests/bettertogether/betocq/batch_all_performance_test_suite.py
@@ -0,0 +1,177 @@
+#  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.
+
+"""This test suite batches all tests to run in sequence.
+
+This requires 3 APs to be ready and configured in testbed.
+2G AP (wifi_2g_ssid): channel 6 (2437)
+5G AP (wifi_5g_ssid): channel 36 (5180)
+DFS 5G AP(wifi_dfs_5g_ssid): channel 52 (5260)
+"""
+
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly import suite_runner
+
+from betocq import base_betocq_suite
+from betocq import nc_constants
+from betocq.cuj_tests import mcc_5g_all_wifi_non_dbs_2g_sta_test
+from betocq.cuj_tests import scc_2g_all_wifi_sta_test
+from betocq.cuj_tests import scc_5g_all_wifi_dbs_2g_sta_test
+from betocq.cuj_tests import scc_5g_all_wifi_sta_test
+from betocq.directed_tests import bt_performance_test
+from betocq.directed_tests import mcc_2g_wfd_indoor_5g_sta_test
+from betocq.directed_tests import mcc_5g_hotspot_dfs_5g_sta_test
+from betocq.directed_tests import mcc_5g_wfd_dfs_5g_sta_test
+from betocq.directed_tests import mcc_5g_wfd_non_dbs_2g_sta_test
+from betocq.directed_tests import scc_2g_wfd_sta_test
+from betocq.directed_tests import scc_2g_wlan_sta_test
+from betocq.directed_tests import scc_5g_wfd_dbs_2g_sta_test
+from betocq.directed_tests import scc_5g_wfd_sta_test
+from betocq.directed_tests import scc_5g_wlan_sta_test
+from betocq.directed_tests import scc_dfs_5g_hotspot_sta_test
+from betocq.directed_tests import scc_dfs_5g_wfd_sta_test
+from betocq.directed_tests import scc_indoor_5g_wfd_sta_test
+from betocq.function_tests import beto_cq_function_group_test
+
+
+class BetoCqPerformanceTestSuite(base_betocq_suite.BaseBetocqSuite):
+  """Add all BetoCQ tests to run in sequence."""
+
+  def setup_suite(self, config):
+    """Add all BetoCQ tests to the suite."""
+    test_parameters = nc_constants.TestParameters.from_user_params(
+        config.user_params
+    )
+
+    if test_parameters.target_cuj_name is nc_constants.TARGET_CUJ_ESIM:
+      self.add_test_class(bt_performance_test.BtPerformanceTest)
+      return
+
+    # add function tests if required
+    if (
+        test_parameters.run_function_tests_with_performance_tests
+        or test_parameters.use_auto_controlled_wifi_ap
+    ):
+      self.add_test_class(
+          beto_cq_function_group_test.BetoCqFunctionGroupTest
+      )
+
+    # add bt and ble test
+    self.add_test_class(bt_performance_test.BtPerformanceTest)
+
+    # TODO(kaishi): enable BLE test when it is ready
+
+    # add directed/cuj tests which requires 2G wlan AP - channel 6
+    if (
+        test_parameters.wifi_2g_ssid
+        or test_parameters.use_auto_controlled_wifi_ap
+    ):
+      config = self._config.copy()
+      config.user_params['wifi_channel'] = 6
+
+      self.add_test_class(
+          clazz=mcc_5g_wfd_non_dbs_2g_sta_test.Mcc5gWfdNonDbs2gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_2g_wfd_sta_test.Scc2gWfdStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_2g_wlan_sta_test.Scc2gWlanStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_5g_wfd_dbs_2g_sta_test.Scc5gWfdDbs2gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=mcc_5g_all_wifi_non_dbs_2g_sta_test.Mcc5gAllWifiNonDbs2gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_2g_all_wifi_sta_test.Scc2gAllWifiStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_5g_all_wifi_dbs_2g_sta_test.Scc5gAllWifiDbs2gStaTest,
+          config=config,
+      )
+
+    # add directed tests which requires 5G wlan AP - channel 36
+    if (
+        test_parameters.wifi_5g_ssid
+        or test_parameters.use_auto_controlled_wifi_ap
+    ):
+      config = self._config.copy()
+      config.user_params['wifi_channel'] = 36
+
+      self.add_test_class(
+          clazz=mcc_2g_wfd_indoor_5g_sta_test.Mcc2gWfdIndoor5gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_5g_wfd_sta_test.Scc5gWfdStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_5g_wlan_sta_test.Scc5gWifiLanStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_indoor_5g_wfd_sta_test.SccIndoor5gWfdStaTest,
+          config=config,
+      )
+
+    # add directed/cuj tests which requires DFS 5G wlan AP - channel 52
+    if (
+        test_parameters.wifi_dfs_5g_ssid
+        or test_parameters.use_auto_controlled_wifi_ap
+    ):
+      config = self._config.copy()
+      config.user_params['wifi_channel'] = 52
+
+      self.add_test_class(
+          clazz=mcc_5g_hotspot_dfs_5g_sta_test.Mcc5gHotspotDfs5gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=mcc_5g_wfd_dfs_5g_sta_test.Mcc5gWfdDfs5gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_dfs_5g_hotspot_sta_test.SccDfs5gHotspotStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_dfs_5g_wfd_sta_test.SccDfs5gWfdStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          clazz=scc_5g_all_wifi_sta_test.Scc5gAllWifiStaTest,
+          config=config,
+      )
+
+
+if __name__ == '__main__':
+  # Use suite_runner's `main`.
+  suite_runner.run_suite_class()
diff --git a/tests/bettertogether/betocq/batch_performance_testing_with_2g_ap_test_suite.py b/tests/bettertogether/betocq/batch_performance_testing_with_2g_ap_test_suite.py
new file mode 100644
index 000000000..fd0767fbc
--- /dev/null
+++ b/tests/bettertogether/betocq/batch_performance_testing_with_2g_ap_test_suite.py
@@ -0,0 +1,103 @@
+#  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.
+
+"""This test suite batches all tests to run in sequence.
+
+This requires a 2G AP to be ready and configured in testbed.
+2G AP (wifi_2g_ssid): channel 6 (2437)
+"""
+
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly import suite_runner
+
+from betocq import base_betocq_suite
+from betocq import nc_constants
+from betocq.cuj_tests import mcc_5g_all_wifi_non_dbs_2g_sta_test
+from betocq.cuj_tests import scc_2g_all_wifi_sta_test
+from betocq.cuj_tests import scc_5g_all_wifi_dbs_2g_sta_test
+from betocq.directed_tests import ble_performance_test
+from betocq.directed_tests import bt_performance_test
+from betocq.directed_tests import mcc_5g_wfd_non_dbs_2g_sta_test
+from betocq.directed_tests import scc_2g_wfd_sta_test
+from betocq.directed_tests import scc_2g_wlan_sta_test
+from betocq.directed_tests import scc_5g_wfd_dbs_2g_sta_test
+from betocq.function_tests import beto_cq_function_group_test
+
+
+class BetoCq2gApPerformanceTestSuite(base_betocq_suite.BaseBetocqSuite):
+  """Add all BetoCQ tests which requires 2G STA AP or no AP to run in sequence."""
+
+  def setup_suite(self, config):
+    """Add all BetoCQ tests which requires 2G STA AP or no AP to the suite."""
+    test_parameters = nc_constants.TestParameters.from_user_params(
+        config.user_params
+    )
+
+    # add function tests if required
+    if test_parameters.run_function_tests_with_performance_tests:
+      self.add_test_class(
+          beto_cq_function_group_test.BetoCqFunctionGroupTest,
+          config=config,
+      )
+
+    # add bt and ble test
+    self.add_test_class(
+        bt_performance_test.BtPerformanceTest, config=config
+    )
+    # TODO(kaishi): enable BLE test when it is ready
+
+    # add directed/cuj tests which requires 2G wlan AP - channel 6
+    if (
+        test_parameters.wifi_2g_ssid
+        or test_parameters.use_auto_controlled_wifi_ap
+    ):
+      self.add_test_class(
+          mcc_5g_wfd_non_dbs_2g_sta_test.Mcc5gWfdNonDbs2gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          scc_2g_wfd_sta_test.Scc2gWfdStaTest, config=config
+      )
+      self.add_test_class(
+          scc_2g_wlan_sta_test.Scc2gWlanStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          scc_5g_wfd_dbs_2g_sta_test.Scc5gWfdDbs2gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          mcc_5g_all_wifi_non_dbs_2g_sta_test.Mcc5gAllWifiNonDbs2gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          scc_2g_all_wifi_sta_test.Scc2gAllWifiStaTest, config=config
+      )
+      self.add_test_class(
+          scc_5g_all_wifi_dbs_2g_sta_test.Scc5gAllWifiDbs2gStaTest,
+          config=config,
+      )
+
+if __name__ == '__main__':
+  # Use suite_runner's `main`.
+  suite_runner.run_suite_class()
+
diff --git a/tests/bettertogether/betocq/batch_performance_testing_with_5g_ap_test_suite.py b/tests/bettertogether/betocq/batch_performance_testing_with_5g_ap_test_suite.py
new file mode 100644
index 000000000..8b50d82cd
--- /dev/null
+++ b/tests/bettertogether/betocq/batch_performance_testing_with_5g_ap_test_suite.py
@@ -0,0 +1,74 @@
+#  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.
+
+"""This test suite batches all tests to run in sequence.
+
+This requires a 5G AP to be ready and configured in testbed.
+5G AP (wifi_5g_ssid): channel 36 (5180)
+"""
+
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly import suite_runner
+
+from betocq import base_betocq_suite
+from betocq import nc_constants
+from betocq.directed_tests import mcc_2g_wfd_indoor_5g_sta_test
+from betocq.directed_tests import scc_5g_wfd_sta_test
+from betocq.directed_tests import scc_5g_wlan_sta_test
+from betocq.directed_tests import scc_indoor_5g_wfd_sta_test
+
+
+class BetoCq5gApPerformanceTestSuite(base_betocq_suite.BaseBetocqSuite):
+  """Add BetoCQ tests which requires a 5G STA AP to run in sequence."""
+
+  def setup_suite(self, config):
+    """Add BetoCQ tests which requires a 5G STA AP to the suite."""
+    test_parameters = nc_constants.TestParameters.from_user_params(
+        config.user_params
+    )
+
+    # add directed tests which requires 5G wlan AP - channel 36
+    if (
+        test_parameters.wifi_5g_ssid
+        or test_parameters.use_auto_controlled_wifi_ap
+    ):
+      self.add_test_class(
+          mcc_2g_wfd_indoor_5g_sta_test.Mcc2gWfdIndoor5gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          scc_5g_wfd_sta_test.Scc5gWfdStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          scc_5g_wlan_sta_test.Scc5gWifiLanStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          scc_indoor_5g_wfd_sta_test.SccIndoor5gWfdStaTest,
+          config=config,
+      )
+
+if __name__ == '__main__':
+  # Use suite_runner's `main`.
+  suite_runner.run_suite_class()
+
diff --git a/tests/bettertogether/betocq/batch_performance_testing_with_dfs_5g_ap_test_suite.py b/tests/bettertogether/betocq/batch_performance_testing_with_dfs_5g_ap_test_suite.py
new file mode 100644
index 000000000..754183bec
--- /dev/null
+++ b/tests/bettertogether/betocq/batch_performance_testing_with_dfs_5g_ap_test_suite.py
@@ -0,0 +1,78 @@
+#  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.
+
+"""This test suite batches all tests to run in sequence.
+
+This requires a DFS 5G AP to be ready and configured in testbed.
+DFS 5G AP(wifi_dfs_5g_ssid): channel 52 (5260)
+"""
+
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly import suite_runner
+
+from betocq import base_betocq_suite
+from betocq import nc_constants
+
+from betocq.cuj_tests import scc_5g_all_wifi_sta_test
+from betocq.directed_tests import mcc_5g_hotspot_dfs_5g_sta_test
+from betocq.directed_tests import mcc_5g_wfd_dfs_5g_sta_test
+from betocq.directed_tests import scc_dfs_5g_hotspot_sta_test
+from betocq.directed_tests import scc_dfs_5g_wfd_sta_test
+
+
+class BetoCqPerformanceTestSuite(base_betocq_suite.BaseBetocqSuite):
+  """Add all BetoCQ tests to run in sequence."""
+
+  def setup_suite(self, config):
+    """Add all BetoCQ tests to the suite."""
+    test_parameters = nc_constants.TestParameters.from_user_params(
+        config.user_params
+    )
+    # add directed/cuj tests which requires DFS 5G wlan AP - channel 52
+    if (
+        test_parameters.wifi_dfs_5g_ssid
+        or test_parameters.use_auto_controlled_wifi_ap
+    ):
+      self.add_test_class(
+          mcc_5g_hotspot_dfs_5g_sta_test.Mcc5gHotspotDfs5gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          mcc_5g_wfd_dfs_5g_sta_test.Mcc5gWfdDfs5gStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          scc_dfs_5g_hotspot_sta_test.SccDfs5gHotspotStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          scc_dfs_5g_wfd_sta_test.SccDfs5gWfdStaTest,
+          config=config,
+      )
+      self.add_test_class(
+          scc_5g_all_wifi_sta_test.Scc5gAllWifiStaTest,
+          config=config,
+      )
+
+if __name__ == '__main__':
+  # Use suite_runner's `main`.
+  suite_runner.run_suite_class()
diff --git a/tests/bettertogether/betocq/cuj_tests/bt_2g_wifi_coex_test.py b/tests/bettertogether/betocq/cuj_tests/bt_2g_wifi_coex_test.py
new file mode 100644
index 000000000..4370eea23
--- /dev/null
+++ b/tests/bettertogether/betocq/cuj_tests/bt_2g_wifi_coex_test.py
@@ -0,0 +1,101 @@
+#  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.
+
+"""This Test is to test the bluetooth and wifi 2G coex.
+
+The AP requirements:
+  wifi channel: 6 (2437)
+"""
+
+import datetime
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+_PERFORMANCE_TEST_COUNT = 100
+_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR = 10
+
+
+class Bt2gWifiCoexTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for BT and 2G wifi coex with a complicated stress test."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    self.test_parameters.requires_bt_multiplex = True
+    super().setup_class()
+    self.performance_test_iterations = getattr(
+        self.test_bt_2g_wifi_coex, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_bt_2g_wifi_coex(self):
+    """Test the BT and 2G wifi coex with a stress test."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_ALL_WIFI,
+        wifi_ssid=self.test_parameters.wifi_2g_ssid,
+        wifi_password=self.test_parameters.wifi_2g_password,
+    )
+
+  def _get_transfer_file_size(self) -> int:
+    # For 2G wifi medium
+    return nc_constants.TRANSFER_FILE_SIZE_20MB
+
+  def _get_file_transfer_timeout(self) -> datetime.timedelta:
+    return nc_constants.WIFI_2G_20M_PAYLOAD_TRANSFER_TIMEOUT
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The Wifi Direct connection might be broken, check related log.'
+    )
+
+  def _get_throughout_benchmark(self) -> int:
+    # no requirement for throughput.
+    return 0
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        f'{self._throughput_low_string}.'
+        ' This should never happen, you may ignore this error, this is'
+        ' not required for this case.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_2g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return True
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/cuj_tests/local_dev_cuj_testbed.yml b/tests/bettertogether/betocq/cuj_tests/local_dev_cuj_testbed.yml
new file mode 100644
index 000000000..2eeee66bd
--- /dev/null
+++ b/tests/bettertogether/betocq/cuj_tests/local_dev_cuj_testbed.yml
@@ -0,0 +1,77 @@
+x-test-params: &test-params
+  # target_cuj_name: "quick_start"
+  allow_unrooted_device: False
+  wifi_2g_ssid: "NETGEAR62"
+  wifi_2g_password: "widetree588"
+  wifi_5g_ssid: "NETGEAR62-5G-1"
+  wifi_5g_password: "widetree588"
+  wifi_dfs_5g_ssid: "GoogleGuest"
+  wifi_dfs_5g_password: ""
+  # use the AP controlled by programming dynamically
+  use_auto_controlled_wifi_ap: False
+
+x-controllers: &controllers
+  Controllers:
+    AndroidDevice:
+      - serial: "R3CN90YNAR"  # "R3CN90YNAR"
+        role: "source_device"
+        # The max number of spatial streams
+        max_num_streams: 2
+        # The max PHY rate at 5G, in Mbps
+        max_phy_rate_5g_mbps: 2402
+        # The max PHY rate at 2G, in Mbps
+        max_phy_rate_2g_mbps: 287
+        # if the device supports 5G Wifi
+        supports_5g: True  # True
+        # if the device supports DBS (Dual Band Simultaneous) in STA + WiFi-Direct concurrency mode
+        supports_dbs_sta_wfd: True
+        # The max number of spatial streams in DBS mode.
+        max_num_streams_dbs: 2
+        # if the device supports to start WFD group owner at a STA-associated DFS channel
+        enable_sta_dfs_channel_for_peer_network: False
+        # if the device supports to start WFD group owner at a STA-associated indoor channel
+        enable_sta_indoor_channel_for_peer_network: False
+        # 14 - U, 13 - T, 12 - S
+        android_version: 14
+      - serial: "2B031FDJH0002G"  # "ABCDEF123456"
+        role: "target_device"
+        # The max number of spatial streams
+        max_num_streams: 2
+        # The max PHY rate at 5G, in Mbps
+        max_phy_rate_5g_mbps: 2402
+        # The max PHY rate at 2G, in Mbps
+        max_phy_rate_2g_mbps: 287
+        # if the device supports 5G Wifi
+        supports_5g: True  # True
+        # if the device supports DBS (Dual Band Simultaneous) in STA + WiFi-Direct concurrency mode
+        supports_dbs_sta_wfd: True
+        # The max number of spatial streams in DBS mode.
+        max_num_streams_dbs: 2
+        # if the device supports to start WFD group owner at a STA-associated DFS channel
+        enable_sta_dfs_channel_for_peer_network: True
+        # if the device supports to start WFD group owner at a STA-associated indoor channel
+        enable_sta_indoor_channel_for_peer_network: True
+        # 14 - U, 13 - T, 12 - S
+        android_version: 14
+
+TestBeds:
+- Name: LocalPerformanceDefaultTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
+
+- Name: LocalQuickstartPerformanceTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
+      target_cuj_name: "quick_start"
+      # before the performance test, run the function tests first
+      run_function_tests_with_performance_tests: True
+      # if the function tests is failed, abort the performance test
+      abort_all_tests_on_function_tests_fail: True
+
+- Name: LocalQuicksharePerformanceTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
+      target_cuj_name: "quick_share"
\ No newline at end of file
diff --git a/tests/bettertogether/betocq/cuj_tests/mcc_5g_all_wifi_non_dbs_2g_sta_test.py b/tests/bettertogether/betocq/cuj_tests/mcc_5g_all_wifi_non_dbs_2g_sta_test.py
new file mode 100644
index 000000000..f49cb09aa
--- /dev/null
+++ b/tests/bettertogether/betocq/cuj_tests/mcc_5g_all_wifi_non_dbs_2g_sta_test.py
@@ -0,0 +1,112 @@
+#  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.
+
+"""This Test is to test the Wifi MCC in a general case.
+
+In this case, even though the expected wifi medium is the WFD, but the wifi D2D
+could be any wifi mediums, such as WFD, HOTSPOT, WifiLan; Once the WFD is
+failed, other meidums will be tried. As DBS is not supported by both devices,
+and the STA is coonected with 2G channel, D2D medium is using a 5G channel,
+this cause the MCC case.
+
+The device requirements:
+  support 5G: true
+  support DBS (target device): false
+The AP requirements:
+  wifi channel: 6 (2437)
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Mcc5gAllWifiNonDbs2gStaTest(
+    d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for CUJ MCC with 5G D2D medium and 2G WLAN test."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self._is_mcc = True
+    self.performance_test_iterations = getattr(
+        self.test_mcc_5g_all_wifi_non_dbs_2g_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.MCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.MCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_mcc_5g_all_wifi_non_dbs_2g_sta(self):
+    self._test_connection_medium_performance(
+        upgrade_medium_under_test=nc_constants.NearbyMedium.UPGRADE_TO_ALL_WIFI,
+        wifi_ssid=self.test_parameters.wifi_2g_ssid,
+        wifi_password=self.test_parameters.wifi_2g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    upgraded_medium_name = None
+    if (self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium
+        is not None):
+      upgraded_medium_name = (
+          self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium.name
+      )
+
+    return (
+        f'The upgraded wifi medium {upgraded_medium_name} might be broken, '
+        f'check the related log, Or {self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    upgraded_medium_name = None
+    if (self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium
+        is not None):
+      upgraded_medium_name = (
+          self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium.name
+      )
+    return (
+        f'{self._throughput_low_string}. The upgraded medium is'
+        f' {upgraded_medium_name} Check with the wifi chip vendor for any FW'
+        ' issue in MCC mode'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_2g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return (
+        self.discoverer.supports_5g and self.advertiser.supports_5g
+        and not self.advertiser.supports_dbs_sta_wfd
+    )
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/cuj_tests/scc_2g_all_wifi_sta_test.py b/tests/bettertogether/betocq/cuj_tests/scc_2g_all_wifi_sta_test.py
new file mode 100644
index 000000000..8d425eb79
--- /dev/null
+++ b/tests/bettertogether/betocq/cuj_tests/scc_2g_all_wifi_sta_test.py
@@ -0,0 +1,113 @@
+#  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.
+
+"""This Test is to test the Wifi SCC in a general case.
+
+In this case, even though the expected wifi medium is the WFD, but the wifi D2D
+could be any technologies, such as WFD, HOTSPOT, WifiLAN; Once the WFD is
+failed, other meidums will be tried. Both the D2D and STA are using the same 2G
+channel.
+
+The device requirements:
+  support 5G: false
+The AP requirements:
+  wifi channel: 6 (2437)
+"""
+
+import datetime
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Scc2gAllWifiStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for Wifi SCC 2G test associated with a specified CUJ."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self._is_2g_d2d_wifi_medium = True
+    self.performance_test_iterations = getattr(
+        self.test_scc_2g_all_wifi_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_2g_all_wifi_sta(self):
+    """Test the 2G SCC case, both the wifi D2D medium and STA are using 2G."""
+    self._test_connection_medium_performance(
+        upgrade_medium_under_test=nc_constants.NearbyMedium.UPGRADE_TO_ALL_WIFI,
+        wifi_ssid=self.test_parameters.wifi_2g_ssid,
+        wifi_password=self.test_parameters.wifi_2g_password)
+
+  def _get_transfer_file_size(self) -> int:
+    # For 2G wifi medium
+    return nc_constants.TRANSFER_FILE_SIZE_20MB
+
+  def _get_file_transfer_timeout(self) -> datetime.timedelta:
+    return nc_constants.WIFI_2G_20M_PAYLOAD_TRANSFER_TIMEOUT
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    upgraded_medium_name = None
+    if (self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium
+        is not None):
+      upgraded_medium_name = (
+          self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium.name
+      )
+    return (
+        f'The upgraded wifi medium {upgraded_medium_name} might be broken, '
+        f'check the related log; Or {self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    upgraded_medium_name = None
+    if (self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium
+        is not None):
+      upgraded_medium_name = (
+          self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium.name
+      )
+    return (
+        f'{self._throughput_low_string}. The upgraded medium is'
+        f' {upgraded_medium_name}, this is a 2G SCC case. Check with the wifi'
+        ' chip vendor for any FW issue in this mode.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_2g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return not self.discoverer.supports_5g or not self.advertiser.supports_5g
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/cuj_tests/scc_5g_all_wifi_dbs_2g_sta_test.py b/tests/bettertogether/betocq/cuj_tests/scc_5g_all_wifi_dbs_2g_sta_test.py
new file mode 100644
index 000000000..1408b61b4
--- /dev/null
+++ b/tests/bettertogether/betocq/cuj_tests/scc_5g_all_wifi_dbs_2g_sta_test.py
@@ -0,0 +1,112 @@
+#  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.
+
+"""This Test is to test the Wifi SCC in a general case.
+
+In this case, even though the expected wifi medium is the WFD, but the wifi D2D
+could be any mediums, such as WFD, HOTSPOT, STA; Once the WFD is failed, other
+meidums will be tried. Also, though the WLAN is connected with 2G channel,
+as the devices support DBS, which don't need to switch between 5G and 2G, it is
+still a SCC case.
+
+The device requirements:
+  support 5G: true
+  support DBS (target device): true
+The AP requirements:
+  wifi channel: 6 (2437)
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Scc5gAllWifiDbs2gStaTest(
+    d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for CUJ SCC with 5G D2D medium and 2G WLAN test.
+  """
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self.performance_test_iterations = getattr(
+        self.test_scc_5g_all_wifi_dbs_2g_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_5g_all_wifi_dbs_2g_sta(self):
+    self._test_connection_medium_performance(
+        upgrade_medium_under_test=nc_constants.NearbyMedium.UPGRADE_TO_ALL_WIFI,
+        wifi_ssid=self.test_parameters.wifi_2g_ssid,
+        wifi_password=self.test_parameters.wifi_2g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    upgraded_medium_name = None
+    if (self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium
+        is not None):
+      upgraded_medium_name = (
+          self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium.name
+      )
+    return (
+        f'The upgraded wifi medium {upgraded_medium_name} might be broken, '
+        f'check the related log, Or {self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    upgraded_medium_name = None
+    if (self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium
+        is not None):
+      upgraded_medium_name = (
+          self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium.name
+      )
+    return (
+        f'{self._throughput_low_string}. The upgraded medium is'
+        f' {upgraded_medium_name}. This is a 2G SCC DBS case, In the'
+        ' configuration file, DBS support is set to true on the target side.'
+        ' Check if the device does support DBS with STA + WFD concurrency.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_2g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return (
+        self.discoverer.supports_5g and self.advertiser.supports_5g
+        and self.advertiser.supports_dbs_sta_wfd
+    )
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/cuj_tests/scc_5g_all_wifi_sta_test.py b/tests/bettertogether/betocq/cuj_tests/scc_5g_all_wifi_sta_test.py
new file mode 100644
index 000000000..348d8621a
--- /dev/null
+++ b/tests/bettertogether/betocq/cuj_tests/scc_5g_all_wifi_sta_test.py
@@ -0,0 +1,106 @@
+#  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.
+
+"""This Test is to test the Wifi SCC in a general case.
+
+In this case, even though the expected wifi medium is the WFD, but the wifi D2D
+could be any technologies, such as WFD, HOTSPOT, STA; Once the WFD is failed,
+other meidums will be tried. Both the D2D medium and STA are using the same 5G
+channel.
+
+The device requirements:
+  support 5G: true
+  country code: US
+The AP requirements:
+  wifi channel: 36 (5180)
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Scc5gAllWifiStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for Wifi SCC 5G test associated with a specified CUJ."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self.performance_test_iterations = getattr(
+        self.test_scc_5g_all_wifi_sta_test, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_5g_all_wifi_sta_test(self):
+    self._test_connection_medium_performance(
+        upgrade_medium_under_test=nc_constants.NearbyMedium.UPGRADE_TO_ALL_WIFI,
+        wifi_ssid=self.test_parameters.wifi_5g_ssid,
+        wifi_password=self.test_parameters.wifi_5g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    upgraded_medium_name = None
+    if (self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium
+        is not None):
+      upgraded_medium_name = (
+          self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium.name
+      )
+    return (
+        f'The upgraded wifi medium {upgraded_medium_name} might be broken, '
+        f'check the related logs; Or {self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    upgraded_medium_name = None
+    if (self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium
+        is not None):
+      upgraded_medium_name = (
+          self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium.name
+      )
+    return (
+        f'{self._throughput_low_string}. The upgraded medium is'
+        f' {upgraded_medium_name}.'
+        ' This is a 5G SCC case. Check with the wifi'
+        ' chip vendor for any possible FW issue.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_5g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return self.discoverer.supports_5g and self.advertiser.supports_5g
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/d2d_performance_test_base.py b/tests/bettertogether/betocq/d2d_performance_test_base.py
new file mode 100644
index 000000000..915f2eef6
--- /dev/null
+++ b/tests/bettertogether/betocq/d2d_performance_test_base.py
@@ -0,0 +1,705 @@
+#  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.
+
+"""Nearby Connection E2E stress tests for D2D wifi performance."""
+
+import abc
+import datetime
+import logging
+import time
+from typing import Any
+
+from mobly import asserts
+from mobly.controllers import android_device
+
+from betocq import nc_base_test
+from betocq import nc_constants
+from betocq import nearby_connection_wrapper
+from betocq import setup_utils
+
+
+_DELAY_BETWEEN_EACH_TEST_CYCLE = datetime.timedelta(seconds=5)
+_INVALID_FREQ = -1
+_INVALID_MAX_LINK_SPEED = -1
+_BITS_PER_BYTE = 8
+
+
+class D2dPerformanceTestBase(nc_base_test.NCBaseTestClass, abc.ABC):
+  """Abstract class for D2D performance test for different connection meidums."""
+
+  performance_test_iterations: int
+
+  # @typing.override
+  def __init__(self, configs):
+    super().__init__(configs)
+    self._is_mcc: bool = False
+    self._is_2g_d2d_wifi_medium: bool = False
+    self._is_dbs_mode: bool = False
+    self._throughput_low_string: str = ''
+    self._upgrade_medium_under_test: nc_constants.NearbyMedium = None
+    self._current_test_result: nc_constants.SingleTestResult = (
+        nc_constants.SingleTestResult()
+    )
+    self._performance_test_metrics: nc_constants.NcPerformanceTestMetrics = (
+        nc_constants.NcPerformanceTestMetrics()
+    )
+    self._prior_bt_nc_fail_reason: nc_constants.SingleTestFailureReason = (
+        nc_constants.SingleTestFailureReason.UNINITIALIZED
+    )
+    self._active_nc_fail_reason: nc_constants.SingleTestFailureReason = (
+        nc_constants.SingleTestFailureReason.UNINITIALIZED
+    )
+    self._finished_test_iteration: int = 0
+    self._use_prior_bt: bool = False
+    self._test_results: list[nc_constants.SingleTestResult] = []
+
+  # @typing.override
+  def setup_test(self):
+    self._current_test_result: nc_constants.SingleTestResult = (
+        nc_constants.SingleTestResult()
+    )
+    self._prior_bt_nc_fail_reason = (
+        nc_constants.SingleTestFailureReason.UNINITIALIZED
+    )
+    self._active_nc_fail_reason = (
+        nc_constants.SingleTestFailureReason.UNINITIALIZED
+    )
+    super().setup_test()
+
+  def teardown_test(self):
+    self._write_current_test_report()
+    self._collect_current_test_metrics()
+    super().teardown_test()
+    time.sleep(_DELAY_BETWEEN_EACH_TEST_CYCLE.total_seconds())
+
+  # @typing.override
+  def _get_skipped_test_class_reason(self) -> str | None:
+    if not self._is_wifi_ap_ready():
+      return 'Wifi AP is not ready for this test.'
+    if not self._are_devices_capabilities_ok():
+      return 'The test is not required per the device capabilities.'
+    return None
+
+  @abc.abstractmethod
+  def _is_wifi_ap_ready(self) -> bool:
+    pass
+
+  @abc.abstractmethod
+  def _are_devices_capabilities_ok(self) -> bool:
+    pass
+
+  def _get_throughout_benchmark(self) -> int:
+    """Gets the throughout benchmark as KBps."""
+    max_num_streams = min(
+        self.discoverer.max_num_streams, self.advertiser.max_num_streams
+    )
+
+    sta_frequency = int(
+        self.advertiser.nearby.wifiGetConnectionInfo().get(
+            'mFrequency', _INVALID_FREQ
+        )
+    )
+
+    sta_max_link_speed_mbps = int(
+        self.advertiser.nearby.wifiGetConnectionInfo().get(
+            'mMaxSupportedTxLinkSpeed', _INVALID_MAX_LINK_SPEED
+        )
+    )
+
+    if self._is_2g_d2d_wifi_medium:
+      max_phy_rate_mbps = min(
+          self.discoverer.max_phy_rate_2g_mbps,
+          self.advertiser.max_phy_rate_2g_mbps,
+      )
+      max_phy_rate_mbps = min(
+          max_phy_rate_mbps,
+          max_num_streams * nc_constants.MAX_PHY_RATE_PER_STREAM_N_20_MBPS,
+      )
+      min_throughput_mbyte_per_sec = int(
+          max_phy_rate_mbps
+          * nc_constants.MAX_PHY_RATE_TO_MIN_THROUGHPUT_RATIO_2G
+          / _BITS_PER_BYTE
+      )
+    else:  # 5G wifi medium
+      max_phy_rate_mbps = min(
+          self.discoverer.max_phy_rate_5g_mbps,
+          self.advertiser.max_phy_rate_5g_mbps,
+      )
+      # max_num_streams could be smaller in DBS mode
+      if self._is_dbs_mode:
+        max_num_streams = self.advertiser.max_num_streams_dbs
+
+      max_phy_rate_ac80 = (
+          max_num_streams * nc_constants.MAX_PHY_RATE_PER_STREAM_AC_80_MBPS
+      )
+      max_phy_rate_mbps = min(max_phy_rate_mbps, max_phy_rate_ac80)
+
+      # if STA is connected to 5G AP with channel BW < 80,
+      # limit the max phy rate to AC 40.
+      if (sta_frequency > 5000
+          and sta_max_link_speed_mbps > 0
+          and sta_max_link_speed_mbps < max_phy_rate_ac80
+          ):
+        max_phy_rate_mbps = min(
+            max_phy_rate_mbps,
+            max_num_streams * nc_constants.MAX_PHY_RATE_PER_STREAM_AC_40_MBPS,
+        )
+
+      min_throughput_mbyte_per_sec = int(
+          max_phy_rate_mbps
+          * nc_constants.MAX_PHY_RATE_TO_MIN_THROUGHPUT_RATIO_5G
+          / _BITS_PER_BYTE
+      )
+      if self._is_mcc:
+        min_throughput_mbyte_per_sec = int(
+            min_throughput_mbyte_per_sec
+            * nc_constants.MCC_THROUGHPUT_MULTIPLIER
+        )
+      if (
+          self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium
+          == nc_constants.NearbyConnectionMedium.WIFI_HOTSPOT
+      ):
+        min_throughput_mbyte_per_sec = int(
+            min_throughput_mbyte_per_sec
+            * nc_constants.WIFI_HOTSPOT_THROUGHPUT_MULTIPLIER
+        )
+
+    self.advertiser.log.info(
+        f'STA frequency = {sta_frequency}, '
+        f'STA max link speed (Mb/s) = {sta_max_link_speed_mbps}, '
+        f'max D2D phy rate (Mb/s) = {max_phy_rate_mbps}, '
+        f'min D2D throughput (MB/s) = {min_throughput_mbyte_per_sec}'
+    )
+    return min_throughput_mbyte_per_sec * 1024
+
+  def _test_connection_medium_performance(
+      self,
+      upgrade_medium_under_test: nc_constants.NearbyMedium,
+      wifi_ssid: str = '',
+      wifi_password: str = '',
+      force_disable_bt_multiplex: bool = False,
+      connection_medium: nc_constants.NearbyMedium = nc_constants.NearbyMedium.BT_ONLY,
+  ) -> None:
+    """Test the D2D performance with the specified upgrade medium."""
+    self._upgrade_medium_under_test = upgrade_medium_under_test
+
+    if self.test_parameters.toggle_airplane_mode_target_side:
+      setup_utils.toggle_airplane_mode(self.advertiser)
+    if self.test_parameters.reset_wifi_connection:
+      self._reset_wifi_connection()
+    # 1. discoverer connect to wifi STA/AP
+    self._current_test_result = nc_constants.SingleTestResult()
+    if wifi_ssid:
+      self._active_nc_fail_reason = (
+          nc_constants.SingleTestFailureReason.SOURCE_WIFI_CONNECTION
+      )
+      discoverer_wifi_sta_latency = (
+          setup_utils.connect_to_wifi_sta_till_success(
+              self.discoverer, wifi_ssid, wifi_password
+          )
+      )
+      self._active_nc_fail_reason = nc_constants.SingleTestFailureReason.SUCCESS
+      self.discoverer.log.info(
+          'connecting to wifi in '
+          f'{round(discoverer_wifi_sta_latency.total_seconds())} s'
+      )
+      self._current_test_result.discoverer_sta_expected = True
+      self._current_test_result.discoverer_sta_latency = (
+          discoverer_wifi_sta_latency
+      )
+
+    # 2. set up BT connection if required
+    advertising_discovery_medium = nc_constants.NearbyMedium.BLE_ONLY
+
+    connection_setup_timeouts = nc_constants.ConnectionSetupTimeouts(
+        nc_constants.FIRST_DISCOVERY_TIMEOUT,
+        nc_constants.FIRST_CONNECTION_INIT_TIMEOUT,
+        nc_constants.FIRST_CONNECTION_RESULT_TIMEOUT,
+    )
+    prior_bt_snippet = None
+    if (
+        not force_disable_bt_multiplex
+        and self.test_parameters.requires_bt_multiplex
+    ):
+      logging.info('set up a prior BT connection.')
+      self._use_prior_bt = True
+      prior_bt_snippet = nearby_connection_wrapper.NearbyConnectionWrapper(
+          self.advertiser,
+          self.discoverer,
+          self.advertiser.nearby2,
+          self.discoverer.nearby2,
+          advertising_discovery_medium=advertising_discovery_medium,
+          connection_medium=nc_constants.NearbyMedium.BT_ONLY,
+          upgrade_medium=nc_constants.NearbyMedium.BT_ONLY,
+      )
+
+      try:
+        prior_bt_snippet.start_nearby_connection(
+            timeouts=connection_setup_timeouts,
+            medium_upgrade_type=nc_constants.MediumUpgradeType.NON_DISRUPTIVE,
+        )
+      finally:
+        self._prior_bt_nc_fail_reason = (
+            prior_bt_snippet.test_failure_reason
+        )
+        self._current_test_result.prior_nc_setup_quality_info = (
+            prior_bt_snippet.connection_quality_info
+        )
+
+    # set up Wifi connection and transfer
+    # 3. advertiser connect to wifi STA/AP
+    if wifi_ssid:
+      self._active_nc_fail_reason = (
+          nc_constants.SingleTestFailureReason.TARGET_WIFI_CONNECTION
+      )
+      advertiser_wifi_sta_latency = (
+          setup_utils.connect_to_wifi_sta_till_success(
+              self.advertiser, wifi_ssid, wifi_password
+          )
+      )
+      self.advertiser.log.info(
+          'connecting to wifi in '
+          f'{round(advertiser_wifi_sta_latency.total_seconds())} s'
+      )
+      self.advertiser.log.info(
+          self.advertiser.nearby.wifiGetConnectionInfo().get('mFrequency')
+      )
+      self._current_test_result.advertiser_wifi_expected = True
+      self._current_test_result.advertiser_sta_latency = (
+          advertiser_wifi_sta_latency
+      )
+
+    # 4. set up the D2D nearby connection
+    logging.info('set up a nearby connection for file transfer.')
+    active_snippet = nearby_connection_wrapper.NearbyConnectionWrapper(
+        self.advertiser,
+        self.discoverer,
+        self.advertiser.nearby,
+        self.discoverer.nearby,
+        advertising_discovery_medium=advertising_discovery_medium,
+        connection_medium=connection_medium,
+        upgrade_medium=upgrade_medium_under_test,
+    )
+    if prior_bt_snippet:
+      connection_setup_timeouts = nc_constants.ConnectionSetupTimeouts(
+          nc_constants.SECOND_DISCOVERY_TIMEOUT,
+          nc_constants.SECOND_CONNECTION_INIT_TIMEOUT,
+          nc_constants.SECOND_CONNECTION_RESULT_TIMEOUT,
+      )
+    try:
+      active_snippet.start_nearby_connection(
+          timeouts=connection_setup_timeouts,
+          medium_upgrade_type=nc_constants.MediumUpgradeType.DISRUPTIVE,
+          keep_alive_timeout_ms=self.test_parameters.keep_alive_timeout_ms,
+          keep_alive_interval_ms=self.test_parameters.keep_alive_interval_ms,
+      )
+    finally:
+      self._active_nc_fail_reason = active_snippet.test_failure_reason
+      self._current_test_result.file_transfer_nc_setup_quality_info = (
+          active_snippet.connection_quality_info
+      )
+
+    # 5. transfer file through the nearby connection
+    try:
+      self._current_test_result.file_transfer_throughput_kbps = (
+          active_snippet.transfer_file(
+              self._get_transfer_file_size(),
+              self._get_file_transfer_timeout(),
+              self.test_parameters.payload_type,
+          )
+      )
+    finally:
+      self._active_nc_fail_reason = active_snippet.test_failure_reason
+      if (
+          self._active_nc_fail_reason
+          is nc_constants.SingleTestFailureReason.SUCCESS
+      ):
+        self._throughout_benchmark_kbyte_per_sec = (
+            self._get_throughout_benchmark()
+        )
+
+        if (
+            self._current_test_result.file_transfer_throughput_kbps
+            < self._throughout_benchmark_kbyte_per_sec
+        ):
+          self._active_nc_fail_reason = (
+              nc_constants.SingleTestFailureReason.FILE_TRANSFER_THROUGHPUT_LOW
+          )
+          file_transfer_throughput_kbps = str(
+              self._current_test_result.file_transfer_throughput_kbps
+          )
+          self._throughput_low_string = (
+              'The measured throughput'
+              f' {file_transfer_throughput_kbps} is'
+              ' lower than the expected'
+              f' {self._throughout_benchmark_kbyte_per_sec} KBps'
+          )
+          asserts.fail(self._throughput_low_string)
+
+    # 6. disconnect prior BT connection if required
+    if prior_bt_snippet:
+      prior_bt_snippet.disconnect_endpoint()
+    # 7. disconnect D2D active connection
+    active_snippet.disconnect_endpoint()
+
+  def _get_transfer_file_size(self) -> int:
+    return nc_constants.TRANSFER_FILE_SIZE_200MB
+
+  def _get_file_transfer_timeout(self) -> datetime.timedelta:
+    return nc_constants.WIFI_200M_PAYLOAD_TRANSFER_TIMEOUT
+
+  def _write_current_test_report(self) -> None:
+    """Writes test report for each iteration."""
+    self._current_test_result.test_iteration = self._finished_test_iteration
+    self._finished_test_iteration += 1
+    if (
+        self._use_prior_bt
+        and self._prior_bt_nc_fail_reason
+        is not nc_constants.SingleTestFailureReason.SUCCESS
+    ):
+      self._current_test_result.is_failed_with_prior_bt = True
+      self._current_test_result.failure_reason = self._prior_bt_nc_fail_reason
+    else:
+      self._current_test_result.failure_reason = self._active_nc_fail_reason
+    result_message = self._get_current_test_result_message()
+    self._current_test_result.result_message = result_message
+    self._test_results.append(self._current_test_result)
+
+    quality_info: list[Any] = []
+    if self._use_prior_bt:
+      quality_info.append(
+          'prior_bt_connection:'
+          f'{self._current_test_result.prior_nc_setup_quality_info.get_dict()}'
+      )
+    quality_info.append(
+        'file_transfer_connection_quality_info:'
+        f'{self._current_test_result.file_transfer_nc_setup_quality_info.get_dict()}'
+    )
+    quality_info.append(
+        'file_transfer_speed: '
+        f'{round(self._current_test_result.file_transfer_throughput_kbps/1024, 1)}'
+        'MBps'
+    )
+
+    if self._current_test_result.discoverer_sta_expected:
+      src_connection_latency = round(
+          self._current_test_result.discoverer_sta_latency.total_seconds()
+      )
+      quality_info.append(f'src_wifi_connection: {src_connection_latency}s')
+    if self._current_test_result.advertiser_wifi_expected:
+      tgt_connection_latency = round(
+          self._current_test_result.advertiser_sta_latency.total_seconds()
+      )
+      quality_info.append(f'tgt_wifi_connection: {tgt_connection_latency}s')
+
+    test_report = {
+        'result': result_message,
+        'quality_info': quality_info,
+    }
+
+    self.discoverer.log.info(test_report)
+    self.record_data({
+        'Test Class': self.TAG,
+        'Test Name': self.current_test_info.name,
+        'sponge_properties': test_report,
+    })
+
+  def _get_current_test_result_message(self) -> str:
+    if (
+        self._active_nc_fail_reason
+        == nc_constants.SingleTestFailureReason.SUCCESS
+    ):
+      return 'PASS'
+    if (
+        self._active_nc_fail_reason
+        == nc_constants.SingleTestFailureReason.SOURCE_WIFI_CONNECTION
+    ):
+      return ''.join([
+          f'FAIL: {self._active_nc_fail_reason.name} - ',
+          nc_constants.COMMON_TRIAGE_TIP.get(
+              self._active_nc_fail_reason
+          ),
+      ])
+
+    if (self._use_prior_bt and
+        self._prior_bt_nc_fail_reason
+        is not nc_constants.SingleTestFailureReason.SUCCESS):
+      return ''.join([
+          'FAIL (The prior BT connection): ',
+          f'{self._prior_bt_nc_fail_reason.name} - ',
+          nc_constants.COMMON_TRIAGE_TIP.get(
+              self._prior_bt_nc_fail_reason
+          ),
+      ])
+
+    if (
+        self._active_nc_fail_reason
+        is nc_constants.SingleTestFailureReason.WIFI_MEDIUM_UPGRADE
+    ):
+      return ''.join([
+          f'FAIL: {self._active_nc_fail_reason.name} - ',
+          self._get_medium_upgrade_failure_tip(),
+      ])
+    if (
+        self._active_nc_fail_reason
+        is nc_constants.SingleTestFailureReason.FILE_TRANSFER_FAIL
+    ):
+      return ''.join([
+          f'{self._active_nc_fail_reason.name} - ',
+          self._get_file_transfer_failure_tip(),
+      ])
+    if (
+        self._active_nc_fail_reason
+        is nc_constants.SingleTestFailureReason.FILE_TRANSFER_THROUGHPUT_LOW
+    ):
+      return ''.join([
+          f'{self._active_nc_fail_reason.name} - ',
+          self._get_throughput_low_tip(),
+      ])
+
+    return ''.join([
+        f'{self._active_nc_fail_reason.name} - ',
+        nc_constants.COMMON_TRIAGE_TIP.get(self._active_nc_fail_reason),
+    ])
+
+  def _get_medium_upgrade_failure_tip(self) -> str:
+    return nc_constants.MEDIUM_UPGRADE_FAIL_TRIAGE_TIPS.get(
+        self._upgrade_medium_under_test,
+        f'unexpected upgrade medium - {self._upgrade_medium_under_test}')
+
+  @abc.abstractmethod
+  def _get_file_transfer_failure_tip(self) -> str:
+    pass
+
+  @abc.abstractmethod
+  def _get_throughput_low_tip(self) -> str:
+    pass
+
+  def _collect_current_test_metrics(self) -> None:
+    """Collects test result metrics for each iteration."""
+    if self._use_prior_bt:
+      self._performance_test_metrics.prior_bt_discovery_latencies.append(
+          self._current_test_result.prior_nc_setup_quality_info.discovery_latency
+      )
+      self._performance_test_metrics.prior_bt_connection_latencies.append(
+          self._current_test_result.prior_nc_setup_quality_info.connection_latency
+      )
+
+    self._performance_test_metrics.file_transfer_discovery_latencies.append(
+        self._current_test_result.file_transfer_nc_setup_quality_info.discovery_latency
+    )
+    self._performance_test_metrics.file_transfer_connection_latencies.append(
+        self._current_test_result.file_transfer_nc_setup_quality_info.connection_latency
+    )
+    self._performance_test_metrics.upgraded_wifi_transfer_mediums.append(
+        self._current_test_result.file_transfer_nc_setup_quality_info.upgrade_medium
+    )
+    self._performance_test_metrics.file_transfer_throughputs_kbps.append(
+        self._current_test_result.file_transfer_throughput_kbps
+    )
+    self._performance_test_metrics.discoverer_wifi_sta_latencies.append(
+        self._current_test_result.discoverer_sta_latency
+    )
+    self._performance_test_metrics.advertiser_wifi_sta_latencies.append(
+        self._current_test_result.advertiser_sta_latency
+    )
+    if self._current_test_result.file_transfer_nc_setup_quality_info.medium_upgrade_expected:
+      self._performance_test_metrics.medium_upgrade_latencies.append(
+          self._current_test_result.file_transfer_nc_setup_quality_info.medium_upgrade_latency
+      )
+
+  def __get_median_throughput(
+      self,
+      throughput_indicators: list[float],
+  ) -> tuple[int, int]:
+    """get the median throughput from iterations which finished file transfer."""
+    filtered = [
+        x
+        for x in throughput_indicators
+        if x != nc_constants.UNSET_THROUGHPUT_KBPS
+    ]
+    if not filtered:
+      # all test cases are failed
+      return (0, 0)
+    # use the descenting order of the throughput
+    filtered.sort(reverse=True)
+    return (len(filtered),
+            int(filtered[int(len(filtered) * nc_constants.PERCENTILE_50_FACTOR)]
+                ))
+
+  def __get_median_latency(
+      self, latency_indicators: list[datetime.timedelta]
+  ) -> tuple[int, float]:
+    filtered = [
+        latency.total_seconds()
+        for latency in latency_indicators
+        if latency != nc_constants.UNSET_LATENCY
+    ]
+    if not filtered:
+      # All test cases are failed.
+      return (0, 0.0)
+
+    filtered.sort()
+
+    percentile_50 = round(
+        filtered[int(len(filtered) * nc_constants.PERCENTILE_50_FACTOR)],
+        nc_constants.LATENCY_PRECISION_DIGITS,
+    )
+    return (len(filtered), percentile_50)
+
+  # @typing.override
+  def _summary_test_results(self) -> None:
+    """Summarizes test results of all iterations."""
+    success_count = sum(
+        test_result.failure_reason
+        == nc_constants.SingleTestFailureReason.SUCCESS
+        for test_result in self._test_results
+    )
+    # round down the passing test iterations
+    passed = success_count >= int(
+        self.performance_test_iterations * nc_constants.SUCCESS_RATE_TARGET
+    )
+    final_result_message = (
+        'PASS' if passed else f'FAIL: low successes - {success_count}'
+    )
+    detailed_stats = [
+        f'Required Iterations: {self.performance_test_iterations}',
+        f'Finished Iterations: {len(self._test_results)}']
+    detailed_stats.append('Failed Iterations:')
+    detailed_stats.extend(self.__get_failed_iteration_messages())
+    detailed_stats.append('File Transfer Connection Stats:')
+    detailed_stats.extend(self.__get_file_transfer_connection_stats())
+
+    if self._use_prior_bt:
+      detailed_stats.append('Prior BT Connection Stats:')
+      detailed_stats.extend(self.__get_prior_bt_connection_stats())
+
+    self.record_data({
+        'Test Class': self.TAG,
+        'sponge_properties': {
+            '01_test_result': final_result_message,
+            '02_source_device': '\n'.join(
+                self.__get_device_attributes(self.discoverer)
+            ),
+            '03_target_device': '\n'.join(
+                self.__get_device_attributes(self.advertiser)
+            ),
+            '04_detailed_stats': '\n'.join(detailed_stats),
+        },
+    })
+
+    asserts.assert_true(passed, final_result_message)
+
+  def __get_failed_iteration_messages(self) -> list[str]:
+    stats = []
+    for test_result in self._test_results:
+      if (
+          test_result.failure_reason
+          is not nc_constants.SingleTestFailureReason.SUCCESS
+      ):
+        stats.append(f'  - {test_result.test_iteration}: '
+                     f'{test_result.result_message}')
+
+    if stats:
+      return stats
+    else:
+      return ['  - NA']
+
+  def __get_prior_bt_connection_stats(self) -> list[str]:
+    if not self._use_prior_bt:
+      return []
+    (prior_bt_discovery_successes, prior_bt_discovery_median_latency) = (
+        self.__get_median_latency(
+            self._performance_test_metrics.prior_bt_discovery_latencies
+        )
+    )
+    (prior_bt_connection_successes, prior_bt_connection_median_latency) = (
+        self.__get_median_latency(
+            self._performance_test_metrics.prior_bt_connection_latencies
+        )
+    )
+    return [
+        f'  - Median Discovery Latency ({prior_bt_discovery_successes}): '
+        f'{prior_bt_discovery_median_latency}s',
+        f'  - Median Connection Latency ({prior_bt_connection_successes}): '
+        f'{prior_bt_connection_median_latency}s',
+    ]
+
+  def __get_file_transfer_connection_stats(self) -> list[str]:
+    (active_discovery_successes, active_discovery_median_latency) = (
+        self.__get_median_latency(
+            self._performance_test_metrics.file_transfer_discovery_latencies)
+    )
+    (active_connection_successes, active_connection_median_latency) = (
+        self.__get_median_latency(
+            self._performance_test_metrics.file_transfer_connection_latencies)
+    )
+    (file_transfer_successes, file_transfer_median_throughput) = (
+        self.__get_median_throughput(
+            self._performance_test_metrics.file_transfer_throughputs_kbps)
+        )
+    stats = [
+        f'  - Median Discovery Latency ({active_discovery_successes}): '
+        f'{active_discovery_median_latency}s',
+        f'  - Median Connection Latency ({active_connection_successes}): '
+        f'{active_connection_median_latency}s',
+        f'  - Median File Transfer Speed ({file_transfer_successes}): '
+        f'{round(file_transfer_median_throughput/1024, 1)}MBps',
+    ]
+    if nc_constants.is_high_quality_medium(self._upgrade_medium_under_test):
+      (medium_upgrade_successes, medium_upgrade_median_latency) = (
+          self.__get_median_latency(
+              self._performance_test_metrics.medium_upgrade_latencies)
+      )
+      stats.extend([
+          f'  - Median Upgrade Latency ({medium_upgrade_successes}): '
+          f'{medium_upgrade_median_latency}s',
+          '  - Upgrade Medium Stats:'
+      ])
+      stats.extend(self._summary_upgraded_wifi_transfer_mediums())
+
+    return stats
+
+  def _summary_upgraded_wifi_transfer_mediums(self) -> list[str]:
+    medium_counts = {}
+    for (
+        upgraded_medium
+    ) in self._performance_test_metrics.upgraded_wifi_transfer_mediums:
+      if upgraded_medium:
+        medium_counts[upgraded_medium.name] = (
+            medium_counts.get(upgraded_medium.name, 0) + 1
+        )
+    return [f'    - {name}: {count}' for name, count in medium_counts.items()]
+
+  def __get_device_attributes(
+      self, ad: android_device.AndroidDevice
+  ) -> list[str]:
+    return [
+        f'Device Serial: {ad.serial}',
+        f'Device Model: {ad.model}',
+        f'Supports 5G Wifi: {ad.supports_5g}',
+        f'Supports DBS: {ad.supports_dbs_sta_wfd}',
+        (
+            'Enable STA DFS channel for peer network:'
+            f' {ad.enable_sta_dfs_channel_for_peer_network}'
+        ),
+        (
+            'Enable STA Indoor channel for peer network:'
+            f' {ad.enable_sta_indoor_channel_for_peer_network}'
+        ),
+        f'Max num of streams: {ad.max_num_streams}',
+        f'Max num of streams (DBS): {ad.max_num_streams_dbs}',
+        f'Android Version: {ad.android_version}',
+        f'GMS_version: {setup_utils.dump_gms_version(ad)}',
+    ]
diff --git a/tests/bettertogether/betocq/directed_tests/ble_performance_test.py b/tests/bettertogether/betocq/directed_tests/ble_performance_test.py
new file mode 100644
index 000000000..8db2f10b7
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/ble_performance_test.py
@@ -0,0 +1,96 @@
+#  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.
+
+"""This Test is to test the BLE performance."""
+
+import datetime
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class BlePerformanceTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for the BLE connection performance."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self.performance_test_iterations = getattr(
+        self.test_ble_performance, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.BT_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.BT_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_ble_performance(self):
+    """Test the performance of the BLE."""
+    self._test_connection_medium_performance(
+        upgrade_medium_under_test=nc_constants.NearbyMedium.BLE_ONLY,
+        force_disable_bt_multiplex=True,
+        connection_medium=nc_constants.NearbyMedium.BLE_ONLY,
+    )
+
+  def _get_transfer_file_size(self) -> int:
+    return nc_constants.TRANSFER_FILE_SIZE_1MB
+
+  def _get_file_transfer_timeout(self) -> datetime.timedelta:
+    return nc_constants.BLE_1M_PAYLOAD_TRANSFER_TIMEOUT
+
+  def _get_throughout_benchmark(self) -> int:
+    return nc_constants.BLE_MEDIUM_THROUGHPUT_BENCHMARK
+
+  def _get_medium_upgrade_failure_tip(self) -> str:
+    return 'Not Applied'  # No medium upgrade required for BLE.
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The BLE connection might be broken, check the related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        f'{self._throughput_low_string}. Check with the chip vendor if there is'
+        ' any BT firmware issue.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    # don't require wifi STA.
+    return True
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    # no special capabilities is required.
+    return True
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/bt_performance_test.py b/tests/bettertogether/betocq/directed_tests/bt_performance_test.py
new file mode 100644
index 000000000..689e001b7
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/bt_performance_test.py
@@ -0,0 +1,95 @@
+#  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.
+
+"""This Test is to test the classic Bluetooth performance."""
+
+import datetime
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class BtPerformanceTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for the classic Bluetooth connection performance."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self.performance_test_iterations = getattr(
+        self.test_classic_bt_performance, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.BT_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.BT_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_classic_bt_performance(self):
+    """Test the performance of the classic BT connetion."""
+    self._test_connection_medium_performance(
+        upgrade_medium_under_test=nc_constants.NearbyMedium.BT_ONLY,
+        force_disable_bt_multiplex=True
+    )
+
+  def _get_transfer_file_size(self) -> int:
+    return nc_constants.TRANSFER_FILE_SIZE_1MB
+
+  def _get_file_transfer_timeout(self) -> datetime.timedelta:
+    return nc_constants.BT_1M_PAYLOAD_TRANSFER_TIMEOUT
+
+  def _get_throughout_benchmark(self) -> int:
+    return nc_constants.CLASSIC_BT_MEDIUM_THROUGHPUT_BENCHMARK
+
+  def _get_medium_upgrade_failure_tip(self) -> str:
+    return 'Not Applied'  # No medium upgrade required for BT.
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The classic Bluetooth connection might be broken, check related log, '
+        f' {self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        f'{self._throughput_low_string}. Check with the chip vendor if there is'
+        ' any BT firmware issue.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    # don't require wifi STA.
+    return True
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    # no special capabilities is required.
+    return True
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/local_dev_directed_testbed.yml b/tests/bettertogether/betocq/directed_tests/local_dev_directed_testbed.yml
new file mode 100644
index 000000000..ea3e03016
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/local_dev_directed_testbed.yml
@@ -0,0 +1,77 @@
+x-test-params: &test-params
+  target_cuj_name: "quick_start"
+  allow_unrooted_device: False
+  wifi_2g_ssid: "NETGEAR62"
+  wifi_2g_password: "widetree588"
+  wifi_5g_ssid: "NETGEAR62-5G-1"
+  wifi_5g_password: "widetree588"
+  wifi_dfs_5g_ssid: "GoogleGuest"
+  wifi_dfs_5g_password: ""
+  # use the AP controlled by programming dynamically
+  use_auto_controlled_wifi_ap: False
+
+x-controllers: &controllers
+  Controllers:
+    AndroidDevice:
+      - serial: "R3CN90YNAR"  # "R3CN90YNAR"
+        role: "source_device"
+        # The max number of spatial streams
+        max_num_streams: 2
+        # The max PHY rate at 5G, in Mbps
+        max_phy_rate_5g_mbps: 2402
+        # The max PHY rate at 2G, in Mbps
+        max_phy_rate_2g_mbps: 287
+        # if the device supports 5G Wifi
+        supports_5g: True  # True
+        # if the device supports DBS (Dual Band Simultaneous) in STA + WiFi-Direct concurrency mode
+        supports_dbs_sta_wfd: True
+        # The max number of spatial streams in DBS mode.
+        max_num_streams_dbs: 2
+        # if the device supports to start WFD group owner at a STA-associated DFS channel
+        enable_sta_dfs_channel_for_peer_network: False
+        # if the device supports to start WFD group owner at a STA-associated indoor channel
+        enable_sta_indoor_channel_for_peer_network: False
+        # 14 - U, 13 - T, 12 - S
+        android_version: 14
+      - serial: "2B031FDJH0002G"  # "ABCDEF123456"
+        role: "target_device"
+        # The max number of spatial streams
+        max_num_streams: 2
+        # The max PHY rate at 5G, in Mbps
+        max_phy_rate_5g_mbps: 2402
+        # The max PHY rate at 2G, in Mbps
+        max_phy_rate_2g_mbps: 287
+        # if the device supports 5G Wifi
+        supports_5g: True  # True
+        # if the device supports DBS (Dual Band Simultaneous) in STA + WiFi-Direct concurrency mode
+        supports_dbs_sta_wfd: True
+        # The max number of spatial streams in DBS mode.
+        max_num_streams_dbs: 2
+        # if the device supports to start WFD group owner at a STA-associated DFS channel
+        enable_sta_dfs_channel_for_peer_network: True
+        # if the device supports to start WFD group owner at a STA-associated indoor channel
+        enable_sta_indoor_channel_for_peer_network: True
+        # 14 - U, 13 - T, 12 - S
+        android_version: 14
+
+TestBeds:
+- Name: LocalPerformanceDefaultTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
+
+- Name: LocalQuickstartPerformanceTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
+      target_cuj_name: "quick_start"
+      # before the performance test, run the function tests first
+      run_function_tests_with_performance_tests: True
+      # if the function tests is failed, abort the performance test
+      abort_all_tests_on_function_tests_fail: True
+
+- Name: LocalQuicksharePerformanceTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
+      target_cuj_name: "quick_share"
diff --git a/tests/bettertogether/betocq/directed_tests/mcc_2g_wfd_indoor_5g_sta_test.py b/tests/bettertogether/betocq/directed_tests/mcc_2g_wfd_indoor_5g_sta_test.py
new file mode 100644
index 000000000..6a263a9a3
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/mcc_2g_wfd_indoor_5g_sta_test.py
@@ -0,0 +1,111 @@
+#  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.
+
+"""This Test is to test the Wifi MCC with the indoor channels case.
+
+This is about the feature - using indoor channels for WFD, for details, refer to
+https://docs.google.com/presentation/d/18Fl0fY4piq_sfXfo3rCr2Ca55AJHEOvB7rC-rV3SQ9E/edit?usp=sharing
+and config_wifiEnableStaIndoorChannelForPeerNetwork -
+https://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Wifi/service/ServiceWifiResources/res/values/config.xml;l=1147
+In this case, the feature is disable for the device; The WFD will be started on
+a 2G channel, but the STA is using the 5G channel.
+
+The device requirements:
+  support 5G: true
+  using indoor channels for peer network: false
+The AP requirements:
+  wifi channel: 36 (5180)
+"""
+
+import datetime
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Mcc2gWfdIndoor5gStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for wifi MCC with 2G WFD and indoor 5G STA."""
+
+  def _get_country_code(self) -> str:
+    return 'JP'
+
+  def setup_class(self):
+    super().setup_class()
+    self._is_mcc = True
+    self._is_2g_d2d_wifi_medium = True
+    self.performance_test_iterations = getattr(
+        self.test_mcc_2g_wfd_indoor_5g_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_mcc_2g_wfd_indoor_5g_sta(self):
+    """Test the performance for wifi MCC with 2G WFD and indoor 5G STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIDIRECT,
+        wifi_ssid=self.test_parameters.wifi_5g_ssid,
+        wifi_password=self.test_parameters.wifi_5g_password,
+    )
+
+  def _get_transfer_file_size(self) -> int:
+    # For 2G wifi medium
+    return nc_constants.TRANSFER_FILE_SIZE_20MB
+
+  def _get_file_transfer_timeout(self) -> datetime.timedelta:
+    return nc_constants.WIFI_2G_20M_PAYLOAD_TRANSFER_TIMEOUT
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The Wifi Direct connection might be broken, check related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        f'{self._throughput_low_string}. This is a MCC test case where WFD uses'
+        ' a 2G channel and the STA uses a 5G indoor channel. Check with the'
+        ' wifi chip vendor about the possible firmware Tx/Rx issues in MCC'
+        ' mode.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_5g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return (
+        self.discoverer.supports_5g
+        and self.advertiser.supports_5g
+        and not self.advertiser.enable_sta_indoor_channel_for_peer_network
+    )
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/mcc_5g_hotspot_dfs_5g_sta_test.py b/tests/bettertogether/betocq/directed_tests/mcc_5g_hotspot_dfs_5g_sta_test.py
new file mode 100644
index 000000000..93f7bee6d
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/mcc_5g_hotspot_dfs_5g_sta_test.py
@@ -0,0 +1,106 @@
+#  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.
+
+"""This Test is to test the Wifi MCC with the DFS channels case.
+
+This is about the feature - using DFS channels for Hotspot, for details, refer
+to
+https://docs.google.com/presentation/d/18Fl0fY4piq_sfXfo3rCr2Ca55AJHEOvB7rC-rV3SQ9E/edit?usp=sharing
+andconfig_wifiEnableStaDfsChannelForPeerNetwork -
+https://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Wifi/service/ServiceWifiResources/res/values/config.xml;l=1151
+In this case, the feature is disable for the device; The WLAN is using the DFS
+5G channel, but the hotspot will be started on another non DFS 5G channel.
+
+The device requirements:
+  support 5G: true
+  using DFS channels for peer network (target device): false
+The AP requirements:
+  wifi channel: 52 (5260)
+  TODO(kaishi): verify DFS channel.
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Mcc5gHotspotDfs5gStaTest(
+    d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for MCC with 5G HOTSPOT and DFS 5G STA."""
+
+  def _get_country_code(self) -> str:
+    return 'GB'
+
+  def setup_class(self):
+    super().setup_class()
+    self._is_mcc = True
+    self.performance_test_iterations = getattr(
+        self.test_mcc_5g_hotspot_dfs_5g_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.MCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.MCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_mcc_5g_hotspot_dfs_5g_sta(self):
+    """Test the performance for wifi MCC with 5G HOTSPOT and DFS 5G STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIHOTSPOT,
+        wifi_ssid=self.test_parameters.wifi_dfs_5g_ssid,
+        wifi_password=self.test_parameters.wifi_dfs_5g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The hotspot connection might be broken, check the related log, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        f'{self._throughput_low_string}. This is a MCC test case where hotspot'
+        ' uses a 5G non-DFS channel and STA uses a5G DFS channel. Note that in'
+        ' hotspot mode, the target acts as a WFD GOwhile the source device'
+        ' acts as the legcy STA. Check with the wifi chip vendorabout the'
+        ' possible firmware Tx/Rx issues in MCC mode.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_dfs_5g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return (
+        self.discoverer.supports_5g
+        and self.advertiser.supports_5g
+        and not self.advertiser.enable_sta_dfs_channel_for_peer_network
+    )
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/mcc_5g_wfd_dfs_5g_sta_test.py b/tests/bettertogether/betocq/directed_tests/mcc_5g_wfd_dfs_5g_sta_test.py
new file mode 100644
index 000000000..bf3444060
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/mcc_5g_wfd_dfs_5g_sta_test.py
@@ -0,0 +1,101 @@
+#  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.
+
+"""This Test is to test the Wifi MCC with the DFS channels case.
+
+This is about the feature - using DFS channels for WFD, for details, refer to
+https://docs.google.com/presentation/d/18Fl0fY4piq_sfXfo3rCr2Ca55AJHEOvB7rC-rV3SQ9E/edit?usp=sharing
+and config_wifiEnableStaDfsChannelForPeerNetwork -
+https://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Wifi/service/ServiceWifiResources/res/values/config.xml;l=1151
+In this case, the feature is disable for the device; The STA is using the DFS 5G
+channel, but the WFD will be started on another 5G channel.
+
+The device requirements:
+  support 5G: true
+  using DFS channels for peer network: false
+The AP requirements:
+  wifi channel: 52 (5260)
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Mcc5gWfdDfs5gStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for MCC with 5G WFD and DFS 5G STA."""
+
+  def _get_country_code(self) -> str:
+    return 'GB'
+
+  def setup_class(self):
+    super().setup_class()
+    self._is_mcc = True
+    self.performance_test_iterations = getattr(
+        self.test_mcc_5g_wfd_dfs_5g_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.MCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.MCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_mcc_5g_wfd_dfs_5g_sta(self):
+    """Test the performance for wifi MCC with 5G WFD and DFS 5G STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIDIRECT,
+        wifi_ssid=self.test_parameters.wifi_dfs_5g_ssid,
+        wifi_password=self.test_parameters.wifi_dfs_5g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The Wifi Direct connection might be broken, check the related log, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        'f{self._throughput_low_string}. This is a MCC test case where WFD uses'
+        ' a 5G non-DFS channel and STA uses a5G DFS channel. Check with the'
+        ' wifi chip vendorabout the possible firmware Tx/Rx issues in MCC mode.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_dfs_5g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return (
+        self.discoverer.supports_5g
+        and self.advertiser.supports_5g
+        and not self.advertiser.enable_sta_dfs_channel_for_peer_network
+    )
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/mcc_5g_wfd_non_dbs_2g_sta_test.py b/tests/bettertogether/betocq/directed_tests/mcc_5g_wfd_non_dbs_2g_sta_test.py
new file mode 100644
index 000000000..2ab994988
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/mcc_5g_wfd_non_dbs_2g_sta_test.py
@@ -0,0 +1,100 @@
+#  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.
+
+"""This Test is to test the Wifi MCC in a general case.
+
+In this case, the WFD is using the 5G channel, but STA is connected to 2G
+channel, as the device(don't support DBS) can not handle the 5G and 2G at
+the same time, there is concurrent contention for the 5G channel and 2G
+channel handling in firmware, the firmware needs to switch 5G and 2G from time
+to time.
+
+The device requirements:
+  support 5G: true
+  support DBS(Target Device): False
+The AP requirements:
+  wifi channel: 6 (2437)
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Mcc5gWfdNonDbs2gStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for MCC case with 5G WFD and 2G STA."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self._is_mcc = True
+    self.performance_test_iterations = getattr(
+        self.test_mcc_5g_wfd_non_dbs_2g_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.MCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.MCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_mcc_5g_wfd_non_dbs_2g_sta(self):
+    """Test the performance for wifi MCC with 5G WFD and 2G STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIDIRECT,
+        wifi_ssid=self.test_parameters.wifi_2g_ssid,
+        wifi_password=self.test_parameters.wifi_2g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The Wifi Direct connection might be broken, check related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        f'{self._throughput_low_string}.'
+        'This is a MCC test case where WFD uses a 5G channel and STA uses a'
+        '2G channel. Check with the wifi chip vendor'
+        'about the possible firmware Tx/Rx issues in MCC mode.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_2g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return (
+        self.discoverer.supports_5g and self.advertiser.supports_5g
+        and not self.advertiser.supports_dbs_sta_wfd
+    )
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/scc_2g_wfd_sta_test.py b/tests/bettertogether/betocq/directed_tests/scc_2g_wfd_sta_test.py
new file mode 100644
index 000000000..7bb56f1fa
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/scc_2g_wfd_sta_test.py
@@ -0,0 +1,100 @@
+#  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.
+
+"""This Test is to test the Wifi SCC in a general case.
+
+In this case, both the WFD and WLAN are using the 2G channel, the WFD and WLAN
+should use the same channel.
+
+The device requirements:
+  support 5G: false
+The AP requirements:
+  wifi channel: 6 (2437)
+"""
+
+import datetime
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Scc2gWfdStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for Wifi SCC with 2G WFD and STA."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self._is_2g_d2d_wifi_medium = True
+    self.performance_test_iterations = getattr(
+        self.test_scc_2g_wfd_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_2g_wfd_sta(self):
+    """Test the performance for Wifi SCC with 2G WFD and STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIDIRECT,
+        wifi_ssid=self.test_parameters.wifi_2g_ssid,
+        wifi_password=self.test_parameters.wifi_2g_password,
+    )
+
+  def _get_transfer_file_size(self) -> int:
+    # For 2G wifi medium
+    return nc_constants.TRANSFER_FILE_SIZE_20MB
+
+  def _get_file_transfer_timeout(self) -> datetime.timedelta:
+    return nc_constants.WIFI_2G_20M_PAYLOAD_TRANSFER_TIMEOUT
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The Wifi Direct connection might be broken, check related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        'f{self._throughput_low_string}.'
+        'This is a SCC 2G test case with WFD medium. Check with the wifi chip'
+        'vendor about the possible firmware Tx/Rx issues in this mode.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_2g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return not self.discoverer.supports_5g or not self.advertiser.supports_5g
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/scc_2g_wlan_sta_test.py b/tests/bettertogether/betocq/directed_tests/scc_2g_wlan_sta_test.py
new file mode 100644
index 000000000..8e3dccc97
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/scc_2g_wlan_sta_test.py
@@ -0,0 +1,103 @@
+#  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.
+
+"""This Test is to test the Wifi LAN 2G SCC in a general case.
+
+In this case, both the Wifi LAN and STA(internet AP) are using the 2G channel,
+the Wifi LAN and STA should use the same channel.
+Wifi LAN - The target device starts WFD GO as the role of AP, the source device
+is connected to the AP.
+
+The device requirements:
+  support 5G: false
+The AP requirements:
+  wifi channel: 6 (2437)
+"""
+
+import datetime
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Scc2gWlanStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for Wifi SCC with 2G WifiLAN and STA."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self._is_2g_d2d_wifi_medium = True
+    self.performance_test_iterations = getattr(
+        self.test_scc_2g_wifilan_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_2g_wifilan_sta(self):
+    """Test the performance for Wifi SCC with 2G WifiLAN and STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.WIFILAN_ONLY,
+        wifi_ssid=self.test_parameters.wifi_2g_ssid,
+        wifi_password=self.test_parameters.wifi_2g_password,
+    )
+
+  def _get_transfer_file_size(self) -> int:
+    # For 2G wifi medium
+    return nc_constants.TRANSFER_FILE_SIZE_20MB
+
+  def _get_file_transfer_timeout(self) -> datetime.timedelta:
+    return nc_constants.WIFI_2G_20M_PAYLOAD_TRANSFER_TIMEOUT
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The WLAN connection might be broken, check related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        f'{self._throughput_low_string}.'
+        'This is a SCC 2G test case with WLAN medium. Check with the wifi chip'
+        'vendor if TDLS is supported correctly. Also check if'
+        'the AP has the firewall which could block the mDNS traffic.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_2g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return not self.discoverer.supports_5g or not self.advertiser.supports_5g
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/scc_5g_wfd_dbs_2g_sta_test.py b/tests/bettertogether/betocq/directed_tests/scc_5g_wfd_dbs_2g_sta_test.py
new file mode 100644
index 000000000..1e7b50517
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/scc_5g_wfd_dbs_2g_sta_test.py
@@ -0,0 +1,99 @@
+#  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.
+
+"""This Test is to test the Wifi MCC in a general case.
+
+In this case, even though the STA is connected to a 2G channel, and the WFD is
+using the 5G channel, as both devices support DBS, which can handle the 5G and
+2G at the same time, this is still a SCC case.
+
+The device requirements:
+  support 5G: true
+  support DBS(target): true
+The AP requirements:
+  wifi channel: 6 (2437)
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Scc5gWfdDbs2gStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for SCC with 5G WFD and 2G STA."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self._is_dbs_mode = True
+    self.performance_test_iterations = getattr(
+        self.test_scc_5g_wfd_dbs_2g_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_5g_wfd_dbs_2g_sta(self):
+    """Test the performance for wifi SCC with 5G WFD and 2G STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIDIRECT,
+        wifi_ssid=self.test_parameters.wifi_2g_ssid,
+        wifi_password=self.test_parameters.wifi_2g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The Wifi Direct connection might be broken, check related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        f'{self._throughput_low_string}. This is a SCC 5G test case with WFD'
+        ' medium operating at 5G and STA operatingat 2G. In the configuration'
+        ' file, DBS support is set to true. Check if thedevice does support'
+        ' DBS with STA + WFD concurrency. Check with the wifi chipvendor about'
+        ' the possible firmware Tx/Rx issues in this mode.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_2g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return (
+        self.discoverer.supports_5g and self.advertiser.supports_5g
+        and self.advertiser.supports_dbs_sta_wfd
+    )
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/scc_5g_wfd_sta_test.py b/tests/bettertogether/betocq/directed_tests/scc_5g_wfd_sta_test.py
new file mode 100644
index 000000000..7bc46b83f
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/scc_5g_wfd_sta_test.py
@@ -0,0 +1,92 @@
+#  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.
+
+"""This Test is to test the Wifi SCC in a general case.
+
+In this case, both the WFD and WLAN are using the 5G channel, the WFD and WLAN
+should use the same channel.
+
+The device requirements:
+  support 5G: true
+The AP requirements:
+  wifi channel: 36 (5180)
+"""
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Scc5gWfdStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for Wifi SCC with 5G WFD and STA."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self.performance_test_iterations = getattr(
+        self.test_scc_5g_wfd_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_5g_wfd_sta(self):
+    """Test the performance for Wifi SCC with 5G WFD and STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIDIRECT,
+        wifi_ssid=self.test_parameters.wifi_5g_ssid,
+        wifi_password=self.test_parameters.wifi_5g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The Wifi Direct connection might be broken, check related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        'f{self._throughput_low_string}. This is a SCC 5G test case with WFD'
+        ' and STA operating at the same 5G channel.Check with the wifi chip'
+        ' vendor about the possible firmware Tx/Rx issues inthis mode. Also'
+        ' check if the AP channel is set correctly and is supported bythe used'
+        ' wifi medium.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_5g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return self.discoverer.supports_5g and self.advertiser.supports_5g
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/scc_5g_wlan_sta_test.py b/tests/bettertogether/betocq/directed_tests/scc_5g_wlan_sta_test.py
new file mode 100644
index 000000000..14cec0bff
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/scc_5g_wlan_sta_test.py
@@ -0,0 +1,91 @@
+#  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.
+
+"""This Test is to test the Wifi SCC of LAN in a general case.
+
+In this case, both the D2D_LAN and WLAN are using the 5G channel, the D2d_LAN
+and WLAN should use the same channel.
+
+The device requirements:
+  support 5G: true
+The AP requirements:
+  wifi channel: 36 (5180)
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class Scc5gWifiLanStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for Wifi SCC with 5G WifiLAN and STA."""
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def setup_class(self):
+    super().setup_class()
+    self.performance_test_iterations = getattr(
+        self.test_scc_5g_wifilan_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_5g_wifilan_sta(self):
+    """Test the performance for Wifi SCC with 5G WifiLAN and STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.WIFILAN_ONLY,
+        wifi_ssid=self.test_parameters.wifi_5g_ssid,
+        wifi_password=self.test_parameters.wifi_5g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The WLAN connection might be broken, check related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        f'{self._throughput_low_string}. This is 5G WLAN test case. Check with'
+        ' the wifi chip vendor if TDLS issupported correctly. Also check if'
+        ' the AP has the firewall which could blockthe mDNS traffic.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_5g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return self.discoverer.supports_5g and self.advertiser.supports_5g
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/scc_dfs_5g_hotspot_sta_test.py b/tests/bettertogether/betocq/directed_tests/scc_dfs_5g_hotspot_sta_test.py
new file mode 100644
index 000000000..36599c8f4
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/scc_dfs_5g_hotspot_sta_test.py
@@ -0,0 +1,103 @@
+#  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.
+
+"""This Test is to test the Wifi SCC with the DFS channels case.
+
+This is about the feature - using DFS channels for Hotspot, for details, refer
+to
+https://docs.google.com/presentation/d/18Fl0fY4piq_sfXfo3rCr2Ca55AJHEOvB7rC-rV3SQ9E/edit?usp=sharing
+and config_wifiEnableStaDfsChannelForPeerNetwork -
+https://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Wifi/service/ServiceWifiResources/res/values/config.xml;l=1151
+In this case, the feature is enabled for the target device.
+
+The device requirements:
+  support 5G: true
+  using DFS channels for peer network(target device): true
+The AP requirements:
+  wifi channel: 52 (5260)
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class SccDfs5gHotspotStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for SCC with Hotspot on a 5G DFS channel."""
+
+  def _get_country_code(self) -> str:
+    return 'GB'
+
+  def setup_class(self):
+    super().setup_class()
+    self.performance_test_iterations = getattr(
+        self.test_scc_dfs_5g_hotspot_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_dfs_5g_hotspot_sta(self):
+    """Test the performance for wifi SCC with DFS 5G Hotspot and STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIHOTSPOT,
+        wifi_ssid=self.test_parameters.wifi_dfs_5g_ssid,
+        wifi_password=self.test_parameters.wifi_dfs_5g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The Wifi Hotspot connection might be broken, check related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        f'{self._throughput_low_string}. This is 5G SCC DFS hotspot test case.'
+        ' In the configuration fileenable_sta_dfs_channel_for_peer_network is'
+        ' set to true.Check if the target device does support WFD group owner'
+        ' in the STA-associatedDFS channel. Check if'
+        ' config_wifiEnableStaDfsChannelForPeerNetworkhttps://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Wifi/service/ServiceWifiResources/res/values/config.xml;l=1151)is'
+        ' set to true and has the correct driver/FW implementation.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_5g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return (
+        self.discoverer.supports_5g
+        and self.advertiser.supports_5g
+        and self.advertiser.enable_sta_dfs_channel_for_peer_network
+    )
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/scc_dfs_5g_wfd_sta_test.py b/tests/bettertogether/betocq/directed_tests/scc_dfs_5g_wfd_sta_test.py
new file mode 100644
index 000000000..eb281ef00
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/scc_dfs_5g_wfd_sta_test.py
@@ -0,0 +1,103 @@
+#  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.
+
+"""This Test is to test the Wifi SCC with the DFS channels case.
+
+This is about the feature - using DFS channels for WFD, for details, refer to
+https://docs.google.com/presentation/d/18Fl0fY4piq_sfXfo3rCr2Ca55AJHEOvB7rC-rV3SQ9E/edit?usp=sharing
+and config_wifiEnableStaDfsChannelForPeerNetwork -
+https://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Wifi/service/ServiceWifiResources/res/values/config.xml;l=1151
+In this case, the feature is enabled for the target device.
+
+The device requirements:
+  support 5G: true
+  using DFS channels for peer network: true
+The AP requirements:
+  wifi channel: 52 (5260)
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class SccDfs5gWfdStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """Test class for Wifi SCC with the DFS 5G channels case."""
+
+  def _get_country_code(self) -> str:
+    return 'GB'
+
+  def setup_class(self):
+    super().setup_class()
+    self.performance_test_iterations = getattr(
+        self.test_scc_dfs_5g_wfd_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_dfs_5g_wfd_sta(self):
+    """Test the performance for Wifi SCC with DFS 5G WFD and STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIDIRECT,
+        wifi_ssid=self.test_parameters.wifi_dfs_5g_ssid,
+        wifi_password=self.test_parameters.wifi_dfs_5g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The Wifi Direct connection might be broken, check related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        'f{self._throughput_low_string}. This is 5G SCC DFS WFD test case. In'
+        ' the configuration file,enable_sta_dfs_channel_for_peer_network is set'
+        ' to true for both src/target.Check if both device do support WFD group'
+        ' owner in the STA-associatedDFS channel. Check if'
+        ' config_wifiEnableStaDfsChannelForPeerNetworkhttps://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Wifi/service/ServiceWifiResources/res/values/config.xml;l=1151),is'
+        ' set to true and has the correct driver/FW implementation.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_5g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return (
+        self.discoverer.supports_5g
+        and self.advertiser.supports_5g
+        and self.discoverer.enable_sta_dfs_channel_for_peer_network
+        and self.advertiser.enable_sta_dfs_channel_for_peer_network
+    )
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/directed_tests/scc_indoor_5g_wfd_sta_test.py b/tests/bettertogether/betocq/directed_tests/scc_indoor_5g_wfd_sta_test.py
new file mode 100644
index 000000000..71d163e77
--- /dev/null
+++ b/tests/bettertogether/betocq/directed_tests/scc_indoor_5g_wfd_sta_test.py
@@ -0,0 +1,103 @@
+#  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.
+
+"""This Test is to test the Wifi SCC with the indoor channels case.
+
+This is about the feature - using indoor channels for WFD, for details, refer to
+https://docs.google.com/presentation/d/18Fl0fY4piq_sfXfo3rCr2Ca55AJHEOvB7rC-rV3SQ9E/edit?usp=sharing
+and config_wifiEnableStaIndoorChannelForPeerNetwork -
+https://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Wifi/service/ServiceWifiResources/res/values/config.xml;l=1147
+In this case, the feature is enabled on both devices; the WFD and STA both are
+operating on the same 5G indoor channel.
+
+The device requirements:
+  support 5G: true
+  using indoor channels for WFD: true
+The AP requirements:
+  wifi channel: 36 (5180)
+"""
+
+import logging
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly  import base_test
+from mobly import test_runner
+
+from betocq import d2d_performance_test_base
+from betocq import nc_constants
+
+
+class SccIndoor5gWfdStaTest(d2d_performance_test_base.D2dPerformanceTestBase):
+  """This class is to test the Wifi SCC with the indoor channel case."""
+
+  def _get_country_code(self) -> str:
+    return 'JP'
+
+  def setup_class(self):
+    super().setup_class()
+    self.performance_test_iterations = getattr(
+        self.test_scc_indoor_5g_wfd_sta, base_test.ATTR_REPEAT_CNT
+    )
+    logging.info(
+        'performance test iterations: %s', self.performance_test_iterations
+    )
+
+  @base_test.repeat(
+      count=nc_constants.SCC_PERFORMANCE_TEST_COUNT,
+      max_consecutive_error=nc_constants.SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR,
+  )
+  def test_scc_indoor_5g_wfd_sta(self):
+    """Test the performance for Wifi SCC with 5G indoor WFD and 5G STA."""
+    self._test_connection_medium_performance(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIDIRECT,
+        wifi_ssid=self.test_parameters.wifi_5g_ssid,
+        wifi_password=self.test_parameters.wifi_5g_password,
+    )
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    return (
+        'The Wifi Direct connection might be broken, check related logs, '
+        f'{self._get_throughput_low_tip()}'
+    )
+
+  def _get_throughput_low_tip(self) -> str:
+    return (
+        'f{self._throughput_low_string}. This is 5G SCC indoor WFD test case.'
+        ' In the configuration file,enable_sta_indoor_channel_for_peer_network'
+        ' is set to true.Check if the target device does support WFD group'
+        ' owner in the STA-associatedindoor channel. Check if'
+        ' config_wifiEnableStaIndoorChannelForPeerNetworkhttps://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Wifi/service/ServiceWifiResources/res/values/config.xml;l=1147),is'
+        ' set to true and has the correct driver/FW implementation.'
+    )
+
+  def _is_wifi_ap_ready(self) -> bool:
+    return True if self.test_parameters.wifi_5g_ssid else False
+
+  def _are_devices_capabilities_ok(self) -> bool:
+    return (
+        self.discoverer.supports_5g
+        and self.advertiser.supports_5g
+        and self.advertiser.enable_sta_indoor_channel_for_peer_network
+    )
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/function_tests/beto_cq_function_group_test.py b/tests/bettertogether/betocq/function_tests/beto_cq_function_group_test.py
new file mode 100644
index 000000000..656ed95cc
--- /dev/null
+++ b/tests/bettertogether/betocq/function_tests/beto_cq_function_group_test.py
@@ -0,0 +1,139 @@
+#  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.
+
+"""Group all function tests."""
+
+import os
+import sys
+
+# Allows local imports to be resolved via relative path, so the test can be run
+# without building.
+_betocq_dir = os.path.dirname(os.path.dirname(__file__))
+if _betocq_dir not in sys.path:
+  sys.path.append(_betocq_dir)
+
+from mobly import test_runner
+
+from betocq import nc_base_test
+from betocq import nc_constants
+from betocq import setup_utils
+from betocq import version
+from betocq.function_tests import bt_ble_function_test_actor
+from betocq.function_tests import bt_multiplex_function_test_actor
+from betocq.function_tests import fixed_wifi_medium_function_test_actor
+from betocq.function_tests import function_test_actor_base
+
+
+class BetoCqFunctionGroupTest(nc_base_test.NCBaseTestClass):
+  """The test class to group all function tests in one mobly test."""
+
+  def __init__(self, configs):
+    super().__init__(configs)
+
+    self._test_result_messages: dict[str, str] = {}
+
+  def test_bt_ble_function(self):
+    """Test the NC with the BT/BLE medium only."""
+    self._current_test_actor = self.bt_ble_test_actor
+    self.bt_ble_test_actor.test_bt_ble_connection()
+
+  def test_wifilan_function(self):
+    """Test the NC with upgrading to the Wifi LAN medium.
+
+    step 1: connect to wifi
+    step 2: set up a nearby connection with the WifiLAN medium and transfer a
+    small file.
+    """
+    self._current_test_actor = self.fixed_wifi_medium_test_actor
+    self.fixed_wifi_medium_test_actor.connect_to_wifi()
+    self.fixed_wifi_medium_test_actor.run_fixed_wifi_medium_test(
+        nc_constants.NearbyMedium.WIFILAN_ONLY)
+
+  def test_d2d_hotspot_function(self):
+    """Test the NC with upgrading to the HOTSPOT as connection medium.
+    """
+    self._current_test_actor = self.fixed_wifi_medium_test_actor
+    self.fixed_wifi_medium_test_actor.run_fixed_wifi_medium_test(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIHOTSPOT)
+
+  def test_wifi_direct_function(self):
+    """Test the NC with upgrading to the WiFi Direct as connection medium.
+    """
+    self._current_test_actor = self.fixed_wifi_medium_test_actor
+    self.fixed_wifi_medium_test_actor.run_fixed_wifi_medium_test(
+        nc_constants.NearbyMedium.UPGRADE_TO_WIFIDIRECT)
+
+  def test_bt_multiplex_connections(self):
+    """Test the BT multiplex function of nearby connection.
+
+    set up 2 Bluetooth connections with NearbyConnection APIs.
+    """
+    self._current_test_actor = self.bt_multiplex_test_actor
+    self.bt_multiplex_test_actor.test_bt_multiplex_connections()
+
+  def setup_class(self):
+    super().setup_class()
+
+    self.bt_ble_test_actor = bt_ble_function_test_actor.BtBleFunctionTestActor(
+        self.test_parameters, self.discoverer, self.advertiser
+    )
+    self.fixed_wifi_medium_test_actor = (
+        fixed_wifi_medium_function_test_actor.FixedWifiMediumFunctionTestActor(
+            self.test_parameters, self.discoverer, self.advertiser
+        )
+    )
+    self.bt_multiplex_test_actor = (
+        bt_multiplex_function_test_actor.BtMultiplexFunctionTestActor(
+            self.test_parameters, self.discoverer, self.advertiser
+        )
+    )
+    self._current_test_actor: function_test_actor_base.FunctionTestActorBase = (
+        None
+    )
+
+  def teardown_test(self) -> None:
+    self._test_result_messages[self.current_test_info.name] = (
+        self._current_test_actor.get_test_result_message()
+    )
+    self.record_data({
+        'Test Name': self.current_test_info.name,
+        'sponge_properties': {
+            'result': self._current_test_actor.get_test_result_message(),
+        },
+    })
+    super().teardown_test()
+
+  # @typing.override
+  def _summary_test_results(self):
+    """Summarizes test results of all function tests."""
+
+    self.record_data({
+        'Test Class': self.TAG,
+        'sponge_properties': {
+            '00_test_script_verion': version.TEST_SCRIPT_VERSION,
+            '01_source_device_serial': self.discoverer.serial,
+            '02_target_device_serial': self.advertiser.serial,
+            '03_source_GMS_version': setup_utils.dump_gms_version(
+                self.discoverer
+            ),
+            '04_target_GMS_version': setup_utils.dump_gms_version(
+                self.advertiser
+            ),
+            '05_test_result': self._test_result_messages,
+        },
+    })
+
+
+if __name__ == '__main__':
+  test_runner.main()
diff --git a/tests/bettertogether/betocq/function_tests/bt_ble_function_test_actor.py b/tests/bettertogether/betocq/function_tests/bt_ble_function_test_actor.py
new file mode 100644
index 000000000..92b89e262
--- /dev/null
+++ b/tests/bettertogether/betocq/function_tests/bt_ble_function_test_actor.py
@@ -0,0 +1,88 @@
+#  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.
+
+"""Bluetooth nearby connection functional test actor."""
+
+from betocq import nc_constants
+from betocq import nearby_connection_wrapper
+from betocq.function_tests import function_test_actor_base
+
+
+class BtBleFunctionTestActor(function_test_actor_base.FunctionTestActorBase):
+  """The actor for running the BT/BlE function test."""
+
+  def test_bt_ble_connection(self):
+    """Test the basic BT and BLE connection.
+
+    Use BLE as discovery/advertising medium
+    Use BT as connection and upgrade medium
+    """
+
+    # 1. set up BT connection
+    advertising_discovery_medium = nc_constants.NearbyMedium.BLE_ONLY
+
+    nearby_snippet = nearby_connection_wrapper.NearbyConnectionWrapper(
+        self.advertiser,
+        self.discoverer,
+        self.advertiser.nearby,
+        self.discoverer.nearby,
+        advertising_discovery_medium=advertising_discovery_medium,
+        connection_medium=nc_constants.NearbyMedium.BT_ONLY,
+        upgrade_medium=nc_constants.NearbyMedium.BT_ONLY,
+    )
+    connection_setup_timeouts = nc_constants.ConnectionSetupTimeouts(
+        nc_constants.FIRST_DISCOVERY_TIMEOUT,
+        nc_constants.FIRST_CONNECTION_INIT_TIMEOUT,
+        nc_constants.FIRST_CONNECTION_RESULT_TIMEOUT)
+    try:
+      nearby_snippet.start_nearby_connection(
+          timeouts=connection_setup_timeouts,
+          medium_upgrade_type=nc_constants.MediumUpgradeType.NON_DISRUPTIVE)
+    finally:
+      self._test_failure_reason = nearby_snippet.test_failure_reason
+      self._test_result.file_transfer_nc_setup_quality_info = (
+          nearby_snippet.connection_quality_info
+      )
+
+    # 2. transfer file through bluetooth
+    try:
+      self._test_result.file_transfer_throughput_kbps = (
+          nearby_snippet.transfer_file(
+              nc_constants.TRANSFER_FILE_SIZE_1KB,
+              nc_constants.BT_1K_PAYLOAD_TRANSFER_TIMEOUT,
+              nc_constants.PayloadType.FILE,
+          )
+      )
+    finally:
+      self._test_failure_reason = nearby_snippet.test_failure_reason
+
+    # 3. disconnect
+    nearby_snippet.disconnect_endpoint()
+
+  def get_test_result_message(self) -> str:
+    """Gets the test result of current test."""
+    if (
+        self._test_failure_reason
+        == nc_constants.SingleTestFailureReason.SUCCESS
+    ):
+      return 'PASS'
+    if (
+        self._test_failure_reason
+        is nc_constants.SingleTestFailureReason.FILE_TRANSFER_FAIL
+    ):
+      return 'The Bluetooth performance is really bad or unknown reason.'
+    return ''.join([
+        f'FAIL: due to {self._test_failure_reason.name} - ',
+        f'{nc_constants.COMMON_TRIAGE_TIP.get(self._test_failure_reason)}'
+        ])
diff --git a/tests/bettertogether/betocq/function_tests/bt_multiplex_function_test_actor.py b/tests/bettertogether/betocq/function_tests/bt_multiplex_function_test_actor.py
new file mode 100644
index 000000000..2f5980416
--- /dev/null
+++ b/tests/bettertogether/betocq/function_tests/bt_multiplex_function_test_actor.py
@@ -0,0 +1,133 @@
+#  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.
+
+"""Bluetooth nearby connection functional test actor."""
+
+from mobly import asserts
+
+from betocq import nc_constants
+from betocq import nearby_connection_wrapper
+from betocq.function_tests import function_test_actor_base
+
+
+class BtMultiplexFunctionTestActor(
+    function_test_actor_base.FunctionTestActorBase):
+  """The actor for running BT/BlE multiplex function test.
+
+  It allows to setup multiple BT connections.
+  """
+
+  def test_bt_multiplex_connections(self):
+    """Test the capability of setting up two BT connections (multiplex).
+
+    This is only required for specific CUJ: quick start
+    step 1: set up 2 BT connections
+    step 2: transfer a small file with the 2nd BT connection
+    """
+    if (
+        self.test_parameters.target_cuj_name
+        != nc_constants.TARGET_CUJ_QUICK_START
+        and not self.test_parameters.requires_bt_multiplex
+    ):
+      self._skipped = True
+      asserts.skip(
+          'BT multiplex is not required for this CUJ -'
+          f' {self.test_parameters.target_cuj_name}'
+      )
+
+    # 1. set up 1st BT connection
+    advertising_discovery_medium = nc_constants.NearbyMedium.BLE_ONLY
+    nearby_snippet_2 = nearby_connection_wrapper.NearbyConnectionWrapper(
+        self.advertiser,
+        self.discoverer,
+        self.advertiser.nearby2,
+        self.discoverer.nearby2,
+        advertising_discovery_medium=advertising_discovery_medium,
+        connection_medium=nc_constants.NearbyMedium.BT_ONLY,
+        upgrade_medium=nc_constants.NearbyMedium.BT_ONLY,
+    )
+    connection_setup_timeouts = nc_constants.ConnectionSetupTimeouts(
+        nc_constants.FIRST_DISCOVERY_TIMEOUT,
+        nc_constants.FIRST_CONNECTION_INIT_TIMEOUT,
+        nc_constants.FIRST_CONNECTION_RESULT_TIMEOUT)
+
+    try:
+      nearby_snippet_2.start_nearby_connection(
+          timeouts=connection_setup_timeouts,
+          medium_upgrade_type=nc_constants.MediumUpgradeType.NON_DISRUPTIVE)
+    finally:
+      self._test_failure_reason = nearby_snippet_2.test_failure_reason
+      self._test_result.prior_nc_setup_quality_info = (
+          nearby_snippet_2.connection_quality_info
+      )
+    # 2nd bt connection
+    nearby_snippet = nearby_connection_wrapper.NearbyConnectionWrapper(
+        self.advertiser,
+        self.discoverer,
+        self.advertiser.nearby,
+        self.discoverer.nearby,
+        advertising_discovery_medium=advertising_discovery_medium,
+        connection_medium=nc_constants.NearbyMedium.BT_ONLY,
+        upgrade_medium=nc_constants.NearbyMedium.BT_ONLY,
+    )
+
+    second_connection_setup_timeouts = nc_constants.ConnectionSetupTimeouts(
+        nc_constants.SECOND_DISCOVERY_TIMEOUT,
+        nc_constants.SECOND_CONNECTION_INIT_TIMEOUT,
+        nc_constants.SECOND_CONNECTION_RESULT_TIMEOUT)
+    try:
+      nearby_snippet.start_nearby_connection(
+          timeouts=second_connection_setup_timeouts,
+          medium_upgrade_type=nc_constants.MediumUpgradeType.NON_DISRUPTIVE)
+    finally:
+      self._test_failure_reason = nearby_snippet.test_failure_reason
+      self._test_result.file_transfer_nc_setup_quality_info = (
+          nearby_snippet.connection_quality_info
+      )
+
+    # 2. transfer file through bluetooth
+    try:
+      self._test_result.file_transfer_throughput_kbps = (
+          nearby_snippet.transfer_file(
+              nc_constants.TRANSFER_FILE_SIZE_1KB,
+              nc_constants.BT_1K_PAYLOAD_TRANSFER_TIMEOUT,
+              nc_constants.PayloadType.FILE))
+    finally:
+      self._test_failure_reason = nearby_snippet.test_failure_reason
+
+    # 3. disconnect
+    nearby_snippet_2.disconnect_endpoint()
+    nearby_snippet.disconnect_endpoint()
+
+  def get_test_result_message(self) -> str:
+    if self._skipped:
+      return (
+          'SKIPPED - not required for the target CUJ: '
+          f'{self.test_parameters.target_cuj_name}'
+      )
+    if (
+        self._test_failure_reason
+        == nc_constants.SingleTestFailureReason.SUCCESS
+    ):
+      return 'PASS'
+    if (
+        self._test_failure_reason
+        is nc_constants.SingleTestFailureReason.FILE_TRANSFER_FAIL
+    ):
+      return 'The Bluetooth performance is really bad.'
+    else:
+      return ''.join([
+          f'FAIL: due to {self._test_failure_reason.name} - ',
+          f'{nc_constants.COMMON_TRIAGE_TIP.get(self._test_failure_reason)}'
+          ])
diff --git a/tests/bettertogether/betocq/function_tests/fixed_wifi_medium_function_test_actor.py b/tests/bettertogether/betocq/function_tests/fixed_wifi_medium_function_test_actor.py
new file mode 100644
index 000000000..2f035399f
--- /dev/null
+++ b/tests/bettertogether/betocq/function_tests/fixed_wifi_medium_function_test_actor.py
@@ -0,0 +1,150 @@
+#  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.
+
+"""Base class for all fixed wifi medium function test actors."""
+import logging
+from mobly import asserts
+
+from betocq import nc_constants
+from betocq import nearby_connection_wrapper
+from betocq import setup_utils
+from betocq.function_tests import function_test_actor_base
+
+
+class FixedWifiMediumFunctionTestActor(
+    function_test_actor_base.FunctionTestActorBase):
+  """The base of actors for running the specified fixed wifi D2D medium."""
+
+  def run_fixed_wifi_medium_test(
+      self, wifi_medium: nc_constants.NearbyMedium) -> None:
+    """upgrade the medium from BT to the specified medium and transfer a sample data."""
+    self._wifi_medium_under_test = wifi_medium
+    self._test_result = nc_constants.SingleTestResult()
+
+    # 1. set up BT and WiFi connection
+    advertising_discovery_medium = nc_constants.NearbyMedium(
+        self.test_parameters.advertising_discovery_medium
+    )
+    nearby_snippet = nearby_connection_wrapper.NearbyConnectionWrapper(
+        self.advertiser,
+        self.discoverer,
+        self.advertiser.nearby,
+        self.discoverer.nearby,
+        advertising_discovery_medium=advertising_discovery_medium,
+        connection_medium=nc_constants.NearbyMedium.BT_ONLY,
+        upgrade_medium=self._wifi_medium_under_test,
+    )
+
+    connection_setup_timeouts = nc_constants.ConnectionSetupTimeouts(
+        nc_constants.FIRST_DISCOVERY_TIMEOUT,
+        nc_constants.FIRST_CONNECTION_INIT_TIMEOUT,
+        nc_constants.FIRST_CONNECTION_RESULT_TIMEOUT)
+
+    try:
+      nearby_snippet.start_nearby_connection(
+          timeouts=connection_setup_timeouts,
+          medium_upgrade_type=nc_constants.MediumUpgradeType.NON_DISRUPTIVE)
+    finally:
+      self._test_failure_reason = nearby_snippet.test_failure_reason
+      self._test_result.file_transfer_nc_setup_quality_info = (
+          nearby_snippet.connection_quality_info
+      )
+
+    # 2. transfer file through WiFi
+    try:
+      self._test_result.file_transfer_throughput_kbps = (
+          nearby_snippet.transfer_file(
+              nc_constants.TRANSFER_FILE_SIZE_1KB,
+              nc_constants.WIFI_1K_PAYLOAD_TRANSFER_TIMEOUT,
+              nc_constants.PayloadType.FILE))
+    finally:
+      self._test_failure_reason = nearby_snippet.test_failure_reason
+
+    # 3. disconnect
+    nearby_snippet.disconnect_endpoint()
+
+  def connect_to_wifi(self):
+    if self.test_parameters.toggle_airplane_mode_target_side:
+      setup_utils.toggle_airplane_mode(self.advertiser)
+
+    wifi_ssid, wifi_password = self._get_wifi_ssid_password()
+
+    if not wifi_ssid:
+      self._test_failure_reason = (
+          nc_constants.SingleTestFailureReason.AP_IS_NOT_CONFIGURED
+      )
+      asserts.fail('Wifi AP must be specified.')
+
+    logging.info('connect to wifi: %s', wifi_ssid)
+
+    # source device
+    self._test_failure_reason = (
+        nc_constants.SingleTestFailureReason.SOURCE_WIFI_CONNECTION
+    )
+    discoverer_wifi_latency = setup_utils.connect_to_wifi_sta_till_success(
+        self.discoverer, wifi_ssid, wifi_password
+    )
+    self.discoverer.log.info(
+        'connecting to wifi in '
+        f'{round(discoverer_wifi_latency.total_seconds())} s'
+    )
+    # target device
+    self._test_failure_reason = (
+        nc_constants.SingleTestFailureReason.TARGET_WIFI_CONNECTION
+    )
+    advertiser_wlan_latency = setup_utils.connect_to_wifi_sta_till_success(
+        self.advertiser, wifi_ssid, wifi_password)
+    self.advertiser.log.info(
+        'connecting to wifi in '
+        f'{round(advertiser_wlan_latency.total_seconds())} s')
+    self.advertiser.log.info(
+        self.advertiser.nearby.wifiGetConnectionInfo().get('mFrequency')
+    )
+    self._test_failure_reason = (
+        nc_constants.SingleTestFailureReason.SUCCESS
+    )
+
+  def get_test_result_message(self) -> str:
+    if (
+        self._test_failure_reason
+        == nc_constants.SingleTestFailureReason.SUCCESS
+    ):
+      return 'PASS'
+    if (
+        self._test_failure_reason
+        == nc_constants.SingleTestFailureReason.WIFI_MEDIUM_UPGRADE
+    ):
+      return f'{self._test_failure_reason.name} - '.join(
+          self._get_medium_upgrade_failure_tip()
+      )
+    if (
+        self._test_failure_reason
+        == nc_constants.SingleTestFailureReason.FILE_TRANSFER_FAIL
+    ):
+      return f'{self._test_failure_reason.name} - '.join(
+          self._get_file_transfer_failure_tip()
+      )
+    return ''.join([
+        f'{self._test_failure_reason.name} - ',
+        nc_constants.COMMON_TRIAGE_TIP.get(self._test_failure_reason),
+    ])
+
+  def _get_medium_upgrade_failure_tip(self) -> str:
+    return nc_constants.MEDIUM_UPGRADE_FAIL_TRIAGE_TIPS.get(
+        self._wifi_medium_under_test, 'unsupported test medium')
+
+  def _get_file_transfer_failure_tip(self) -> str:
+    if self._wifi_medium_under_test is not None:
+      return f'{self._wifi_medium_under_test.name}: the performance is too bad.'
+    asserts.fail('unexpected calling of _get_file_transfer_failure_tip')
diff --git a/tests/bettertogether/betocq/function_tests/function_test_actor_base.py b/tests/bettertogether/betocq/function_tests/function_test_actor_base.py
new file mode 100644
index 000000000..c0ecf122d
--- /dev/null
+++ b/tests/bettertogether/betocq/function_tests/function_test_actor_base.py
@@ -0,0 +1,74 @@
+#  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.
+
+"""Provides the actor base for all function tests."""
+
+from typing import Tuple
+
+from mobly.controllers import android_device
+
+from betocq import nc_constants
+
+
+class FunctionTestActorBase:
+  """Base class of actors for running all function tests."""
+
+  def __init__(self,
+               test_parameters: nc_constants.TestParameters,
+               discoverer: android_device.AndroidDevice,
+               advertiser: android_device.AndroidDevice
+               ):
+    self.test_parameters: nc_constants.TestParameters = test_parameters
+    self.advertiser: android_device.AndroidDevice = advertiser
+    self.discoverer: android_device.AndroidDevice = discoverer
+    self._test_result: nc_constants.SingleTestResult = (
+        nc_constants.SingleTestResult()
+    )
+    self._test_failure_reason: nc_constants.SingleTestFailureReason = (
+        nc_constants.SingleTestFailureReason.UNINITIALIZED
+    )
+    self._wifi_medium_under_test = None
+    self._skipped: bool = False
+
+  def _get_wifi_ssid_password(self) -> Tuple[str, str]:
+    """Returns the available wifi username and password."""
+    if self.test_parameters.wifi_ssid:
+      return (
+          self.test_parameters.wifi_ssid,
+          self.test_parameters.wifi_password,
+      )
+    if self.test_parameters.wifi_5g_ssid:
+      return (
+          self.test_parameters.wifi_5g_ssid,
+          self.test_parameters.wifi_5g_password,
+      )
+    if self.test_parameters.wifi_dfs_5g_ssid:
+      return (
+          self.test_parameters.wifi_dfs_5g_ssid,
+          self.test_parameters.wifi_dfs_5g_password,
+      )
+    if self.test_parameters.wifi_2g_ssid:
+      return (
+          self.test_parameters.wifi_2g_ssid,
+          self.test_parameters.wifi_2g_password,
+      )
+    return ('', '')
+
+  def get_test_result_message(self) -> str:
+    """Returns the message about the test result."""
+    return 'Unknown'
+
+  def _get_test_failure_reason(self) -> nc_constants.SingleTestFailureReason:
+    """Returns the test failure reason."""
+    return self._test_failure_reason
diff --git a/tests/bettertogether/betocq/function_tests/local_dev_function_group_testbed.yml b/tests/bettertogether/betocq/function_tests/local_dev_function_group_testbed.yml
new file mode 100644
index 000000000..ce7eac557
--- /dev/null
+++ b/tests/bettertogether/betocq/function_tests/local_dev_function_group_testbed.yml
@@ -0,0 +1,26 @@
+# Test bed for the function tests only.
+x-test-params: &test-params
+  wifi_ssid: "anyAp"
+  wifi_password: "anyAp"
+  target_cuj_name: "quick_start"
+
+x-controllers: &controllers
+  Controllers:
+    AndroidDevice:
+      - serial: "123456ABCDEF"
+        role: "source_device"
+      - serial: "ABCDEF123456"
+        role: "target_device"
+
+TestBeds:
+- Name: LocalDevTestbed
+  # only for development, DO NOT port to AOSP.
+  Controllers:
+    AndroidDevice: '*'
+  TestParams:
+      <<: *test-params
+
+- Name: LocalFunctionGroupTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
\ No newline at end of file
diff --git a/tests/bettertogether/betocq/gms_auto_updates_util.py b/tests/bettertogether/betocq/gms_auto_updates_util.py
new file mode 100644
index 000000000..f8dab4faf
--- /dev/null
+++ b/tests/bettertogether/betocq/gms_auto_updates_util.py
@@ -0,0 +1,176 @@
+#  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.
+
+"""class to enable/disable GMS auto update."""
+
+import logging
+import os
+import tempfile
+from xml import etree
+from mobly.controllers import android_device
+from mobly.controllers.android_device_lib import adb
+
+
+_FINSKY_CONFIG_FILE = '/data/data/com.android.vending/shared_prefs/finsky.xml'
+_FINSKY_CONFIG_NAME = 'auto_update_enabled'
+_FINSKY_CONFIG_VALUE_DISABLE = 'false'
+_FINSKY_CONFIG_VALUE_ENABLE = 'true'
+_VENDING_CONFIG_FILE = '/data/data/com.android.vending/shared_prefs/com.android.vending_preferences.xml'
+_VENDING_CONFIG_NAME = 'auto-update-mode'
+_VENDING_CONFIG_VALUE_DISABLE = 'AUTO_UPDATE_NEVER'
+_VENDING_CONFIG_VALUE_ENABLE = 'AUTO_UPDATE_WIFI'
+_BLANK_CONFIG = '<?xml version="1.0" encoding="utf-8"?><map></map>'
+_XML_BOOL_TYPE = 'boolean'
+_XML_STRING_TYPE = 'string'
+_ENABLE_GSERVICES_CMD_TEMPLATE = [
+    (
+        'am broadcast '
+        '-a com.google.gservices.intent.action.GSERVICES_OVERRIDE '
+        '-e finsky.play_services_auto_update_enabled {}'
+    ),
+    (
+        'am broadcast '
+        '-a com.google.gservices.intent.action.GSERVICES_OVERRIDE '
+        '-e finsky.setup_wizard_additional_account_vpa_enable {}'
+    ),
+]
+
+
+class GmsAutoUpdatesUtil:
+  """class to enable/disable GMS auto updates."""
+
+  def __init__(self, ad: android_device.AndroidDevice):
+    self._device: android_device.AndroidDevice = ad
+
+  def enable_gms_auto_updates(self) -> None:
+    self._config_gms_auto_updates(True)
+
+  def disable_gms_auto_updates(self) -> None:
+    self._config_gms_auto_updates(False)
+
+  def _config_gms_auto_updates(self, enable_updates: bool) -> None:
+    """Configures GMS auto updates."""
+    if not self._device.is_adb_root:
+      self._device.log.info(
+          f'failed to set the play store auto updates as {enable_updates}'
+          'you should enable/disable it manually on an unrooted device.')
+    else:
+      if enable_updates:
+        self._configure_play_store_updates(
+            _FINSKY_CONFIG_VALUE_ENABLE, _VENDING_CONFIG_VALUE_ENABLE
+        )
+      else:
+        self._configure_play_store_updates(
+            _FINSKY_CONFIG_VALUE_DISABLE, _VENDING_CONFIG_VALUE_DISABLE
+        )
+    self._configure_gservice_updates(enable_updates)
+
+  def _configure_gservice_updates(self, enable_updates: bool) -> None:
+    """Overwites Gservice to enable/disable updates."""
+    for cmd in _ENABLE_GSERVICES_CMD_TEMPLATE:
+      self._device.adb.shell(
+          cmd.format('true' if enable_updates else 'false')
+      )
+
+  def _create_or_update_play_store_config(
+      self,
+      tmp_dir: str,
+      value_type: str,
+      name: str,
+      value: str,
+      device_path: str,
+  ) -> str:
+    """Creates or updates a Play Store configuration file.
+
+    The function retrieves the Play Store configuration file from the device
+    then update it. If the file does not exist, it creates a new one.
+
+    Args:
+        tmp_dir: The temporary directory to store the configuration file.
+        value_type: The type of the configuration field.
+        name: The name of the configuration field.
+        value: The value of the configuration field.
+        device_path: The path to the configuration file on the device.
+
+    Returns:
+        The path to the updated configuration file.
+    """
+    path = os.path.join(tmp_dir, f'play_store_config_{name}.xml')
+    try:
+      self._device.adb.pull([device_path, path])
+    except adb.AdbError as e:
+      self._device.log.warning('failed to pull %s: %s', device_path, e)
+
+    config_doc = etree.ElementTree.parse(path) if os.path.isfile(path) else None
+
+    changing_element = None
+    root = (
+        etree.ElementTree.fromstring(_BLANK_CONFIG.encode())
+        if config_doc is None
+        else config_doc.getroot()
+    )
+
+    # find the element, xPath doesn't work as the name is a reserved word.
+    for child in root:
+      if child.attrib['name'] == name:
+        changing_element = child
+        break
+    if changing_element is None:
+      if value_type == _XML_BOOL_TYPE:
+        changing_element = etree.ElementTree.SubElement(root, 'boolean')
+      else:
+        changing_element = etree.ElementTree.SubElement(root, 'string')
+    logging.info('element for %s is %s, %s', name, changing_element.tag,
+                 changing_element.attrib)
+    if value_type == _XML_BOOL_TYPE:
+      changing_element.set('name', name)
+      changing_element.set('value', value)
+    else:
+      changing_element.attrib['name'] = name
+      changing_element.text = value
+
+    tree = etree.ElementTree.ElementTree(root)
+    tree.write(path, xml_declaration=True, encoding='utf-8')
+    return path
+
+  def _configure_play_store_updates(
+      self, finsky_config_value: str, vending_config_value: str
+  ) -> None:
+    """Configures the Play Store update related settings."""
+    with tempfile.TemporaryDirectory() as tmp_dir:
+      finsky_config = self._create_or_update_play_store_config(
+          tmp_dir,
+          _XML_BOOL_TYPE,
+          _FINSKY_CONFIG_NAME,
+          finsky_config_value,
+          _FINSKY_CONFIG_FILE,
+      )
+      self._device.adb.push([finsky_config, _FINSKY_CONFIG_FILE])
+      try:
+        os.remove(finsky_config)
+      except OSError as e:
+        logging.warning('failed to remove %s: %s', finsky_config, e)
+
+      vending_config = self._create_or_update_play_store_config(
+          tmp_dir,
+          _XML_STRING_TYPE,
+          _VENDING_CONFIG_NAME,
+          vending_config_value,
+          _VENDING_CONFIG_FILE,
+      )
+      self._device.adb.push([vending_config, _VENDING_CONFIG_FILE])
+      try:
+        os.remove(vending_config)
+      except OSError as e:
+        logging.warning('failed to remove %s: %s', vending_config, e)
diff --git a/tests/bettertogether/betocq/local_dev_testbed.yml b/tests/bettertogether/betocq/local_dev_testbed.yml
new file mode 100644
index 000000000..e5b21581a
--- /dev/null
+++ b/tests/bettertogether/betocq/local_dev_testbed.yml
@@ -0,0 +1,102 @@
+x-test-params: &test-params
+  allow_unrooted_device: False
+  # for 2G wifi SSID - channel 6, frequency 2437, comment out if it is not available.
+  wifi_2g_ssid: "AP2437"
+  wifi_2g_password: "AP2437"
+  # for 5G wifi SSID - channel 36, frequency 5180, comment out if it is not available.
+  wifi_5g_ssid: "AP5180"
+  wifi_5g_password: "AP5180"
+  # for DFS 5G wifi SSID - channel 52, frequency 5260, comment out if it is not available.
+  wifi_dfs_5g_ssid: "AP5260"
+  wifi_dfs_5g_password: "AP5260"
+  # use the AP controlled by programming dynamically
+  use_auto_controlled_wifi_ap: False
+  run_function_tests_with_performance_tests: True
+
+x-controllers: &controllers
+  Controllers:
+    AndroidDevice:
+      - serial: "123456ABCDEF"
+        role: "source_device"
+        # The max number of spatial streams
+        max_num_streams: 2
+        # The max PHY rate at 5G, in Mbps
+        max_phy_rate_5g_mbps: 2402
+        # The max PHY rate at 2G, in Mbps
+        max_phy_rate_2g_mbps: 287
+        # if the device supports 5G Wifi
+        supports_5g: True
+        # if the device supports DBS (Dual Band Simultaneous) in STA + WiFi-Direct concurrency mode
+        supports_dbs_sta_wfd: True
+        # The max number of spatial streams in DBS mode.
+        max_num_streams_dbs: 2
+        # if the device supports to start WFD group owner at a STA-associated DFS channel
+        enable_sta_dfs_channel_for_peer_network: False
+        # if the device supports to start WFD group owner at a STA-associated indoor channel
+        enable_sta_indoor_channel_for_peer_network: False
+        # 14 - U, 13 - T, 12 - S
+        android_version: 14
+      - serial: "ABCDEF123456"
+        role: "target_device"
+        # The max number of spatial streams
+        max_num_streams: 2
+        # The max PHY rate at 5G, in Mbps
+        max_phy_rate_5g_mbps: 2402
+        # The max PHY rate at 2G, in Mbps
+        max_phy_rate_2g_mbps: 287
+        # if the device supports 5G Wifi
+        supports_5g: True  # True
+        # if the device supports DBS (Dual Band Simultaneous) in STA + WiFi-Direct concurrency mode
+        supports_dbs_sta_wfd: True
+        # The max number of spatial streams in DBS mode.
+        max_num_streams_dbs: 2
+        # if the device supports to start WFD group owner at a STA-associated DFS channel
+        enable_sta_dfs_channel_for_peer_network: True
+        # if the device supports to start WFD group owner at a STA-associated indoor channel
+        enable_sta_indoor_channel_for_peer_network: True
+        # 14 - U, 13 - T, 12 - S
+        android_version: 14
+
+TestBeds:
+- Name: LocalPerformanceDefaultTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
+
+- Name: LocalQuickstartPerformanceTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
+      target_cuj_name: "quick_start"
+      # before the performance test, run the function tests first
+      run_function_tests_with_performance_tests: True
+      # if the function tests is failed, abort the performance test
+      abort_all_tests_on_function_tests_fail: True
+
+- Name: LocalQuicksharePerformanceTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
+      target_cuj_name: "quick_share"
+
+- Name: LocalEsimTransferPerformanceTestbed
+  <<: *controllers
+  TestParams:
+      <<: *test-params
+      target_cuj_name: "setting_based_esim_transfer"
+
+# used only for function test
+- Name: LocalQuickstartFunctionTestbed
+  Controllers:
+    <<: *controllers
+  TestParams:
+      <<: *test-params
+      target_cuj_name: "quick_start"
+
+# used only for function test
+- Name: LocalQuickshareFunctionTestbed
+  Controllers:
+    <<: *controllers
+  TestParams:
+      <<: *test-params
+      target_cuj_name: "quick_share"
\ No newline at end of file
diff --git a/tests/bettertogether/betocq/nc_base_test.py b/tests/bettertogether/betocq/nc_base_test.py
new file mode 100644
index 000000000..716f65a34
--- /dev/null
+++ b/tests/bettertogether/betocq/nc_base_test.py
@@ -0,0 +1,271 @@
+#  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.
+
+"""Mobly base test class for Neaby Connections.
+
+Override the NCBaseTestClass#_get_country_code method if the test requires
+a special country code, the 'US' is used by default.
+"""
+
+import logging
+import os
+import time
+
+from mobly import asserts
+from mobly import base_test
+from mobly import records
+from mobly import utils
+from mobly.controllers import android_device
+from mobly.controllers.android_device_lib import errors
+import yaml
+
+from betocq import android_wifi_utils
+from betocq import nc_constants
+from betocq import setup_utils
+
+NEARBY_SNIPPET_PACKAGE_NAME = 'com.google.android.nearby.mobly.snippet'
+NEARBY_SNIPPET_2_PACKAGE_NAME = 'com.google.android.nearby.mobly.snippet.second'
+
+_CONFIG_EXTERNAL_PATH = 'TBD'
+
+
+class NCBaseTestClass(base_test.BaseTestClass):
+  """The Base of Nearby Connection E2E tests."""
+
+  def __init__(self, configs):
+    super().__init__(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._nearby_snippet_apk_path: str = None
+    self._nearby_snippet_2_apk_path: str = None
+    self.performance_test_iterations: int = 1
+    self.num_bug_reports: int = 0
+    self._requires_2_snippet_apks = False
+    self.__loaded_2_nearby_snippets = False
+    self.__skipped_test_class = False
+
+  def _get_skipped_test_class_reason(self) -> str | None:
+    return None
+
+  def setup_class(self) -> None:
+    self._setup_openwrt_wifi()
+    self.ads = self.register_controller(android_device, min_number=2)
+    try:
+      self.discoverer = android_device.get_device(
+          self.ads, role='source_device'
+      )
+      self.advertiser = android_device.get_device(
+          self.ads, role='target_device'
+      )
+    except errors.Error:
+      logging.warning(
+          'The source,target devices are not specified in testbed;'
+          'The result may not be expected.'
+      )
+      self.advertiser, self.discoverer = self.ads
+
+    utils.concurrent_exec(
+        self._setup_android_hw_capability,
+        param_list=[[ad] for ad in self.ads],
+        raise_on_exception=True,
+    )
+
+    skipped_test_class_reason = self._get_skipped_test_class_reason()
+    if skipped_test_class_reason:
+      self.__skipped_test_class = True
+      asserts.abort_class(skipped_test_class_reason)
+
+    file_tag = 'files' if 'files' in self.user_params else 'mh_files'
+    self._nearby_snippet_apk_path = self.user_params.get(file_tag, {}).get(
+        'nearby_snippet', ['']
+    )[0]
+    if self.test_parameters.requires_bt_multiplex:
+      self._requires_2_snippet_apks = True
+      self._nearby_snippet_2_apk_path = self.user_params.get(file_tag, {}).get(
+          'nearby_snippet_2', ['']
+      )[0]
+
+    # disconnect from all wifi automatically
+    utils.concurrent_exec(
+        android_wifi_utils.forget_all_wifi,
+        param_list=[[ad] for ad in self.ads],
+        raise_on_exception=True,
+    )
+
+    utils.concurrent_exec(
+        self._setup_android_device,
+        param_list=[[ad] for ad in self.ads],
+        raise_on_exception=True,
+    )
+
+  def _setup_openwrt_wifi(self):
+    """Sets up the wifi connection with OpenWRT."""
+    if not self.user_params.get('use_auto_controlled_wifi_ap', False):
+      return
+
+    # TODO(xianyuanjia): Add OpenWRT setup logic here
+    return
+
+  def _setup_android_hw_capability(
+      self, ad: android_device.AndroidDevice
+  ) -> None:
+    ad.android_version = int(ad.adb.getprop('ro.build.version.release'))
+
+    if not os.path.isfile(_CONFIG_EXTERNAL_PATH):
+      return
+
+    config_path = _CONFIG_EXTERNAL_PATH
+    with open(config_path, 'r') as f:
+      rule = yaml.safe_load(f).get(ad.model, None)
+      asserts.assert_is_not_none(
+          rule, f'{ad} Model {ad.model} is not supported in config file'
+      )
+      for key, value in rule.items():
+        setattr(ad, key, value)
+
+  def _get_country_code(self) -> str:
+    return 'US'
+
+  def _setup_android_device(self, ad: android_device.AndroidDevice) -> None:
+    ad.debug_tag = ad.serial + '(' + ad.adb.getprop('ro.product.model') + ')'
+    if not ad.is_adb_root:
+      if self.test_parameters.allow_unrooted_device:
+        ad.log.info('Unrooted device is detected. Test coverage is limited')
+      else:
+        asserts.abort_all('The test only can run on rooted device.')
+
+    setup_utils.disable_gms_auto_updates(ad)
+
+    ad.debug_tag = ad.serial + '(' + ad.adb.getprop('ro.product.model') + ')'
+    ad.log.info('try to install nearby_snippet_apk')
+    if self._nearby_snippet_apk_path:
+      setup_utils.install_apk(ad, self._nearby_snippet_apk_path)
+    else:
+      ad.log.warning(
+          'nearby_snippet apk is not specified, '
+          'make sure it is installed in the device'
+      )
+
+    ad.log.info('grant manage external storage permission')
+    setup_utils.grant_manage_external_storage_permission(
+        ad, NEARBY_SNIPPET_PACKAGE_NAME
+    )
+    ad.load_snippet('nearby', NEARBY_SNIPPET_PACKAGE_NAME)
+
+    if self._requires_2_snippet_apks:
+      ad.log.info('try to install nearby_snippet_2_apk')
+      if self._nearby_snippet_2_apk_path:
+        setup_utils.install_apk(ad, self._nearby_snippet_2_apk_path)
+      else:
+        ad.log.warning(
+            'nearby_snippet_2 apk is not specified, '
+            'make sure it is installed in the device'
+        )
+      setup_utils.grant_manage_external_storage_permission(
+          ad, NEARBY_SNIPPET_2_PACKAGE_NAME
+      )
+      setup_utils.enable_bluetooth_multiplex(ad)
+      ad.load_snippet('nearby2', NEARBY_SNIPPET_2_PACKAGE_NAME)
+      self.__loaded_2_nearby_snippets = True
+    if not ad.nearby.wifiIsEnabled():
+      ad.nearby.wifiEnable()
+    setup_utils.disconnect_from_wifi(ad)
+    setup_utils.enable_logs(ad)
+
+    setup_utils.disable_redaction(ad)
+
+    setup_utils.set_country_code(ad, self._get_country_code())
+
+  def setup_test(self):
+    self.record_data({
+        'Test Name': self.current_test_info.name,
+        'sponge_properties': {
+            'beto_team': 'Nearby Connections',
+            'beto_feature': 'Nearby Connections',
+        },
+    })
+    self._reset_nearby_connection()
+
+  def _reset_wifi_connection(self) -> None:
+    """Resets wifi connections on both devices."""
+    self.discoverer.nearby.wifiClearConfiguredNetworks()
+    self.advertiser.nearby.wifiClearConfiguredNetworks()
+    time.sleep(nc_constants.WIFI_DISCONNECTION_DELAY.total_seconds())
+
+  def _reset_nearby_connection(self) -> None:
+    """Resets nearby connection."""
+    self.discoverer.nearby.stopDiscovery()
+    self.discoverer.nearby.stopAllEndpoints()
+    self.advertiser.nearby.stopAdvertising()
+    self.advertiser.nearby.stopAllEndpoints()
+    if self.__loaded_2_nearby_snippets:
+      self.discoverer.nearby2.stopDiscovery()
+      self.discoverer.nearby2.stopAllEndpoints()
+      self.advertiser.nearby2.stopAdvertising()
+      self.advertiser.nearby2.stopAllEndpoints()
+    time.sleep(nc_constants.NEARBY_RESET_WAIT_TIME.total_seconds())
+
+  def _teardown_device(self, ad: android_device.AndroidDevice) -> None:
+    ad.nearby.transferFilesCleanup()
+    setup_utils.enable_gms_auto_updates(ad)
+
+    if self.test_parameters.disconnect_wifi_after_test:
+      setup_utils.disconnect_from_wifi(ad)
+
+    ad.unload_snippet('nearby')
+    if self.__loaded_2_nearby_snippets:
+      ad.unload_snippet('nearby2')
+
+  def teardown_test(self) -> None:
+    utils.concurrent_exec(
+        lambda d: d.services.create_output_excerpts_all(self.current_test_info),
+        param_list=[[ad] for ad in self.ads],
+        raise_on_exception=True,
+    )
+
+  def teardown_class(self) -> None:
+    if self.__skipped_test_class:
+      logging.info('Skipping teardown class.')
+      return
+
+    # handle summary results
+    self._summary_test_results()
+
+    utils.concurrent_exec(
+        self._teardown_device,
+        param_list=[[ad] for ad in self.ads],
+        raise_on_exception=True,
+    )
+
+    if hasattr(self, 'openwrt') and hasattr(self, 'wifi_info'):
+      self.openwrt.stop_wifi(self.wifi_info)
+
+  def _summary_test_results(self) -> None:
+    pass
+
+  def on_fail(self, record: records.TestResultRecord) -> None:
+    if self.__skipped_test_class:
+      logging.info('Skipping on_fail.')
+      return
+    self.num_bug_reports = self.num_bug_reports + 1
+    if self.num_bug_reports <= nc_constants.MAX_NUM_BUG_REPORT:
+      logging.info('take bug report for failure')
+      android_device.take_bug_reports(
+          self.ads,
+          destination=self.current_test_info.output_path,
+      )
diff --git a/tests/bettertogether/betocq/nc_constants.py b/tests/bettertogether/betocq/nc_constants.py
new file mode 100644
index 000000000..e3eaba740
--- /dev/null
+++ b/tests/bettertogether/betocq/nc_constants.py
@@ -0,0 +1,366 @@
+#  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.
+
+"""Constants for Nearby Connection."""
+
+import ast
+import dataclasses
+import datetime
+import enum
+import logging
+from typing import Any
+
+SUCCESS_RATE_TARGET = 0.95  # 95%
+MCC_PERFORMANCE_TEST_COUNT = 100
+MCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR = 5
+SCC_PERFORMANCE_TEST_COUNT = 10
+SCC_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR = 2
+BT_PERFORMANCE_TEST_COUNT = 100
+BT_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR = 5
+
+NEARBY_RESET_WAIT_TIME = datetime.timedelta(seconds=5)
+WIFI_DISCONNECTION_DELAY = datetime.timedelta(seconds=3)
+
+FIRST_DISCOVERY_TIMEOUT = datetime.timedelta(seconds=30)
+FIRST_CONNECTION_INIT_TIMEOUT = datetime.timedelta(seconds=30)
+FIRST_CONNECTION_RESULT_TIMEOUT = datetime.timedelta(seconds=35)
+BT_1K_PAYLOAD_TRANSFER_TIMEOUT = datetime.timedelta(seconds=20)
+BT_1M_PAYLOAD_TRANSFER_TIMEOUT = datetime.timedelta(seconds=50)
+BLE_1M_PAYLOAD_TRANSFER_TIMEOUT = datetime.timedelta(seconds=50)
+SECOND_DISCOVERY_TIMEOUT = datetime.timedelta(seconds=35)
+SECOND_CONNECTION_INIT_TIMEOUT = datetime.timedelta(seconds=10)
+SECOND_CONNECTION_RESULT_TIMEOUT = datetime.timedelta(seconds=25)
+CONNECTION_BANDWIDTH_CHANGED_TIMEOUT = datetime.timedelta(seconds=25)
+WIFI_1K_PAYLOAD_TRANSFER_TIMEOUT = datetime.timedelta(seconds=20)
+WIFI_2G_20M_PAYLOAD_TRANSFER_TIMEOUT = datetime.timedelta(seconds=20)
+WIFI_200M_PAYLOAD_TRANSFER_TIMEOUT = datetime.timedelta(seconds=100)
+WIFI_STA_CONNECTING_TIME_OUT = datetime.timedelta(seconds=25)
+DISCONNECTION_TIMEOUT = datetime.timedelta(seconds=15)
+
+MAX_PHY_RATE_PER_STREAM_AC_80_MBPS = 433
+MAX_PHY_RATE_PER_STREAM_AC_40_MBPS = 200
+MAX_PHY_RATE_PER_STREAM_N_20_MBPS = 72
+
+MCC_THROUGHPUT_MULTIPLIER = 0.25
+MAX_PHY_RATE_TO_MIN_THROUGHPUT_RATIO_5G = 0.37
+MAX_PHY_RATE_TO_MIN_THROUGHPUT_RATIO_2G = 0.4
+WIFI_HOTSPOT_THROUGHPUT_MULTIPLIER = 0.7
+
+CLASSIC_BT_MEDIUM_THROUGHPUT_BENCHMARK = 20  # 20KBps
+BLE_MEDIUM_THROUGHPUT_BENCHMARK = 20  # 20KBps
+
+KEEP_ALIVE_TIMEOUT_BT_MS = 30000
+KEEP_ALIVE_INTERVAL_BT_MS = 5000
+
+KEEP_ALIVE_TIMEOUT_WIFI_MS = 10000
+KEEP_ALIVE_INTERVAL_WIFI_MS = 3000
+
+PERCENTILE_50_FACTOR = 0.5
+LATENCY_PRECISION_DIGITS = 1
+
+UNSET_LATENCY = datetime.timedelta.max
+UNSET_THROUGHPUT_KBPS = -1.0
+MAX_NUM_BUG_REPORT = 5
+
+TRANSFER_FILE_SIZE_200MB = 200 * 1024  # kB
+TRANSFER_FILE_SIZE_20MB = 20 * 1024  # kB
+TRANSFER_FILE_SIZE_1MB = 1024  # kB
+TRANSFER_FILE_SIZE_1KB = 1  # kB
+
+TARGET_CUJ_QUICK_START = 'quick_start'
+TARGET_CUJ_ESIM = 'setting_based_esim_transfer'
+TARGET_CUJ_QUICK_SHARE = 'quick_share'
+
+
+@enum.unique
+class PayloadType(enum.IntEnum):
+  FILE = 2
+  STREAM = 3
+
+
+@enum.unique
+class NearbyMedium(enum.IntEnum):
+  """Medium options for discovery, advertising, connection and upgrade."""
+
+  AUTO = 0
+  BT_ONLY = 1
+  BLE_ONLY = 2
+  WIFILAN_ONLY = 3
+  WIFIAWARE_ONLY = 4
+  UPGRADE_TO_WEBRTC = 5
+  UPGRADE_TO_WIFIHOTSPOT = 6
+  UPGRADE_TO_WIFIDIRECT = 7
+  BLE_L2CAP_ONLY = 8
+  # including WIFI_LAN, WIFI_HOTSPOT, WIFI_DIRECT
+  UPGRADE_TO_ALL_WIFI = 9
+
+
+@dataclasses.dataclass(frozen=False)
+class TestParameters:
+  """Test parameters to be customized for Nearby Connection."""
+
+  target_cuj_name: str = 'unspecified'
+  requires_bt_multiplex: bool = False
+  run_function_tests_with_performance_tests: bool = False
+  abort_all_tests_on_function_tests_fail: bool = True
+  fast_fail_on_any_error: bool = False
+  use_auto_controlled_wifi_ap: bool = False
+  wifi_2g_ssid: str = ''
+  wifi_2g_password: str = ''
+  wifi_5g_ssid: str = ''
+  wifi_5g_password: str = ''
+  wifi_dfs_5g_ssid: str = ''
+  wifi_dfs_5g_password: str = ''
+  wifi_ssid: str = ''  # optional, for tests which can use any wifi
+  wifi_password: str = ''
+  advertising_discovery_medium: NearbyMedium = NearbyMedium.BLE_ONLY
+  toggle_airplane_mode_target_side: bool = False
+  reset_wifi_connection: bool = True
+  disconnect_bt_after_test: bool = False
+  disconnect_wifi_after_test: bool = False
+  payload_type: PayloadType = PayloadType.FILE
+  allow_unrooted_device: bool = False
+  keep_alive_timeout_ms: int = KEEP_ALIVE_TIMEOUT_WIFI_MS
+  keep_alive_interval_ms: int = KEEP_ALIVE_INTERVAL_WIFI_MS
+
+  @classmethod
+  def from_user_params(
+      cls,
+      user_params: dict[str, Any]) -> 'TestParameters':
+    """convert the parameters from the testbed to the test parameter."""
+
+    # Convert user int parameter in str format to int.
+    for key, value in user_params.items():
+      if key == 'mh_files':
+        continue
+      logging.info('[Test Parameters] %s: %s', key, value)
+      if isinstance(value, str) and value.isdigit():
+        user_params[key] = ast.literal_eval(value)
+
+    test_parameters_names = {
+        field.name for field in dataclasses.fields(cls)
+    }
+    test_parameters = cls(
+        **{
+            key: val
+            for key, val in user_params.items()
+            if key in test_parameters_names
+        }
+    )
+
+    if test_parameters.target_cuj_name == TARGET_CUJ_QUICK_START:
+      test_parameters.requires_bt_multiplex = True
+
+    return test_parameters
+
+
+@enum.unique
+class NearbyConnectionMedium(enum.IntEnum):
+  """The final connection medium selected, see BandWidthInfo.Medium."""
+  UNKNOWN = 0
+  # reserved 1, it's Medium.MDNS, not used now
+  BLUETOOTH = 2
+  WIFI_HOTSPOT = 3
+  BLE = 4
+  WIFI_LAN = 5
+  WIFI_AWARE = 6
+  NFC = 7
+  WIFI_DIRECT = 8
+  WEB_RTC = 9
+  # 10 is reserved.
+  USB = 11
+
+
+def is_high_quality_medium(medium: NearbyMedium) -> bool:
+  return medium in {
+      NearbyMedium.WIFILAN_ONLY,
+      NearbyMedium.WIFIAWARE_ONLY,
+      NearbyMedium.UPGRADE_TO_WEBRTC,
+      NearbyMedium.UPGRADE_TO_WIFIHOTSPOT,
+      NearbyMedium.UPGRADE_TO_WIFIDIRECT,
+      NearbyMedium.UPGRADE_TO_ALL_WIFI,
+  }
+
+
+@enum.unique
+class MediumUpgradeType(enum.IntEnum):
+  DEFAULT = 0
+  DISRUPTIVE = 1
+  NON_DISRUPTIVE = 2
+
+
+@enum.unique
+class WifiD2DType(enum.IntEnum):
+  SCC_2G = 0
+  SCC_5G = 1
+  MCC_2G_WFD_5G_STA = 2
+  MCC_2G_WFD_5G_INDOOR_STA = 3
+  MCC_5G_WFD_5G_DFS_STA = 4
+  MCC_5G_HS_5G_DFS_STA = 5
+
+
+@enum.unique
+class SingleTestFailureReason(enum.IntEnum):
+  """The failure reasons for a nearby connect connection test."""
+  UNINITIALIZED = 0
+  SOURCE_START_DISCOVERY = 1
+  TARGET_START_ADVERTISING = 2
+  SOURCE_REQUEST_CONNECTION = 3
+  TARGET_ACCEPT_CONNECTION = 4
+  WIFI_MEDIUM_UPGRADE = 5
+  FILE_TRANSFER_FAIL = 6
+  FILE_TRANSFER_THROUGHPUT_LOW = 7
+  SOURCE_WIFI_CONNECTION = 8
+  TARGET_WIFI_CONNECTION = 9
+  AP_IS_NOT_CONFIGURED = 10
+  SUCCESS = 11
+
+
+COMMON_TRIAGE_TIP: dict[SingleTestFailureReason, str] = {
+    SingleTestFailureReason.UNINITIALIZED: (
+        'not executed, the whole test was exited earlier; the devices may be'
+        'disconnected from the host, abnormal things, such as system crash, '
+        'mobly snippet was killed; Or something wrong with the script, check'
+        'the test running log and the corresponding bugreport log.'
+    ),
+    SingleTestFailureReason.SUCCESS: 'success!',
+    SingleTestFailureReason.SOURCE_START_DISCOVERY: (
+        'The source device can not start BLE scan.'
+    ),
+    SingleTestFailureReason.TARGET_START_ADVERTISING: (
+        'The target device can not start BLE advertising.'
+    ),
+    SingleTestFailureReason.SOURCE_REQUEST_CONNECTION: (
+        'The source device can not connect to the target device.'),
+    SingleTestFailureReason.TARGET_ACCEPT_CONNECTION: (
+        'The target device can not accept the connection.'),
+    SingleTestFailureReason.SOURCE_WIFI_CONNECTION: (
+        'The source device can not connect to the wifi network. '
+        '1) Check if the wifi ssid or password is correct;'
+        '2) Try to remove any saved wifi network from wifi settings;'
+        '3) Check if other device can connect to the same AP'
+        '4) Check the wifi related log on the source device.'
+    ),
+    SingleTestFailureReason.TARGET_WIFI_CONNECTION: (
+        '1) Check if the wifi ssid or password is correct;'
+        '2) Try to remove any saved wifi network from wifi settings;'
+        '3) Check if other device can connect to the same AP'
+        '4) Check the wifi related log on the target device.'
+    ),
+    SingleTestFailureReason.AP_IS_NOT_CONFIGURED: (
+        'The test AP is not set correctly in the test configuration file.'
+    ),
+}
+
+
+MEDIUM_UPGRADE_FAIL_TRIAGE_TIPS: dict[NearbyMedium, str] = {
+    NearbyMedium.WIFILAN_ONLY: (
+        ' WLAN, check if AP blocks mDNS traffic'
+    ),
+    NearbyMedium.UPGRADE_TO_WIFIHOTSPOT: (
+        ' HOTSPOT, check if the WFD group owner fails to start on'
+        ' the target side or the legacy STA fails to connect on the source side'
+    ),
+    NearbyMedium.UPGRADE_TO_WIFIDIRECT: (
+        ' WFD, check if the WFD group owner fails to start on the'
+        ' target side or WFD group client fails to connect on the source side'
+    ),
+    NearbyMedium.UPGRADE_TO_ALL_WIFI: (
+        ' all WiFI mediums, check if WFD, WLAN and HOTSPOT mediums are tried'
+        ' and if the failure is on the target or source side'
+    ),
+}
+
+
+@dataclasses.dataclass(frozen=True)
+class ConnectionSetupTimeouts:
+  """The timeouts of the nearby connection setup."""
+  discovery_timeout: datetime.timedelta | None = None
+  connection_init_timeout: datetime.timedelta | None = None
+  connection_result_timeout: datetime.timedelta | None = None
+
+
+@dataclasses.dataclass(frozen=False)
+class ConnectionSetupQualityInfo:
+  """The quality information of the nearby connection setup."""
+  discovery_latency: datetime.timedelta = UNSET_LATENCY
+  connection_latency: datetime.timedelta = UNSET_LATENCY
+  medium_upgrade_latency: datetime.timedelta = UNSET_LATENCY
+  medium_upgrade_expected: bool = False
+  upgrade_medium: NearbyConnectionMedium | None = None
+
+  def get_dict(self) -> dict[str, str]:
+    dict_repr = {
+        'discovery': f'{round(self.discovery_latency.total_seconds(), 1)}s',
+        'connection': f'{round(self.connection_latency.total_seconds(), 1)}s'
+    }
+    if self.medium_upgrade_expected:
+      dict_repr['upgrade'] = (
+          f'{round(self.medium_upgrade_latency.total_seconds(), 1)}s'
+      )
+    if self.upgrade_medium:
+      dict_repr['medium'] = self.upgrade_medium.name
+    return dict_repr
+
+
+@dataclasses.dataclass(frozen=False)
+class SingleTestResult:
+  """The test result of a single iteration."""
+
+  test_iteration: int = 0
+  is_failed_with_prior_bt: bool = False
+  failure_reason: SingleTestFailureReason = (
+      SingleTestFailureReason.UNINITIALIZED
+  )
+  result_message: str = ''
+  prior_nc_setup_quality_info: ConnectionSetupQualityInfo = dataclasses.field(
+      default_factory=ConnectionSetupQualityInfo
+  )
+  discoverer_sta_latency: datetime.timedelta = UNSET_LATENCY
+  file_transfer_nc_setup_quality_info: ConnectionSetupQualityInfo = (
+      dataclasses.field(default_factory=ConnectionSetupQualityInfo)
+  )
+  file_transfer_throughput_kbps: float = UNSET_THROUGHPUT_KBPS
+  advertiser_sta_latency: datetime.timedelta = UNSET_LATENCY
+  discoverer_sta_expected: bool = False
+  advertiser_wifi_expected: bool = False
+
+
+@dataclasses.dataclass(frozen=False)
+class NcPerformanceTestMetrics:
+  """Metrics data for quick start test."""
+
+  prior_bt_discovery_latencies: list[datetime.timedelta] = dataclasses.field(
+      default_factory=list[datetime.timedelta]
+  )
+  prior_bt_connection_latencies: list[datetime.timedelta] = dataclasses.field(
+      default_factory=list[datetime.timedelta]
+  )
+  discoverer_wifi_sta_latencies: list[datetime.timedelta] = dataclasses.field(
+      default_factory=list[datetime.timedelta]
+  )
+  file_transfer_discovery_latencies: list[datetime.timedelta] = (
+      dataclasses.field(default_factory=list[datetime.timedelta])
+  )
+  file_transfer_connection_latencies: list[datetime.timedelta] = (
+      dataclasses.field(default_factory=list[datetime.timedelta])
+  )
+  medium_upgrade_latencies: list[datetime.timedelta] = dataclasses.field(
+      default_factory=list[datetime.timedelta])
+  advertiser_wifi_sta_latencies: list[datetime.timedelta] = dataclasses.field(
+      default_factory=list[datetime.timedelta])
+  file_transfer_throughputs_kbps: list[float] = dataclasses.field(
+      default_factory=list[float])
+  upgraded_wifi_transfer_mediums: list[NearbyConnectionMedium] = (
+      dataclasses.field(default_factory=list[NearbyConnectionMedium]))
diff --git a/tests/bettertogether/betocq/nearby_connection_wrapper.py b/tests/bettertogether/betocq/nearby_connection_wrapper.py
new file mode 100644
index 000000000..8e0dd8fea
--- /dev/null
+++ b/tests/bettertogether/betocq/nearby_connection_wrapper.py
@@ -0,0 +1,417 @@
+#  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.
+
+"""Utils for handling Nearby Connection rpc."""
+
+import datetime
+import random
+import time
+
+from mobly import asserts
+from mobly import utils
+from mobly.controllers import android_device
+from mobly.controllers.android_device_lib import callback_handler_v2
+from mobly.controllers.android_device_lib import snippet_client_v2
+from mobly.snippet import callback_event
+
+from betocq import nc_constants
+
+# This number should be large enough to cover advertising interval, firmware
+# scheduling timing interval and user action delay
+ADVERTISING_TO_DISCOVERY_MAX_DELAY_SEC = 4
+
+
+class NearbyConnectionWrapper:
+  """Wrapper for Nearby Connection Snippet Client Operations."""
+
+  def __init__(
+      self,
+      advertiser: android_device.AndroidDevice,
+      discoverer: android_device.AndroidDevice,
+      advertiser_nearby: snippet_client_v2.SnippetClientV2,
+      discoverer_nearby: snippet_client_v2.SnippetClientV2,
+      advertising_discovery_medium: nc_constants.NearbyMedium = (
+          nc_constants.NearbyMedium.BLE_ONLY
+      ),
+      connection_medium: nc_constants.NearbyMedium = (
+          nc_constants.NearbyMedium.BT_ONLY
+      ),
+      upgrade_medium: nc_constants.NearbyMedium = (
+          nc_constants.NearbyMedium.BT_ONLY
+      ),
+  ):
+    self.advertiser = advertiser
+    self.discoverer = discoverer
+    self.service_id = utils.rand_ascii_str(8)
+    self.advertising_discovery_medium = advertising_discovery_medium
+    self.connection_medium = connection_medium
+    self.upgrade_medium = upgrade_medium
+    self.discoverer_nearby = discoverer_nearby
+    self.advertiser_nearby = advertiser_nearby
+    self.test_failure_reason = (
+        nc_constants.SingleTestFailureReason.UNINITIALIZED
+        )
+
+    self.connection_quality_info: nc_constants.ConnectionSetupQualityInfo = (
+        nc_constants.ConnectionSetupQualityInfo())
+
+    self._advertiser_connection_lifecycle_callback: (
+        callback_handler_v2.CallbackHandlerV2) = None
+    self._discoverer_endpoint_discovery_callback: (
+        callback_handler_v2.CallbackHandlerV2) = None
+    self._discoverer_connection_lifecycle_callback: (
+        callback_handler_v2.CallbackHandlerV2) = None
+    self._advertiser_payload_callback: (
+        callback_handler_v2.CallbackHandlerV2) = None
+    self._discoverer_payload_callback: (
+        callback_handler_v2.CallbackHandlerV2) = None
+    self._advertiser_endpoint_id: str = None
+    self._discoverer_endpoint_id: str = None
+
+  def start_advertising(self) -> None:
+    """Starts Nearby Connection advertising."""
+    advertiser_callback = self.advertiser_nearby.startAdvertising(
+        self.advertiser.serial,
+        self.service_id,
+        self.advertising_discovery_medium.value,
+        self.upgrade_medium.value,
+    )
+    self.advertiser.log.info(
+        f'Start advertising {self.advertising_discovery_medium.value}'
+    )
+    self._advertiser_connection_lifecycle_callback = advertiser_callback
+    self._advertiser_endpoint_id = self.advertiser_nearby.getLocalEndpointId()
+
+  def start_discovery(self, timeout: datetime.timedelta) -> None:
+    """Starts Nearby Connection discovery."""
+    self.discoverer.log.info(
+        f'Start discovery {self.advertising_discovery_medium.value}'
+    )
+    self._discoverer_endpoint_discovery_callback = (
+        self.discoverer_nearby.startDiscovery(
+            self.service_id, self.advertising_discovery_medium.value
+        )
+    )
+
+    endpoint_found_event = (
+        self._discoverer_endpoint_discovery_callback.waitAndGet(
+            'onEndpointFound', timeout=timeout.total_seconds()
+        )
+    )
+    endpoint_info = endpoint_found_event.data['discoveredEndpointInfo']
+    self.connection_quality_info.discovery_latency = datetime.timedelta(
+        microseconds=endpoint_found_event.data['discoveryTimeNs'] / 1_000
+    )
+    asserts.assert_equal(
+        endpoint_info['endpointName'], self.advertiser.serial,
+        'Received an unexpected endpoint during discovery: '
+        f'{endpoint_found_event}')
+
+    asserts.assert_equal(
+        endpoint_info['serviceId'], self.service_id,
+        f'Received an unexpected service id during discovery: '
+        f'{endpoint_found_event}')
+
+  def stop_advertising(self) -> None:
+    """Stops Nearby Connection advertising."""
+    self.advertiser_nearby.stopAdvertising()
+    self.advertiser.log.info('Stop advertising')
+
+  def stop_discovery(self) -> None:
+    """Stops Nearby Connection discovery."""
+    self.discoverer_nearby.stopDiscovery()
+    self.discoverer.log.info('Stop discovery')
+
+  def request_connection(
+      self,
+      medium_upgrade_type: nc_constants.MediumUpgradeType,
+      timeout: datetime.timedelta,
+      keep_alive_timeout_ms: int = nc_constants.KEEP_ALIVE_TIMEOUT_BT_MS,
+      keep_alive_interval_ms: int = nc_constants.KEEP_ALIVE_INTERVAL_BT_MS,
+  ) -> None:
+    """Requests Nearby Connection."""
+
+    self.discoverer.log.info(
+        'Start connection request with keep_alive_timeout_ms'
+        f' {keep_alive_timeout_ms}'
+    )
+    self._discoverer_connection_lifecycle_callback = (
+        self.discoverer_nearby.requestConnection(
+            self.discoverer.serial,
+            self._advertiser_endpoint_id,
+            self.connection_medium.value,
+            self.upgrade_medium.value,
+            medium_upgrade_type.value,
+            keep_alive_timeout_ms,
+            keep_alive_interval_ms,
+        )
+    )
+
+    d_connection_init_event = (
+        self._discoverer_connection_lifecycle_callback.waitAndGet(
+            'onConnectionInitiated', timeout.total_seconds()
+        )
+    )
+    self.connection_quality_info.connection_latency = datetime.timedelta(
+        microseconds=d_connection_init_event.data['connectionTimeNs'] / 1_000
+    )
+
+    d_connection_info = d_connection_init_event.data['connectionInfo']
+    asserts.assert_false(
+        d_connection_info['isIncomingConnection'],
+        f'Received an incoming connection: {d_connection_init_event}'
+        'but expected an outgoing connection')
+
+    asserts.assert_equal(
+        d_connection_info['endpointName'],
+        self.advertiser.serial,
+        f'Received an unexpected endpoint: {d_connection_init_event}')
+
+    self._discoverer_endpoint_id = self.discoverer_nearby.getLocalEndpointId()
+
+    # wait for the advertiser connection initialized.
+    a_connection_init_event = (
+        self._advertiser_connection_lifecycle_callback.waitAndGet(
+            'onConnectionInitiated', timeout=timeout.total_seconds()
+        )
+    )
+    a_connection_info = a_connection_init_event.data['connectionInfo']
+    asserts.assert_true(
+        a_connection_info['isIncomingConnection'],
+        f'Received an outgoing connection: {d_connection_init_event}'
+        'but expected an incoming connection')
+
+    asserts.assert_equal(
+        a_connection_info['endpointName'],
+        self.discoverer.serial,
+        f'Received an unexpected endpoint: {a_connection_init_event}')
+
+  def accept_connection(
+      self, timeout: datetime.timedelta
+  ) -> None:
+    """Accepts Nearby Connection."""
+    self._advertiser_payload_callback = (
+        self.advertiser_nearby.acceptConnection(
+            self._discoverer_endpoint_id
+        )
+    )
+    self.advertiser.log.info('Start connection accept')
+    self._discoverer_payload_callback = (
+        self.discoverer_nearby.acceptConnection(
+            self._advertiser_endpoint_id
+        )
+    )
+    self.discoverer.log.info('Start connection accept')
+
+    advertiser_connection_event = (
+        self._advertiser_connection_lifecycle_callback.waitAndGet(
+            'onConnectionResult', timeout=timeout.total_seconds()
+        )
+    )
+
+    asserts.assert_true(
+        advertiser_connection_event.data['isSuccess'],
+        f'Received an unsuccessful event: {advertiser_connection_event}')
+
+    asserts.assert_equal(
+        advertiser_connection_event.data['endpointId'],
+        self._discoverer_endpoint_id,
+        f'Received an unexpected endpoint: {advertiser_connection_event}')
+
+    discoverer_connection_event = (
+        self._discoverer_connection_lifecycle_callback.waitAndGet(
+            'onConnectionResult', timeout=timeout.total_seconds()
+        )
+    )
+    asserts.assert_true(
+        discoverer_connection_event.data['isSuccess'],
+        f'Received an unsuccessful event: {discoverer_connection_event}')
+
+    asserts.assert_equal(
+        discoverer_connection_event.data['endpointId'],
+        self._advertiser_endpoint_id,
+        f'Received an unexpected endpoint: {discoverer_connection_event}')
+
+    if nc_constants.is_high_quality_medium(self.upgrade_medium):
+      self.test_failure_reason = (
+          nc_constants.SingleTestFailureReason.WIFI_MEDIUM_UPGRADE
+      )
+      upgrade_start_time = datetime.datetime.now()
+      wait_high_quality = True
+      while wait_high_quality:
+        discoverer_medium_upgrade_event = self._discoverer_connection_lifecycle_callback.waitAndGet(
+            'onBandwidthChanged',
+            nc_constants.CONNECTION_BANDWIDTH_CHANGED_TIMEOUT.total_seconds(),
+        )
+        self.discoverer.log.info(
+            f'medium upgrade to {discoverer_medium_upgrade_event.data}'
+        )
+        if discoverer_medium_upgrade_event.data['isHighBwQuality']:
+          wait_high_quality = False
+          self.connection_quality_info.medium_upgrade_latency = (
+              datetime.datetime.now() - upgrade_start_time)
+          self.connection_quality_info.upgrade_medium = (
+              nc_constants.NearbyConnectionMedium(
+                  discoverer_medium_upgrade_event.data['medium']))
+          self.connection_quality_info.medium_upgrade_expected = True
+          self.discoverer.log.info(
+              f'upgraded to high quality medium: '
+              f'{self.connection_quality_info.upgrade_medium.name}')
+        else:
+          latency = datetime.datetime.now() - upgrade_start_time
+          if latency >= nc_constants.CONNECTION_BANDWIDTH_CHANGED_TIMEOUT:
+            raise TimeoutError('medium upgrade timeout')
+
+  def disconnect_endpoint(self) -> None:
+    """Disconnects Nearby Connection endpoint."""
+    if self:
+      self.discoverer_nearby.disconnectFromEndpoint(
+          self._advertiser_endpoint_id
+      )
+      self.discoverer.log.info(
+          f'Start disconnecting from endpoint: {self._advertiser_endpoint_id}'
+      )
+    else:
+      self.discoverer.log.info('no nearby connecty setup yet')
+      return nc_constants.OpResult(nc_constants.Result.SUCCESS)
+
+    if self._discoverer_connection_lifecycle_callback is not None:
+      disconnected_event = (
+          self._discoverer_connection_lifecycle_callback.waitAndGet(
+              'onDisconnected',
+              timeout=nc_constants.DISCONNECTION_TIMEOUT.total_seconds(),
+          )
+      )
+      asserts.assert_equal(
+          disconnected_event.data['endpointId'],
+          self._advertiser_endpoint_id,
+          f'Receive unexpected event on disconnect: {disconnected_event}')
+    self.discoverer.log.info(
+        f'disconnected with endpoint: {self._advertiser_endpoint_id}'
+    )
+
+  def start_nearby_connection(
+      self,
+      timeouts: nc_constants.ConnectionSetupTimeouts,
+      medium_upgrade_type: nc_constants.MediumUpgradeType = nc_constants.MediumUpgradeType.DEFAULT,
+      keep_alive_timeout_ms: int = nc_constants.KEEP_ALIVE_TIMEOUT_BT_MS,
+      keep_alive_interval_ms: int = nc_constants.KEEP_ALIVE_INTERVAL_BT_MS,
+  ) -> None:
+    """Starts Nearby Connection between two Android devices."""
+    self.test_failure_reason = (
+        nc_constants.SingleTestFailureReason.TARGET_START_ADVERTISING)
+    # Start advertising.
+    self.start_advertising()
+    # Add a random delay between adversting and discovery
+    # to mimic the random delay between two devices' user action
+    time.sleep(ADVERTISING_TO_DISCOVERY_MAX_DELAY_SEC * random.random())
+
+    self.test_failure_reason = (
+        nc_constants.SingleTestFailureReason.SOURCE_START_DISCOVERY)
+    # Start discovery.
+    self.start_discovery(timeout=timeouts.discovery_timeout)
+
+    # Request connection.
+    self.test_failure_reason = (
+        nc_constants.SingleTestFailureReason.SOURCE_REQUEST_CONNECTION)
+    self.request_connection(
+        medium_upgrade_type=medium_upgrade_type,
+        timeout=timeouts.connection_init_timeout,
+        keep_alive_timeout_ms=keep_alive_timeout_ms,
+        keep_alive_interval_ms=keep_alive_interval_ms)
+
+    # Stop discovery.
+    self.stop_discovery()
+
+    # Accept connection.
+    self.test_failure_reason = (
+        nc_constants.SingleTestFailureReason.TARGET_ACCEPT_CONNECTION)
+    self.accept_connection(timeout=timeouts.connection_result_timeout)
+
+    # Stop advertising.
+    self.stop_advertising()
+    self.test_failure_reason = nc_constants.SingleTestFailureReason.SUCCESS
+
+  def transfer_file(
+      self,
+      file_size_kb: int,
+      timeout: datetime.timedelta,
+      payload_type: nc_constants.PayloadType,
+  ) -> float:
+    """Sends payloads and returns the transfer speed in kBS."""
+    try:
+      self.test_failure_reason = (
+          nc_constants.SingleTestFailureReason.FILE_TRANSFER_FAIL
+      )
+      transfer_speed_kbs = self._transfer_file(
+          file_size_kb, timeout, payload_type
+      )
+      self.test_failure_reason = nc_constants.SingleTestFailureReason.SUCCESS
+    finally:
+      # clean up
+      utils.concurrent_exec(
+          lambda nb: nb.transferFilesCleanup(),
+          param_list=[[self.discoverer_nearby], [self.advertiser_nearby]],
+          raise_on_exception=True)
+    return transfer_speed_kbs
+
+  def _transfer_file(
+      self, file_size_kb: int, timeout: datetime.timedelta,
+      payload_type: nc_constants.PayloadType
+  ) -> float:
+    """Sends payloads and returns the transfer speed in kBS."""
+    # Creates a file and send it to the advertiser.
+    file_name = utils.rand_ascii_str(8)
+    self.discoverer.log.info(
+        f'Start sending payloads with type: {payload_type.name}'
+    )
+    payload_id = self.discoverer_nearby.sendPayloadWithType(
+        self._advertiser_endpoint_id, file_name, file_size_kb, payload_type
+    )
+
+    # Waits for the advertiser received.
+    def on_receive(event: callback_event.CallbackEvent) -> bool:
+      return (
+          event.data['endpointId'] == self._discoverer_endpoint_id
+          and event.data['payload']['id'] == payload_id
+      )
+
+    asserts.assert_is_not_none(
+        self._advertiser_payload_callback,
+        'No nearby connection is set up, advertiser payload cb is none.')
+    asserts.assert_is_not_none(
+        self._discoverer_payload_callback,
+        'No nearby connection is set up, discoverer payload cb is none.')
+
+    self._advertiser_payload_callback.waitForEvent(
+        'onPayloadReceived',
+        predicate=on_receive,
+        timeout=timeout.total_seconds())
+
+    # Waits for complete transfer.
+    self._advertiser_payload_callback.waitForEvent(
+        'onPayloadTransferUpdate',
+        predicate=lambda event: event.data['update']['isSuccess'],
+        timeout=timeout.total_seconds())
+
+    payload_transfer_event = self._discoverer_payload_callback.waitForEvent(
+        'onPayloadTransferUpdate',
+        predicate=lambda event: event.data['update']['isSuccess'],
+        timeout=timeout.total_seconds(),
+    )
+    self.advertiser.log.info('payload received')
+
+    transfer_time = datetime.timedelta(
+        microseconds=payload_transfer_event.data['transferTimeNs'] / 1_000)
+    return round(file_size_kb/transfer_time.total_seconds())
diff --git a/tests/bettertogether/betocq/setup_utils.py b/tests/bettertogether/betocq/setup_utils.py
new file mode 100644
index 000000000..b8ae7505c
--- /dev/null
+++ b/tests/bettertogether/betocq/setup_utils.py
@@ -0,0 +1,390 @@
+#  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.
+
+"""Android Nearby device setup."""
+
+import datetime
+import time
+from typing import Mapping
+
+from mobly.controllers import android_device
+from mobly.controllers.android_device_lib import adb
+
+from betocq import gms_auto_updates_util
+
+WIFI_COUNTRYCODE_CONFIG_TIME_SEC = 3
+TOGGLE_AIRPLANE_MODE_WAIT_TIME_SEC = 2
+PH_FLAG_WRITE_WAIT_TIME_SEC = 3
+WIFI_DISCONNECTION_DELAY_SEC = 3
+ADB_RETRY_WAIT_TIME_SEC = 2
+
+_DISABLE_ENABLE_GMS_UPDATE_WAIT_TIME_SEC = 2
+
+LOG_TAGS = [
+    'Nearby',
+    'NearbyMessages',
+    'NearbyDiscovery',
+    'NearbyConnections',
+    'NearbyMediums',
+    'NearbySetup',
+]
+
+
+def set_country_code(
+    ad: android_device.AndroidDevice, country_code: str
+) -> None:
+  """Sets Wi-Fi and Telephony country code.
+
+  When you set the phone to EU or JP, the available 5GHz channels shrinks.
+  Some phones, like Pixel 2, can't use Wi-Fi Direct or Hotspot on 5GHz
+  in these countries. Pixel 3+ can, but only on some channels.
+  Not all of them. So, test Nearby Share or Nearby Connections without
+  Wi-Fi LAN to catch any bugs and make sure we don't break it later.
+
+  Args:
+    ad: AndroidDevice, Mobly Android Device.
+    country_code: WiFi and Telephony Country Code.
+  """
+  try:
+    _do_set_country_code(ad, country_code)
+  except adb.AdbError:
+    ad.log.exception(
+        f'Failed to set country code on device "{ad.serial}, try again.'
+    )
+    time.sleep(ADB_RETRY_WAIT_TIME_SEC)
+    _do_set_country_code(ad, country_code)
+
+
+def _do_set_country_code(
+    ad: android_device.AndroidDevice, country_code: str
+) -> None:
+  """Sets Wi-Fi and Telephony country code."""
+  if not ad.is_adb_root:
+    ad.log.info(
+        f'Skipped setting wifi country code on device "{ad.serial}" '
+        'because we do not set country code on unrooted phone.'
+    )
+    return
+
+  ad.log.info(f'Set Wi-Fi and Telephony country code to {country_code}.')
+  ad.adb.shell('cmd wifi set-wifi-enabled disabled')
+  time.sleep(WIFI_COUNTRYCODE_CONFIG_TIME_SEC)
+  ad.adb.shell(
+      'am broadcast -a com.android.internal.telephony.action.COUNTRY_OVERRIDE'
+      f' --es country {country_code}'
+  )
+  ad.adb.shell(f'cmd wifi force-country-code enabled {country_code}')
+  enable_airplane_mode(ad)
+  time.sleep(WIFI_COUNTRYCODE_CONFIG_TIME_SEC)
+  disable_airplane_mode(ad)
+  ad.adb.shell('cmd wifi set-wifi-enabled enabled')
+  telephony_country_code = (
+      ad.adb.shell('dumpsys wifi | grep mTelephonyCountryCode')
+      .decode('utf-8')
+      .strip()
+  )
+  ad.log.info(f'Telephony country code: {telephony_country_code}')
+
+
+def enable_logs(ad: android_device.AndroidDevice) -> None:
+  """Enables Nearby related logs."""
+  ad.log.info('Enable Nearby loggings.')
+  for tag in LOG_TAGS:
+    ad.adb.shell(f'setprop log.tag.{tag} VERBOSE')
+
+
+def grant_manage_external_storage_permission(
+    ad: android_device.AndroidDevice, package_name: str
+) -> None:
+  """Grants MANAGE_EXTERNAL_STORAGE permission to Nearby snippet."""
+  try:
+    _do_grant_manage_external_storage_permission(ad, package_name)
+  except adb.AdbError:
+    ad.log.exception(
+        'Failed to grant MANAGE_EXTERNAL_STORAGE permission on device'
+        f' "{ad.serial}", try again.'
+    )
+    time.sleep(ADB_RETRY_WAIT_TIME_SEC)
+    _do_grant_manage_external_storage_permission(ad, package_name)
+
+
+def _do_grant_manage_external_storage_permission(
+    ad: android_device.AndroidDevice, package_name: str
+) -> None:
+  """Grants MANAGE_EXTERNAL_STORAGE permission to Nearby snippet."""
+  build_version_sdk = int(ad.build_info['build_version_sdk'])
+  if build_version_sdk < 30:
+    return
+  ad.log.info(
+      f'Grant MANAGE_EXTERNAL_STORAGE permission on device "{ad.serial}".'
+  )
+  _grant_manage_external_storage_permission(ad, package_name)
+
+
+def dump_gms_version(ad: android_device.AndroidDevice) -> Mapping[str, str]:
+  """Dumps GMS version from dumpsys to sponge properties."""
+  try:
+    gms_version = _do_dump_gms_version(ad)
+  except adb.AdbError:
+    ad.log.exception(
+        f'Failed to dump GMS version on device "{ad.serial}", try again.'
+    )
+    time.sleep(ADB_RETRY_WAIT_TIME_SEC)
+    gms_version = _do_dump_gms_version(ad)
+  return gms_version
+
+
+def _do_dump_gms_version(ad: android_device.AndroidDevice) -> Mapping[str, str]:
+  """Dumps GMS version from dumpsys to sponge properties."""
+  out = (
+      ad.adb.shell(
+          'dumpsys package com.google.android.gms | grep "versionCode="'
+      )
+      .decode('utf-8')
+      .strip()
+  )
+  return {f'GMS core version on {ad.serial}': out}
+
+
+def toggle_airplane_mode(ad: android_device.AndroidDevice) -> None:
+  """Toggles airplane mode on the given device."""
+  ad.log.info('turn on airplane mode')
+  enable_airplane_mode(ad)
+  ad.log.info('turn off airplane mode')
+  disable_airplane_mode(ad)
+
+
+def connect_to_wifi_sta_till_success(
+    ad: android_device.AndroidDevice, wifi_ssid: str, wifi_password: str
+) -> datetime.timedelta:
+  """Connecting to the specified wifi STA/AP."""
+  ad.log.info('Start connecting to wifi STA/AP')
+  wifi_connect_start = datetime.datetime.now()
+  if not wifi_password:
+    wifi_password = None
+  connect_to_wifi(ad, wifi_ssid, wifi_password)
+  return datetime.datetime.now() - wifi_connect_start
+
+
+def connect_to_wifi(
+    ad: android_device.AndroidDevice,
+    ssid: str,
+    password: str | None = None,
+) -> None:
+  if not ad.nearby.wifiIsEnabled():
+    ad.nearby.wifiEnable()
+  # return until the wifi is connected.
+  ad.nearby.wifiConnectSimple(ssid, password)
+
+
+def disconnect_from_wifi(ad: android_device.AndroidDevice) -> None:
+  if not ad.is_adb_root:
+    ad.log.info("Can't clear wifi network in non-rooted device")
+    return
+  ad.nearby.wifiClearConfiguredNetworks()
+  time.sleep(WIFI_DISCONNECTION_DELAY_SEC)
+
+
+def _grant_manage_external_storage_permission(
+    ad: android_device.AndroidDevice, package_name: str
+) -> None:
+  """Grants MANAGE_EXTERNAL_STORAGE permission to Nearby snippet.
+
+  This permission will not grant automatically by '-g' option of adb install,
+  you can check the all permission granted by:
+  am start -a android.settings.APPLICATION_DETAILS_SETTINGS
+           -d package:{YOUR_PACKAGE}
+
+  Reference for MANAGE_EXTERNAL_STORAGE:
+  https://developer.android.com/training/data-storage/manage-all-files
+
+  This permission will reset to default "Allow access to media only" after
+  reboot if you never grant "Allow management of all files" through system UI.
+  The appops command and MANAGE_EXTERNAL_STORAGE only available on API 30+.
+
+  Args:
+    ad: AndroidDevice, Mobly Android Device.
+    package_name: The nearbu snippet package name.
+  """
+  try:
+    ad.adb.shell(
+        f'appops set --uid {package_name} MANAGE_EXTERNAL_STORAGE allow'
+    )
+  except adb.Error:
+    ad.log.info('Failed to grant MANAGE_EXTERNAL_STORAGE permission.')
+
+
+def enable_airplane_mode(ad: android_device.AndroidDevice) -> None:
+  """Enables airplane mode on the given device."""
+  try:
+    _do_enable_airplane_mode(ad)
+  except adb.AdbError:
+    ad.log.exception(
+        f'Failed to enable airplane mode on device "{ad.serial}", try again.'
+    )
+    time.sleep(ADB_RETRY_WAIT_TIME_SEC)
+    _do_enable_airplane_mode(ad)
+
+
+def _do_enable_airplane_mode(ad: android_device.AndroidDevice) -> None:
+  if (ad.is_adb_root):
+    ad.adb.shell(['settings', 'put', 'global', 'airplane_mode_on', '1'])
+    ad.adb.shell([
+        'am', 'broadcast', '-a', 'android.intent.action.AIRPLANE_MODE', '--ez',
+        'state', 'true'
+    ])
+  ad.adb.shell(['svc', 'wifi', 'disable'])
+  ad.adb.shell(['svc', 'bluetooth', 'disable'])
+  time.sleep(TOGGLE_AIRPLANE_MODE_WAIT_TIME_SEC)
+
+
+def disable_airplane_mode(ad: android_device.AndroidDevice) -> None:
+  """Disables airplane mode on the given device."""
+  try:
+    _do_disable_airplane_mode(ad)
+  except adb.AdbError:
+    ad.log.exception(
+        f'Failed to disable airplane mode on device "{ad.serial}", try again.'
+    )
+    time.sleep(ADB_RETRY_WAIT_TIME_SEC)
+    _do_disable_airplane_mode(ad)
+
+
+def _do_disable_airplane_mode(ad: android_device.AndroidDevice) -> None:
+  if (ad.is_adb_root):
+    ad.adb.shell(['settings', 'put', 'global', 'airplane_mode_on', '0'])
+    ad.adb.shell([
+        'am', 'broadcast', '-a', 'android.intent.action.AIRPLANE_MODE', '--ez',
+        'state', 'false'
+    ])
+  ad.adb.shell(['svc', 'wifi', 'enable'])
+  ad.adb.shell(['svc', 'bluetooth', 'enable'])
+  time.sleep(TOGGLE_AIRPLANE_MODE_WAIT_TIME_SEC)
+
+
+def check_if_ph_flag_committed(
+    ad: android_device.AndroidDevice,
+    pname: str,
+    flag_name: str,
+) -> bool:
+  """Check if P/H flag is committed."""
+  sql_str = (
+      'sqlite3 /data/data/com.google.android.gms/databases/phenotype.db'
+      ' "select name, quote(coalesce(intVal, boolVal, floatVal, stringVal,'
+      ' extensionVal)) from FlagOverrides where committed=1 AND'
+      f' packageName=\'{pname}\';"'
+  )
+  flag_result = ad.adb.shell(sql_str).decode('utf-8').strip()
+  return flag_name in flag_result
+
+
+def write_ph_flag(
+    ad: android_device.AndroidDevice,
+    pname: str,
+    flag_name: str,
+    flag_type: str,
+    flag_value: str,
+) -> None:
+  """Write P/H flag."""
+  ad.adb.shell(
+      'am broadcast -a "com.google.android.gms.phenotype.FLAG_OVERRIDE" '
+      f'--es package "{pname}" --es user "*" '
+      f'--esa flags "{flag_name}" '
+      f'--esa types "{flag_type}" --esa values "{flag_value}" '
+      'com.google.android.gms'
+  )
+  time.sleep(PH_FLAG_WRITE_WAIT_TIME_SEC)
+
+
+def check_and_try_to_write_ph_flag(
+    ad: android_device.AndroidDevice,
+    pname: str,
+    flag_name: str,
+    flag_type: str,
+    flag_value: str,
+) -> None:
+  """Check and try to enable the given flag on the given device."""
+  if(not ad.is_adb_root):
+    ad.log.info(
+        "Can't read or write P/H flag value in non-rooted device. Use Mobile"
+        ' Utility app to config instead.'
+    )
+    return
+
+  if check_if_ph_flag_committed(ad, pname, flag_name):
+    ad.log.info(f'{flag_name} is already committed.')
+    return
+  ad.log.info(f'write {flag_name}.')
+  write_ph_flag(ad, pname, flag_name, flag_type, flag_value)
+
+  if check_if_ph_flag_committed(ad, pname, flag_name):
+    ad.log.info(f'{flag_name} is configured successfully.')
+  else:
+    ad.log.info(f'failed to configure {flag_name}.')
+
+
+def enable_bluetooth_multiplex(ad: android_device.AndroidDevice) -> None:
+  """Enable bluetooth multiplex on the given device."""
+  pname = 'com.google.android.gms.nearby'
+  flag_name = 'mediums_supports_bluetooth_multiplex_socket'
+  flag_type = 'boolean'
+  flag_value = 'true'
+  check_and_try_to_write_ph_flag(ad, pname, flag_name, flag_type, flag_value)
+
+
+def enable_wifi_aware(ad: android_device.AndroidDevice) -> None:
+  """Enable wifi aware on the given device."""
+  pname = 'com.google.android.gms.nearby'
+  flag_name = 'mediums_supports_wifi_aware'
+  flag_type = 'boolean'
+  flag_value = 'true'
+
+  check_and_try_to_write_ph_flag(ad, pname, flag_name, flag_type, flag_value)
+
+
+def disable_redaction(ad: android_device.AndroidDevice) -> None:
+  """Disable info log redaction on the given device."""
+  pname = 'com.google.android.gms'
+  flag_name = 'ClientLogging__enable_info_log_redaction'
+  flag_type = 'boolean'
+  flag_value = 'false'
+
+  check_and_try_to_write_ph_flag(ad, pname, flag_name, flag_type, flag_value)
+
+
+def install_apk(ad: android_device.AndroidDevice, apk_path: str) -> None:
+  """Installs the apk on the given device."""
+  ad.adb.install(['-r', '-g', '-t', apk_path])
+
+
+def disable_gms_auto_updates(ad: android_device.AndroidDevice) -> None:
+  """Disable GMS auto updates on the given device."""
+  if not ad.is_adb_root:
+    ad.log.warning(
+        'You should disable the play store auto updates manually on a'
+        'unrooted device, otherwise the test may be broken unexpected')
+  ad.log.info('try to disable GMS Auto Updates.')
+  gms_auto_updates_util.GmsAutoUpdatesUtil(ad).disable_gms_auto_updates()
+  time.sleep(_DISABLE_ENABLE_GMS_UPDATE_WAIT_TIME_SEC)
+
+
+def enable_gms_auto_updates(ad: android_device.AndroidDevice) -> None:
+  """Enable GMS auto updates on the given device."""
+  if not ad.is_adb_root:
+    ad.log.warning(
+        'You may enable the play store auto updates manually on a'
+        'unrooted device after test.')
+  ad.log.info('try to enable GMS Auto Updates.')
+  gms_auto_updates_util.GmsAutoUpdatesUtil(ad).enable_gms_auto_updates()
+  time.sleep(_DISABLE_ENABLE_GMS_UPDATE_WAIT_TIME_SEC)
diff --git a/tests/bettertogether/betocq/version.py b/tests/bettertogether/betocq/version.py
new file mode 100644
index 000000000..a21f32a58
--- /dev/null
+++ b/tests/bettertogether/betocq/version.py
@@ -0,0 +1,21 @@
+#  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.
+
+"""Define the Beto CQ test script version."""
+
+
+TEST_SCRIPT_VERSION = '2.0.0'
+
+# VERSION_LOG (only add new description for new version, keep the history log)
+# '2.0.0': 'initial version'