New test cases and util related to data connection re-establish are added into class Nsa5gDSDSVoiceTest.

16 new test cases with idle:
- test_msim_long_voice_esim_mo_5g_nsa_volte_psim_5g_nsa_volte_dds_0
- test_msim_long_voice_esim_mt_5g_nsa_volte_psim_5g_nsa_volte_dds_0
- test_msim_long_voice_esim_mo_volte_psim_5g_nsa_volte_dds_0
- test_msim_long_voice_esim_mt_volte_psim_5g_nsa_volte_dds_0
- test_msim_long_voice_esim_mo_5g_nsa_volte_psim_volte_dds_0
- test_msim_long_voice_esim_mt_5g_nsa_volte_psim_volte_dds_0
- test_msim_long_voice_psim_mo_5g_nsa_volte_esim_5g_nsa_volte_dds_1
- test_msim_long_voice_psim_mt_5g_nsa_volte_esim_5g_nsa_volte_dds_1
- test_msim_long_voice_psim_mo_5g_nsa_volte_esim_volte_dds_1
- test_msim_long_voice_psim_mt_5g_nsa_volte_esim_volte_dds_1
- test_msim_long_voice_psim_mo_volte_esim_5g_nsa_volte_dds_1
- test_msim_long_voice_psim_mt_volte_esim_5g_nsa_volte_dds_1
- test_msim_long_voice_esim_mo_volte_psim_volte_dds_0
- test_msim_long_voice_esim_mt_volte_psim_volte_dds_0
- test_msim_long_voice_psim_mo_volte_esim_volte_dds_1
- test_msim_long_voice_psim_mt_volte_esim_volte_dds_1

16 new test cases with streaming:
- test_msim_long_voice_streaming_esim_mo_5g_nsa_volte_psim_5g_nsa_volte_dds_0
- test_msim_long_voice_streaming_esim_mt_5g_nsa_volte_psim_5g_nsa_volte_dds_0
- test_msim_long_voice_streaming_esim_mo_volte_psim_5g_nsa_volte_dds_0
- test_msim_long_voice_streaming_esim_mt_volte_psim_5g_nsa_volte_dds_0
- test_msim_long_voice_streaming_esim_mo_5g_nsa_volte_psim_volte_dds_0
- test_msim_long_voice_streaming_esim_mt_5g_nsa_volte_psim_volte_dds_0
- test_msim_long_voice_streaming_psim_mo_5g_nsa_volte_esim_5g_nsa_volte_dds_1
- test_msim_long_voice_streaming_psim_mt_5g_nsa_volte_esim_5g_nsa_volte_dds_1
- test_msim_long_voice_streaming_psim_mo_5g_nsa_volte_esim_volte_dds_1
- test_msim_long_voice_streaming_psim_mt_5g_nsa_volte_esim_volte_dds_1
- test_msim_long_voice_streaming_psim_mo_volte_esim_5g_nsa_volte_dds_1
- test_msim_long_voice_streaming_psim_mt_volte_esim_5g_nsa_volte_dds_1
- test_msim_long_voice_streaming_esim_mo_volte_psim_volte_dds_0
- test_msim_long_voice_streaming_esim_mt_volte_psim_volte_dds_0
- test_msim_long_voice_streaming_psim_mo_volte_esim_volte_dds_1
- test_msim_long_voice_streaming_psim_mt_volte_esim_volte_dds_1

Bug: 204406125
Test: Yes, locally
Change-Id: I091519fe2598c0f231ffb162a043638387688ca4
diff --git a/acts_tests/acts_contrib/test_utils/tel/tel_dsds_utils.py b/acts_tests/acts_contrib/test_utils/tel/tel_dsds_utils.py
index 0a93e8f..6b15886 100644
--- a/acts_tests/acts_contrib/test_utils/tel/tel_dsds_utils.py
+++ b/acts_tests/acts_contrib/test_utils/tel/tel_dsds_utils.py
@@ -14,14 +14,18 @@
 #   See the License for the specific language governing permissions and
 #   limitations under the License.
 
+from datetime import datetime, timedelta
 import re
 import time
-from datetime import datetime, timedelta
+from typing import Optional, Sequence
 
 from acts import signals
+from acts import tracelogger
+from acts.controllers.android_device import AndroidDevice
 from acts.utils import rand_ascii_str
 from acts.libs.utils.multithread import multithread_func
 from acts_contrib.test_utils.tel.loggers.protos.telephony_metric_pb2 import TelephonyVoiceTestResult
+from acts_contrib.test_utils.tel.loggers.telephony_metric_logger import TelephonyMetricLogger
 from acts_contrib.test_utils.tel.tel_defines import INVALID_SUB_ID
 from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_SMS_RECEIVE
 from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_ANDROID_STATE_SETTLING
@@ -76,6 +80,237 @@
 CallResult = TelephonyVoiceTestResult.CallResult.Value
 
 
+def dsds_long_call_streaming_test(
+    log: tracelogger.TraceLogger,
+    tel_logger: TelephonyMetricLogger.for_test_case,
+    ads: Sequence[AndroidDevice],
+    test_rat: list,
+    test_slot: int,
+    dds_slot: int,
+    direction: str = "mo",
+    duration: int = 360,
+    streaming: bool = True,
+    is_airplane_mode = False,
+    wfc_mode: list[str, str] = [
+        WFC_MODE_CELLULAR_PREFERRED,
+        WFC_MODE_CELLULAR_PREFERRED],
+    wifi_network_ssid: Optional[str] = None,
+    wifi_network_pass: Optional[str] = None,
+    turn_off_wifi_in_the_end: bool = False,
+    turn_off_airplane_mode_in_the_end: bool = False) -> bool:
+    """Make MO/MT call at specific slot in specific RAT with DDS at specific
+    slot for the given time.
+
+    Args:
+        log: Logger object.
+        tel_logger: Logger object for telephony proto.
+        ads: A list of Android device objects.
+        test_rat: RAT for both slots of primary device.
+        test_slot: The slot which make/receive MO/MT call of primary device.
+        dds_slot: Preferred data slot of primary device.
+        direction: The direction of call("mo" or "mt").
+        streaming: True for playing Youtube and False on the contrary.
+        is_airplane_mode: True or False for WFC setup
+        wfc_mode: Cellular preferred or Wi-Fi preferred.
+        wifi_network_ssid: SSID of Wi-Fi AP.
+        wifi_network_pass: Password of Wi-Fi AP SSID.
+        turn_off_wifi_in_the_end: True to turn off Wi-Fi and False not to turn
+            off Wi-Fi in the end of the function.
+        turn_off_airplane_mode_in_the_end: True to turn off airplane mode and
+            False not to turn off airplane mode in the end of the function.
+
+    Returns:
+        TestFailure if failed.
+    """
+    log.info("Step 1: Switch DDS.")
+    if not set_dds_on_slot(ads[0], dds_slot):
+        ads[0].log.error(
+            "Failed to set DDS at slot %s on %s",(dds_slot, ads[0].serial))
+        return False
+
+    log.info("Step 2: Check HTTP connection after DDS switch.")
+    if not verify_http_connection(log, ads[0]):
+        ads[0].log.error("Failed to verify http connection.")
+        return False
+    else:
+        ads[0].log.info("Verify http connection successfully.")
+
+    log.info("Step 3: Set up phones in desired RAT.")
+    if direction == "mo":
+        # setup voice subid on primary device.
+        ad_mo = ads[0]
+        mo_sub_id = get_subid_from_slot_index(log, ad_mo, test_slot)
+        if mo_sub_id == INVALID_SUB_ID:
+            ad_mo.log.warning("Failed to get sub ID at slot %s.", test_slot)
+            return False
+        mo_other_sub_id = get_subid_from_slot_index(
+            log, ad_mo, 1-test_slot)
+        sub_id_list = [mo_sub_id, mo_other_sub_id]
+        set_voice_sub_id(ad_mo, mo_sub_id)
+        ad_mo.log.info("Sub ID for outgoing call at slot %s: %s", test_slot,
+        get_outgoing_voice_sub_id(ad_mo))
+
+        # setup voice subid on secondary device.
+        ad_mt = ads[1]
+        _, mt_sub_id, _ = get_subid_on_same_network_of_host_ad(ads)
+        if mt_sub_id == INVALID_SUB_ID:
+            ad_mt.log.warning("Failed to get sub ID at default voice slot.")
+            return False
+        mt_slot = get_slot_index_from_subid(ad_mt, mt_sub_id)
+        set_voice_sub_id(ad_mt, mt_sub_id)
+        ad_mt.log.info("Sub ID for incoming call at slot %s: %s", mt_slot,
+        get_outgoing_voice_sub_id(ad_mt))
+
+        # setup the rat on non-test slot(primary device).
+        phone_setup_on_rat(
+            log,
+            ad_mo,
+            test_rat[1-test_slot],
+            mo_other_sub_id,
+            is_airplane_mode,
+            wfc_mode[1-test_slot],
+            wifi_network_ssid,
+            wifi_network_pass)
+        # assign phone setup argv for test slot.
+        mo_phone_setup_func_argv = (
+            log,
+            ad_mo,
+            test_rat[test_slot],
+            mo_sub_id,
+            is_airplane_mode,
+            wfc_mode[test_slot],
+            wifi_network_ssid,
+            wifi_network_pass)
+        verify_caller_func = is_phone_in_call_on_rat(
+            log, ad_mo, test_rat[test_slot], only_return_fn=True)
+        mt_phone_setup_func_argv = (log, ad_mt, 'general')
+        verify_callee_func = is_phone_in_call_on_rat(
+            log, ad_mt, 'general', only_return_fn=True)
+    else:
+        # setup voice subid on primary device.
+        ad_mt = ads[0]
+        mt_sub_id = get_subid_from_slot_index(log, ad_mt, test_slot)
+        if mt_sub_id == INVALID_SUB_ID:
+            ad_mt.log.warning("Failed to get sub ID at slot %s.", test_slot)
+            return False
+        mt_other_sub_id = get_subid_from_slot_index(
+            log, ad_mt, 1-test_slot)
+        sub_id_list = [mt_sub_id, mt_other_sub_id]
+        set_voice_sub_id(ad_mt, mt_sub_id)
+        ad_mt.log.info("Sub ID for incoming call at slot %s: %s", test_slot,
+        get_outgoing_voice_sub_id(ad_mt))
+
+        # setup voice subid on secondary device.
+        ad_mo = ads[1]
+        _, mo_sub_id, _ = get_subid_on_same_network_of_host_ad(ads)
+        if mo_sub_id == INVALID_SUB_ID:
+            ad_mo.log.warning("Failed to get sub ID at default voice slot.")
+            return False
+        mo_slot = get_slot_index_from_subid(ad_mo, mo_sub_id)
+        set_voice_sub_id(ad_mo, mo_sub_id)
+        ad_mo.log.info("Sub ID for outgoing call at slot %s: %s", mo_slot,
+        get_outgoing_voice_sub_id(ad_mo))
+
+        # setup the rat on non-test slot(primary device).
+        phone_setup_on_rat(
+            log,
+            ad_mt,
+            test_rat[1-test_slot],
+            mt_other_sub_id,
+            is_airplane_mode,
+            wfc_mode[1-test_slot],
+            wifi_network_ssid,
+            wifi_network_pass)
+        # assign phone setup argv for test slot.
+        mt_phone_setup_func_argv = (
+            log,
+            ad_mt,
+            test_rat[test_slot],
+            mt_sub_id,
+            is_airplane_mode,
+            wfc_mode[test_slot],
+            wifi_network_ssid,
+            wifi_network_pass)
+        verify_callee_func = is_phone_in_call_on_rat(
+            log, ad_mt, test_rat[test_slot], only_return_fn=True)
+        mo_phone_setup_func_argv = (log, ad_mo, 'general')
+        verify_caller_func = is_phone_in_call_on_rat(
+            log, ad_mo, 'general', only_return_fn=True)
+
+    tasks = [(phone_setup_on_rat, mo_phone_setup_func_argv),
+             (phone_setup_on_rat, mt_phone_setup_func_argv)]
+    if not multithread_func(log, tasks):
+        log.error("Phone Failed to Set Up Properly.")
+        tel_logger.set_result(CallResult("CALL_SETUP_FAILURE"))
+        raise signals.TestFailure("Failed",
+            extras={"fail_reason": "Phone Failed to Set Up Properly."})
+    if streaming:
+        log.info("Step 4-0: Start Youtube streaming.")
+        if not start_youtube_video(ads[0]):
+            raise signals.TestFailure("Failed",
+                extras={"fail_reason": "Fail to bring up youtube video."})
+        time.sleep(10)
+
+    log.info("Step 4: Make voice call.")
+    result = call_setup_teardown(log,
+                                 ad_mo,
+                                 ad_mt,
+                                 ad_hangup=ad_mo,
+                                 verify_caller_func=verify_caller_func,
+                                 verify_callee_func=verify_callee_func,
+                                 wait_time_in_call=duration)
+    tel_logger.set_result(result.result_value)
+
+    if not result:
+        log.error(
+            "Failed to make %s call from %s slot %s to %s slot %s",
+                direction, ad_mo.serial, mo_slot, ad_mt.serial, mt_slot)
+        raise signals.TestFailure("Failed",
+            extras={"fail_reason": str(result.result_value)})
+
+    log.info("Step 5: Verify RAT and HTTP connection.")
+    # For the tese cases related to WFC in which airplane mode will be turned
+    # off in the end.
+    if turn_off_airplane_mode_in_the_end:
+        log.info("Step 5-1: Turning off airplane mode......")
+        if not toggle_airplane_mode(log, ads[0], False):
+            ads[0].log.error('Failed to toggle off airplane mode.')
+
+    # For the tese cases related to WFC in which Wi-Fi will be turned off in the
+    # end.
+
+    rat_list = [test_rat[test_slot], test_rat[1-test_slot]]
+
+    if turn_off_wifi_in_the_end:
+        log.info("Step 5-2: Turning off Wi-Fi......")
+        if not wifi_toggle_state(log, ads[0], False):
+            ads[0].log.error('Failed to toggle off Wi-Fi.')
+            return False
+
+        for index, value in enumerate(rat_list):
+            if value == '5g_wfc':
+                rat_list[index] = '5g'
+            elif value == 'wfc':
+                rat_list[index] = '4g'
+
+    for rat, sub_id in zip(rat_list, sub_id_list):
+        if not wait_for_network_idle(log, ads[0], rat, sub_id):
+            raise signals.TestFailure(
+                "Failed",
+                extras={
+                    "fail_reason": "Idle state of sub ID %s does not match the "
+                    "given RAT %s." % (sub_id, rat)})
+
+    if not verify_http_connection(log, ads[0]):
+        ads[0].log.error("Failed to verify http connection.")
+        return False
+    else:
+        ads[0].log.info("Verify http connection successfully.")
+
+    if streaming:
+        ads[0].force_stop_apk(YOUTUBE_PACKAGE_NAME)
+
+
 def dsds_voice_call_test(
         log,
         tel_logger,