tests/test_e2ap.py

   1 # *******************************************************************************
   2 #  * Copyright 2020 Samsung Electronics All Rights Reserved.
   3 #  *
   4 #  * Licensed under the Apache License, Version 2.0 (the "License");
   5 #  * you may not use this file except in compliance with the License.
   6 #  * You may obtain a copy of the License at
   7 #  *
   8 #  * http://www.apache.org/licenses/LICENSE-2.0
   9 #  *
  10 #  * Unless required by applicable law or agreed to in writing, software
  11 #  * distributed under the License is distributed on an "AS IS" BASIS,
  12 #  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #  * See the License for the specific language governing permissions and
  14 #  * limitations under the License.
  15 #  *
  16 #  *******************************************************************************
  17 from ricxappframe.e2ap.asn1 import IndicationMsg, SubResponseMsg, SubRequestMsg, ControlRequestMsg, ActionDefinition, SubsequentAction, ARRAY, c_uint8
  18
  19 """
  20 fake class for c-type Structure
  21 """
  22
  23
  24 class indication_msg_type:
  25     def __init__(self):
  26         self.contents = _indication_contents()
  27
  28
  29 class _indication_contents:
  30     def __init__(self):
  31         self.request_id = 0
  32         self.request_sequence_number = 0
  33         self.function_id = 0
  34         self.action_id = 0
  35         self.indication_sequence_number = 0
  36         self.indication_type = 0
  37         self.indication_header = ARRAY(c_uint8, 1)()
  38         self.indication_header_length = 1
  39         self.indication_message = ARRAY(c_uint8, 1)()
  40         self.indication_message_length = 1
  41         self.call_process_id = ARRAY(c_uint8, 1)()
  42         self.call_process_id_length = 1
  43
  44
  45 class actionAdmittedList_msg_type:
  46     def __init__(self):
  47         self.request_id = []
  48         self.count = 0
  49
  50
  51 class causeItem_msg_type:
  52     def __init__(self):
  53         self.cause_type = 0
  54         self.cause_id = 0
  55
  56
  57 class actionNotAdmittedList_msg_type:
  58     def __init__(self):
  59         self.request_id = []
  60         self.cause = []
  61         self.count = 0
  62
  63
  64 class subResp_msg_type:
  65     def __init__(self):
  66         self.contents = _subResp_contents()
  67
  68
  69 class _subResp_contents:
  70
  71     def __init__(self):
  72         self.request_id = 0
  73         self.request_sequence_number = 0
  74         self.function_id = 0
  75         self.action_admitted_list = actionAdmittedList_msg_type()
  76         self.action_not_admitted_list = actionNotAdmittedList_msg_type()
  77
  78
  79 def test_call_decode_indication_and_clib_return_none_expect_error_raise(monkeypatch):
  80     '''
  81     test the decode of IndicationMsg class with invalid payload from rmr
  82     '''
  83     def mock_decode_return_none(payload: bytes, size):
  84         return None
  85
  86     monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_indicationMsg",
  87                         mock_decode_return_none)
  88
  89     indication = IndicationMsg()
  90
  91     try:
  92         indication.decode(bytes(0))
  93         assert False
  94     except BaseException:
  95         assert True
  96
  97
  98 def test_call_decode_indication_expect_success(monkeypatch):
  99     '''
 100     test the decode of IndicationMsg class
 101     '''
 102     def mock_decode_return_valid_indication(payload: bytes, size: int):
 103         indication_msg = indication_msg_type()
 104         indication_msg.contents.request_id = 1
 105         indication_msg.contents.request_sequence_number = 1
 106         indication_msg.contents.function_id = 1
 107         indication_msg.contents.action_id = 1
 108         indication_msg.contents.indication_sequence_number = 1
 109         indication_msg.contents.indication_type = 1
 110
 111         indication_header = ARRAY(c_uint8, 1)()
 112         indication_message = ARRAY(c_uint8, 1)()
 113         call_process_id = ARRAY(c_uint8, 1)()
 114
 115         indication_msg.contents.indication_header = indication_header
 116         indication_msg.contents.indication_message = indication_message
 117         indication_msg.contents.call_process_id = call_process_id
 118         return indication_msg
 119
 120     monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_indicationMsg",
 121                         mock_decode_return_valid_indication)
 122
 123     indication = IndicationMsg()
 124     try:
 125         indication.decode(bytes(0))
 126         assert indication.request_id == 1
 127         assert indication.request_sequence_number == 1
 128         assert indication.indication_type == 1
 129         assert indication.function_id == 1
 130         assert indication.action_id == 1
 131         assert indication.indication_sequence_number == 1
 132         assert indication.indication_header == bytes(b'\x00')
 133         assert indication.indication_message == bytes(b'\x00')
 134         assert indication.call_process_id == bytes(b'\x00')
 135     except BaseException:
 136         assert False
 137
 138
 139 def test_call_decode_sub_response_and_clib_return_none_expect_error_raise(monkeypatch):
 140     '''
 141     test the decode of SubResponseMsg class with invalid payload from rmr
 142     '''
 143     def mock_decode_return_none(payload: bytes, size):
 144         return None
 145
 146     monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_subRespMsg",
 147                         mock_decode_return_none)
 148
 149     sub_response = SubResponseMsg()
 150     try:
 151         sub_response.decode(bytes(0))
 152         assert False
 153     except BaseException:
 154         assert True
 155
 156
 157 def test_call_decode_sub_response_expect_success(monkeypatch):
 158     '''
 159     test the decode of SubResponseMsg class
 160     '''
 161     def mock_decode_return_valid_sub_response(payload: bytes, size: int):
 162         subResp_msg = subResp_msg_type()
 163         subResp_msg.contents.request_id = 1
 164         subResp_msg.contents.request_sequence_number = 1
 165         subResp_msg.contents.function_id = 1
 166
 167         action_admitted_list_msg = actionAdmittedList_msg_type()
 168         action_admitted_list_msg.request_id = [1]
 169         action_admitted_list_msg.count = 1
 170
 171         casue_item_msg = causeItem_msg_type()
 172         casue_item_msg.cause_id = 1
 173         casue_item_msg.cause_type = 1
 174
 175         action_not_admitted_list_msg = actionNotAdmittedList_msg_type()
 176         action_not_admitted_list_msg.request_id = [1]
 177         action_not_admitted_list_msg.count = 1
 178         action_not_admitted_list_msg.cause = [casue_item_msg]
 179
 180         subResp_msg.contents.action_admitted_list = action_admitted_list_msg
 181         subResp_msg.contents.action_not_admitted_list = action_not_admitted_list_msg
 182
 183         return subResp_msg
 184
 185     monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_subRespMsg",
 186                         mock_decode_return_valid_sub_response)
 187
 188     sub_response = SubResponseMsg()
 189     try:
 190         sub_response.decode(bytes(0))
 191         assert sub_response.request_id == 1
 192         assert sub_response.request_sequence_number == 1
 193         assert sub_response.function_id == 1
 194         assert sub_response.action_admitted_list.request_id[0] == 1
 195         assert sub_response.action_admitted_list.count == 1
 196         assert sub_response.action_not_admitted_list.request_id[0] == 1
 197         assert sub_response.action_not_admitted_list.count == 1
 198         assert sub_response.action_not_admitted_list.cause[0].cause_id == 1
 199         assert sub_response.action_not_admitted_list.cause[0].cause_type == 1
 200     except BaseException:
 201         assert False
 202
 203
 204 def test_call_encode_sub_request_and_clib_return_error_expect_error_raise(monkeypatch):
 205     '''
 206     test the encode of SubRequestMsg class with invalid param
 207     '''
 208     def mock_encode_return_error(buf, buf_size, requestor_id, request_sequence_number,
 209                                  ran_function_id, event_trigger_definition_array, event_definition_count,
 210                                  action_count, action_id_array, action_type_array, acttion_definition_array,
 211                                  subsequent_action_array):
 212         return -1
 213
 214     monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_subReqMsg",
 215                         mock_encode_return_error)
 216
 217     sub_request = SubRequestMsg()
 218     try:
 219         sub_request.encode(1, 1, 1, bytes([1]), [1], [1], [], [])
 220         assert False
 221     except BaseException:
 222         assert True
 223
 224
 225 def test_call_encode_sub_request_expect_success(monkeypatch):
 226     '''
 227     test the encode of SubRequestMsg class
 228     '''
 229     def mock_encode_return_success(buf, buf_size, requestor_id, request_sequence_number,
 230                                    ran_function_id, event_trigger_definition_array, event_definition_count,
 231                                    action_count, action_id_array, action_type_array, acttion_definition_array,
 232                                    subsequent_action_array):
 233         assert buf_size.value == 1024
 234         assert requestor_id.value == 1
 235         assert request_sequence_number.value == 1
 236         assert ran_function_id.value == 1
 237         assert event_trigger_definition_array[0] == 1
 238         assert event_definition_count.value == 1
 239         assert action_count.value == 1
 240         assert action_type_array[0] == 1
 241         assert acttion_definition_array[0].action_definition[0] == 1
 242         assert acttion_definition_array[0].size == 1
 243         assert subsequent_action_array[0].is_valid == 1
 244         assert subsequent_action_array[0].subsequent_action_type == 1
 245         assert subsequent_action_array[0].time_to_wait == 1
 246         return 1
 247
 248     monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_subReqMsg",
 249                         mock_encode_return_success)
 250
 251     action_definitions = list()
 252
 253     action_definition = ActionDefinition()
 254     action_definition.action_definition = bytes([1])
 255     action_definition.size = len(action_definition.action_definition)
 256
 257     action_definitions.append(action_definition)
 258
 259     subsequent_actions = list()
 260
 261     subsequent_action = SubsequentAction()
 262     subsequent_action.is_valid = 1
 263     subsequent_action.subsequent_action_type = 1
 264     subsequent_action.time_to_wait = 1
 265
 266     subsequent_actions.append(subsequent_action)
 267     sub_request = SubRequestMsg()
 268     try:
 269         sub_request.encode(1, 1, 1, bytes([1]), [1], [1],
 270                            action_definitions, subsequent_actions)
 271     except BaseException:
 272         assert False
 273
 274
 275 def test_call_encode_control_request_and_clib_return_error_expect_error_raise(monkeypatch):
 276     '''
 277     test the encode of ControlRequestMsg class with invalid param
 278     '''
 279     def mock_encode_return_error(buf, buf_size, requestor_id, request_sequence_number,
 280                                  ran_function_id, event_trigger_definition_array, event_definition_count,
 281                                  action_count, action_id_array, action_type_array, acttion_definition_array,
 282                                  subsequent_action_array):
 283         return -1
 284
 285     monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_controlReqMsg",
 286                         mock_encode_return_error)
 287
 288     control_request = ControlRequestMsg()
 289     try:
 290         control_request.encode(1, 1, 1, bytes([1]), bytes([1]), bytes([1]), 1)
 291         assert False
 292     except BaseException:
 293         assert True
 294
 295
 296 def test_call_encode_control_request_expect_success(monkeypatch):
 297     '''
 298     test the encode of ControlRequestMsg class
 299     '''
 300     def mock_encode_return_success(buf, buf_size, requestor_id, request_sequence_number,
 301                                    ran_function_id, call_process_id_buffer, call_process_id_buffer_count,
 302                                    call_header_buffer, call_header_buffer_count, call_message_buffer, call_message_buffer_count,
 303                                    control_ack_request):
 304         assert buf_size.value == 1024
 305         assert requestor_id.value == 1
 306         assert request_sequence_number.value == 1
 307         assert ran_function_id.value == 1
 308         assert call_process_id_buffer[0] == 1
 309         assert call_process_id_buffer_count.value == 1
 310         assert call_header_buffer[0] == 1
 311         assert call_header_buffer_count.value == 1
 312         assert call_message_buffer[0] == 1
 313         assert call_message_buffer_count.value == 1
 314         assert control_ack_request.value == 1
 315         return 1
 316
 317     monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_controlReqMsg",
 318                         mock_encode_return_success)
 319
 320     action_definitions = list()
 321
 322     action_definition = ActionDefinition()
 323     action_definition.action_definition = bytes([1])
 324     action_definition.size = len(action_definition.action_definition)
 325
 326     action_definitions.append(action_definition)
 327
 328     subsequent_actions = list()
 329
 330     subsequent_action = SubsequentAction()
 331     subsequent_action.is_valid = 1
 332     subsequent_action.subsequent_action_type = 1
 333     subsequent_action.time_to_wait = 1
 334
 335     subsequent_actions.append(subsequent_action)
 336     control_request = ControlRequestMsg()
 337     try:
 338         control_request.encode(1, 1, 1, bytes([1]), bytes([1]), bytes([1]), 1)
 339     except BaseException:
 340         assert False