X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=tests%2Ftest_e2ap.py;fp=tests%2Ftest_e2ap.py;h=99f6cd8f696dedf69f9f97064dae75f23ef18a53;hb=26f15d649bf3687b2b6864f0e05c9fb88eec5456;hp=0000000000000000000000000000000000000000;hpb=6b6b79b0ac66347c3e6b70fd82796390d0ddddf1;p=ric-plt%2Fxapp-frame-py.git

diff --git a/tests/test_e2ap.py b/tests/test_e2ap.py
new file mode 100644
index 0000000..99f6cd8
--- /dev/null
+++ b/tests/test_e2ap.py
@@ -0,0 +1,340 @@
+# *******************************************************************************
+#  * 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