1 # *******************************************************************************
2 # * Copyright 2020 Samsung Electronics All Rights Reserved.
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
8 # * http://www.apache.org/licenses/LICENSE-2.0
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.
16 # *******************************************************************************
17 from ricxappframe.e2ap.asn1 import IndicationMsg, SubResponseMsg, SubRequestMsg, ControlRequestMsg, ActionDefinition, SubsequentAction, ARRAY, c_uint8
20 fake class for c-type Structure
24 class indication_msg_type:
26 self.contents = _indication_contents()
29 class _indication_contents:
32 self.request_sequence_number = 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
45 class actionAdmittedList_msg_type:
51 class causeItem_msg_type:
57 class actionNotAdmittedList_msg_type:
64 class subResp_msg_type:
66 self.contents = _subResp_contents()
69 class _subResp_contents:
73 self.request_sequence_number = 0
75 self.action_admitted_list = actionAdmittedList_msg_type()
76 self.action_not_admitted_list = actionNotAdmittedList_msg_type()
79 def test_call_decode_indication_and_clib_return_none_expect_error_raise(monkeypatch):
81 test the decode of IndicationMsg class with invalid payload from rmr
83 def mock_decode_return_none(payload: bytes, size):
86 monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_indicationMsg",
87 mock_decode_return_none)
89 indication = IndicationMsg()
92 indication.decode(bytes(0))
98 def test_call_decode_indication_expect_success(monkeypatch):
100 test the decode of IndicationMsg class
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
111 indication_header = ARRAY(c_uint8, 1)()
112 indication_message = ARRAY(c_uint8, 1)()
113 call_process_id = ARRAY(c_uint8, 1)()
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
120 monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_indicationMsg",
121 mock_decode_return_valid_indication)
123 indication = IndicationMsg()
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:
139 def test_call_decode_sub_response_and_clib_return_none_expect_error_raise(monkeypatch):
141 test the decode of SubResponseMsg class with invalid payload from rmr
143 def mock_decode_return_none(payload: bytes, size):
146 monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_subRespMsg",
147 mock_decode_return_none)
149 sub_response = SubResponseMsg()
151 sub_response.decode(bytes(0))
153 except BaseException:
157 def test_call_decode_sub_response_expect_success(monkeypatch):
159 test the decode of SubResponseMsg class
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
167 action_admitted_list_msg = actionAdmittedList_msg_type()
168 action_admitted_list_msg.request_id = [1]
169 action_admitted_list_msg.count = 1
171 casue_item_msg = causeItem_msg_type()
172 casue_item_msg.cause_id = 1
173 casue_item_msg.cause_type = 1
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]
180 subResp_msg.contents.action_admitted_list = action_admitted_list_msg
181 subResp_msg.contents.action_not_admitted_list = action_not_admitted_list_msg
185 monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_decode_subRespMsg",
186 mock_decode_return_valid_sub_response)
188 sub_response = SubResponseMsg()
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:
204 def test_call_encode_sub_request_and_clib_return_error_expect_error_raise(monkeypatch):
206 test the encode of SubRequestMsg class with invalid param
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):
214 monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_subReqMsg",
215 mock_encode_return_error)
217 sub_request = SubRequestMsg()
219 sub_request.encode(1, 1, 1, bytes([1]), [1], [1], [], [])
221 except BaseException:
225 def test_call_encode_sub_request_expect_success(monkeypatch):
227 test the encode of SubRequestMsg class
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
248 monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_subReqMsg",
249 mock_encode_return_success)
251 action_definitions = list()
253 action_definition = ActionDefinition()
254 action_definition.action_definition = bytes([1])
255 action_definition.size = len(action_definition.action_definition)
257 action_definitions.append(action_definition)
259 subsequent_actions = list()
261 subsequent_action = SubsequentAction()
262 subsequent_action.is_valid = 1
263 subsequent_action.subsequent_action_type = 1
264 subsequent_action.time_to_wait = 1
266 subsequent_actions.append(subsequent_action)
267 sub_request = SubRequestMsg()
269 sub_request.encode(1, 1, 1, bytes([1]), [1], [1],
270 action_definitions, subsequent_actions)
271 except BaseException:
275 def test_call_encode_control_request_and_clib_return_error_expect_error_raise(monkeypatch):
277 test the encode of ControlRequestMsg class with invalid param
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):
285 monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_controlReqMsg",
286 mock_encode_return_error)
288 control_request = ControlRequestMsg()
290 control_request.encode(1, 1, 1, bytes([1]), bytes([1]), bytes([1]), 1)
292 except BaseException:
296 def test_call_encode_control_request_expect_success(monkeypatch):
298 test the encode of ControlRequestMsg class
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
317 monkeypatch.setattr("ricxappframe.e2ap.asn1._asn1_encode_controlReqMsg",
318 mock_encode_return_success)
320 action_definitions = list()
322 action_definition = ActionDefinition()
323 action_definition.action_definition = bytes([1])
324 action_definition.size = len(action_definition.action_definition)
326 action_definitions.append(action_definition)
328 subsequent_actions = list()
330 subsequent_action = SubsequentAction()
331 subsequent_action.is_valid = 1
332 subsequent_action.subsequent_action_type = 1
333 subsequent_action.time_to_wait = 1
335 subsequent_actions.append(subsequent_action)
336 control_request = ControlRequestMsg()
338 control_request.encode(1, 1, 1, bytes([1]), bytes([1]), bytes([1]), 1)
339 except BaseException: