text: """
  Copyright 2020 The Magma Authors.
  
  This source code is licensed under the BSD-style license found in the
  LICENSE file in the root directory of this source tree.
  
  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.
  """
  
  import time
  import unittest
  
  import s1ap_types
  import s1ap_wrapper
  from s1ap_utils import MagmadUtil
  
  
  class TestAttachDetachWithSctpdRestart(unittest.TestCase):
      def setUp(self):
          self._s1ap_wrapper = s1ap_wrapper.TestWrapper(
              stateless_mode=MagmadUtil.stateless_cmds.ENABLE,
          )
  
      def tearDown(self):
          self._s1ap_wrapper.cleanup()
  
      def test_attach_detach(self):
          """
          Attach/detach test with two UEs and Sctpd restarting after each attach.
          A new attach has to happen after Sctpd restarts before UE can detach.
          """
          num_ues = 2
          detach_type = [
              s1ap_types.ueDetachType_t.UE_NORMAL_DETACH.value,
              s1ap_types.ueDetachType_t.UE_SWITCHOFF_DETACH.value,
          ]
          wait_for_s1 = [True, False]
          self._s1ap_wrapper.configUEDevice(num_ues)
  
          for i in range(num_ues):
              req = self._s1ap_wrapper.ue_req
              print(
                  "************************* Running End to End attach for ",
                  "UE id ",
                  req.ue_id,
              )
              # Now actually complete the attach
              self._s1ap_wrapper._s1_util.attach(
                  req.ue_id,
                  s1ap_types.tfwCmd.UE_END_TO_END_ATTACH_REQUEST,
                  s1ap_types.tfwCmd.UE_ATTACH_ACCEPT_IND,
                  s1ap_types.ueAttachAccept_t,
              )
  
              # Wait on EMM Information from MME
              self._s1ap_wrapper._s1_util.receive_emm_info()
  
              print(
                  "************************* Restarting Sctpd service on",
                  "gateway",
              )
  
              self._s1ap_wrapper.magmad_util.restart_sctpd()
  
              # Re-establish S1 connection between eNB and MME
              self._s1ap_wrapper._s1setup()
  
              print(
                  "************************* Re-running End to End attach for ",
                  "UE id ",
                  req.ue_id,
              )
  
              # Repeat the attach
              self._s1ap_wrapper._s1_util.attach(
                  req.ue_id,
                  s1ap_types.tfwCmd.UE_END_TO_END_ATTACH_REQUEST,
                  s1ap_types.tfwCmd.UE_ATTACH_ACCEPT_IND,
                  s1ap_types.ueAttachAccept_t,
              )
  
              # Wait on EMM Information from MME
              self._s1ap_wrapper._s1_util.receive_emm_info()
  
              # Now detach the UE
              print(
                  "************************* Running UE detach for UE id ",
                  req.ue_id,
              )
              self._s1ap_wrapper.s1_util.detach(
                  req.ue_id, detach_type[i], wait_for_s1[i],
              )
  
              if i == 0:
                  break
  
              for j in range(15):
                  print("Connecting next UE in", 15 - j, "seconds")
                  time.sleep(1)
  
  
  if __name__ == "__main__":
      unittest.main()
  