"""
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 ipaddress
import time
import unittest

import s1ap_types
from integ_tests.s1aptests import s1ap_wrapper
from integ_tests.s1aptests.s1ap_utils import SpgwUtil
from s1ap_utils import MagmadUtil


class Test3485TimerForDedicatedBearerWithMmeRestart(unittest.TestCase):
    """Test 3485 timer expiry for dedicated bearer setup while mme restarts"""

    def setUp(self):
        """Initialize"""
        self._s1ap_wrapper = s1ap_wrapper.TestWrapper(
            stateless_mode=MagmadUtil.stateless_cmds.ENABLE,
        )
        self._spgw_util = SpgwUtil()

    def tearDown(self):
        """Cleanup"""
        self._s1ap_wrapper.cleanup()

    def test_3485_timer_for_dedicated_bearer_with_mme_restart(self):
        """Test case validates the functionality of 3485 timer for

        Dedicated bearer while MME restarts
        Step1: UE attaches to network
        Step2: Send an indication to initiate Dedicated bearer activation
        Step3: Send command to Magma to restart mme service
        Step4: TFW shall receive re-transmitted Activate Dedicated EPS
        Bearer Context Request and send Activate Dedicated EPS Bearer
        Context Accept
        Step5: TFW shall initiate de-activation of dedicated bearer and then
        initiate Detach procedure.
        """
        num_ues = 1
        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 idx 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
            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,
            )

            addr = attach.esmInfo.pAddr.addrInfo
            default_ip = ipaddress.ip_address(bytes(addr[:4]))

            # Wait on EMM Information from MME
            self._s1ap_wrapper._s1_util.receive_emm_info()

            time.sleep(2)
            print(
                '********************** Adding dedicated bearer to IMSI',
                ''.join([str(idx) for idx in req.imsi]),
            )

            # Create default flow list
            flow_list = self._spgw_util.create_default_ipv4_flows()
            self._spgw_util.create_bearer(
                'IMSI' + ''.join([str(idx) for idx in req.imsi]),
                attach.esmInfo.epsBearerId,
                flow_list,
            )

            response = self._s1ap_wrapper.s1_util.get_response()
            self.assertEqual(
                response.msg_type, s1ap_types.tfwCmd.UE_ACT_DED_BER_REQ.value,
            )
            print(
                '*******************Received first Activate dedicated bearer '
                'request',
            )

            print('***** Restarting MME service on gateway')
            self._s1ap_wrapper.magmad_util.restart_services(['mme'])

            wait_for_restart = 20
            for j in range(wait_for_restart):
                print('Waiting for', j, 'seconds')
                time.sleep(1)

            response = self._s1ap_wrapper.s1_util.get_response()
            act_ded_ber_ctxt_req = response.cast(
                s1ap_types.UeActDedBearCtxtReq_t,
            )
            self._s1ap_wrapper.sendActDedicatedBearerAccept(
                req.ue_id, act_ded_ber_ctxt_req.bearerId,
            )
            print('*******************Send Activate dedicated bearer accept')
            time.sleep(4)

            print(
                '********************** Deleting dedicated bearer for IMSI',
                ''.join([str(idx) for idx in req.imsi]),
            )
            self._spgw_util.delete_bearer(
                'IMSI' + ''.join([str(idx) for idx in req.imsi]),
                attach.esmInfo.epsBearerId, act_ded_ber_ctxt_req.bearerId,
            )

            # During wait time, mme may send multiple times Activate EPS bearer
            # context request, after sending response for first message ignore
            # the subsequent messages

            response = self._s1ap_wrapper.s1_util.get_response()
            msg_type = s1ap_types.tfwCmd.UE_DEACTIVATE_BER_REQ.value
            while (response.msg_type != msg_type):
                response = self._s1ap_wrapper.s1_util.get_response()
            self.assertEqual(
                response.msg_type,
                s1ap_types.tfwCmd.UE_DEACTIVATE_BER_REQ.value,
            )
            print('******************* Received deactivate eps bearer context')

            deactv_bearer_req = response.cast(s1ap_types.UeDeActvBearCtxtReq_t)
            self._s1ap_wrapper.sendDeactDedicatedBearerAccept(
                req.ue_id, deactv_bearer_req.bearerId,
            )
            # Verify if the rule for dedicated bearer is deleted
            dl_flow_rules = {
                default_ip: [],
            }
            # 1 UL flow for default bearer
            num_ul_flows = 1
            self._s1ap_wrapper.s1_util.verify_flow_rules(
                num_ul_flows, dl_flow_rules,
            )

            time.sleep(5)
            print(
                '********************** Running UE detach for UE id ',
                req.ue_id,
            )
            # Now detach the UE
            self._s1ap_wrapper.s1_util.detach(
                req.ue_id, detach_type[idx], wait_for_s1[idx],
            )


if __name__ == '__main__':
    unittest.main()

"""
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 ipaddress
import time
import unittest

import s1ap_types
from integ_tests.s1aptests import s1ap_wrapper
from integ_tests.s1aptests.s1ap_utils import SpgwUtil
from s1ap_utils import MagmadUtil


class Test3485TimerForDedicatedBearerWithMmeRestart(unittest.TestCase):
    """Test 3485 timer expiry for dedicated bearer setup while mme restarts"""

    def setUp(self):
        """Initialize"""
        self._s1ap_wrapper = s1ap_wrapper.TestWrapper(
            stateless_mode=MagmadUtil.stateless_cmds.ENABLE,
        )
        self._spgw_util = SpgwUtil()

    def tearDown(self):
        """Cleanup"""
        self._s1ap_wrapper.cleanup()

    def test_3485_timer_for_dedicated_bearer_with_mme_restart(self):
        """Test case validates the functionality of 3485 timer for

        Dedicated bearer while MME restarts
        Step1: UE attaches to network
        Step2: Send an indication to initiate Dedicated bearer activation
        Step3: Send command to Magma to restart mme service
        Step4: TFW shall receive re-transmitted Activate Dedicated EPS
        Bearer Context Request and send Activate Dedicated EPS Bearer
        Context Accept
        Step5: TFW shall initiate de-activation of dedicated bearer and then
        initiate Detach procedure.
        """
        num_ues = 1
        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 idx 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
            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,
            )

            addr = attach.esmInfo.pAddr.addrInfo
            default_ip = ipaddress.ip_address(bytes(addr[:4]))

            # Wait on EMM Information from MME
            self._s1ap_wrapper._s1_util.receive_emm_info()

            time.sleep(2)
            print(
                '********************** Adding dedicated bearer to IMSI',
                ''.join([str(idx) for idx in req.imsi]),
            )

            # Create default flow list
            flow_list = self._spgw_util.create_default_ipv4_flows()
            self._spgw_util.create_bearer(
                'IMSI' + ''.join([str(idx) for idx in req.imsi]),
                attach.esmInfo.epsBearerId,
                flow_list,
            )

            response = self._s1ap_wrapper.s1_util.get_response()
            self.assertEqual(
                response.msg_type, s1ap_types.tfwCmd.UE_ACT_DED_BER_REQ.value,
            )
            print(
                '*******************Received first Activate dedicated bearer '
                'request',
            )

            print('***** Restarting MME service on gateway')
            self._s1ap_wrapper.magmad_util.restart_services(['mme'])

            wait_for_restart = 20
            for j in range(wait_for_restart):
                print('Waiting for', j, 'seconds')
                time.sleep(1)

            response = self._s1ap_wrapper.s1_util.get_response()
            act_ded_ber_ctxt_req = response.cast(
                s1ap_types.UeActDedBearCtxtReq_t,
            )
            self._s1ap_wrapper.sendActDedicatedBearerAccept(
                req.ue_id, act_ded_ber_ctxt_req.bearerId,
            )
            print('*******************Send Activate dedicated bearer accept')
            time.sleep(4)

            print(
                '********************** Deleting dedicated bearer for IMSI',
                ''.join([str(idx) for idx in req.imsi]),
            )
            self._spgw_util.delete_bearer(
                'IMSI' + ''.join([str(idx) for idx in req.imsi]),
                attach.esmInfo.epsBearerId, act_ded_ber_ctxt_req.bearerId,
            )

            # During wait time, mme may send multiple times Activate EPS bearer
            # context request, after sending response for first message ignore
            # the subsequent messages

            response = self._s1ap_wrapper.s1_util.get_response()
            msg_type = s1ap_types.tfwCmd.UE_DEACTIVATE_BER_REQ.value
            while (response.msg_type != msg_type):
                response = self._s1ap_wrapper.s1_util.get_response()
            self.assertEqual(
                response.msg_type,
                s1ap_types.tfwCmd.UE_DEACTIVATE_BER_REQ.value,
            )
            print('******************* Received deactivate eps bearer context')

            deactv_bearer_req = response.cast(s1ap_types.UeDeActvBearCtxtReq_t)
            self._s1ap_wrapper.sendDeactDedicatedBearerAccept(
                req.ue_id, deactv_bearer_req.bearerId,
            )
            # Verify if the rule for dedicated bearer is deleted
            dl_flow_rules = {
                default_ip: [],
            }
            # 1 UL flow for default bearer
            num_ul_flows = 1
            self._s1ap_wrapper.s1_util.verify_flow_rules(
                num_ul_flows, dl_flow_rules,
            )

            time.sleep(5)
            print(
                '********************** Running UE detach for UE id ',
                req.ue_id,
            )
            # Now detach the UE
            self._s1ap_wrapper.s1_util.detach(
                req.ue_id, detach_type[idx], wait_for_s1[idx],
            )


if __name__ == '__main__':
    unittest.main()

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. Test 3485 timer expiry for dedicated bearer setup while mme restarts Initialize Cleanup Test case validates the functionality of 3485 timer for          Dedicated bearer while MME restarts         Step1: UE attaches to network         Step2: Send an indication to initiate Dedicated bearer activation         Step3: Send command to Magma to restart mme service         Step4: TFW shall receive re-transmitted Activate Dedicated EPS         Bearer Context Request and send Activate Dedicated EPS Bearer         Context Accept         Step5: TFW shall initiate de-activation of dedicated bearer and then         initiate Detach procedure. # Now actually complete the attach # Wait on EMM Information from MME # Create default flow list # During wait time, mme may send multiple times Activate EPS bearer # context request, after sending response for first message ignore # the subsequent messages # Verify if the rule for dedicated bearer is deleted # 1 UL flow for default bearer # Now detach the UE