tests/test_e2ap.py

# *******************************************************************************
#  * Copyright 2020 Samsung Electronics All Rights Reserved.
#  *
#  * 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 ricxappframe.e2ap.asn1 import IndicationMsg, SubResponseMsg, SubRequestMsg, ControlRequestMsg, ActionDefinition, SubsequentAction, ARRAY, c_uint8

"""
fake class for c-type Structure
"""


class indication_msg_type:
    def __init__(self):
        self.contents = _indication_contents()


class _indication_contents:
    def __init__(self):
        self.request_id = 0
        self.request_sequence_number = 0
        self.function_id = 0
        self.action_id = 0
        self.indication_sequence_number = 0
        self.indication_type = 0
        self.indication_header = ARRAY(c_uint8, 1)()
        self.indication_header_length = 1
        self.indication_message = ARRAY(c_uint8, 1)()
        self.indication_message_length = 1
        self.call_process_id = ARRAY(c_uint8, 1)()
        self.call_process_id_length = 1


class actionAdmittedList_msg_type:
    def __init__(self):
        self.request_id = []
        self.count = 0


class causeItem_msg_type:
    def __init__(self):
        self.cause_type = 0
        self.cause_id = 0


class actionNotAdmittedList_msg_type:
    def __init__(self):
        self.request_id = []
        self.cause = []
        self.count = 0


class subResp_msg_type:
    def __init__(self):
        self.contents = _subResp_contents()


class _subResp_contents:

    def __init__(self):
        self.request_id = 0
        self.request_sequence_number = 0
        self.function_id = 0
        self.action_admitted_list = actionAdmittedList_msg_type()
        self.action_not_admitted_list = actionNotAdmittedList_msg_type()


def test_call_decode_indication_and_clib_return_none_expect_error_raise(monkeypatch):
    '''
    test the decode of IndicationMsg class with invalid payload from rmr
    '''
    def mock_decode_return_none(payload: bytes, size):
        return None

    monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_indicationMsg",
                        mock_decode_return_none)

    indication = IndicationMsg()

    try:
        indication.decode(bytes(0))
        assert False
    except BaseException:
        assert True


def test_call_decode_indication_expect_success(monkeypatch):
    '''
    test the decode of IndicationMsg class
    '''
    def mock_decode_return_valid_indication(payload: bytes, size: int):
        indication_msg = indication_msg_type()
        indication_msg.contents.request_id = 1
        indication_msg.contents.request_sequence_number = 1
        indication_msg.contents.function_id = 1
        indication_msg.contents.action_id = 1
        indication_msg.contents.indication_sequence_number = 1
        indication_msg.contents.indication_type = 1

        indication_header = ARRAY(c_uint8, 1)()
        indication_message = ARRAY(c_uint8, 1)()
        call_process_id = ARRAY(c_uint8, 1)()

        indication_msg.contents.indication_header = indication_header
        indication_msg.contents.indication_message = indication_message
        indication_msg.contents.call_process_id = call_process_id
        return indication_msg

    monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_indicationMsg",
                        mock_decode_return_valid_indication)

    indication = IndicationMsg()
    try:
        indication.decode(bytes(0))
        assert indication.request_id == 1
        assert indication.request_sequence_number == 1
        assert indication.indication_type == 1
        assert indication.function_id == 1
        assert indication.action_id == 1
        assert indication.indication_sequence_number == 1
        assert indication.indication_header == bytes(b'\x00')
        assert indication.indication_message == bytes(b'\x00')
        assert indication.call_process_id == bytes(b'\x00')
    except BaseException:
        assert False


def test_call_decode_sub_response_and_clib_return_none_expect_error_raise(monkeypatch):
    '''
    test the decode of SubResponseMsg class with invalid payload from rmr
    '''
    def mock_decode_return_none(payload: bytes, size):
        return None

    monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_subRespMsg",
                        mock_decode_return_none)

    sub_response = SubResponseMsg()
    try:
        sub_response.decode(bytes(0))
        assert False
    except BaseException:
        assert True


def test_call_decode_sub_response_expect_success(monkeypatch):
    '''
    test the decode of SubResponseMsg class
    '''
    def mock_decode_return_valid_sub_response(payload: bytes, size: int):
        subResp_msg = subResp_msg_type()
        subResp_msg.contents.request_id = 1
        subResp_msg.contents.request_sequence_number = 1
        subResp_msg.contents.function_id = 1

        action_admitted_list_msg = actionAdmittedList_msg_type()
        action_admitted_list_msg.request_id = [1]
        action_admitted_list_msg.count = 1

        casue_item_msg = causeItem_msg_type()
        casue_item_msg.cause_id = 1
        casue_item_msg.cause_type = 1

        action_not_admitted_list_msg = actionNotAdmittedList_msg_type()
        action_not_admitted_list_msg.request_id = [1]
        action_not_admitted_list_msg.count = 1
        action_not_admitted_list_msg.cause = [casue_item_msg]

        subResp_msg.contents.action_admitted_list = action_admitted_list_msg
        subResp_msg.contents.action_not_admitted_list = action_not_admitted_list_msg

        return subResp_msg

    monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_subRespMsg",
                        mock_decode_return_valid_sub_response)

    sub_response = SubResponseMsg()
    try:
        sub_response.decode(bytes(0))
        assert sub_response.request_id == 1
        assert sub_response.request_sequence_number == 1
        assert sub_response.function_id == 1
        assert sub_response.action_admitted_list.request_id[0] == 1
        assert sub_response.action_admitted_list.count == 1
        assert sub_response.action_not_admitted_list.request_id[0] == 1
        assert sub_response.action_not_admitted_list.count == 1
        assert sub_response.action_not_admitted_list.cause[0].cause_id == 1
        assert sub_response.action_not_admitted_list.cause[0].cause_type == 1
    except BaseException:
        assert False


def test_call_encode_sub_request_and_clib_return_error_expect_error_raise(monkeypatch):
    '''
    test the encode of SubRequestMsg class with invalid param
    '''
    def mock_encode_return_error(buf, buf_size, requestor_id, request_sequence_number,
                                 ran_function_id, event_trigger_definition_array, event_definition_count,
                                 action_count, action_id_array, action_type_array, acttion_definition_array,
                                 subsequent_action_array):
        return -1

    monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_subReqMsg",
                        mock_encode_return_error)

    sub_request = SubRequestMsg()
    try:
        sub_request.encode(1, 1, 1, bytes([1]), [1], [1], [], [])
        assert False
    except BaseException:
        assert True


def test_call_encode_sub_request_expect_success(monkeypatch):
    '''
    test the encode of SubRequestMsg class
    '''
    def mock_encode_return_success(buf, buf_size, requestor_id, request_sequence_number,
                                   ran_function_id, event_trigger_definition_array, event_definition_count,
                                   action_count, action_id_array, action_type_array, acttion_definition_array,
                                   subsequent_action_array):
        assert buf_size.value == 1024
        assert requestor_id.value == 1
        assert request_sequence_number.value == 1
        assert ran_function_id.value == 1
        assert event_trigger_definition_array[0] == 1
        assert event_definition_count.value == 1
        assert action_count.value == 1
        assert action_type_array[0] == 1
        assert acttion_definition_array[0].action_definition[0] == 1
        assert acttion_definition_array[0].size == 1
        assert subsequent_action_array[0].is_valid == 1
        assert subsequent_action_array[0].subsequent_action_type == 1
        assert subsequent_action_array[0].time_to_wait == 1
        return 1

    monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_subReqMsg",
                        mock_encode_return_success)

    action_definitions = list()

    action_definition = ActionDefinition()
    action_definition.action_definition = bytes([1])
    action_definition.size = len(action_definition.action_definition)

    action_definitions.append(action_definition)

    subsequent_actions = list()

    subsequent_action = SubsequentAction()
    subsequent_action.is_valid = 1
    subsequent_action.subsequent_action_type = 1
    subsequent_action.time_to_wait = 1

    subsequent_actions.append(subsequent_action)
    sub_request = SubRequestMsg()
    try:
        sub_request.encode(1, 1, 1, bytes([1]), [1], [1],
                           action_definitions, subsequent_actions)
    except BaseException:
        assert False


def test_call_encode_control_request_and_clib_return_error_expect_error_raise(monkeypatch):
    '''
    test the encode of ControlRequestMsg class with invalid param
    '''
    def mock_encode_return_error(buf, buf_size, requestor_id, request_sequence_number,
                                 ran_function_id, event_trigger_definition_array, event_definition_count,
                                 action_count, action_id_array, action_type_array, acttion_definition_array,
                                 subsequent_action_array):
        return -1

    monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_controlReqMsg",
                        mock_encode_return_error)

    control_request = ControlRequestMsg()
    try:
        control_request.encode(1, 1, 1, bytes([1]), bytes([1]), bytes([1]), 1)
        assert False
    except BaseException:
        assert True


def test_call_encode_control_request_expect_success(monkeypatch):
    '''
    test the encode of ControlRequestMsg class
    '''
    def mock_encode_return_success(buf, buf_size, requestor_id, request_sequence_number,
                                   ran_function_id, call_process_id_buffer, call_process_id_buffer_count,
                                   call_header_buffer, call_header_buffer_count, call_message_buffer, call_message_buffer_count,
                                   control_ack_request):
        assert buf_size.value == 1024
        assert requestor_id.value == 1
        assert request_sequence_number.value == 1
        assert ran_function_id.value == 1
        assert call_process_id_buffer[0] == 1
        assert call_process_id_buffer_count.value == 1
        assert call_header_buffer[0] == 1
        assert call_header_buffer_count.value == 1
        assert call_message_buffer[0] == 1
        assert call_message_buffer_count.value == 1
        assert control_ack_request.value == 1
        return 1

    monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_controlReqMsg",
                        mock_encode_return_success)

    action_definitions = list()

    action_definition = ActionDefinition()
    action_definition.action_definition = bytes([1])
    action_definition.size = len(action_definition.action_definition)

    action_definitions.append(action_definition)

    subsequent_actions = list()

    subsequent_action = SubsequentAction()
    subsequent_action.is_valid = 1
    subsequent_action.subsequent_action_type = 1
    subsequent_action.time_to_wait = 1

    subsequent_actions.append(subsequent_action)
    control_request = ControlRequestMsg()
    try:
        control_request.encode(1, 1, 1, bytes([1]), bytes([1]), bytes([1]), 1)
    except BaseException:
        assert False