blob: 4d88496691db4901808279216e3418756957884b [file] [log] [blame]
Wei-Yu Chen49950b92021-11-08 19:19:18 +08001"""
2Copyright 2020 The Magma Authors.
3
4This source code is licensed under the BSD-style license found in the
5LICENSE file in the root directory of this source tree.
6
7Unless required by applicable law or agreed to in writing, software
8distributed under the License is distributed on an "AS IS" BASIS,
9WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10See the License for the specific language governing permissions and
11limitations under the License.
12"""
13import logging
14import os
15from unittest.mock import Mock, call, patch
16
17from lte.protos.mconfig import mconfigs_pb2
18from common.service import MagmaService
19from data_models.data_model_parameters import ParameterName
20from devices.device_map import get_device_handler_from_name
21from devices.device_utils import EnodebDeviceName
22from devices.freedomfi_one import (
23 FreedomFiOneConfigurationInitializer,
24 SASParameters,
25 StatusParameters,
26)
27from tests.test_utils.config_builder import EnodebConfigBuilder
28from tests.test_utils.enb_acs_builder import (
29 EnodebAcsStateMachineBuilder,
30)
31from tests.test_utils.enodeb_handler import EnodebHandlerTestCase
32from tests.test_utils.tr069_msg_builder import Tr069MessageBuilder
33from tr069 import models
34
35SRC_CONFIG_DIR = os.path.join(
36 os.environ.get('MAGMA_ROOT'),
37 'lte/gateway/configs',
38)
39
40
41class FreedomFiOneTests(EnodebHandlerTestCase):
42
43 def _get_freedomfi_one_read_only_param_values_response(
44 self,
45 ) -> models.GetParameterValuesResponse:
46 msg = models.GetParameterValuesResponse()
47 param_val_list = []
48 param_val_list.append(
49 Tr069MessageBuilder.get_parameter_value_struct(
50 name='Device.X_000E8F_DeviceFeature.X_000E8F_NEStatus'
51 '.X_000E8F_Sync_Status',
52 val_type='string',
53 data='InSync',
54 ),
55 )
56 param_val_list.append(
57 Tr069MessageBuilder.get_parameter_value_struct(
58 name='Device.X_000E8F_DeviceFeature.X_000E8F_NEStatus'
59 '.X_000E8F_SAS_Status',
60 val_type='string',
61 data='SUCCESS',
62 ),
63 )
64 param_val_list.append(
65 Tr069MessageBuilder.get_parameter_value_struct(
66 name='Device.X_000E8F_DeviceFeature.X_000E8F_NEStatus'
67 '.X_000E8F_eNB_Status',
68 val_type='string',
69 data='SUCCESS',
70 ),
71 )
72 param_val_list.append(
73 Tr069MessageBuilder.get_parameter_value_struct(
74 name='Device.X_000E8F_DeviceFeature.X_000E8F_NEStatus'
75 '.X_000E8F_DEFGW_Status',
76 val_type='string',
77 data='SUCCESS',
78 ),
79 )
80 param_val_list.append(
81 Tr069MessageBuilder.get_parameter_value_struct(
82 name='Device.FAP.GPS.ScanStatus',
83 val_type='string',
84 data='SUCCESS',
85 ),
86 )
87 param_val_list.append(
88 Tr069MessageBuilder.get_parameter_value_struct(
89 name='Device.FAP.GPS.LockedLongitude',
90 val_type='int',
91 data='-105272892',
92 ),
93 )
94 param_val_list.append(
95 Tr069MessageBuilder.get_parameter_value_struct(
96 name='Device.FAP.GPS.LockedLatitude',
97 val_type='int',
98 data='40019606',
99 ),
100 )
101 msg.ParameterList = models.ParameterValueList()
102 msg.ParameterList.ParameterValueStruct = param_val_list
103 return msg
104
105 def _get_freedomfi_one_param_values_response(self):
106 msg = models.GetParameterValuesResponse()
107 param_val_list = []
108 msg.ParameterList = models.ParameterValueList()
109 msg.ParameterList.ParameterValueStruct = param_val_list
110
111 param_val_list.append(
112 Tr069MessageBuilder.get_parameter_value_struct(
113 name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.EARFCNDL',
114 val_type='int',
115 data='56240',
116 ),
117 )
118 param_val_list.append(
119 Tr069MessageBuilder.get_parameter_value_struct(
120 name='Device.FAP.GPS.ScanOnBoot',
121 val_type='boolean',
122 data='1',
123 ),
124 )
125 param_val_list.append(
126 Tr069MessageBuilder.get_parameter_value_struct(
127 name='Device.Services.FAPService.1.FAPControl.LTE.AdminState',
128 val_type='boolean',
129 data='1',
130 ),
131 )
132 param_val_list.append(
133 Tr069MessageBuilder.get_parameter_value_struct(
134 name='Device.FAP.PerfMgmt.Config.1.Enable',
135 val_type='boolean',
136 data='1',
137 ),
138 )
139 param_val_list.append(
140 Tr069MessageBuilder.get_parameter_value_struct(
141 name='Device.Services.FAPService.1.FAPControl.LTE.Gateway.S1SigLinkServerList',
142 val_type='string',
143 data='10.0.2.1',
144 ),
145 )
146 param_val_list.append(
147 Tr069MessageBuilder.get_parameter_value_struct(
148 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_RRMConfig.X_000E8F_Cell_Number',
149 val_type='int',
150 data='2',
151 ),
152 )
153 param_val_list.append(
154 Tr069MessageBuilder.get_parameter_value_struct(
155 name='Device.Services.FAPService.1.CellConfig.LTE.EPC.TAC',
156 val_type='int',
157 data='1',
158 ),
159 )
160 param_val_list.append(
161 Tr069MessageBuilder.get_parameter_value_struct(
162 name='Device.FAP.PerfMgmt.Config.1.PeriodicUploadInterval',
163 val_type='int',
164 data='60',
165 ),
166 )
167 param_val_list.append(
168 Tr069MessageBuilder.get_parameter_value_struct(
169 name='Device.DeviceInfo.SoftwareVersion',
170 val_type='string',
171 data='TEST3920@210901',
172 ),
173 )
174 param_val_list.append(
175 Tr069MessageBuilder.get_parameter_value_struct(
176 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.UserContactInformation',
177 val_type='string',
178 data='M0LK4T',
179 ),
180 )
181 param_val_list.append(
182 Tr069MessageBuilder.get_parameter_value_struct(
183 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.ProtectionLevel',
184 val_type='string',
185 data='GAA',
186 ),
187 )
188 param_val_list.append(
189 Tr069MessageBuilder.get_parameter_value_struct(
190 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.CertSubject',
191 val_type='string',
192 data='/C=TW/O=Sercomm/OU=WInnForum CBSD Certificate/CN=P27-SCE4255W:%s',
193 ),
194 )
195 param_val_list.append(
196 Tr069MessageBuilder.get_parameter_value_struct(
197 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.HeightType',
198 val_type='string',
199 data='AMSL',
200 ),
201 )
202 param_val_list.append(
203 Tr069MessageBuilder.get_parameter_value_struct(
204 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.Category',
205 val_type='string',
206 data='A',
207 ),
208 )
209 param_val_list.append(
210 Tr069MessageBuilder.get_parameter_value_struct(
211 name='Device.FAP.GPS.ScanStatus',
212 val_type='string',
213 data='Success',
214 ),
215 )
216 param_val_list.append(
217 Tr069MessageBuilder.get_parameter_value_struct(
218 name='Device.ManagementServer.PeriodicInformInterval',
219 val_type='int',
220 data='60',
221 ),
222 )
223 param_val_list.append(
224 Tr069MessageBuilder.get_parameter_value_struct(
225 name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.FreqBandIndicator',
226 val_type='unsignedInt',
227 data='48',
228 ),
229 )
230 param_val_list.append(
231 Tr069MessageBuilder.get_parameter_value_struct(
232 name='Device.Services.FAPService.1.CellConfig.LTE.RAN.Common.CellIdentity',
233 val_type='unsignedInt',
234 data='101',
235 ),
236 )
237 param_val_list.append(
238 Tr069MessageBuilder.get_parameter_value_struct(
239 name='Device.FAP.GPS.LockedLongitude',
240 val_type='string',
241 data='-105272892',
242 ),
243 )
244 param_val_list.append(
245 Tr069MessageBuilder.get_parameter_value_struct(
246 name='Device.FAP.GPS.LockedLatitude',
247 val_type='string',
248 data='40019606',
249 ),
250 )
251 param_val_list.append(
252 Tr069MessageBuilder.get_parameter_value_struct(
253 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.CPIEnable',
254 val_type='boolean',
255 data='0',
256 ),
257 )
258 param_val_list.append(
259 Tr069MessageBuilder.get_parameter_value_struct(
260 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_RRMConfig.X_000E8F_CA_Enable',
261 val_type='boolean',
262 data='1',
263 ),
264 )
265 param_val_list.append(
266 Tr069MessageBuilder.get_parameter_value_struct(
267 name='Device.FAP.GPS.ScanOnBoot',
268 val_type='boolean',
269 data='1',
270 ),
271 )
272 param_val_list.append(
273 Tr069MessageBuilder.get_parameter_value_struct(
274 name='Device.X_000E8F_DeviceFeature.X_000E8F_NEStatus.X_000E8F_Sync_Status',
275 val_type='string',
276 data='InSync',
277 ),
278 )
279 param_val_list.append(
280 Tr069MessageBuilder.get_parameter_value_struct(
281 name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.PhyCellID',
282 val_type='string',
283 data='101,102',
284 ),
285 )
286 param_val_list.append(
287 Tr069MessageBuilder.get_parameter_value_struct(
288 name='Device.FAP.PerfMgmt.Config.1.URL',
289 val_type='string',
290 data=None,
291 ),
292 )
293 param_val_list.append(
294 Tr069MessageBuilder.get_parameter_value_struct(
295 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.Location',
296 val_type='string',
297 data='indoor',
298 ),
299 )
300 param_val_list.append(
301 Tr069MessageBuilder.get_parameter_value_struct(
302 name='Device.Services.FAPService.1.CellConfig.LTE.RAN.PHY.TDDFrame.SubFrameAssignment',
303 val_type='boolean',
304 data='2',
305 ),
306 )
307 param_val_list.append(
308 Tr069MessageBuilder.get_parameter_value_struct(
309 name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.IsPrimary',
310 val_type='boolean',
311 data='1',
312 ),
313 )
314 param_val_list.append(
315 Tr069MessageBuilder.get_parameter_value_struct(
316 name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.Enable',
317 val_type='boolean',
318 data='1',
319 ),
320 )
321 param_val_list.append(
322 Tr069MessageBuilder.get_parameter_value_struct(
323 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.Server',
324 val_type='string',
325 data='https://spectrum-connect.federatedwireless.com/v1.2/',
326 ),
327 )
328 param_val_list.append(
329 Tr069MessageBuilder.get_parameter_value_struct(
330 name='Device.X_000E8F_DeviceFeature.X_000E8F_WebServerEnable',
331 val_type='boolean',
332 data='1',
333 ),
334 )
335 param_val_list.append(
336 Tr069MessageBuilder.get_parameter_value_struct(
337 name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.CellReservedForOperatorUse',
338 val_type='boolean',
339 data='0',
340 ),
341 )
342 param_val_list.append(
343 Tr069MessageBuilder.get_parameter_value_struct(
344 name='Device.Services.FAPService.1.CellConfig.LTE.Tunnel.1.TunnelRef',
345 val_type='string',
346 data='Device.IP.Interface.1.IPv4Address.1.',
347 ),
348 )
349 param_val_list.append(
350 Tr069MessageBuilder.get_parameter_value_struct(
351 name='Device.Services.FAPService.1.REM.X_000E8F_tfcsManagerConfig.primSrc',
352 val_type='string',
353 data='GNSS',
354 ),
355 )
356 param_val_list.append(
357 Tr069MessageBuilder.get_parameter_value_struct(
358 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.Enable',
359 val_type='boolean',
360 data='1',
361 ),
362 )
363 param_val_list.append(
364 Tr069MessageBuilder.get_parameter_value_struct(
365 name='Device.ManagementServer.PeriodicInformEnable',
366 val_type='boolean',
367 data='1',
368 ),
369 )
370 param_val_list.append(
371 Tr069MessageBuilder.get_parameter_value_struct(
372 name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNListNumberOfEntries',
373 val_type='int',
374 data='1',
375 ),
376 )
377 param_val_list.append(
378 Tr069MessageBuilder.get_parameter_value_struct(
379 name='Device.Services.FAPService.1.CellConfig.LTE.RAN.PHY.TDDFrame.SpecialSubframePatterns',
380 val_type='int',
381 data='7',
382 ),
383 )
384 param_val_list.append(
385 Tr069MessageBuilder.get_parameter_value_struct(
386 name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_RRMConfig.X_000E8F_CELL_Freq_Contiguous',
387 val_type='int',
388 data='0',
389 ),
390 )
391 param_val_list.append(
392 Tr069MessageBuilder.get_parameter_value_struct(
393 name='Device.Services.FAPService.1.FAPControl.LTE.Gateway.S1SigLinkPort',
394 val_type='int',
395 data='36412',
396 ),
397 )
398 return msg
399
400 def _get_service_config(self):
401 return {
402 "tr069": {
403 "interface": "eth1",
404 "port": 48080,
405 "perf_mgmt_port": 8081,
406 "public_ip": "192.88.99.142",
407 },
408 "reboot_enodeb_on_mme_disconnected": True,
409 "s1_interface": "eth1",
410 "sas": {
411 "sas_enabled": True,
412 "sas_server_url":
413 "https://spectrum-connect.federatedwireless.com/v1.2/",
414 "sas_uid": "M0LK4T",
415 "sas_category": "A",
416 "sas_channel_type": "GAA",
417 "sas_cert_subject": "/C=TW/O=Sercomm/OU=WInnForum CBSD "
418 "Certificate/CN=P27-SCE4255W:%s",
419 "sas_icg_group_id": "",
420 "sas_location": "indoor",
421 "sas_height_type": "AMSL",
422 },
423 }
424
425 def build_freedomfi_one_acs_state_machine(self):
426 service = EnodebAcsStateMachineBuilder.build_magma_service(
427 mconfig=EnodebConfigBuilder.get_mconfig(),
428 service_config=self._get_service_config(),
429 )
430 handler_class = get_device_handler_from_name(
431 EnodebDeviceName.FREEDOMFI_ONE,
432 )
433 acs_state_machine = handler_class(service)
434 return acs_state_machine
435
436 def test_provision(self) -> None:
437 """
438 Test the basic provisioning workflow
439 1 - enodeb sends Inform, enodebd sends InformResponse
440 2 - enodeb sends empty HTTP message,
441 3 - enodebd sends get transient params, updates the device state.
442 4 - enodebd sends get param values, updates the device state
443 5 - enodebd, sets fields including SAS fields.
444 """
445 logging.root.level = logging.DEBUG
446 acs_state_machine = self.build_freedomfi_one_acs_state_machine()
447
448 inform = Tr069MessageBuilder.get_inform(
449 oui="000E8F",
450 sw_version="TEST3920@210901",
451 enb_serial="2006CW5000023",
452 )
453 resp = acs_state_machine.handle_tr069_message(inform)
454 self.assertTrue(
455 isinstance(resp, models.InformResponse),
456 'Should respond with an InformResponse',
457 )
458
459 # Send an empty http request
460 req = models.DummyInput()
461 resp = acs_state_machine.handle_tr069_message(req)
462
463 # Expect a request for read-only params
464 self.assertTrue(
465 isinstance(resp, models.GetParameterValues),
466 'State machine should be requesting param values',
467 )
468 for tr_69_nodes in StatusParameters.STATUS_PARAMETERS.values():
469 self.assertIn(tr_69_nodes.path, resp.ParameterNames.string)
470
471 req = self._get_freedomfi_one_read_only_param_values_response()
472 get_resp = acs_state_machine.handle_tr069_message(req)
473
474 self.assertTrue(
475 isinstance(get_resp, models.GetParameterValues),
476 'State machine should be requesting param values',
477 )
478 req = self._get_freedomfi_one_param_values_response()
479 resp = acs_state_machine.handle_tr069_message(req)
480 self.assertTrue(
481 isinstance(resp, models.SetParameterValues),
482 'State machine should be setting parameters',
483 )
484 self.assertIsNotNone(
485 resp.ParameterKey.Data,
486 'ParameterKey should be set for FreedomFiOne eNB',
487 )
488
489 msg = models.SetParameterValuesResponse()
490 msg.Status = 1
491 get_resp = acs_state_machine.handle_tr069_message(msg)
492 self.assertTrue(
493 isinstance(get_resp, models.GetParameterValues),
494 'We should read back all parameters',
495 )
496
497 req = self._get_freedomfi_one_param_values_response()
498 resp = acs_state_machine.handle_tr069_message(req)
499 self.assertTrue(
500 isinstance(resp, models.DummyInput),
501 'Provisioning completed with Dummy response',
502 )
503
504 def test_manual_reboot_during_provisioning(self) -> None:
505 """
506 Test a scenario where a Magma user goes through the enodebd CLI to
507 reboot the Baicells eNodeB.
508
509 This checks the scenario where the command is sent in the middle
510 of a TR-069 provisioning session.
511 """
512 logging.root.level = logging.DEBUG
513 acs_state_machine = self.build_freedomfi_one_acs_state_machine()
514
515 # Send an Inform message, wait for an InformResponse
516 inform = Tr069MessageBuilder.get_inform(
517 oui="000E8F",
518 sw_version="TEST3920@210901",
519 enb_serial="2006CW5000023",
520 )
521 resp = acs_state_machine.handle_tr069_message(inform)
522 self.assertTrue(
523 isinstance(resp, models.InformResponse),
524 'Should respond with an InformResponse',
525 )
526
527 # Send an empty http request to kick off the rest of provisioning
528 req = models.DummyInput()
529 resp = acs_state_machine.handle_tr069_message(req)
530
531 # Expect a request for an optional parameter, three times
532 self.assertTrue(
533 isinstance(resp, models.GetParameterValues),
534 'State machine should be requesting param values',
535 )
536 req = Tr069MessageBuilder.get_fault()
537
538 # User uses the CLI tool to get eNodeB to reboot
539 acs_state_machine.reboot_asap()
540
541 resp = acs_state_machine.handle_tr069_message(req)
542 self.assertTrue(
543 isinstance(resp, models.Reboot),
544 'In reboot sequence, state machine should send a '
545 'Reboot message.',
546 )
547 req = Tr069MessageBuilder.get_reboot_response()
548 resp = acs_state_machine.handle_tr069_message(req)
549 self.assertTrue(
550 isinstance(resp, models.DummyInput),
551 'State machine should end TR-069 session after '
552 'receiving a RebootResponse',
553 )
554
555 def test_post_processing(self) -> None:
556 """ Test FreedomFi One specific post processing functionality"""
557
558 acs_state_machine = self.build_freedomfi_one_acs_state_machine()
559 cfg_desired = Mock()
560 acs_state_machine.device_cfg.set_parameter(
561 ParameterName.SERIAL_NUMBER,
562 "2006CW5000023",
563 )
564
565 cfg_init = FreedomFiOneConfigurationInitializer(acs_state_machine)
566 cfg_init.postprocess(
567 EnodebConfigBuilder.get_mconfig(),
568 self._get_service_config(), cfg_desired,
569 )
570 expected = [
571 call.delete_parameter('EARFCNDL'),
572 call.delete_parameter('DL bandwidth'),
573 call.delete_parameter('UL bandwidth'),
574 call.set_parameter(
575 'tunnel_ref',
576 'Device.IP.Interface.1.IPv4Address.1.',
577 ),
578 call.set_parameter('prim_src', 'GNSS'),
579 call.set_parameter('carrier_agg_enable', True),
580 call.set_parameter('carrier_number', 2),
581 call.set_parameter('contiguous_cc', 0),
582 call.set_parameter('web_ui_enable', False),
583 call.set_parameter('sas_enabled', True),
584 call.set_parameter(
585 'sas_server_url',
586 'https://spectrum-connect.federatedwireless.com/v1.2/',
587 ),
588 call.set_parameter('sas_uid', 'M0LK4T'),
589 call.set_parameter('sas_category', 'A'),
590 call.set_parameter('sas_channel_type', 'GAA'),
591 call.set_parameter(
592 'sas_cert_subject',
593 '/C=TW/O=Sercomm/OU=WInnForum CBSD Certificate/CN=P27-SCE4255W:%s',
594 ),
595 call.set_parameter('sas_location', 'indoor'),
596 call.set_parameter('sas_height_type', 'AMSL'),
597 ]
598 self.assertEqual(cfg_desired.mock_calls.sort(), expected.sort())
599
600 # Check without sas config
601 service_cfg = {
602 "tr069": {
603 "interface": "eth1",
604 "port": 48080,
605 "perf_mgmt_port": 8081,
606 "public_ip": "192.88.99.142",
607 },
608 "reboot_enodeb_on_mme_disconnected": True,
609 "s1_interface": "eth1",
610 }
611 cfg_desired = Mock()
612 cfg_init.postprocess(
613 EnodebConfigBuilder.get_mconfig(),
614 service_cfg, cfg_desired,
615 )
616 expected = [
617 call.delete_parameter('EARFCNDL'),
618 call.delete_parameter('DL bandwidth'),
619 call.delete_parameter('UL bandwidth'),
620 call.set_parameter(
621 'tunnel_ref',
622 'Device.IP.Interface.1.IPv4Address.1.',
623 ),
624 call.set_parameter('prim_src', 'GNSS'),
625 call.set_parameter('carrier_agg_enable', True),
626 call.set_parameter('carrier_number', 2),
627 call.set_parameter('contiguous_cc', 0),
628 call.set_parameter('web_ui_enable', False),
629 ]
630 self.assertEqual(cfg_desired.mock_calls.sort(), expected.sort())
631
632 service_cfg['web_ui_enable_list'] = ["2006CW5000023"]
633
634 expected = [
635 call.delete_parameter('EARFCNDL'),
636 call.delete_parameter('DL bandwidth'),
637 call.delete_parameter('UL bandwidth'),
638 call.set_parameter(
639 'tunnel_ref',
640 'Device.IP.Interface.1.IPv4Address.1.',
641 ),
642 call.set_parameter('prim_src', 'GNSS'),
643 call.set_parameter('carrier_agg_enable', True),
644 call.set_parameter('carrier_number', 2),
645 call.set_parameter('contiguous_cc', 0),
646 call.set_parameter('web_ui_enable', False),
647 call.set_parameter('web_ui_enable', True),
648 ]
649 cfg_desired = Mock()
650 cfg_init.postprocess(
651 EnodebConfigBuilder.get_mconfig(),
652 service_cfg, cfg_desired,
653 )
654 print(cfg_desired.mock_calls)
655 print(type(cfg_desired.mock_calls))
656 self.assertEqual(cfg_desired.mock_calls.sort(), expected.sort())
657
658 @patch('magma.configuration.service_configs.CONFIG_DIR', SRC_CONFIG_DIR)
659 def test_service_cfg_parsing(self):
660 """ Test the parsing of the service config file for enodebd.yml"""
661 service = MagmaService('enodebd', mconfigs_pb2.EnodebD())
662 service_cfg = service.config
663 service_cfg_1 = self._get_service_config()
664 service_cfg_1['web_ui_enable_list'] = []
665 service_cfg_1[FreedomFiOneConfigurationInitializer.SAS_KEY][
666 SASParameters.SAS_UID
667 ] = "INVALID_ID"
668 service_cfg_1[FreedomFiOneConfigurationInitializer.SAS_KEY][
669 SASParameters.SAS_CERT_SUBJECT
670 ] = "INVALID_CERT_SUBJECT"
671 self.assertEqual(service_cfg, service_cfg_1)
672
673 def test_status_nodes(self):
674 """ Test that the status of the node is valid"""
675 status = StatusParameters()
676
677 # Happy path
678 n1 = {
679 StatusParameters.DEFAULT_GW: "SUCCESS",
680 StatusParameters.SYNC_STATUS: "InSync",
681 StatusParameters.ENB_STATUS: "Success",
682 StatusParameters.SAS_STATUS: "Success",
683 StatusParameters.GPS_SCAN_STATUS: "SUCCESS",
684 ParameterName.GPS_LONG: "1",
685 ParameterName.GPS_LAT: "1",
686 }
687 device_config = Mock()
688 status.set_magma_device_cfg(n1, device_config)
689 expected = [
690 call.set_parameter(param_name='RF TX status', value=True),
691 call.set_parameter(param_name='GPS status', value=True),
692 call.set_parameter(param_name='PTP status', value=True),
693 call.set_parameter(param_name='MME status', value=True),
694 call.set_parameter(param_name='Opstate', value=True),
695 call.set_parameter('GPS lat', '1'),
696 call.set_parameter('GPS long', '1'),
697 ]
698 self.assertEqual(expected, device_config.mock_calls)
699
700 n2 = n1.copy()
701 # Verify we can handle specific none params
702 n2[StatusParameters.DEFAULT_GW] = None
703 n3 = n1.copy()
704 n3[StatusParameters.SYNC_STATUS] = None
705 n4 = n1.copy()
706 n4[StatusParameters.ENB_STATUS] = None
707 n5 = n1.copy()
708 n5[StatusParameters.SAS_STATUS] = None
709 n6 = n1.copy()
710 n6[StatusParameters.GPS_SCAN_STATUS] = None
711 n7 = n1.copy()
712 n7[ParameterName.GPS_LONG] = None
713 n8 = n1.copy()
714 n8[ParameterName.GPS_LAT] = None
715
716 device_config = Mock()
717 expected = [
718 call.set_parameter(param_name='RF TX status', value=True),
719 call.set_parameter(param_name='GPS status', value=True),
720 call.set_parameter(param_name='PTP status', value=True),
721 call.set_parameter(param_name='MME status', value=False),
722 call.set_parameter(param_name='Opstate', value=True),
723 call.set_parameter('GPS lat', '1'),
724 call.set_parameter('GPS long', '1'),
725 ]
726 status.set_magma_device_cfg(n2, device_config)
727 self.assertEqual(expected, device_config.mock_calls)
728
729 device_config = Mock()
730 expected = [
731 call.set_parameter(param_name='RF TX status', value=True),
732 call.set_parameter(param_name='GPS status', value=True),
733 call.set_parameter(param_name='PTP status', value=False),
734 call.set_parameter(param_name='MME status', value=True),
735 call.set_parameter(param_name='Opstate', value=True),
736 call.set_parameter('GPS lat', '1'),
737 call.set_parameter('GPS long', '1'),
738 ]
739 status.set_magma_device_cfg(n3, device_config)
740 self.assertEqual(expected, device_config.mock_calls)
741
742 device_config = Mock()
743 expected = [
744 call.set_parameter(param_name='RF TX status', value=True),
745 call.set_parameter(param_name='GPS status', value=True),
746 call.set_parameter(param_name='PTP status', value=True),
747 call.set_parameter(param_name='MME status', value=True),
748 call.set_parameter(param_name='Opstate', value=False),
749 call.set_parameter('GPS lat', '1'),
750 call.set_parameter('GPS long', '1'),
751 ]
752 status.set_magma_device_cfg(n4, device_config)
753 self.assertEqual(expected, device_config.mock_calls)
754
755 device_config = Mock()
756 expected = [
757 call.set_parameter(param_name='RF TX status', value=False),
758 call.set_parameter(param_name='GPS status', value=True),
759 call.set_parameter(param_name='PTP status', value=True),
760 call.set_parameter(param_name='MME status', value=True),
761 call.set_parameter(param_name='Opstate', value=True),
762 call.set_parameter('GPS lat', '1'),
763 call.set_parameter('GPS long', '1'),
764 ]
765 status.set_magma_device_cfg(n5, device_config)
766 self.assertEqual(expected, device_config.mock_calls)
767
768 device_config = Mock()
769 expected = [
770 call.set_parameter(param_name='RF TX status', value=True),
771 call.set_parameter(param_name='GPS status', value=False),
772 call.set_parameter(param_name='PTP status', value=False),
773 call.set_parameter(param_name='MME status', value=True),
774 call.set_parameter(param_name='Opstate', value=True),
775 call.set_parameter('GPS lat', '1'),
776 call.set_parameter('GPS long', '1'),
777 ]
778 status.set_magma_device_cfg(n6, device_config)
779 self.assertEqual(expected, device_config.mock_calls)
780
781 device_config = Mock()
782 expected = [
783 call.set_parameter(param_name='RF TX status', value=True),
784 call.set_parameter(param_name='GPS status', value=True),
785 call.set_parameter(param_name='PTP status', value=True),
786 call.set_parameter(param_name='MME status', value=True),
787 call.set_parameter(param_name='Opstate', value=True),
788 call.set_parameter('GPS lat', '1'),
789 call.set_parameter('GPS long', None),
790 ]
791 status.set_magma_device_cfg(n7, device_config)
792 self.assertEqual(expected, device_config.mock_calls)
793
794 device_config = Mock()
795 expected = [
796 call.set_parameter(param_name='RF TX status', value=True),
797 call.set_parameter(param_name='GPS status', value=True),
798 call.set_parameter(param_name='PTP status', value=True),
799 call.set_parameter(param_name='MME status', value=True),
800 call.set_parameter(param_name='Opstate', value=True),
801 call.set_parameter('GPS lat', None),
802 call.set_parameter('GPS long', '1'),
803 ]
804 status.set_magma_device_cfg(n8, device_config)
805 self.assertEqual(expected, device_config.mock_calls)