Merge "Add support for various callbox network options"
diff --git a/acts/framework/acts/controllers/anritsu_lib/md8475_cellular_simulator.py b/acts/framework/acts/controllers/anritsu_lib/md8475_cellular_simulator.py
index fbb9de5..79b35a6 100644
--- a/acts/framework/acts/controllers/anritsu_lib/md8475_cellular_simulator.py
+++ b/acts/framework/acts/controllers/anritsu_lib/md8475_cellular_simulator.py
@@ -707,7 +707,7 @@
     LTE_MAX_CARRIERS = 4
 
     # The maximum power that the equipment is able to transmit
-    MAX_DL_POWER = -30
+    MAX_DL_POWER = -10
 
     # Simulation config files in the callbox computer.
     # These should be replaced in the future by setting up
diff --git a/acts/framework/acts/controllers/cellular_lib/LteSimulation.py b/acts/framework/acts/controllers/cellular_lib/LteSimulation.py
index 89a8c4d..250a0d1 100644
--- a/acts/framework/acts/controllers/cellular_lib/LteSimulation.py
+++ b/acts/framework/acts/controllers/cellular_lib/LteSimulation.py
@@ -94,7 +94,7 @@
     # RSRP signal levels thresholds (as reported by Android) in dBm/15KHz.
     # Excellent is set to -75 since callbox B Tx power is limited to -30 dBm
     DOWNLINK_SIGNAL_LEVEL_DICTIONARY = {
-        'excellent': -75,
+        'excellent': -62,
         'high': -110,
         'medium': -115,
         'weak': -120,
diff --git a/acts/framework/acts/controllers/rohdeschwarz_lib/cmx500_cellular_simulator.py b/acts/framework/acts/controllers/rohdeschwarz_lib/cmx500_cellular_simulator.py
index 6336b7a..a59a43a 100644
--- a/acts/framework/acts/controllers/rohdeschwarz_lib/cmx500_cellular_simulator.py
+++ b/acts/framework/acts/controllers/rohdeschwarz_lib/cmx500_cellular_simulator.py
@@ -15,6 +15,7 @@
 #   limitations under the License.
 
 import time
+from enum import Enum
 
 from acts.controllers.rohdeschwarz_lib import cmx500
 from acts.controllers.rohdeschwarz_lib.cmx500 import LteBandwidth
@@ -43,6 +44,10 @@
 }
 
 
+class ConfigurationMode(Enum):
+    Power = "Power"
+
+
 class CMX500CellularSimulator(cc.AbstractCellularSimulator):
     """ A cellular simulator for telephony simulations based on the CMX 500
     controller. """
@@ -50,12 +55,16 @@
     # The maximum power that the equipment is able to transmit
     MAX_DL_POWER = -25
 
-    def __init__(self, ip_address, port='5025'):
+    def __init__(self,
+                 ip_address,
+                 port='5025',
+                 config_mode=ConfigurationMode.Power):
         """ Initializes the cellular simulator.
 
         Args:
             ip_address: the ip address of the CMX500
             port: the port number for the CMX500 controller
+            config_mode: A pre-defined configuration mode to use.
         """
         super().__init__()
         try:
@@ -63,6 +72,7 @@
         except:
             raise cc.CellularSimulatorError('Error when Initializes CMX500.')
 
+        self._config_mode = config_mode
         self.bts = self.cmx.bts
 
     def destroy(self):
@@ -339,6 +349,14 @@
         self.bts[1].attach_as_secondary_cell()
         time.sleep(10)
 
+        if self._config_mode and self._config_mode == ConfigurationMode.Power:
+            self.configure_for_power_measurement()
+
+        self.log.info('The radio connectivity is {}'.format(
+            self.cmx.dut.state.radio_connectivity))
+
+    def configure_for_power_measurement(self):
+        """ Applies a pre-defined configuration for PDCCH power testing."""
         self.log.info('set lte cdrx for nr nsa scenario')
         self.bts[0].set_cdrx_config()
         time.sleep(5)
@@ -360,9 +378,6 @@
         self.bts[1].disable_all_ul_slots()
         time.sleep(5)
 
-        self.log.info('The radio connectivity is {}'.format(
-            self.cmx.dut.state.radio_connectivity))
-
     def wait_until_attached(self, timeout=120):
         """ Waits until the DUT is attached to the primary carrier.
 
@@ -401,6 +416,15 @@
         self.log.info('wait for rrc off state')
         return self.cmx.wait_for_rrc_state(cmx500.RrcState.RRC_OFF, timeout)
 
+    def wait_until_quiet(self, timeout=120):
+        """Waits for all pending operations to finish on the simulator.
+
+        Args:
+            timeout: after this amount of time the method will raise a
+                CellularSimulatorError exception. Default is 120 seconds.
+        """
+        self.cmx._network.apply_changes()
+
     def detach(self):
         """ Turns off all the base stations so the DUT loose connection."""
         self.log.info('Bypass simulator detach step for now')
diff --git a/acts/framework/acts/controllers/rohdeschwarz_lib/cmx500_events.py b/acts/framework/acts/controllers/rohdeschwarz_lib/cmx500_events.py
new file mode 100644
index 0000000..f4abb0c
--- /dev/null
+++ b/acts/framework/acts/controllers/rohdeschwarz_lib/cmx500_events.py
@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+#
+#   Copyright 2023 - 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.
+
+from acts import logger as acts_logger
+
+logger = acts_logger.create_logger()
+
+
+def on_emm_registered(callback):
+    """Registers a callback to watch for EMM attach events.
+
+    Args:
+        callback: a callback to be invoked on EMM attach events.
+
+    Returns:
+        cancel: a callback to deregister the event watcher.
+    """
+    from rs_mrt.testenvironment.signaling.sri.nas.eps.pubsub import EmmAttachPub
+    from rs_mrt.testenvironment.signaling.sri.nas.eps import EmmRegistrationState
+
+    def wrapped(msg):
+        logger.debug("CMX received EMM registration state: {}".format(
+            msg.registration_state))
+        if msg.registration_state in (
+                EmmRegistrationState.COMBINED_REGISTERED, ):
+            callback()
+
+    sub = EmmAttachPub.multi_subscribe(callback=wrapped)
+
+    return lambda: sub.unsubscribe()
diff --git a/acts_tests/tests/google/nr/nsa5gmmw/Nsa5gMmwSmsTest.py b/acts_tests/tests/google/nr/nsa5gmmw/Nsa5gMmwSmsTest.py
index b0e88b5..cbd7df2 100755
--- a/acts_tests/tests/google/nr/nsa5gmmw/Nsa5gMmwSmsTest.py
+++ b/acts_tests/tests/google/nr/nsa5gmmw/Nsa5gMmwSmsTest.py
@@ -332,4 +332,151 @@
                             mt_rat='5g_nsa_mmwave',
                             long_msg=True)
 
+    @test_tracker_info(uuid="bee8b263-fe61-4a5c-a857-eb849426f0c7")
+    @TelephonyBaseTest.tel_test_wrap
+    def test_5g_nsa_mmw_sms_mo_mt_in_call_volte(self):
+        """ Test MO SMS during a MO VoLTE call over 5G NSA MMW.
+
+        Provision devices on VoLTE
+        Provision devices in 5g NSA MMW
+        Make a Voice call from PhoneA to PhoneB
+        Send and Verify SMS from PhoneA to PhoneB
+        Verify both devices are still on 5g NSA
+
+        Returns:
+            True if pass; False if fail.
+        """
+        return message_test(self.log,
+                            self.android_devices[0],
+                            self.android_devices[1],
+                            mo_rat='5g_nsa_mmw_volte',
+                            mt_rat='5g_nsa_mmw_volte',
+                            msg_in_call=True)
+
+    @test_tracker_info(uuid="1a5ca91a-91f6-4b5d-b4b9-f436bb3870ff")
+    @TelephonyBaseTest.tel_test_wrap
+    def test_5g_nsa_mmw_sms_mo_in_call_volte(self):
+        """ Test MO SMS during a MO VoLTE call over 5G NSA MMW.
+
+        Provision PhoneA on VoLTE
+        Provision PhoneA in 5g NSA MMW
+        Make a Voice call from PhoneA to PhoneB
+        Send and Verify SMS from PhoneA to PhoneB
+        Verify phoneA is still on 5g NSA
+
+        Returns:
+            True if pass; False if fail.
+        """
+        return message_test(self.log,
+                            self.android_devices[0],
+                            self.android_devices[1],
+                            mo_rat='5g_nsa_mmw_volte',
+                            mt_rat='default',
+                            msg_in_call=True)
+
+    @test_tracker_info(uuid="8a334ce7-a431-4fdc-bd65-f094d5ae727b")
+    @TelephonyBaseTest.tel_test_wrap
+    def test_5g_nsa_mmw_sms_mt_in_call_volte(self):
+        """ Test MT SMS during a MT VoLTE call over 5G NSA.
+
+        Provision PhoneA on VoLTE
+        Provision PhoneA in 5g NSA MMW
+        Make a Voice call from PhoneB to PhoneA
+        Send and Verify SMS from PhoneB to PhoneA
+        Verify phoneA is still on 5g NSA
+
+        Returns:
+            True if pass; False if fail.
+        """
+        return message_test(self.log,
+                            self.android_devices[1],
+                            self.android_devices[0],
+                            mo_rat='default',
+                            mt_rat='5g_nsa_mmw_volte',
+                            msg_in_call=True)
+
+    @test_tracker_info(uuid="ff29b86c-df34-4c6e-b880-baa321442fcf")
+    @TelephonyBaseTest.tel_test_wrap
+    def test_5g_nsa_mmw_sms_mo_mt_iwlan(self):
+        """ Test SMS text function between two phones,
+        Phones in APM, WiFi connected, WFC Cell Preferred mode.
+
+        Disable APM on both devices
+        Provision devices in 5g NSA MMW
+        Provision devices for WFC Cell Pref with APM ON
+        Send and Verify SMS from PhoneA to PhoneB
+
+        Returns:
+            True if pass; False if fail.
+        """
+        apm_mode = [
+            toggle_airplane_mode(self.log, ad, False)
+            for ad in self.android_devices
+        ]
+        return message_test(self.log,
+                            self.android_devices[0],
+                            self.android_devices[1],
+                            mo_rat='5g_nsa_mmw_wfc',
+                            mt_rat='5g_nsa_mmw_wfc',
+                            is_airplane_mode=True,
+                            wfc_mode=WFC_MODE_CELLULAR_PREFERRED,
+                            wifi_ssid=self.wifi_network_ssid,
+                            wifi_pwd=self.wifi_network_pass)
+
+    @test_tracker_info(uuid="010a85d9-84d5-4893-aa43-478490bbf03c")
+    @TelephonyBaseTest.tel_test_wrap
+    def test_5g_nsa_mmw_sms_mo_mt_iwlan_apm_off(self):
+        """ Test MO SMS, Phone in APM off, WiFi connected, WFC WiFi Preferred mode.
+
+        Disable APM on both devices
+        Provision devices in 5g NSA MMW
+        Provision devices for WFC Wifi Pref with APM OFF
+        Send and Verify SMS from PhoneA to PhoneB
+        Verify 5g NSA MMW attach for both devices
+
+        Returns:
+            True if pass; False if fail.
+        """
+        apm_mode = [
+            toggle_airplane_mode(self.log, ad, False)
+            for ad in self.android_devices
+        ]
+        return message_test(self.log,
+                            self.android_devices[0],
+                            self.android_devices[1],
+                            mo_rat='5g_nsa_mmw_wfc',
+                            mt_rat='5g_nsa_mmw_wfc',
+                            wfc_mode=WFC_MODE_WIFI_PREFERRED,
+                            wifi_ssid=self.wifi_network_ssid,
+                            wifi_pwd=self.wifi_network_pass)
+
+    @test_tracker_info(uuid="40c11bd3-e0a2-46f9-b7f0-ccb8366e85b7")
+    @TelephonyBaseTest.tel_test_wrap
+    def test_5g_nsa_mmw_sms_mo_mt_in_call_iwlan(self):
+        """ Test MO SMS, Phone in APM, WiFi connected, WFC WiFi Preferred mode.
+
+        Disable APM on both devices
+        Provision devices in 5g NSA MMW
+        Provision devices for WFC Wifi Pref with APM ON
+        Make a Voice call from PhoneA to PhoneB
+        Send and Verify SMS from PhoneA to PhoneB
+
+        Returns:
+            True if pass; False if fail.
+        """
+        apm_mode = [
+            toggle_airplane_mode(self.log, ad, False)
+            for ad in self.android_devices
+        ]
+        return message_test(self.log,
+                            self.android_devices[0],
+                            self.android_devices[1],
+                            mo_rat='5g_nsa_mmw_wfc',
+                            mt_rat='5g_nsa_mmw_wfc',
+                            msg_in_call=True,
+                            is_airplane_mode=True,
+                            wfc_mode=WFC_MODE_WIFI_PREFERRED,
+                            wifi_ssid=self.wifi_network_ssid,
+                            wifi_pwd=self.wifi_network_pass)
+
     """ Tests End """
diff --git a/acts_tests/tests/google/tel/live/TelLiveStressTest.py b/acts_tests/tests/google/tel/live/TelLiveStressTest.py
index 2c19b7a..7c01952 100644
--- a/acts_tests/tests/google/tel/live/TelLiveStressTest.py
+++ b/acts_tests/tests/google/tel/live/TelLiveStressTest.py
@@ -273,7 +273,8 @@
             ad.log.info("Phone VOLTE is enabled successfully.")
             # TODO: b/186865335 Move 5G methods to NR directory
             if self.nsa_5g_for_stress:
-                if not provision_device_for_5g(self.log, ad):
+                if not provision_device_for_5g(
+                        self.log, ad, nr_type=self.nr_type):
                     ad.log.error("Phone failed to attach 5G NSA.")
                     return False
                 ad.log.info("Phone 5G NSA VOLTE is enabled successfully.")
@@ -1361,7 +1362,14 @@
     @test_tracker_info(uuid="4212d0e0-fb87-47e5-ba48-9df9a4a6bb9b")
     @TelephonyBaseTest.tel_test_wrap
     def test_voice_performance_stress(self):
-        """ Vocie Performance stress test"""
+        """ Voice Performance stress test"""
         return self.performance_tests()
 
+    @test_tracker_info(uuid="a126793e-6e78-4920-a8c4-b382a444c4b7")
+    @TelephonyBaseTest.tel_test_wrap
+    def test_voice_performance_stress_nr(self):
+        """ Voice Performance stress test"""
+        return self.performance_tests(
+            setup_func=self._setup_lte_volte_enabled)
+
     """ Tests End """