Init commit for standalone enodebd
Change-Id: I88eeef5135dd7ba8551ddd9fb6a0695f5325337b
diff --git a/tests/baicells_old_tests.py b/tests/baicells_old_tests.py
new file mode 100644
index 0000000..e693790
--- /dev/null
+++ b/tests/baicells_old_tests.py
@@ -0,0 +1,581 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+from devices.device_utils import EnodebDeviceName
+from tests.test_utils.enb_acs_builder import (
+ EnodebAcsStateMachineBuilder,
+)
+from tests.test_utils.enodeb_handler import EnodebHandlerTestCase
+from tests.test_utils.tr069_msg_builder import Tr069MessageBuilder
+from tr069 import models
+
+
+class BaicellsOldHandlerTests(EnodebHandlerTestCase):
+ def test_initial_enb_bootup(self) -> None:
+ """
+ Baicells does not support configuration during initial bootup of
+ eNB device. This is because it is in a REM process, and we just need
+ to wait for this process to finish, ~10 minutes. Attempting to
+ configure the device during this period will cause undefined
+ behavior.
+ As a result of this, end any provisoning sessions, which we can do
+ by just sending empty HTTP responses, not even using an
+ InformResponse.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS_OLD)
+
+ # Send an Inform message
+ inform_msg = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC002',
+ '120200002618AGP0003', ['1 BOOT'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'Should respond with an InformResponse',
+ )
+
+ def test_manual_reboot(self) -> None:
+ """
+ Test a scenario where a Magma user goes through the enodebd CLI to
+ reboot the Baicells eNodeB.
+
+ This checks the scenario where the command is not sent in the middle
+ of a TR-069 provisioning session.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS_OLD)
+
+ # User uses the CLI tool to get eNodeB to reboot
+ acs_state_machine.reboot_asap()
+
+ # And now the Inform message arrives from the eNodeB
+ inform_msg = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC002',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'In reboot sequence, state machine should still '
+ 'respond to an Inform with InformResponse.',
+ )
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.Reboot),
+ 'In reboot sequence, state machine should send a '
+ 'Reboot message.',
+ )
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'State machine should end TR-069 session after '
+ 'receiving a RebootResponse',
+ )
+
+ def test_manual_reboot_during_provisioning(self) -> None:
+ """
+ Test a scenario where a Magma user goes through the enodebd CLI to
+ reboot the Baicells eNodeB.
+
+ This checks the scenario where the command is sent in the middle
+ of a TR-069 provisioning session.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS_OLD)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC002',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+
+ # User uses the CLI tool to get eNodeB to reboot
+ acs_state_machine.reboot_asap()
+
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.Reboot),
+ 'In reboot sequence, state machine should send a '
+ 'Reboot message.',
+ )
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'State machine should end TR-069 session after '
+ 'receiving a RebootResponse',
+ )
+
+ def test_provision_without_invasive_changes(self) -> None:
+ """
+ Test the scenario where:
+ - eNodeB has already been powered for 10 minutes without configuration
+ - Setting parameters which are 'non-invasive' on the eNodeB
+
+ 'Invasive' parameters are those which require special behavior to apply
+ the changes for the eNodeB.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS_OLD)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC001',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, five times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+
+ # Send back some typical values
+ # And then SM should request regular parameter values
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ # Send back typical values for the regular parameters
+ req = Tr069MessageBuilder.\
+ get_regular_param_values_response(
+ admin_state=False,
+ earfcndl=39150,
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # SM will be requesting object parameter values
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting object param vals',
+ )
+
+ # Send back some typical values for object parameters
+ req = Tr069MessageBuilder.get_object_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # In this scenario, the ACS and thus state machine will not need
+ # to delete or add objects to the eNB configuration.
+ # SM should then just be attempting to set parameter values
+ self.assertTrue(
+ isinstance(resp, models.SetParameterValues),
+ 'State machine should be setting param values',
+ )
+
+ # Send back confirmation that the parameters were successfully set
+ req = models.SetParameterValuesResponse()
+ req.Status = 0
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+
+ # Send back some typical values
+ # And then SM should continue polling the read-only params
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'State machine should be ending session',
+ )
+
+ # If a different eNB is suddenly plugged in, or the same eNB sends a
+ # new Inform, enodebd should be able to handle it.
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC002',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ def test_reboot_after_invasive_changes(self) -> None:
+ """
+ Test the scenario where:
+ - eNodeB has already been powered for 10 minutes without configuration
+ - Setting parameters which are 'invasive' on the eNodeB
+ - Simulate the scenario up until reboot, and test that enodebd does
+ not try to complete configuration after reboot, because it is
+ waiting for REM process to finish running
+ - This test does not wait the ten minutes to simulate REM process
+ finishing on the Baicells eNodeB
+
+ 'Invasive' parameters are those which require special behavior to apply
+ the changes for the eNodeB.
+
+ In the case of the Baicells eNodeB, properly applying changes to
+ invasive parameters requires rebooting the device.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder\
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS_OLD)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC002',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, five times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+
+ # Send back some typical values
+ # And then SM should request regular parameter values
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ # Send back typical values for the regular parameters
+ req = Tr069MessageBuilder.get_regular_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # SM will be requesting object parameter values
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting object param vals',
+ )
+
+ # Send back some typical values for object parameters
+ req = Tr069MessageBuilder.get_object_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # In this scenario, the ACS and thus state machine will not need
+ # to delete or add objects to the eNB configuration.
+ # SM should then just be attempting to set parameter values
+ self.assertTrue(
+ isinstance(resp, models.SetParameterValues),
+ 'State machine should be setting param values',
+ )
+
+ # Send back confirmation that the parameters were successfully set
+ req = models.SetParameterValuesResponse()
+ req.Status = 0
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Since invasive parameters have been set, then to apply the changes
+ # to the Baicells eNodeB, we need to reboot the device
+ self.assertTrue(isinstance(resp, models.Reboot))
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # After the reboot has been received, enodebd should end the
+ # provisioning session
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'After sending command to reboot the Baicells eNodeB, '
+ 'enodeb should end the TR-069 session.',
+ )
+
+ # At this point, sometime after the eNodeB reboots, we expect it to
+ # send an Inform indicating reboot. Since it should be in REM process,
+ # we hold off on finishing configuration, and end TR-069 sessions.
+ req = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC002',
+ '120200002618AGP0003',
+ ['1 BOOT', 'M Reboot'],
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'After receiving a post-reboot Inform, enodebd '
+ 'should end TR-069 sessions for 10 minutes to wait '
+ 'for REM process to finish.',
+ )
+
+ # Pretend that we have waited, and now we are in normal operation again
+ acs_state_machine.transition('wait_inform_post_reboot')
+ req = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC002',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'After receiving a post-reboot Inform, enodebd '
+ 'should end TR-069 sessions for 10 minutes to wait '
+ 'for REM process to finish.',
+ )
+
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'enodebd should be requesting params',
+ )
+ self.assertTrue(
+ len(resp.ParameterNames.string) > 1,
+ 'Should be requesting transient params.',
+ )
+
+ def test_reboot_without_getting_optional(self) -> None:
+ """
+ The state machine should not skip figuring out which optional
+ parameters are present.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS_OLD)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC002',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # And now reboot the eNodeB
+ acs_state_machine.transition('reboot')
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(isinstance(resp, models.Reboot))
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # After the reboot has been received, enodebd should end the
+ # provisioning session
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'After sending command to reboot the Baicells eNodeB, '
+ 'enodeb should end the TR-069 session.',
+ )
+
+ # At this point, sometime after the eNodeB reboots, we expect it to
+ # send an Inform indicating reboot. Since it should be in REM process,
+ # we hold off on finishing configuration, and end TR-069 sessions.
+ req = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC002',
+ '120200002618AGP0003',
+ ['1 BOOT', 'M Reboot'],
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'After receiving a post-reboot Inform, enodebd '
+ 'should end TR-069 sessions for 10 minutes to wait '
+ 'for REM process to finish.',
+ )
+
+ # Pretend that we have waited, and now we are in normal operation again
+ acs_state_machine.transition('wait_inform_post_reboot')
+ req = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiStation_V100R001C00B110SPC002',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'After receiving a post-reboot Inform, enodebd '
+ 'should end TR-069 sessions for 10 minutes to wait '
+ 'for REM process to finish.',
+ )
+
+ # Since we haven't figured out the presence of optional parameters, the
+ # state machine should be requesting them now. There are three for the
+ # Baicells state machine.
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'enodebd should be requesting params',
+ )
+ self.assertTrue(
+ len(resp.ParameterNames.string) == 1,
+ 'Should be requesting optional params.',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param',
+ )
+ self.assertTrue(
+ len(resp.ParameterNames.string) == 1,
+ 'Should be requesting optional params.',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param',
+ )
+ self.assertTrue(
+ len(resp.ParameterNames.string) == 1,
+ 'Should be requesting optional params.',
+ )
diff --git a/tests/baicells_qafb_tests.py b/tests/baicells_qafb_tests.py
new file mode 100644
index 0000000..0b0168e
--- /dev/null
+++ b/tests/baicells_qafb_tests.py
@@ -0,0 +1,230 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+from devices.device_utils import EnodebDeviceName
+from tests.test_utils.enb_acs_builder import (
+ EnodebAcsStateMachineBuilder,
+)
+from tests.test_utils.enodeb_handler import EnodebHandlerTestCase
+from tests.test_utils.tr069_msg_builder import Tr069MessageBuilder
+from tr069 import models
+
+
+class BaicellsQAFBHandlerTests(EnodebHandlerTestCase):
+ def test_manual_reboot(self) -> None:
+ """
+ Test a scenario where a Magma user goes through the enodebd CLI to
+ reboot the Baicells eNodeB.
+
+ This checks the scenario where the command is not sent in the middle
+ of a TR-069 provisioning session.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS_QAFB)
+
+ # User uses the CLI tool to get eNodeB to reboot
+ acs_state_machine.reboot_asap()
+
+ # And now the Inform message arrives from the eNodeB
+ inform_msg = \
+ Tr069MessageBuilder.get_qafb_inform(
+ '48BF74',
+ 'BaiBS_QAFBv123',
+ '1202000181186TB0006',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'In reboot sequence, state machine should still '
+ 'respond to an Inform with InformResponse.',
+ )
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.Reboot),
+ 'In reboot sequence, state machine should send a '
+ 'Reboot message.',
+ )
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'State machine should end TR-069 session after '
+ 'receiving a RebootResponse',
+ )
+
+ def test_manual_reboot_during_provisioning(self) -> None:
+ """
+ Test a scenario where a Magma user goes through the enodebd CLI to
+ reboot the Baicells eNodeB.
+
+ This checks the scenario where the command is sent in the middle
+ of a TR-069 provisioning session.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS_QAFB)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = \
+ Tr069MessageBuilder.get_qafb_inform(
+ '48BF74',
+ 'BaiBS_QAFBv123',
+ '1202000181186TB0006',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+
+ # User uses the CLI tool to get eNodeB to reboot
+ acs_state_machine.reboot_asap()
+
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.Reboot),
+ 'In reboot sequence, state machine should send a '
+ 'Reboot message.',
+ )
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'State machine should end TR-069 session after '
+ 'receiving a RebootResponse',
+ )
+
+ def test_provision(self) -> None:
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS_QAFB)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = \
+ Tr069MessageBuilder.get_qafb_inform(
+ '48BF74',
+ 'BaiBS_QAFBv123',
+ '1202000181186TB0006',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_qafb_read_only_param_values_response()
+
+ # Send back some typical values
+ # And then SM should request regular parameter values
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ # Send back typical values for the regular parameters
+ req = Tr069MessageBuilder.\
+ get_qafb_regular_param_values_response(
+ admin_state=False,
+ earfcndl=39150,
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # SM will be requesting object parameter values
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting object param vals',
+ )
+
+ # Send back some typical values for object parameters
+ req = Tr069MessageBuilder.get_qafb_object_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ self.assertTrue(
+ isinstance(resp, models.AddObject),
+ 'State machine should be adding objects',
+ )
+
+ def test_get_rpc_methods_cold(self) -> None:
+ """
+ Test the scenario where:
+ - enodeB just booted
+ - enodeB is cold and has no state of ACS RPCMethods
+ - Simulate the enodeB performing the initial Inform and
+ the call for the GetRPCMethods, and the subsequent Empty
+ response for provisioning
+ finishing on the Baicells eNodeB
+
+ Verifies that the ACS will continue into provisioning
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder\
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS_QAFB)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = \
+ Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_QAFBv123',
+ '120200002618AGP0003',
+ ['1 BOOT'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send GetRPCMethods
+ req = models.GetRPCMethods()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetRPCMethodsResponse),
+ 'State machine should be sending RPC methods',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
diff --git a/tests/baicells_tests.py b/tests/baicells_tests.py
new file mode 100644
index 0000000..d992e58
--- /dev/null
+++ b/tests/baicells_tests.py
@@ -0,0 +1,944 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+from data_models.data_model_parameters import ParameterName
+from devices.device_utils import EnodebDeviceName
+from tests.test_utils.enb_acs_builder import (
+ EnodebAcsStateMachineBuilder,
+)
+from tests.test_utils.enodeb_handler import EnodebHandlerTestCase
+from tests.test_utils.tr069_msg_builder import Tr069MessageBuilder
+from tr069 import models
+
+
+class BaicellsHandlerTests(EnodebHandlerTestCase):
+ def test_initial_enb_bootup(self) -> None:
+ """
+ Baicells does not support configuration during initial bootup of
+ eNB device. This is because it is in a REM process, and we just need
+ to wait for this process to finish, ~10 minutes. Attempting to
+ configure the device during this period will cause undefined
+ behavior.
+ As a result of this, end any provisoning sessions, which we can do
+ by just sending empty HTTP responses, not even using an
+ InformResponse.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+
+ # Send an Inform message
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['1 BOOT'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'Should respond with an InformResponse',
+ )
+
+ def test_manual_reboot(self) -> None:
+ """
+ Test a scenario where a Magma user goes through the enodebd CLI to
+ reboot the Baicells eNodeB.
+
+ This checks the scenario where the command is not sent in the middle
+ of a TR-069 provisioning session.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+
+ # User uses the CLI tool to get eNodeB to reboot
+ acs_state_machine.reboot_asap()
+
+ # And now the Inform message arrives from the eNodeB
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'In reboot sequence, state machine should still '
+ 'respond to an Inform with InformResponse.',
+ )
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.Reboot),
+ 'In reboot sequence, state machine should send a '
+ 'Reboot message.',
+ )
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'State machine should end TR-069 session after '
+ 'receiving a RebootResponse',
+ )
+
+ def test_gps_coords(self) -> None:
+ """ Check GPS coordinates are processed and stored correctly """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = models.GetParameterValuesResponse()
+ param_val_list = [
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.X_BAICELLS_COM_GpsSyncEnable',
+ val_type='boolean',
+ data='true',
+ ),
+ ]
+ req.ParameterList = models.ParameterValueList()
+ req.ParameterList.ParameterValueStruct = param_val_list
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = models.GetParameterValuesResponse()
+ param_val_list = [
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.GPS.LockedLatitude',
+ val_type='int',
+ data='37483629',
+ ),
+ ]
+ req.ParameterList = models.ParameterValueList()
+ req.ParameterList.ParameterValueStruct = param_val_list
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = models.GetParameterValuesResponse()
+ param_val_list = [
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.GPS.LockedLongitude',
+ val_type='int',
+ data='-122150583',
+ ),
+ ]
+ req.ParameterList = models.ParameterValueList()
+ req.ParameterList.ParameterValueStruct = param_val_list
+ acs_state_machine.handle_tr069_message(req)
+
+ gps_long = acs_state_machine.get_parameter(ParameterName.GPS_LONG)
+ gps_lat = acs_state_machine.get_parameter(ParameterName.GPS_LAT)
+
+ self.assertTrue(gps_long == '-122.150583', 'Should be valid longitude')
+ self.assertTrue(gps_lat == '37.483629', 'Should be valid latitude')
+
+ def test_manual_reboot_during_provisioning(self) -> None:
+ """
+ Test a scenario where a Magma user goes through the enodebd CLI to
+ reboot the Baicells eNodeB.
+
+ This checks the scenario where the command is sent in the middle
+ of a TR-069 provisioning session.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+
+ # User uses the CLI tool to get eNodeB to reboot
+ acs_state_machine.reboot_asap()
+
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.Reboot),
+ 'In reboot sequence, state machine should send a '
+ 'Reboot message.',
+ )
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'State machine should end TR-069 session after '
+ 'receiving a RebootResponse',
+ )
+
+ def test_missing_param_during_provisioning(self) -> None:
+ """
+ Test the scenario where:
+ - enodebd is configuring the eNodeB
+ - eNB does not send all parameters due to bug
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform()
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+
+ # Send back some typical values
+ # And then SM should request regular parameter values
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ # Send back typical values for the regular parameters
+ # Pretend that here the NumPLMNs was not sent because of a Baicells bug
+ req = Tr069MessageBuilder.\
+ get_regular_param_values_response(
+ admin_state=False,
+ earfcndl=39150,
+ exclude_num_plmns=True,
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # The state machine will fail and go into an error state.
+ # It will send an empty http response to end the session.
+ # Regularly, the SM should be using info on the number
+ # of PLMNs to figure out which object parameter values
+ # to fetch.
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'State machine should be ending session',
+ )
+
+ def test_provision_multi_without_invasive_changes(self) -> None:
+ """
+ Test the scenario where:
+ - eNodeB has already been powered for 10 minutes without configuration
+ - Setting parameters which are 'non-invasive' on the eNodeB
+ - Using enodebd mconfig which has old style config with addition
+ of eNodeB config tied to a serial number
+
+ 'Invasive' parameters are those which require special behavior to apply
+ the changes for the eNodeB.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_multi_enb_acs_state_machine(EnodebDeviceName.BAICELLS)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform()
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+
+ # Send back some typical values
+ # And then SM should request regular parameter values
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ # Send back typical values for the regular parameters
+ req = Tr069MessageBuilder.\
+ get_regular_param_values_response(
+ admin_state=False,
+ earfcndl=39150,
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # SM will be requesting object parameter values
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting object param vals',
+ )
+
+ # Send back some typical values for object parameters
+ req = Tr069MessageBuilder.get_object_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # In this scenario, the ACS and thus state machine will not need
+ # to delete or add objects to the eNB configuration.
+ # SM should then just be attempting to set parameter values
+ self.assertTrue(
+ isinstance(resp, models.SetParameterValues),
+ 'State machine should be setting param values',
+ )
+
+ isEnablingAdminState = False
+ param = 'Device.Services.FAPService.1.FAPControl.LTE.AdminState'
+ for name_value in resp.ParameterList.ParameterValueStruct:
+ if name_value.Name == param:
+ isEnablingAdminState = True
+ self.assertTrue(
+ isEnablingAdminState,
+ 'eNB config is set to enable transmit, '
+ 'while old enodebd config does not '
+ 'enable transmit. Use eNB config.',
+ )
+
+ def test_provision_without_invasive_changes(self) -> None:
+ """
+ Test the scenario where:
+ - eNodeB has already been powered for 10 minutes without configuration
+ - Setting parameters which are 'non-invasive' on the eNodeB
+
+ 'Invasive' parameters are those which require special behavior to apply
+ the changes for the eNodeB.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform()
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+
+ # Send back some typical values
+ # And then SM should request regular parameter values
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ # Send back typical values for the regular parameters
+ req = Tr069MessageBuilder.\
+ get_regular_param_values_response(
+ admin_state=False,
+ earfcndl=39150,
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # SM will be requesting object parameter values
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting object param vals',
+ )
+
+ # Send back some typical values for object parameters
+ req = Tr069MessageBuilder.get_object_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # In this scenario, the ACS and thus state machine will not need
+ # to delete or add objects to the eNB configuration.
+ # SM should then just be attempting to set parameter values
+ self.assertTrue(
+ isinstance(resp, models.SetParameterValues),
+ 'State machine should be setting param values',
+ )
+
+ # Send back confirmation that the parameters were successfully set
+ req = models.SetParameterValuesResponse()
+ req.Status = 0
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+
+ # Send back some typical values
+ # And then SM should continue polling the read-only params
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'State machine should be ending session',
+ )
+
+ # If a different eNB is suddenly plugged in, or the same eNB sends a
+ # new Inform, enodebd should be able to handle it.
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform()
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ def test_reboot_after_invasive_changes(self) -> None:
+ """
+ Test the scenario where:
+ - eNodeB has already been powered for 10 minutes without configuration
+ - Setting parameters which are 'invasive' on the eNodeB
+ - Simulate the scenario up until reboot, and test that enodebd does
+ not try to complete configuration after reboot, because it is
+ waiting for REM process to finish running
+ - This test does not wait the ten minutes to simulate REM process
+ finishing on the Baicells eNodeB
+
+ 'Invasive' parameters are those which require special behavior to apply
+ the changes for the eNodeB.
+
+ In the case of the Baicells eNodeB, properly applying changes to
+ invasive parameters requires rebooting the device.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder\
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+ # Since the test utils pretend the eNB is set to 20MHz, we force this
+ # to 10 MHz, so the state machine sets this value.
+ acs_state_machine.mconfig.bandwidth_mhz = 10
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+
+ # Send back some typical values
+ # And then SM should request regular parameter values
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ # Send back typical values for the regular parameters
+ req = Tr069MessageBuilder.get_regular_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # SM will be requesting object parameter values
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting object param vals',
+ )
+
+ # Send back some typical values for object parameters
+ req = Tr069MessageBuilder.get_object_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # In this scenario, the ACS and thus state machine will not need
+ # to delete or add objects to the eNB configuration.
+ # SM should then just be attempting to set parameter values
+ self.assertTrue(
+ isinstance(resp, models.SetParameterValues),
+ 'State machine should be setting param values',
+ )
+
+ # Send back confirmation that the parameters were successfully set
+ req = models.SetParameterValuesResponse()
+ req.Status = 0
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Since invasive parameters have been set, then to apply the changes
+ # to the Baicells eNodeB, we need to reboot the device
+ self.assertTrue(isinstance(resp, models.Reboot))
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # After the reboot has been received, enodebd should end the
+ # provisioning session
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'After sending command to reboot the Baicells eNodeB, '
+ 'enodeb should end the TR-069 session.',
+ )
+
+ # At this point, sometime after the eNodeB reboots, we expect it to
+ # send an Inform indicating reboot. Since it should be in REM process,
+ # we hold off on finishing configuration, and end TR-069 sessions.
+ req = Tr069MessageBuilder.get_inform(
+ '48BF74', 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['1 BOOT', 'M Reboot'],
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'After receiving a post-reboot Inform, enodebd '
+ 'should end TR-069 sessions for 10 minutes to wait '
+ 'for REM process to finish.',
+ )
+
+ # Pretend that we have waited, and now we are in normal operation again
+ acs_state_machine.transition('wait_inform_post_reboot')
+ req = Tr069MessageBuilder.get_inform(
+ '48BF74', 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'After receiving a post-reboot Inform, enodebd '
+ 'should end TR-069 sessions for 10 minutes to wait '
+ 'for REM process to finish.',
+ )
+
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'enodebd should be requesting params',
+ )
+ self.assertTrue(
+ len(resp.ParameterNames.string) > 1,
+ 'Should be requesting transient params.',
+ )
+
+ def test_reboot_without_getting_optional(self) -> None:
+ """
+ The state machine should not skip figuring out which optional
+ parameters are present.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # And now reboot the eNodeB
+ acs_state_machine.transition('reboot')
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(isinstance(resp, models.Reboot))
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # After the reboot has been received, enodebd should end the
+ # provisioning session
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'After sending command to reboot the Baicells eNodeB, '
+ 'enodeb should end the TR-069 session.',
+ )
+
+ # At this point, sometime after the eNodeB reboots, we expect it to
+ # send an Inform indicating reboot. Since it should be in REM process,
+ # we hold off on finishing configuration, and end TR-069 sessions.
+ req = Tr069MessageBuilder.get_inform(
+ '48BF74', 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['1 BOOT', 'M Reboot'],
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'After receiving a post-reboot Inform, enodebd '
+ 'should end TR-069 sessions for 10 minutes to wait '
+ 'for REM process to finish.',
+ )
+
+ # Pretend that we have waited, and now we are in normal operation again
+ acs_state_machine.transition('wait_inform_post_reboot')
+ req = Tr069MessageBuilder.get_inform(
+ '48BF74', 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'After receiving a post-reboot Inform, enodebd '
+ 'should end TR-069 sessions for 10 minutes to wait '
+ 'for REM process to finish.',
+ )
+
+ # Since we haven't figured out the presence of optional parameters, the
+ # state machine should be requesting them now. There are three for the
+ # Baicells state machine.
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'enodebd should be requesting params',
+ )
+ self.assertTrue(
+ len(resp.ParameterNames.string) == 1,
+ 'Should be requesting optional params.',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param',
+ )
+ self.assertTrue(
+ len(resp.ParameterNames.string) == 1,
+ 'Should be requesting optional params.',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param',
+ )
+ self.assertTrue(
+ len(resp.ParameterNames.string) == 1,
+ 'Should be requesting optional params.',
+ )
+
+ def test_missing_mme_timeout_handler(self) -> None:
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ acs_state_machine.handle_tr069_message(inform_msg)
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ acs_state_machine.handle_tr069_message(req)
+
+ acs_state_machine.mme_timeout_handler.cancel()
+ acs_state_machine.mme_timeout_handler = None
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ acs_state_machine.handle_tr069_message(inform_msg)
+
+ def test_fault_after_set_parameters(self) -> None:
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+
+ # Send back some typical values
+ # And then SM should request regular parameter values
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ # Send back typical values for the regular parameters
+ req = Tr069MessageBuilder.get_regular_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # SM will be requesting object parameter values
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting object param vals',
+ )
+
+ # Send back some typical values for object parameters
+ req = Tr069MessageBuilder.get_object_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # In this scenario, the ACS and thus state machine will not need
+ # to delete or add objects to the eNB configuration.
+ # SM should then just be attempting to set parameter values
+ self.assertTrue(
+ isinstance(resp, models.SetParameterValues),
+ 'State machine should be setting param values',
+ )
+
+ req = models.Fault()
+ req.FaultCode = 12345
+ req.FaultString = 'Test FaultString'
+ acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ 'Error' in acs_state_machine.get_state(),
+ 'Should be in error state',
+ )
+
+ def test_autoremediation_from_fault(self):
+ """
+ Transition the state machine into the unexpected fault state, then
+ verify that it transitions itself back to WaitInform after an Inform
+ is received.
+ """
+ sm = EnodebAcsStateMachineBuilder.build_acs_state_machine(
+ EnodebDeviceName.BAICELLS,
+ )
+
+ # Send an initial inform
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ ['2 PERIODIC'],
+ )
+ resp = sm.handle_tr069_message(inform_msg)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Now send a fault
+ req = models.Fault()
+ req.FaultCode = 12345
+ req.FaultString = 'Test FaultString'
+ sm.handle_tr069_message(req)
+ self.assertTrue('Error' in sm.get_state(), 'Should be in error state')
+
+ # Send the Inform again, verify SM transitions out of fault
+ resp = sm.handle_tr069_message(inform_msg)
+ self.assertTrue(isinstance(resp, models.DummyInput))
+ self.assertEqual('Waiting for an Inform', sm.get_state())
diff --git a/tests/cavium_tests.py b/tests/cavium_tests.py
new file mode 100644
index 0000000..7cc9b9a
--- /dev/null
+++ b/tests/cavium_tests.py
@@ -0,0 +1,171 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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 data_models.data_model_parameters import ParameterName
+# pylint: disable=protected-access
+from devices.device_utils import EnodebDeviceName
+from tests.test_utils.enb_acs_builder import (
+ EnodebAcsStateMachineBuilder,
+)
+from tests.test_utils.enodeb_handler import EnodebHandlerTestCase
+from tests.test_utils.tr069_msg_builder import Tr069MessageBuilder
+from tr069 import models
+
+
+class CaviumHandlerTests(EnodebHandlerTestCase):
+ def test_count_plmns_less(self) -> None:
+ """
+ Tests the Cavium provisioning up to GetObjectParameters.
+
+ In particular tests when the eNB reports NUM_PLMNS less
+ than actually listed. The eNB says there are no PLMNs
+ defined when actually there are two.
+
+ Verifies that the number of PLMNs is correctly accounted.
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.CAVIUM)
+
+ # Send an Inform message
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '000FB7',
+ 'OC-LTE',
+ '120200002618AGP0003',
+ ['1 BOOT'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values: %s' % resp,
+ )
+
+ # Transient config response and request for parameter values
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values: %s' % resp,
+ )
+
+ # Send back typical values for the regular parameters
+ req = Tr069MessageBuilder.get_cavium_param_values_response(num_plmns=0)
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # SM will be requesting object parameter values
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting object param vals',
+ )
+
+ # Send back some object parameters with TWO plmns
+ req = Tr069MessageBuilder.get_cavium_object_param_values_response(
+ num_plmns=2,
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # In this scenario, the ACS and thus state machine will not need
+ # to delete or add objects to the eNB configuration.
+ # SM should then just be attempting to set parameter values
+ self.assertTrue(
+ isinstance(resp, models.SetParameterValues),
+ 'State machine should be setting param values',
+ )
+
+ # Number of PLMNs should reflect object count
+ num_plmns_cur = \
+ acs_state_machine \
+ .device_cfg.get_parameter(ParameterName.NUM_PLMNS)
+ self.assertEqual(num_plmns_cur, 2)
+
+ def test_count_plmns_more_defined(self) -> None:
+ """
+ Tests the Cavium provisioning up to GetObjectParameters.
+
+ In particular tests when the eNB has more PLMNs than is
+ currently defined in our data model (NUM_PLMNS_IN_CONFIG)
+ """
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.CAVIUM)
+
+ # Send an Inform message
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '000FB7',
+ 'OC-LTE',
+ '120200002618AGP0003',
+ ['1 BOOT'],
+ )
+ resp = acs_state_machine.handle_tr069_message(inform_msg)
+
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values: %s' % resp,
+ )
+
+ # Transient config response and request for parameter values
+ req = Tr069MessageBuilder.get_read_only_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values: %s' % resp,
+ )
+
+ # Send back regular parameters, and some absurd number of PLMNS
+ req = Tr069MessageBuilder.get_cavium_param_values_response(
+ num_plmns=100,
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # SM will be requesting object parameter values
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting object param vals',
+ )
+
+ # Send back some object parameters with an absurd number of PLMNs
+ req = Tr069MessageBuilder.get_cavium_object_param_values_response(
+ num_plmns=100,
+ )
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # In this scenario, the ACS and thus state machine will not need
+ # to delete or add objects to the eNB configuration.
+ # SM should then just be attempting to set parameter values
+ self.assertTrue(
+ isinstance(resp, models.SetParameterValues),
+ 'State machine should be setting param values',
+ )
+
+ # Number of PLMNs should reflect data model
+ num_plmns_cur = \
+ acs_state_machine \
+ .device_cfg.get_parameter(ParameterName.NUM_PLMNS)
+ self.assertEqual(num_plmns_cur, 6)
diff --git a/tests/configuration_init_tests.py b/tests/configuration_init_tests.py
new file mode 100644
index 0000000..5db4ffa
--- /dev/null
+++ b/tests/configuration_init_tests.py
@@ -0,0 +1,268 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+from unittest import TestCase
+
+from data_models.data_model_parameters import ParameterName
+from device_config.configuration_init import (
+ _get_enb_config,
+ _set_bandwidth,
+ _set_earfcn_freq_band_mode,
+ _set_management_server,
+ _set_misc_static_params,
+ _set_pci,
+ _set_perf_mgmt,
+ _set_plmnids_tac,
+ _set_s1_connection,
+ _set_tdd_subframe_config,
+)
+from device_config.enodeb_configuration import EnodebConfiguration
+from devices.baicells import BaicellsTrDataModel
+from exceptions import ConfigurationError
+from tests.test_utils.config_builder import EnodebConfigBuilder
+
+
+class EnodebConfigurationFactoryTest(TestCase):
+
+ def setUp(self):
+ self.data_model = BaicellsTrDataModel()
+ self.cfg = EnodebConfiguration(BaicellsTrDataModel())
+ self.device_cfg = EnodebConfiguration(BaicellsTrDataModel())
+
+ def tearDown(self):
+ self.data_model = None
+ self.cfg = None
+ self.device_cfg = None
+
+ def test_set_pci(self):
+ pci = 3
+ _set_pci(self.cfg, pci)
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.PCI), pci,
+ 'PCI value should be same as what was set',
+ )
+ with self.assertRaises(ConfigurationError):
+ _set_pci(self.cfg, 505)
+
+ def test_set_bandwidth(self):
+ mhz = 15
+ _set_bandwidth(self.cfg, self.data_model, mhz)
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.DL_BANDWIDTH),
+ mhz,
+ 'Should have set %s' % ParameterName.DL_BANDWIDTH,
+ )
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.UL_BANDWIDTH),
+ mhz,
+ 'Should have set %s' % ParameterName.UL_BANDWIDTH,
+ )
+
+ def test_set_tdd_subframe_config(self):
+ # Not TDD mode, should not try to set anything
+ self.device_cfg.set_parameter(
+ ParameterName.DUPLEX_MODE_CAPABILITY, 'Not TDDMode',
+ )
+ subframe = 0
+ special_subframe = 0
+ _set_tdd_subframe_config(
+ self.device_cfg, self.cfg, subframe,
+ special_subframe,
+ )
+ self.assertTrue(
+ ParameterName.SUBFRAME_ASSIGNMENT not in
+ self.cfg.get_parameter_names(),
+ )
+
+ # Invalid subframe assignment
+ self.device_cfg.set_parameter(
+ ParameterName.DUPLEX_MODE_CAPABILITY, 'TDDMode',
+ )
+ _set_tdd_subframe_config(
+ self.device_cfg, self.cfg, subframe,
+ special_subframe,
+ )
+ self.assertIn(
+ ParameterName.SUBFRAME_ASSIGNMENT,
+ self.cfg.get_parameter_names(),
+ 'Expected a subframe assignment',
+ )
+
+ def test_set_management_server(self):
+ _set_management_server(self.cfg)
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.PERIODIC_INFORM_ENABLE),
+ True, 'Expected periodic inform enable to be true',
+ )
+ self.assertTrue(
+ isinstance(
+ self.cfg.get_parameter(ParameterName.PERIODIC_INFORM_INTERVAL),
+ int,
+ ),
+ 'Expected periodic inform interval to ani integer',
+ )
+
+ def test_set_s1_connection(self):
+ invalid_mme_ip = 1234
+ invalid_mme_port = '8080'
+ mme_ip = '192.168.0.1'
+ mme_port = 8080
+
+ # MME IP should be a string
+ with self.assertRaises(ConfigurationError):
+ _set_s1_connection(self.cfg, invalid_mme_ip, mme_port)
+
+ # MME Port should be an integer
+ with self.assertRaises(ConfigurationError):
+ _set_s1_connection(self.cfg, mme_ip, invalid_mme_port)
+
+ # Check the ip and port are sort properly
+ _set_s1_connection(self.cfg, mme_ip, mme_port)
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.MME_IP), mme_ip,
+ 'Expected mme ip to be set',
+ )
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.MME_PORT), mme_port,
+ 'Expected mme port to be set',
+ )
+
+ def test_set_perf_mgmt(self):
+ mgmt_ip = '192.168.0.1'
+ mgmt_upload_interval = 300
+ mgmt_port = 8080
+ _set_perf_mgmt(self.cfg, mgmt_ip, mgmt_port)
+ self.assertTrue(
+ self.cfg.get_parameter(ParameterName.PERF_MGMT_ENABLE),
+ 'Expected perf mgmt to be enabled',
+ )
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.PERF_MGMT_UPLOAD_INTERVAL),
+ mgmt_upload_interval, 'Expected upload interval to be set',
+ )
+ expected_url = 'http://192.168.0.1:8080/'
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.PERF_MGMT_UPLOAD_URL),
+ expected_url, 'Incorrect Url',
+ )
+
+ def test_set_misc_static_params(self):
+ # IPSec enable as integer
+ self.device_cfg.set_parameter(ParameterName.IP_SEC_ENABLE, 0)
+ self.data_model.set_parameter_presence(ParameterName.GPS_ENABLE, True)
+ _set_misc_static_params(self.device_cfg, self.cfg, self.data_model)
+ self.assertTrue(
+ isinstance(
+ self.cfg.get_parameter(ParameterName.IP_SEC_ENABLE), int,
+ ),
+ 'Should support an integer IP_SEC_ENABLE parameter',
+ )
+
+ # IPSec enable as boolean
+ self.device_cfg.set_parameter(ParameterName.IP_SEC_ENABLE, 'False')
+ _set_misc_static_params(self.device_cfg, self.cfg, self.data_model)
+ self.assertTrue(
+ isinstance(
+ self.cfg.get_parameter(ParameterName.IP_SEC_ENABLE), bool,
+ ),
+ 'Should support a boolean IP_SEC_ENABLE parameter',
+ )
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.LOCAL_GATEWAY_ENABLE), 0,
+ 'Should be disabled',
+ )
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.CELL_RESERVED), False,
+ 'Should be disabled',
+ )
+ self.assertEqual(
+ self.cfg.get_parameter(ParameterName.MME_POOL_ENABLE), False,
+ 'Should be disabled',
+ )
+
+ def test_set_plmnids_tac(self):
+ # We only handle a single PLMNID for now
+ plmnids = '1, 2, 3, 4'
+ tac = 1
+ with self.assertRaises(ConfigurationError):
+ _set_plmnids_tac(self.cfg, plmnids, tac)
+
+ # Max PLMNID length is 6 characters
+ plmnids = '1234567'
+ with self.assertRaises(ConfigurationError):
+ _set_plmnids_tac(self.cfg, plmnids, tac)
+
+ # Check that only one PLMN element is enabled
+ plmnids = '1'
+ _set_plmnids_tac(self.cfg, plmnids, tac)
+ self.assertTrue(
+ self.cfg.get_parameter_for_object(
+ ParameterName.PLMN_N_ENABLE % 1, ParameterName.PLMN_N % 1,
+ ),
+ 'First PLMN should be enabled',
+ )
+ self.assertFalse(
+ self.cfg.has_object(ParameterName.PLMN_N % 2),
+ 'Second PLMN should be disabled',
+ )
+
+ def test_set_earafcn_freq_band_mode(self):
+ # Invalid earfcndl
+ with self.assertRaises(ConfigurationError):
+ invalid_earfcndl = -1
+ _set_earfcn_freq_band_mode(
+ self.device_cfg, self.cfg,
+ self.data_model, invalid_earfcndl,
+ )
+
+ # Duplex_mode is TDD but capability is FDD
+ with self.assertRaises(ConfigurationError):
+ self.device_cfg.set_parameter(
+ ParameterName.DUPLEX_MODE_CAPABILITY, 'FDDMode',
+ )
+ earfcndl = 38650 # Corresponds to TDD
+ _set_earfcn_freq_band_mode(
+ self.device_cfg, self.cfg,
+ self.data_model, earfcndl,
+ )
+
+ # Duplex_mode is FDD but capability is TDD
+ with self.assertRaises(ConfigurationError):
+ self.device_cfg.set_parameter(
+ ParameterName.DUPLEX_MODE_CAPABILITY, 'TDDMode',
+ )
+ earfcndl = 0 # Corresponds to FDD
+ _set_earfcn_freq_band_mode(
+ self.device_cfg, self.cfg,
+ self.data_model, earfcndl,
+ )
+
+ def test_get_enb_config(self):
+ mconfig = EnodebConfigBuilder.get_mconfig()
+ enb_config = _get_enb_config(mconfig, self.device_cfg)
+ self.assertTrue(
+ enb_config.earfcndl == 39150,
+ "Should give earfcndl from default eNB config",
+ )
+
+ mconfig = EnodebConfigBuilder.get_multi_enb_mconfig()
+ self.device_cfg.set_parameter(
+ ParameterName.SERIAL_NUMBER,
+ '120200002618AGP0003',
+ )
+ enb_config = _get_enb_config(mconfig, self.device_cfg)
+ self.assertTrue(
+ enb_config.earfcndl == 39151,
+ "Should give earfcndl from specific eNB config",
+ )
diff --git a/tests/data_model_tests.py b/tests/data_model_tests.py
new file mode 100644
index 0000000..e813354
--- /dev/null
+++ b/tests/data_model_tests.py
@@ -0,0 +1,137 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+
+from unittest import TestCase
+
+from data_models.data_model_parameters import ParameterName
+from devices.baicells import BaicellsTrDataModel
+
+
+class BaicellsTrDataModelTest(TestCase):
+ """
+ Tests for BaicellsTrDataModel
+ """
+
+ def test_is_parameter_present(self):
+ data_model = BaicellsTrDataModel()
+ with self.assertRaises(KeyError):
+ data_model.is_parameter_present(ParameterName.GPS_LONG)
+
+ result = data_model.is_parameter_present(ParameterName.DEVICE)
+ self.assertTrue(result, "Should have the device parameter")
+
+ def test_get_parameter(self):
+ param_info =\
+ BaicellsTrDataModel.get_parameter(ParameterName.GPS_STATUS)
+ self.assertIsNotNone(
+ param_info,
+ 'Cannot get parameter info on %s' % ParameterName.GPS_STATUS,
+ )
+ path = param_info.path
+ expected_path = 'Device.DeviceInfo.X_BAICELLS_COM_GPS_Status'
+ self.assertEqual(
+ path,
+ expected_path,
+ 'Path for parameter %s has incorrect value' %
+ ParameterName.GPS_STATUS,
+ )
+
+ def test_get_num_plmns(self):
+ n_plmns = BaicellsTrDataModel.get_num_plmns()
+ expected_n_plmns = 6
+ self.assertEqual(n_plmns, expected_n_plmns, 'Incorrect # of PLMNs')
+
+ def test_get_parameter_names(self):
+ name_list = BaicellsTrDataModel.get_parameter_names()
+
+ # Check that some random parameter names we expect are there
+ self.assertIn(
+ ParameterName.MME_STATUS, name_list,
+ 'Should have %s in parameter name list' %
+ ParameterName.MME_STATUS,
+ )
+ self.assertIn(
+ ParameterName.DUPLEX_MODE_CAPABILITY, name_list,
+ 'Should have %s in parameter name list' %
+ ParameterName.DUPLEX_MODE_CAPABILITY,
+ )
+ self.assertIn(
+ ParameterName.EARFCNUL, name_list,
+ 'Should have %s in parameter name list' %
+ ParameterName.EARFCNUL,
+ )
+
+ # Check that some other parameter names are missing
+ self.assertNotIn(
+ ParameterName.PLMN, name_list,
+ 'Should not have %s in parameter name list' %
+ ParameterName.PLMN,
+ )
+ self.assertNotIn(
+ ParameterName.PLMN_N % 1, name_list,
+ 'Should not have %s in parameter name list' %
+ ParameterName.PLMN_N % 1,
+ )
+
+ def test_get_numbered_param_names(self):
+ name_list = list(BaicellsTrDataModel.get_numbered_param_names().keys())
+
+ # Check that unnumbered parameters are missing
+ self.assertNotIn(
+ ParameterName.EARFCNDL, name_list,
+ 'Should not have %s in parameter name list' %
+ ParameterName.EARFCNDL,
+ )
+ self.assertNotIn(
+ ParameterName.MME_PORT, name_list,
+ 'Should not have %s in parameter name list' %
+ ParameterName.MME_PORT,
+ )
+ self.assertNotIn(
+ ParameterName.PERIODIC_INFORM_ENABLE, name_list,
+ 'Should not have %s in parameter name list' %
+ ParameterName.PERIODIC_INFORM_ENABLE,
+ )
+
+ # Check that some numbered parameters are present
+ self.assertIn(
+ ParameterName.PLMN_N % 1, name_list,
+ 'Should have %s in parameter name list' %
+ ParameterName.PLMN_N % 1,
+ )
+ self.assertIn(
+ ParameterName.PLMN_N % 6, name_list,
+ 'Should have %s in parameter name list' %
+ ParameterName.PLMN_N % 6,
+ )
+
+ def test_transform_for_magma(self):
+ gps_lat = str(10 * 1000000)
+ gps_lat_magma = BaicellsTrDataModel.transform_for_magma(
+ ParameterName.GPS_LAT, gps_lat,
+ )
+ expected = str(10.0)
+ self.assertEqual(gps_lat_magma, expected)
+
+ def test_transform_for_enb(self):
+ dl_bandwidth = 15
+ dl_bandwidth_enb = BaicellsTrDataModel.transform_for_enb(
+ ParameterName.DL_BANDWIDTH, dl_bandwidth,
+ )
+ expected = 'n75'
+ self.assertEqual(
+ dl_bandwidth_enb, expected,
+ 'Transform for enb returning incorrect value',
+ )
diff --git a/tests/device_utils_tests.py b/tests/device_utils_tests.py
new file mode 100644
index 0000000..c1d82d2
--- /dev/null
+++ b/tests/device_utils_tests.py
@@ -0,0 +1,94 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+
+from unittest import TestCase
+
+from devices.device_utils import (
+ EnodebDeviceName,
+ _parse_sw_version,
+ get_device_name,
+)
+from exceptions import UnrecognizedEnodebError
+
+
+class EnodebConfigUtilsTest(TestCase):
+ def test_get_device_name(self) -> None:
+ # Baicells
+ oui = '34ED0B'
+ version = 'BaiStation_V100R001C00B110SPC003'
+ data_model = get_device_name(oui, version)
+ expected = EnodebDeviceName.BAICELLS
+ self.assertEqual(data_model, expected, 'Incorrect data model')
+
+ # Baicells before bug-fix
+ oui = '34ED0B'
+ version = 'BaiStation_V100R001C00B110SPC002'
+ data_model = get_device_name(oui, version)
+ expected = EnodebDeviceName.BAICELLS_OLD
+ self.assertEqual(data_model, expected, 'Incorrect data model')
+
+ # Baicells QAFB
+ oui = '48BF74'
+ version = 'BaiBS_QAFB_some_version'
+ data_model = get_device_name(oui, version)
+ expected = EnodebDeviceName.BAICELLS_QAFB
+ self.assertEqual(data_model, expected, 'Incorrect data model')
+
+ # Cavium
+ oui = '000FB7'
+ version = 'Some version of Cavium'
+ data_model = get_device_name(oui, version)
+ expected = EnodebDeviceName.CAVIUM
+ self.assertEqual(data_model, expected, 'Incorrect data model')
+
+ # Unsupported device OUI
+ oui = 'beepboopbeep'
+ version = 'boopboopboop'
+ with self.assertRaises(UnrecognizedEnodebError):
+ get_device_name(oui, version)
+
+ # Unsupported software version for Baicells
+ oui = '34ED0B'
+ version = 'blingblangblong'
+ with self.assertRaises(UnrecognizedEnodebError):
+ get_device_name(oui, version)
+
+ def test_parse_version(self):
+ """ Test that version string is parsed correctly """
+ self.assertEqual(
+ _parse_sw_version('BaiStation_V100R001C00B110SPC003'),
+ [100, 1, 0, 110, 3],
+ )
+ self.assertEqual(
+ _parse_sw_version('BaiStation_V100R001C00B060SPC012'),
+ [100, 1, 0, 60, 12],
+ )
+ self.assertEqual(
+ _parse_sw_version('BaiStation_V100R001C00B060SPC012_FB_3'),
+ [100, 1, 0, 60, 12],
+ )
+ # Incorrect number of digits
+ self.assertEqual(
+ _parse_sw_version('BaiStation_V10R001C00B060SPC012'),
+ None,
+ )
+ self.assertEqual(
+ _parse_sw_version('XYZ123'),
+ None,
+ )
+ self.assertEqual(
+ _parse_sw_version(''),
+ None,
+ )
diff --git a/tests/enb_acs_manager_tests.py b/tests/enb_acs_manager_tests.py
new file mode 100644
index 0000000..c7147e0
--- /dev/null
+++ b/tests/enb_acs_manager_tests.py
@@ -0,0 +1,325 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+from unittest import TestCase
+
+from state_machines.enb_acs_manager import StateMachineManager
+from tests.test_utils.enb_acs_builder import (
+ EnodebAcsStateMachineBuilder,
+)
+from tests.test_utils.spyne_builder import (
+ get_spyne_context_with_ip,
+)
+from tests.test_utils.tr069_msg_builder import Tr069MessageBuilder
+from tr069 import models
+
+
+class StateMachineManagerTests(TestCase):
+ def test_handle_one_ip(self):
+ manager = self._get_manager()
+
+ # Send in an Inform message, and we should get an InformResponse
+ ctx = get_spyne_context_with_ip()
+ inform = Tr069MessageBuilder.get_inform()
+ req = manager.handle_tr069_message(ctx, inform)
+ self.assertTrue(
+ isinstance(req, models.InformResponse),
+ 'State machine handler should reply with an '
+ 'InformResponse',
+ )
+
+ def test_serial_not_found(self):
+ """
+ Test that the SM manager doesn't crash if serial number is not found
+ in an Inform message under any expected param path.
+ """
+ manager = self._get_manager()
+ ctx = get_spyne_context_with_ip("192.168.60.145")
+ inform_msg = models.Inform(
+ DeviceId=models.DeviceIdStruct(
+ Manufacturer='Unused',
+ OUI='48BF74',
+ ProductClass='Unused',
+ ),
+ Event=models.EventList(EventStruct=[]),
+ ParameterList=models.ParameterValueList(
+ ParameterValueStruct=[
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.DeviceInfo.HardwareVersion',
+ val_type='string',
+ data='VER.C',
+ ),
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.DeviceInfo.ManufacturerOUI',
+ val_type='string',
+ data='48BF74',
+ ),
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.DeviceInfo.SoftwareVersion',
+ val_type='string',
+ data='BaiBS_RTS_3.1.6',
+ ),
+ ],
+ ),
+ )
+
+ # No exception should be thrown, and we should return an empty response
+ resp = manager.handle_tr069_message(ctx, inform_msg)
+ self.assertTrue(isinstance(resp, models.DummyInput))
+
+ def test_handle_two_ips(self):
+ manager = self._get_manager()
+ ctx1 = get_spyne_context_with_ip("192.168.60.145")
+ ctx2 = get_spyne_context_with_ip("192.168.60.99")
+
+ ##### Start session for the first IP #####
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0001',
+ )
+ resp1 = manager.handle_tr069_message(ctx1, inform_msg)
+ self.assertTrue(
+ isinstance(resp1, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req1 = models.DummyInput()
+ resp1 = manager.handle_tr069_message(ctx1, req1)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp1, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req1 = Tr069MessageBuilder.get_fault()
+ resp1 = manager.handle_tr069_message(ctx1, req1)
+ self.assertTrue(
+ isinstance(resp1, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ ##### Start session for the second IP #####
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0002',
+ )
+ resp2 = manager.handle_tr069_message(ctx2, inform_msg)
+ self.assertTrue(
+ isinstance(resp2, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ ##### Continue session for the first IP #####
+ req1 = Tr069MessageBuilder.get_fault()
+ resp1 = manager.handle_tr069_message(ctx1, req1)
+ self.assertTrue(
+ isinstance(resp1, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req1 = Tr069MessageBuilder.get_fault()
+ resp1 = manager.handle_tr069_message(ctx1, req1)
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp1, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ ##### Continue session for the second IP #####
+ # Send an empty http request to kick off the rest of provisioning
+ req2 = models.DummyInput()
+ resp2 = manager.handle_tr069_message(ctx2, req2)
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp2, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req2 = Tr069MessageBuilder.get_fault()
+ resp2 = manager.handle_tr069_message(ctx2, req2)
+ self.assertTrue(
+ isinstance(resp2, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req2 = Tr069MessageBuilder.get_fault()
+ resp2 = manager.handle_tr069_message(ctx2, req2)
+ self.assertTrue(
+ isinstance(resp2, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req2 = Tr069MessageBuilder.get_fault()
+ resp2 = manager.handle_tr069_message(ctx2, req2)
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp2, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+
+ def test_handle_registered_enb(self):
+ """
+ When we have a config with eNB registered per serial, we should accept
+ TR-069 sessions from any registered eNB, and ereject from unregistered
+ eNB devices.
+ """
+ manager = self._get_manager_multi_enb()
+ ip1 = "192.168.60.145"
+ ctx1 = get_spyne_context_with_ip(ip1)
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ )
+ resp1 = manager.handle_tr069_message(ctx1, inform_msg)
+ self.assertTrue(
+ isinstance(resp1, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ ip2 = "192.168.60.146"
+ ctx2 = get_spyne_context_with_ip(ip2)
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ 'unregistered_serial',
+ )
+
+ resp2 = manager.handle_tr069_message(ctx2, inform_msg)
+ self.assertTrue(
+ isinstance(resp2, models.DummyInput),
+ 'Should respond with an empty HTTP response',
+ )
+
+ def test_ip_change(self) -> None:
+ manager = self._get_manager()
+
+ # Send an Inform
+ ip1 = "192.168.60.145"
+ ctx1 = get_spyne_context_with_ip(ip1)
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ )
+ resp1 = manager.handle_tr069_message(ctx1, inform_msg)
+ self.assertTrue(
+ isinstance(resp1, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+ handler1 = manager.get_handler_by_ip(ip1)
+
+ # Send an Inform from the same serial, but different IP
+ ip2 = "192.168.60.99"
+ ctx2 = get_spyne_context_with_ip(ip2)
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0003',
+ )
+ resp2 = manager.handle_tr069_message(ctx2, inform_msg)
+ self.assertTrue(
+ isinstance(resp2, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+ handler2 = manager.get_handler_by_ip(ip2)
+
+ # Now check that the serial is associated with the second ip
+ self.assertTrue(
+ (handler1 is handler2),
+ 'After an IP switch, the manager should have moved '
+ 'the handler to a new IP',
+ )
+
+ def test_serial_change(self) -> None:
+ manager = self._get_manager()
+ ip = "192.168.60.145"
+
+ # Send an Inform
+ ctx1 = get_spyne_context_with_ip(ip)
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0001',
+ )
+ resp1 = manager.handle_tr069_message(ctx1, inform_msg)
+ self.assertTrue(
+ isinstance(resp1, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+ handler1 = manager.get_handler_by_ip(ip)
+
+ # Send an Inform from the same serial, but different IP
+ ctx2 = get_spyne_context_with_ip(ip)
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0002',
+ )
+ resp2 = manager.handle_tr069_message(ctx2, inform_msg)
+ self.assertTrue(
+ isinstance(resp2, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+ handler2 = manager.get_handler_by_ip(ip)
+
+ # Now check that the serial is associated with the second ip
+ self.assertTrue(
+ (handler1 is not handler2),
+ 'After an IP switch, the manager should have moved '
+ 'the handler to a new IP',
+ )
+
+ def test_inform_from_baicells_qafb(self) -> None:
+ manager = self._get_manager()
+ ip = "192.168.60.145"
+
+ # Send an Inform
+ ctx1 = get_spyne_context_with_ip(ip)
+ inform_msg = Tr069MessageBuilder.get_qafb_inform(
+ '48BF74',
+ 'BaiBS_QAFB_v1234',
+ '120200002618AGP0001',
+ )
+ resp1 = manager.handle_tr069_message(ctx1, inform_msg)
+ self.assertTrue(
+ isinstance(resp1, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ def test_inform_from_unrecognized(self) -> None:
+ manager = self._get_manager()
+ ip = "192.168.60.145"
+
+ # Send an Inform
+ ctx1 = get_spyne_context_with_ip(ip)
+ inform_msg = Tr069MessageBuilder.get_qafb_inform(
+ '48BF74',
+ 'Unrecognized device',
+ '120200002618AGP0001',
+ )
+ resp1 = manager.handle_tr069_message(ctx1, inform_msg)
+ self.assertTrue(
+ isinstance(resp1, models.DummyInput),
+ 'Should end provisioninng session with empty response',
+ )
+
+ def _get_manager(self) -> StateMachineManager:
+ service = EnodebAcsStateMachineBuilder.build_magma_service()
+ return StateMachineManager(service)
+
+ def _get_manager_multi_enb(self) -> StateMachineManager:
+ service = EnodebAcsStateMachineBuilder.build_multi_enb_magma_service()
+ return StateMachineManager(service)
diff --git a/tests/enodeb_acs_states_tests.py b/tests/enodeb_acs_states_tests.py
new file mode 100644
index 0000000..561007d
--- /dev/null
+++ b/tests/enodeb_acs_states_tests.py
@@ -0,0 +1,226 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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 typing import Dict
+from unittest import TestCase
+from unittest.mock import patch
+
+from common.service import MagmaService
+from data_models.data_model import DataModel
+from devices.device_utils import EnodebDeviceName
+from exceptions import Tr069Error
+from state_machines.enb_acs_impl import BasicEnodebAcsStateMachine
+from state_machines.enb_acs_states import (
+ AcsMsgAndTransition,
+ AcsReadMsgResult,
+ EnodebAcsState,
+ WaitEmptyMessageState,
+ WaitInformState,
+ WaitSetParameterValuesState,
+)
+from tests.test_utils.enb_acs_builder import (
+ EnodebAcsStateMachineBuilder,
+)
+from tr069 import models
+
+
+class DummyDataModel(DataModel):
+ @classmethod
+ def get_parameter(cls, param_name):
+ return None
+
+ @classmethod
+ def _get_magma_transforms(cls):
+ return {}
+
+ @classmethod
+ def _get_enb_transforms(cls):
+ return {}
+
+ @classmethod
+ def get_load_parameters(cls):
+ return []
+
+ @classmethod
+ def get_num_plmns(cls) -> int:
+ return 1
+
+ @classmethod
+ def get_parameter_names(cls):
+ return []
+
+ @classmethod
+ def get_numbered_param_names(cls):
+ return {}
+
+
+class DummyHandler(BasicEnodebAcsStateMachine):
+
+ def __init__(
+ self,
+ service: MagmaService,
+ ) -> None:
+ self._state_map = {}
+ super().__init__(service=service, use_param_key=False)
+
+ def are_invasive_changes_applied(self) -> bool:
+ return False
+
+ def _init_state_map(self) -> None:
+ self._state_map = {
+ 'wait_inform': WaitInformState(
+ self,
+ when_done='wait_empty',
+ when_boot='wait_rem',
+ ),
+ }
+
+ @property
+ def state_map(self) -> Dict[str, EnodebAcsState]:
+ return self._state_map
+
+ @property
+ def disconnected_state_name(self) -> str:
+ return 'wait_inform'
+
+ @property
+ def unexpected_fault_state_name(self) -> str:
+ """ State to handle unexpected Fault messages """
+ return ''
+
+ @property
+ def device_name(self) -> EnodebDeviceName:
+ return "dummy"
+
+ @property
+ def config_postprocessor(self):
+ pass
+
+ def reboot_asap(self) -> None:
+ """
+ Send a request to reboot the eNodeB ASAP
+ """
+ pass
+
+ def is_enodeb_connected(self) -> bool:
+ return True
+
+ @property
+ def data_model_class(self):
+ return DummyDataModel
+
+
+class EnodebStatusTests(TestCase):
+
+ def _get_acs(self):
+ """ Get a dummy ACS statemachine for tests"""
+ service = EnodebAcsStateMachineBuilder.build_magma_service()
+ return DummyHandler(service)
+
+ @patch(
+ 'magma.enodebd.state_machines.enb_acs_states'
+ '.get_param_values_to_set',
+ )
+ @patch(
+ 'magma.enodebd.state_machines.enb_acs_states.get_obj_param_values_to_set',
+ )
+ def test_wait_set_parameter_values_state(
+ self, mock_get_obj_param,
+ mock_get_param,
+ ):
+ """ Test SetParameter return values"""
+ mock_get_param.return_value = {}
+ mock_get_obj_param.return_value = {}
+ test_message_0 = models.SetParameterValuesResponse()
+ test_message_0.Status = 0
+ test_message_1 = models.SetParameterValuesResponse()
+ test_message_1.Status = 1
+ # TC-1: return value is 0. No fault
+ acs_state = WaitSetParameterValuesState(
+ self._get_acs(), 'done',
+ 'invasive',
+ )
+
+ rc = acs_state.read_msg(test_message_0)
+ self.assertEqual(type(rc), AcsReadMsgResult)
+
+ # It raises exception if we return 1
+ self.assertRaises(
+ Tr069Error,
+ acs_state.read_msg, test_message_1,
+ )
+
+ # It passes if we return 1 and pass the non zero flag
+ acs_state = WaitSetParameterValuesState(
+ self._get_acs(), 'done',
+ 'invasive',
+ status_non_zero_allowed=True,
+ )
+ rc = acs_state.read_msg(test_message_1)
+ self.assertEqual(type(rc), AcsReadMsgResult)
+ rc = acs_state.read_msg(test_message_0)
+ self.assertEqual(type(rc), AcsReadMsgResult)
+
+ @patch(
+ 'magma.enodebd.state_machines.enb_acs_states.get_optional_param_to_check',
+ )
+ def test_wait_empty_message_state(
+ self,
+ mock_param_to_check,
+ ):
+ test_message_1 = models.DummyInput()
+ test_message_2 = models.SetParameterValuesResponse()
+ mock_param_to_check.return_value = True
+
+ # test 1: No missing_param_transition
+ # ensure we go to done state even when there are
+ # optional params to check
+ acs_state = WaitEmptyMessageState(
+ self._get_acs(),
+ when_done='done',
+ )
+ rc = acs_state.read_msg(test_message_1)
+ self.assertEqual(type(rc), AcsReadMsgResult)
+ self.assertEqual(rc.next_state, 'done')
+ self.assertEqual(rc.msg_handled, True)
+
+ # test 2: No unknown_param_transition
+ # ensure we go to missing state when there are
+ # optional params to check and missing state is specified
+ acs_state = WaitEmptyMessageState(
+ self._get_acs(),
+ when_done='done',
+ when_missing='missing',
+ )
+ rc = acs_state.read_msg(test_message_1)
+ self.assertEqual(type(rc), AcsReadMsgResult)
+ self.assertEqual(rc.next_state, 'missing')
+ self.assertEqual(rc.msg_handled, True)
+
+ # test 3: Negative test case send a message that is not empty
+ # ensure we return msg_handled is False
+ acs_state = WaitEmptyMessageState(
+ self._get_acs(),
+ when_done='done',
+ when_missing='missing',
+ )
+ rc = acs_state.read_msg(test_message_2)
+ self.assertEqual(type(rc), AcsReadMsgResult)
+ self.assertEqual(rc.next_state, None)
+ self.assertEqual(rc.msg_handled, False)
+
+ # test 4: Test get_msg
+ rc = acs_state.get_msg(test_message_1)
+ self.assertEqual(type(rc), AcsMsgAndTransition)
+ self.assertEqual(type(rc.msg), models.DummyInput)
+ self.assertEqual(rc.next_state, None)
diff --git a/tests/enodeb_configuration_tests.py b/tests/enodeb_configuration_tests.py
new file mode 100644
index 0000000..10742ef
--- /dev/null
+++ b/tests/enodeb_configuration_tests.py
@@ -0,0 +1,112 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+
+from unittest import TestCase
+
+from data_models.data_model_parameters import ParameterName
+from device_config.enodeb_configuration import EnodebConfiguration
+from devices.experimental.cavium import CaviumTrDataModel
+
+
+class EnodebConfigurationTest(TestCase):
+ def setUp(self):
+ self.config = EnodebConfiguration(CaviumTrDataModel)
+
+ def tearDown(self):
+ self.config = None
+
+ def test_data_model(self) -> None:
+ data_model = self.config.data_model
+ expected = CaviumTrDataModel
+ self.assertEqual(data_model, expected, 'Data model fetch incorrect')
+
+ def test_get_has_set_parameter(self) -> None:
+ param = ParameterName.ADMIN_STATE
+ self.config.set_parameter(param, True)
+ self.assertTrue(
+ self.config.has_parameter(param),
+ 'Expected to have parameter',
+ )
+ param_value = self.config.get_parameter(param)
+ expected = True
+ self.assertEqual(
+ param_value, expected,
+ 'Parameter value does not match what was set',
+ )
+
+ def test_add_has_delete_object(self) -> None:
+ object_name = ParameterName.PLMN_N % 1
+ self.assertFalse(self.config.has_object(object_name))
+ self.config.add_object(object_name)
+ self.assertTrue(self.config.has_object(object_name))
+ self.config.delete_object(object_name)
+ self.assertFalse(self.config.has_object(object_name))
+
+ def test_get_parameter_names(self) -> None:
+ # Should start off as an empty list
+ names_list = self.config.get_parameter_names()
+ self.assertEqual(len(names_list), 0, 'Expected 0 names')
+
+ # Should grow as we set parameters
+ self.config.set_parameter(ParameterName.ADMIN_STATE, True)
+ names_list = self.config.get_parameter_names()
+ self.assertEqual(len(names_list), 1, 'Expected 1 name')
+
+ # Parameter names should not include objects
+ self.config.add_object(ParameterName.PLMN)
+ names_list = self.config.get_parameter_names()
+ self.assertEqual(len(names_list), 1, 'Expected 1 name')
+
+ def test_get_object_names(self) -> None:
+ # Should start off as an empty list
+ obj_list = self.config.get_object_names()
+ self.assertEqual(len(obj_list), 0, 'Expected 0 names')
+
+ # Should grow as we set parameters
+ self.config.add_object(ParameterName.PLMN)
+ obj_list = self.config.get_object_names()
+ self.assertEqual(len(obj_list), 1, 'Expected 1 names')
+
+ def test_get_set_parameter_for_object(self) -> None:
+ self.config.add_object(ParameterName.PLMN_N % 1)
+ self.config.set_parameter_for_object(
+ ParameterName.PLMN_N_CELL_RESERVED % 1, True,
+ ParameterName.PLMN_N % 1,
+ )
+ param_value = self.config.get_parameter_for_object(
+ ParameterName.PLMN_N_CELL_RESERVED % 1, ParameterName.PLMN_N % 1,
+ )
+ self.assertTrue(
+ param_value,
+ 'Expected that the param for object was set correctly',
+ )
+
+ def test_get_parameter_names_for_object(self) -> None:
+ # Should start off empty
+ self.config.add_object(ParameterName.PLMN_N % 1)
+ param_list = self.config.get_parameter_names_for_object(
+ ParameterName.PLMN_N % 1,
+ )
+ self.assertEqual(len(param_list), 0, 'Should be an empty param list')
+
+ # Should increment as we set parameters
+ self.config.set_parameter_for_object(
+ ParameterName.PLMN_N_CELL_RESERVED % 1, True,
+ ParameterName.PLMN_N % 1,
+ )
+ param_list = self.config.get_parameter_names_for_object(
+ ParameterName.PLMN_N % 1,
+ )
+ self.assertEqual(len(param_list), 1, 'Should not be an empty list')
diff --git a/tests/enodeb_status_tests.py b/tests/enodeb_status_tests.py
new file mode 100644
index 0000000..3138fcf
--- /dev/null
+++ b/tests/enodeb_status_tests.py
@@ -0,0 +1,122 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+from unittest import TestCase
+
+from lte.protos.enodebd_pb2 import SingleEnodebStatus
+from devices.device_utils import EnodebDeviceName
+from enodeb_status import (
+ get_all_enb_status,
+ get_enb_status,
+ get_service_status_old,
+ get_single_enb_status,
+)
+from state_machines.enb_acs_manager import StateMachineManager
+from tests.test_utils.enb_acs_builder import (
+ EnodebAcsStateMachineBuilder,
+)
+from tests.test_utils.spyne_builder import (
+ get_spyne_context_with_ip,
+)
+from tests.test_utils.tr069_msg_builder import Tr069MessageBuilder
+
+
+class EnodebStatusTests(TestCase):
+ def test_get_service_status_old(self):
+ manager = self._get_manager()
+ status = get_service_status_old(manager)
+ self.assertTrue(
+ status['enodeb_connected'] == '0',
+ 'Should report no eNB connected',
+ )
+
+ ##### Start session for the first IP #####
+ ctx1 = get_spyne_context_with_ip("192.168.60.145")
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0001',
+ )
+ manager.handle_tr069_message(ctx1, inform_msg)
+ status = get_service_status_old(manager)
+ self.assertTrue(
+ status['enodeb_connected'] == '1',
+ 'Should report an eNB as conencted',
+ )
+ self.assertTrue(
+ status['enodeb_serial'] == '120200002618AGP0001',
+ 'eNodeB serial should match the earlier Inform',
+ )
+
+ def test_get_enb_status(self):
+ acs_state_machine = \
+ EnodebAcsStateMachineBuilder\
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+ try:
+ get_enb_status(acs_state_machine)
+ except KeyError:
+ self.fail(
+ 'Getting eNB status should succeed after constructor '
+ 'runs.',
+ )
+
+ def test_get_single_enb_status(self):
+ manager = self._get_manager()
+ ctx1 = get_spyne_context_with_ip("192.168.60.145")
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0001',
+ )
+ manager.handle_tr069_message(ctx1, inform_msg)
+ status = get_single_enb_status('120200002618AGP0001', manager)
+ self.assertEquals(
+ status.connected,
+ SingleEnodebStatus.StatusProperty.Value('ON'),
+ 'Status should be connected.',
+ )
+ self.assertEquals(
+ status.configured,
+ SingleEnodebStatus.StatusProperty.Value('OFF'),
+ 'Status should be not configured.',
+ )
+
+ def test_get_enodeb_all_status(self):
+ manager = self._get_manager()
+
+ ##### Test Empty #####
+ enb_status_by_serial = get_all_enb_status(manager)
+ self.assertTrue(enb_status_by_serial == {}, "No eNB connected")
+
+ ##### Start session for the first IP #####
+ ctx1 = get_spyne_context_with_ip("192.168.60.145")
+ # Send an Inform message, wait for an InformResponse
+ inform_msg = Tr069MessageBuilder.get_inform(
+ '48BF74',
+ 'BaiBS_RTS_3.1.6',
+ '120200002618AGP0001',
+ )
+ manager.handle_tr069_message(ctx1, inform_msg)
+ enb_status_by_serial = get_all_enb_status(manager)
+ enb_status = enb_status_by_serial.get('120200002618AGP0001')
+ self.assertEquals(
+ enb_status.enodeb_connected,
+ SingleEnodebStatus.StatusProperty.Value('ON'),
+ 'Status should be connected.',
+ )
+
+ def _get_manager(self) -> StateMachineManager:
+ service = EnodebAcsStateMachineBuilder.build_magma_service()
+ return StateMachineManager(service)
diff --git a/tests/freedomfi_one_tests.py b/tests/freedomfi_one_tests.py
new file mode 100644
index 0000000..4d88496
--- /dev/null
+++ b/tests/freedomfi_one_tests.py
@@ -0,0 +1,805 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+import logging
+import os
+from unittest.mock import Mock, call, patch
+
+from lte.protos.mconfig import mconfigs_pb2
+from common.service import MagmaService
+from data_models.data_model_parameters import ParameterName
+from devices.device_map import get_device_handler_from_name
+from devices.device_utils import EnodebDeviceName
+from devices.freedomfi_one import (
+ FreedomFiOneConfigurationInitializer,
+ SASParameters,
+ StatusParameters,
+)
+from tests.test_utils.config_builder import EnodebConfigBuilder
+from tests.test_utils.enb_acs_builder import (
+ EnodebAcsStateMachineBuilder,
+)
+from tests.test_utils.enodeb_handler import EnodebHandlerTestCase
+from tests.test_utils.tr069_msg_builder import Tr069MessageBuilder
+from tr069 import models
+
+SRC_CONFIG_DIR = os.path.join(
+ os.environ.get('MAGMA_ROOT'),
+ 'lte/gateway/configs',
+)
+
+
+class FreedomFiOneTests(EnodebHandlerTestCase):
+
+ def _get_freedomfi_one_read_only_param_values_response(
+ self,
+ ) -> models.GetParameterValuesResponse:
+ msg = models.GetParameterValuesResponse()
+ param_val_list = []
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.X_000E8F_DeviceFeature.X_000E8F_NEStatus'
+ '.X_000E8F_Sync_Status',
+ val_type='string',
+ data='InSync',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.X_000E8F_DeviceFeature.X_000E8F_NEStatus'
+ '.X_000E8F_SAS_Status',
+ val_type='string',
+ data='SUCCESS',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.X_000E8F_DeviceFeature.X_000E8F_NEStatus'
+ '.X_000E8F_eNB_Status',
+ val_type='string',
+ data='SUCCESS',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.X_000E8F_DeviceFeature.X_000E8F_NEStatus'
+ '.X_000E8F_DEFGW_Status',
+ val_type='string',
+ data='SUCCESS',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.GPS.ScanStatus',
+ val_type='string',
+ data='SUCCESS',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.GPS.LockedLongitude',
+ val_type='int',
+ data='-105272892',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.GPS.LockedLatitude',
+ val_type='int',
+ data='40019606',
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = param_val_list
+ return msg
+
+ def _get_freedomfi_one_param_values_response(self):
+ msg = models.GetParameterValuesResponse()
+ param_val_list = []
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = param_val_list
+
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.EARFCNDL',
+ val_type='int',
+ data='56240',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.GPS.ScanOnBoot',
+ val_type='boolean',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.AdminState',
+ val_type='boolean',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.PerfMgmt.Config.1.Enable',
+ val_type='boolean',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.Gateway.S1SigLinkServerList',
+ val_type='string',
+ data='10.0.2.1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_RRMConfig.X_000E8F_Cell_Number',
+ val_type='int',
+ data='2',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.TAC',
+ val_type='int',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.PerfMgmt.Config.1.PeriodicUploadInterval',
+ val_type='int',
+ data='60',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.DeviceInfo.SoftwareVersion',
+ val_type='string',
+ data='TEST3920@210901',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.UserContactInformation',
+ val_type='string',
+ data='M0LK4T',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.ProtectionLevel',
+ val_type='string',
+ data='GAA',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.CertSubject',
+ val_type='string',
+ data='/C=TW/O=Sercomm/OU=WInnForum CBSD Certificate/CN=P27-SCE4255W:%s',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.HeightType',
+ val_type='string',
+ data='AMSL',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.Category',
+ val_type='string',
+ data='A',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.GPS.ScanStatus',
+ val_type='string',
+ data='Success',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.ManagementServer.PeriodicInformInterval',
+ val_type='int',
+ data='60',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.FreqBandIndicator',
+ val_type='unsignedInt',
+ data='48',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.Common.CellIdentity',
+ val_type='unsignedInt',
+ data='101',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.GPS.LockedLongitude',
+ val_type='string',
+ data='-105272892',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.GPS.LockedLatitude',
+ val_type='string',
+ data='40019606',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.CPIEnable',
+ val_type='boolean',
+ data='0',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_RRMConfig.X_000E8F_CA_Enable',
+ val_type='boolean',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.GPS.ScanOnBoot',
+ val_type='boolean',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.X_000E8F_DeviceFeature.X_000E8F_NEStatus.X_000E8F_Sync_Status',
+ val_type='string',
+ data='InSync',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.PhyCellID',
+ val_type='string',
+ data='101,102',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.FAP.PerfMgmt.Config.1.URL',
+ val_type='string',
+ data=None,
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.Location',
+ val_type='string',
+ data='indoor',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.PHY.TDDFrame.SubFrameAssignment',
+ val_type='boolean',
+ data='2',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.IsPrimary',
+ val_type='boolean',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.Enable',
+ val_type='boolean',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.Server',
+ val_type='string',
+ data='https://spectrum-connect.federatedwireless.com/v1.2/',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.X_000E8F_DeviceFeature.X_000E8F_WebServerEnable',
+ val_type='boolean',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.CellReservedForOperatorUse',
+ val_type='boolean',
+ data='0',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.Tunnel.1.TunnelRef',
+ val_type='string',
+ data='Device.IP.Interface.1.IPv4Address.1.',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.REM.X_000E8F_tfcsManagerConfig.primSrc',
+ val_type='string',
+ data='GNSS',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_SAS.Enable',
+ val_type='boolean',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.ManagementServer.PeriodicInformEnable',
+ val_type='boolean',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNListNumberOfEntries',
+ val_type='int',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.PHY.TDDFrame.SpecialSubframePatterns',
+ val_type='int',
+ data='7',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.X_000E8F_RRMConfig.X_000E8F_CELL_Freq_Contiguous',
+ val_type='int',
+ data='0',
+ ),
+ )
+ param_val_list.append(
+ Tr069MessageBuilder.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.Gateway.S1SigLinkPort',
+ val_type='int',
+ data='36412',
+ ),
+ )
+ return msg
+
+ def _get_service_config(self):
+ return {
+ "tr069": {
+ "interface": "eth1",
+ "port": 48080,
+ "perf_mgmt_port": 8081,
+ "public_ip": "192.88.99.142",
+ },
+ "reboot_enodeb_on_mme_disconnected": True,
+ "s1_interface": "eth1",
+ "sas": {
+ "sas_enabled": True,
+ "sas_server_url":
+ "https://spectrum-connect.federatedwireless.com/v1.2/",
+ "sas_uid": "M0LK4T",
+ "sas_category": "A",
+ "sas_channel_type": "GAA",
+ "sas_cert_subject": "/C=TW/O=Sercomm/OU=WInnForum CBSD "
+ "Certificate/CN=P27-SCE4255W:%s",
+ "sas_icg_group_id": "",
+ "sas_location": "indoor",
+ "sas_height_type": "AMSL",
+ },
+ }
+
+ def build_freedomfi_one_acs_state_machine(self):
+ service = EnodebAcsStateMachineBuilder.build_magma_service(
+ mconfig=EnodebConfigBuilder.get_mconfig(),
+ service_config=self._get_service_config(),
+ )
+ handler_class = get_device_handler_from_name(
+ EnodebDeviceName.FREEDOMFI_ONE,
+ )
+ acs_state_machine = handler_class(service)
+ return acs_state_machine
+
+ def test_provision(self) -> None:
+ """
+ Test the basic provisioning workflow
+ 1 - enodeb sends Inform, enodebd sends InformResponse
+ 2 - enodeb sends empty HTTP message,
+ 3 - enodebd sends get transient params, updates the device state.
+ 4 - enodebd sends get param values, updates the device state
+ 5 - enodebd, sets fields including SAS fields.
+ """
+ logging.root.level = logging.DEBUG
+ acs_state_machine = self.build_freedomfi_one_acs_state_machine()
+
+ inform = Tr069MessageBuilder.get_inform(
+ oui="000E8F",
+ sw_version="TEST3920@210901",
+ enb_serial="2006CW5000023",
+ )
+ resp = acs_state_machine.handle_tr069_message(inform)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for read-only params
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ for tr_69_nodes in StatusParameters.STATUS_PARAMETERS.values():
+ self.assertIn(tr_69_nodes.path, resp.ParameterNames.string)
+
+ req = self._get_freedomfi_one_read_only_param_values_response()
+ get_resp = acs_state_machine.handle_tr069_message(req)
+
+ self.assertTrue(
+ isinstance(get_resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = self._get_freedomfi_one_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.SetParameterValues),
+ 'State machine should be setting parameters',
+ )
+ self.assertIsNotNone(
+ resp.ParameterKey.Data,
+ 'ParameterKey should be set for FreedomFiOne eNB',
+ )
+
+ msg = models.SetParameterValuesResponse()
+ msg.Status = 1
+ get_resp = acs_state_machine.handle_tr069_message(msg)
+ self.assertTrue(
+ isinstance(get_resp, models.GetParameterValues),
+ 'We should read back all parameters',
+ )
+
+ req = self._get_freedomfi_one_param_values_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'Provisioning completed with Dummy response',
+ )
+
+ def test_manual_reboot_during_provisioning(self) -> None:
+ """
+ Test a scenario where a Magma user goes through the enodebd CLI to
+ reboot the Baicells eNodeB.
+
+ This checks the scenario where the command is sent in the middle
+ of a TR-069 provisioning session.
+ """
+ logging.root.level = logging.DEBUG
+ acs_state_machine = self.build_freedomfi_one_acs_state_machine()
+
+ # Send an Inform message, wait for an InformResponse
+ inform = Tr069MessageBuilder.get_inform(
+ oui="000E8F",
+ sw_version="TEST3920@210901",
+ enb_serial="2006CW5000023",
+ )
+ resp = acs_state_machine.handle_tr069_message(inform)
+ self.assertTrue(
+ isinstance(resp, models.InformResponse),
+ 'Should respond with an InformResponse',
+ )
+
+ # Send an empty http request to kick off the rest of provisioning
+ req = models.DummyInput()
+ resp = acs_state_machine.handle_tr069_message(req)
+
+ # Expect a request for an optional parameter, three times
+ self.assertTrue(
+ isinstance(resp, models.GetParameterValues),
+ 'State machine should be requesting param values',
+ )
+ req = Tr069MessageBuilder.get_fault()
+
+ # User uses the CLI tool to get eNodeB to reboot
+ acs_state_machine.reboot_asap()
+
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.Reboot),
+ 'In reboot sequence, state machine should send a '
+ 'Reboot message.',
+ )
+ req = Tr069MessageBuilder.get_reboot_response()
+ resp = acs_state_machine.handle_tr069_message(req)
+ self.assertTrue(
+ isinstance(resp, models.DummyInput),
+ 'State machine should end TR-069 session after '
+ 'receiving a RebootResponse',
+ )
+
+ def test_post_processing(self) -> None:
+ """ Test FreedomFi One specific post processing functionality"""
+
+ acs_state_machine = self.build_freedomfi_one_acs_state_machine()
+ cfg_desired = Mock()
+ acs_state_machine.device_cfg.set_parameter(
+ ParameterName.SERIAL_NUMBER,
+ "2006CW5000023",
+ )
+
+ cfg_init = FreedomFiOneConfigurationInitializer(acs_state_machine)
+ cfg_init.postprocess(
+ EnodebConfigBuilder.get_mconfig(),
+ self._get_service_config(), cfg_desired,
+ )
+ expected = [
+ call.delete_parameter('EARFCNDL'),
+ call.delete_parameter('DL bandwidth'),
+ call.delete_parameter('UL bandwidth'),
+ call.set_parameter(
+ 'tunnel_ref',
+ 'Device.IP.Interface.1.IPv4Address.1.',
+ ),
+ call.set_parameter('prim_src', 'GNSS'),
+ call.set_parameter('carrier_agg_enable', True),
+ call.set_parameter('carrier_number', 2),
+ call.set_parameter('contiguous_cc', 0),
+ call.set_parameter('web_ui_enable', False),
+ call.set_parameter('sas_enabled', True),
+ call.set_parameter(
+ 'sas_server_url',
+ 'https://spectrum-connect.federatedwireless.com/v1.2/',
+ ),
+ call.set_parameter('sas_uid', 'M0LK4T'),
+ call.set_parameter('sas_category', 'A'),
+ call.set_parameter('sas_channel_type', 'GAA'),
+ call.set_parameter(
+ 'sas_cert_subject',
+ '/C=TW/O=Sercomm/OU=WInnForum CBSD Certificate/CN=P27-SCE4255W:%s',
+ ),
+ call.set_parameter('sas_location', 'indoor'),
+ call.set_parameter('sas_height_type', 'AMSL'),
+ ]
+ self.assertEqual(cfg_desired.mock_calls.sort(), expected.sort())
+
+ # Check without sas config
+ service_cfg = {
+ "tr069": {
+ "interface": "eth1",
+ "port": 48080,
+ "perf_mgmt_port": 8081,
+ "public_ip": "192.88.99.142",
+ },
+ "reboot_enodeb_on_mme_disconnected": True,
+ "s1_interface": "eth1",
+ }
+ cfg_desired = Mock()
+ cfg_init.postprocess(
+ EnodebConfigBuilder.get_mconfig(),
+ service_cfg, cfg_desired,
+ )
+ expected = [
+ call.delete_parameter('EARFCNDL'),
+ call.delete_parameter('DL bandwidth'),
+ call.delete_parameter('UL bandwidth'),
+ call.set_parameter(
+ 'tunnel_ref',
+ 'Device.IP.Interface.1.IPv4Address.1.',
+ ),
+ call.set_parameter('prim_src', 'GNSS'),
+ call.set_parameter('carrier_agg_enable', True),
+ call.set_parameter('carrier_number', 2),
+ call.set_parameter('contiguous_cc', 0),
+ call.set_parameter('web_ui_enable', False),
+ ]
+ self.assertEqual(cfg_desired.mock_calls.sort(), expected.sort())
+
+ service_cfg['web_ui_enable_list'] = ["2006CW5000023"]
+
+ expected = [
+ call.delete_parameter('EARFCNDL'),
+ call.delete_parameter('DL bandwidth'),
+ call.delete_parameter('UL bandwidth'),
+ call.set_parameter(
+ 'tunnel_ref',
+ 'Device.IP.Interface.1.IPv4Address.1.',
+ ),
+ call.set_parameter('prim_src', 'GNSS'),
+ call.set_parameter('carrier_agg_enable', True),
+ call.set_parameter('carrier_number', 2),
+ call.set_parameter('contiguous_cc', 0),
+ call.set_parameter('web_ui_enable', False),
+ call.set_parameter('web_ui_enable', True),
+ ]
+ cfg_desired = Mock()
+ cfg_init.postprocess(
+ EnodebConfigBuilder.get_mconfig(),
+ service_cfg, cfg_desired,
+ )
+ print(cfg_desired.mock_calls)
+ print(type(cfg_desired.mock_calls))
+ self.assertEqual(cfg_desired.mock_calls.sort(), expected.sort())
+
+ @patch('magma.configuration.service_configs.CONFIG_DIR', SRC_CONFIG_DIR)
+ def test_service_cfg_parsing(self):
+ """ Test the parsing of the service config file for enodebd.yml"""
+ service = MagmaService('enodebd', mconfigs_pb2.EnodebD())
+ service_cfg = service.config
+ service_cfg_1 = self._get_service_config()
+ service_cfg_1['web_ui_enable_list'] = []
+ service_cfg_1[FreedomFiOneConfigurationInitializer.SAS_KEY][
+ SASParameters.SAS_UID
+ ] = "INVALID_ID"
+ service_cfg_1[FreedomFiOneConfigurationInitializer.SAS_KEY][
+ SASParameters.SAS_CERT_SUBJECT
+ ] = "INVALID_CERT_SUBJECT"
+ self.assertEqual(service_cfg, service_cfg_1)
+
+ def test_status_nodes(self):
+ """ Test that the status of the node is valid"""
+ status = StatusParameters()
+
+ # Happy path
+ n1 = {
+ StatusParameters.DEFAULT_GW: "SUCCESS",
+ StatusParameters.SYNC_STATUS: "InSync",
+ StatusParameters.ENB_STATUS: "Success",
+ StatusParameters.SAS_STATUS: "Success",
+ StatusParameters.GPS_SCAN_STATUS: "SUCCESS",
+ ParameterName.GPS_LONG: "1",
+ ParameterName.GPS_LAT: "1",
+ }
+ device_config = Mock()
+ status.set_magma_device_cfg(n1, device_config)
+ expected = [
+ call.set_parameter(param_name='RF TX status', value=True),
+ call.set_parameter(param_name='GPS status', value=True),
+ call.set_parameter(param_name='PTP status', value=True),
+ call.set_parameter(param_name='MME status', value=True),
+ call.set_parameter(param_name='Opstate', value=True),
+ call.set_parameter('GPS lat', '1'),
+ call.set_parameter('GPS long', '1'),
+ ]
+ self.assertEqual(expected, device_config.mock_calls)
+
+ n2 = n1.copy()
+ # Verify we can handle specific none params
+ n2[StatusParameters.DEFAULT_GW] = None
+ n3 = n1.copy()
+ n3[StatusParameters.SYNC_STATUS] = None
+ n4 = n1.copy()
+ n4[StatusParameters.ENB_STATUS] = None
+ n5 = n1.copy()
+ n5[StatusParameters.SAS_STATUS] = None
+ n6 = n1.copy()
+ n6[StatusParameters.GPS_SCAN_STATUS] = None
+ n7 = n1.copy()
+ n7[ParameterName.GPS_LONG] = None
+ n8 = n1.copy()
+ n8[ParameterName.GPS_LAT] = None
+
+ device_config = Mock()
+ expected = [
+ call.set_parameter(param_name='RF TX status', value=True),
+ call.set_parameter(param_name='GPS status', value=True),
+ call.set_parameter(param_name='PTP status', value=True),
+ call.set_parameter(param_name='MME status', value=False),
+ call.set_parameter(param_name='Opstate', value=True),
+ call.set_parameter('GPS lat', '1'),
+ call.set_parameter('GPS long', '1'),
+ ]
+ status.set_magma_device_cfg(n2, device_config)
+ self.assertEqual(expected, device_config.mock_calls)
+
+ device_config = Mock()
+ expected = [
+ call.set_parameter(param_name='RF TX status', value=True),
+ call.set_parameter(param_name='GPS status', value=True),
+ call.set_parameter(param_name='PTP status', value=False),
+ call.set_parameter(param_name='MME status', value=True),
+ call.set_parameter(param_name='Opstate', value=True),
+ call.set_parameter('GPS lat', '1'),
+ call.set_parameter('GPS long', '1'),
+ ]
+ status.set_magma_device_cfg(n3, device_config)
+ self.assertEqual(expected, device_config.mock_calls)
+
+ device_config = Mock()
+ expected = [
+ call.set_parameter(param_name='RF TX status', value=True),
+ call.set_parameter(param_name='GPS status', value=True),
+ call.set_parameter(param_name='PTP status', value=True),
+ call.set_parameter(param_name='MME status', value=True),
+ call.set_parameter(param_name='Opstate', value=False),
+ call.set_parameter('GPS lat', '1'),
+ call.set_parameter('GPS long', '1'),
+ ]
+ status.set_magma_device_cfg(n4, device_config)
+ self.assertEqual(expected, device_config.mock_calls)
+
+ device_config = Mock()
+ expected = [
+ call.set_parameter(param_name='RF TX status', value=False),
+ call.set_parameter(param_name='GPS status', value=True),
+ call.set_parameter(param_name='PTP status', value=True),
+ call.set_parameter(param_name='MME status', value=True),
+ call.set_parameter(param_name='Opstate', value=True),
+ call.set_parameter('GPS lat', '1'),
+ call.set_parameter('GPS long', '1'),
+ ]
+ status.set_magma_device_cfg(n5, device_config)
+ self.assertEqual(expected, device_config.mock_calls)
+
+ device_config = Mock()
+ expected = [
+ call.set_parameter(param_name='RF TX status', value=True),
+ call.set_parameter(param_name='GPS status', value=False),
+ call.set_parameter(param_name='PTP status', value=False),
+ call.set_parameter(param_name='MME status', value=True),
+ call.set_parameter(param_name='Opstate', value=True),
+ call.set_parameter('GPS lat', '1'),
+ call.set_parameter('GPS long', '1'),
+ ]
+ status.set_magma_device_cfg(n6, device_config)
+ self.assertEqual(expected, device_config.mock_calls)
+
+ device_config = Mock()
+ expected = [
+ call.set_parameter(param_name='RF TX status', value=True),
+ call.set_parameter(param_name='GPS status', value=True),
+ call.set_parameter(param_name='PTP status', value=True),
+ call.set_parameter(param_name='MME status', value=True),
+ call.set_parameter(param_name='Opstate', value=True),
+ call.set_parameter('GPS lat', '1'),
+ call.set_parameter('GPS long', None),
+ ]
+ status.set_magma_device_cfg(n7, device_config)
+ self.assertEqual(expected, device_config.mock_calls)
+
+ device_config = Mock()
+ expected = [
+ call.set_parameter(param_name='RF TX status', value=True),
+ call.set_parameter(param_name='GPS status', value=True),
+ call.set_parameter(param_name='PTP status', value=True),
+ call.set_parameter(param_name='MME status', value=True),
+ call.set_parameter(param_name='Opstate', value=True),
+ call.set_parameter('GPS lat', None),
+ call.set_parameter('GPS long', '1'),
+ ]
+ status.set_magma_device_cfg(n8, device_config)
+ self.assertEqual(expected, device_config.mock_calls)
diff --git a/tests/pm_file_example.xml b/tests/pm_file_example.xml
new file mode 100644
index 0000000..920d951
--- /dev/null
+++ b/tests/pm_file_example.xml
@@ -0,0 +1,1302 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<PmFile xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="PmFileFormat.xsd">
+ <FileHeader>
+ <InfoModelReferenced>NanoCell-eNB-PM-V2.5.0</InfoModelReferenced>
+ <DnPrefix>CMCC</DnPrefix>
+ <SenderName>CMCC\\,SubNetwork=1\\,ManagementNode=1</SenderName>
+ <VendorName></VendorName>
+ <JobId>{}</JobId>
+ <BeginTime>2018-05-25T14:24:00+08:00</BeginTime>
+ <EndTime>2018-05-25T14:25:00+08:00</EndTime>
+ </FileHeader>
+ <Measurements>
+ <ObjectType>EutranCellTdd</ObjectType>
+ <PmName>
+ <N i="1">RRC.SetupTimeMean</N>
+ <N i="2">RRC.SetupTimeMax</N>
+ <N i="3">RRC.ConnMean</N>
+ <N i="4">RRC.ConnMax</N>
+ <N i="5">RRC.AttConnEstab</N>
+ <N i="6">RRC.AttConnEstab._Cause</N>
+ <N i="7">RRC.SuccConnEstab</N>
+ <N i="8">RRC.SuccConnEstab._Cause</N>
+ <N i="9">RRC.AttConnReestab</N>
+ <N i="10">RRC.AttConnReestab._Cause</N>
+ <N i="11">RRC.SuccConnReestab</N>
+ <N i="12">RRC.SuccConnReestab.NonSrccell</N>
+ <N i="13">RRC.SuccConnReestab._Cause</N>
+ <N i="14">RRC.ConnReleaseCsfb</N>
+ <N i="15">RRC.ConnRelease.RedirectTo2G</N>
+ <N i="16">RRC.ConnRelease.RedirectTo3G</N>
+ <N i="17">ERAB.NbrMeanEstab</N>
+ <N i="18">ERAB.NbrMeanEstab._Qci</N>
+ <N i="19">ERAB.EstabTimeMean</N>
+ <N i="20">ERAB.EstabTimeMax</N>
+ <N i="21">ERAB.NbrHoInc</N>
+ <N i="22">ERAB.NbrHoInc._Qci</N>
+ <N i="23">ERAB.NbrAttEstab</N>
+ <N i="24">ERAB.NbrAttEstab._Qci</N>
+ <N i="25">ERAB.NbrSuccEstab</N>
+ <N i="26">ERAB.NbrSuccEstab._Qci</N>
+ <N i="27">ERAB.NbrFailEstab</N>
+ <N i="28">ERAB.NbrFailEstab.CauseTransport</N>
+ <N i="29">ERAB.NbrFailEstab.CauseRadioResourcesNotAvailable</N>
+ <N i="30">ERAB.NbrFailEstab.CauseFailureInRadioInterfaceProcedure</N>
+ <N i="31">ERAB.NbrFailEstab._Cause</N>
+ <N i="32">ERAB.NbrReqRelEnb</N>
+ <N i="33">ERAB.NbrReqRelEnb.CauseUserInactivity</N>
+ <N i="34">ERAB.NbrReqRelEnb._Cause</N>
+ <N i="35">ERAB.NbrReqRelEnb._Qci</N>
+ <N i="36">ERAB.NbrReqRelEnb.Normal</N>
+ <N i="37">ERAB.NbrReqRelEnb.Normal._Qci</N>
+ <N i="38">ERAB.HoFail</N>
+ <N i="39">ERAB.HoFail._Qci</N>
+ <N i="40">ERAB.NbrLeft</N>
+ <N i="41">ERAB.NbrLeft._Qci</N>
+ <N i="42">CONTEXT.AttInitalSetup</N>
+ <N i="43">CONTEXT.AttInitalSetup.Csfb</N>
+ <N i="44">CONTEXT.SuccInitalSetup</N>
+ <N i="45">CONTEXT.SuccInitalSetup.Csfb</N>
+ <N i="46">CONTEXT.FailInitalSetup</N>
+ <N i="47">CONTEXT.FailInitalSetup._Cause</N>
+ <N i="48">CONTEXT.AttMod</N>
+ <N i="49">CONTEXT.AttMod.Csfb</N>
+ <N i="50">CONTEXT.SuccMod</N>
+ <N i="51">CONTEXT.SuccMod.Csfb</N>
+ <N i="52">CONTEXT.AttRelEnb</N>
+ <N i="53">CONTEXT.AttRelEnb._Cause</N>
+ <N i="54">CONTEXT.AttRelEnb.CauseUserInactivity</N>
+ <N i="55">CONTEXT.AttRelEnb.Normal</N>
+ <N i="56">CONTEXT.NbrLeft</N>
+ <N i="57">HO.AttOutInterEnbS1</N>
+ <N i="58">HO.AttOutInterEnbS1.1</N>
+ <N i="59">HO.PreSuccOutInterEnbS1</N>
+ <N i="60">HO.PreSuccOutInterEnbS1.1</N>
+ <N i="61">HO.ExecOutInterEnbS1</N>
+ <N i="62">HO.ExecOutInterEnbS1.1</N>
+ <N i="63">HO.SuccOutPrepInterEnbS1</N>
+ <N i="64">HO.SuccOutPrepInterEnbS1.1</N>
+ <N i="65">HO.AttOutExecInterEnbS1</N>
+ <N i="66">HO.AttOutExecInterEnbS1.1</N>
+ <N i="67">HO.SuccOutInterEnbS1</N>
+ <N i="68">HO.SuccOutInterEnbS1.1</N>
+ <N i="69">HO.AvgTimeInterEnbS1</N>
+ <N i="70">HO.AttOutInterEnbX2</N>
+ <N i="71">HO.AttOutInterEnbX2.1</N>
+ <N i="72">HO.SuccOutPrepInterEnbX2</N>
+ <N i="73">HO.SuccOutPrepInterEnbX2.1</N>
+ <N i="74">HO.AttOutExecInterEnbX2</N>
+ <N i="75">HO.AttOutExecInterEnbX2.1</N>
+ <N i="76">HO.SuccOutInterEnbX2</N>
+ <N i="77">HO.SuccOutInterEnbX2.1</N>
+ <N i="78">HO.AvgTimeInterEnbX2</N>
+ <N i="79">HO.FailOut</N>
+ <N i="80">HO.FailOut._Cause</N>
+ <N i="81">IRATHO.AttOutGeran</N>
+ <N i="82">IRATHO.SuccPrepOutGeran</N>
+ <N i="83">IRATHO.FailPrepOutGeran</N>
+ <N i="84">IRATHO.FailPrepOutGeran._Cause</N>
+ <N i="85">IRATHO.SuccOutGeran</N>
+ <N i="86">IRATHO.AvgTimeOutGeran</N>
+ <N i="87">IRATHO.AttOutUtran</N>
+ <N i="88">IRATHO.SuccPrepOutUtran</N>
+ <N i="89">IRATHO.FailPrepOutUtran</N>
+ <N i="90">IRATHO.FailPrepOutUtran._Cause</N>
+ <N i="91">IRATHO.SuccOutUtran</N>
+ <N i="92">PAG.PagReceived</N>
+ <N i="93">PAG.PagDiscarded</N>
+ <N i="94">PDCP.UpOctUl</N>
+ <N i="95">PDCP.UpOctUl._Qci</N>
+ <N i="96">PDCP.UpOctDl</N>
+ <N i="97">PDCP.UpOctDl._Qci</N>
+ <N i="98">PDCP.CpOctUl</N>
+ <N i="99">PDCP.CpOctDl</N>
+ <N i="100">PDCP.NbrPktUl</N>
+ <N i="101">PDCP.NbrPktUl._Qci</N>
+ <N i="102">PDCP.NbrPktLossUl</N>
+ <N i="103">PDCP.NbrPktLossUl._Qci</N>
+ <N i="104">PDCP.NbrPktDl</N>
+ <N i="105">PDCP.NbrPktDl._Qci</N>
+ <N i="106">PDCP.NbrPktLossDl</N>
+ <N i="107">PDCP.NbrPktLossDl._Qci</N>
+ <N i="108">PDCP.UpPktDelayDl</N>
+ <N i="109">PDCP.UpPktDelayDl._Qci</N>
+ <N i="110">PDCP.UpPktDiscardDl</N>
+ <N i="111">PDCP.UpPktDiscardDl._Qci</N>
+ <N i="112">RRU.DtchPrbAssnMeanUl</N>
+ <N i="113">RRU.DtchPrbAssnMeanDl</N>
+ <N i="114">RRU.PuschPrbTotMeanUl</N>
+ <N i="115">RRU.PdschPrbTotMeanDl</N>
+ <N i="116">RRU.PdcchCceUtilRatio</N>
+ <N i="117">RRU.PuschPrbMeanTot</N>
+ <N i="118">RRU.PdschPrbMeanTot</N>
+ <N i="119">RRU.TtiTotUl</N>
+ <N i="120">RRU.TtiTotDl</N>
+ <N i="121">RRU.DtchPrbAssnTotalUl.1</N>
+ <N i="122">RRU.DtchPrbAssnTotalDl.1</N>
+ <N i="123">MAC.NbrTbUl</N>
+ <N i="124">MAC.NbrTbUl.1</N>
+ <N i="125">MAC.NbrInitTbUl</N>
+ <N i="126">MAC.NbrInitTbUl.Qpsk</N>
+ <N i="127">MAC.NbrInitTbUl.16Qam</N>
+ <N i="128">MAC.NbrInitTbUl.64Qam</N>
+ <N i="129">MAC.NbrSuccInitTbUl</N>
+ <N i="130">MAC.NbrSuccInitTbUl.Qpsk</N>
+ <N i="131">MAC.NbrSuccInitTbUl.16Qam</N>
+ <N i="132">MAC.NbrSuccInitTbUl.64Qam</N>
+ <N i="133">MAC.NbrSuccInitTbUl.Qpsk.1</N>
+ <N i="134">MAC.NbrSuccInitTbUl.16Qam.1</N>
+ <N i="135">MAC.NbrSuccInitTbUl.64Qam.1</N>
+ <N i="136">MAC.NbrResErrTbUl</N>
+ <N i="137">MAC.NbrTbDl</N>
+ <N i="138">MAC.NbrTbDl.1</N>
+ <N i="139">MAC.NbrTbDl.Rank1</N>
+ <N i="140">MAC.NbrTbDl.Rank2</N>
+ <N i="141">MAC.NbrTbDl.Tm1</N>
+ <N i="142">MAC.NbrTbDl.Tm2</N>
+ <N i="143">MAC.NbrTbDl.Tm3</N>
+ <N i="144">MAC.NbrTbDl.Tm4</N>
+ <N i="145">MAC.NbrTbDl.Tm5</N>
+ <N i="146">MAC.NbrTbDl.Tm6</N>
+ <N i="147">MAC.NbrTbDl.Tm7</N>
+ <N i="148">MAC.NbrTbDl.Tm8</N>
+ <N i="149">MAC.NbrInitTbDl</N>
+ <N i="150">MAC.NbrInitTbDl.Qpsk</N>
+ <N i="151">MAC.NbrInitTbDl.16Qam</N>
+ <N i="152">MAC.NbrInitTbDl.64Qam</N>
+ <N i="153">MAC.NbrSuccInitTbDl</N>
+ <N i="154">MAC.NbrSuccInitTbDl.Qpsk</N>
+ <N i="155">MAC.NbrSuccInitTbDl.16Qam</N>
+ <N i="156">MAC.NbrSuccInitTbDl.64Qam</N>
+ <N i="157">MAC.NbrSuccInitTbDl.Qpsk.1</N>
+ <N i="158">MAC.NbrSuccInitTbDl.16Qam.1</N>
+ <N i="159">MAC.NbrSuccInitTbDl.64Qam.1</N>
+ <N i="160">MAC.NbrResErrTbDl</N>
+ <N i="161">PHY.NbrCqi0</N>
+ <N i="162">PHY.NbrCqi1</N>
+ <N i="163">PHY.NbrCqi2</N>
+ <N i="164">PHY.NbrCqi3</N>
+ <N i="165">PHY.NbrCqi4</N>
+ <N i="166">PHY.NbrCqi5</N>
+ <N i="167">PHY.NbrCqi6</N>
+ <N i="168">PHY.NbrCqi7</N>
+ <N i="169">PHY.NbrCqi8</N>
+ <N i="170">PHY.NbrCqi9</N>
+ <N i="171">PHY.NbrCqi10</N>
+ <N i="172">PHY.NbrCqi11</N>
+ <N i="173">PHY.NbrCqi12</N>
+ <N i="174">PHY.NbrCqi13</N>
+ <N i="175">PHY.NbrCqi14</N>
+ <N i="176">PHY.NbrCqi15</N>
+ <N i="177">PHY.ULMaxNL._PRB</N>
+ <N i="178">PHY.ULMeanNL._PRB</N>
+ <N i="179">PHY.CellMaxTxPower</N>
+ <N i="180">PHY.CellMeanTxPower</N>
+ <N i="181">S1SIG.ConnEstabSucc</N>
+ <N i="182">S1SIG.ConnEstabAtt</N>
+ </PmName>
+ <PmData>
+ <Pm Dn="CMCC,SubNetwork=1,ManagedElement=1,EnbFunction=1,EutranCellTdd=1" UserLabel="BJ01">
+ <V i="1">0</V>
+ <V i="2">0</V>
+ <V i="3">0</V>
+ <V i="4">0</V>
+ <V i="5">123</V>
+ <CV i="6">
+ <SN>RRC.AttConnEstab.Emergency</SN>
+ <SV>0</SV>
+ <SN>RRC.AttConnEstab.HI_PRIO_ACCESS</SN>
+ <SV>0</SV>
+ <SN>RRC.AttConnEstab.MT_ACCESS</SN>
+ <SV>0</SV>
+ <SN>RRC.AttConnEstab.MO_SIGNAL</SN>
+ <SV>0</SV>
+ <SN>RRC.AttConnEstab.MO_DATA</SN>
+ <SV>0</SV>
+ <SN>RRC.AttConnEstab.DeToAccess</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="7">99</V>
+ <CV i="8">
+ <SN>RRC.SuccConnEstab.EMERGENCY</SN>
+ <SV>0</SV>
+ <SN>RRC.SuccConnEstab.HIGHPRIORITYACCES</SN>
+ <SV>0</SV>
+ <SN>RRC.SuccConnEstab.MTACCESS</SN>
+ <SV>0</SV>
+ <SN>RRC.SuccConnEstab.MOSIGNALLING</SN>
+ <SV>0</SV>
+ <SN>RRC.SuccConnEstab.MODATA</SN>
+ <SV>0</SV>
+ <SN>RRC.SuccConnEstab.DE_TO_ACCESS</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="9">0</V>
+ <CV i="10">
+ <SN>RRC.AttConnReestab.RECONF_FAIL</SN>
+ <SV>654</SV>
+ <SN>RRC.AttConnReestab.HO_FAIL</SN>
+ <SV>321</SV>
+ <SN>RRC.AttConnReestab.OTHER</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="11">0</V>
+ <V i="12">0</V>
+ <CV i="13">
+ <SN>RRC.SuccConnReestab.RECONF_FAIL</SN>
+ <SV>0</SV>
+ <SN>RRC.SuccConnReestab.HO_FAIL</SN>
+ <SV>0</SV>
+ <SN>RRC.SuccConnReestab.OTHER</SN>
+ <SV>0</SV>
+ <SN>RRC.SuccConnReestab.Sum</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="14">0</V>
+ <V i="15">0</V>
+ <V i="16">0</V>
+ <V i="17">0</V>
+ <CV i="18">
+ <SN>ERAB.NbrMeanEstab.1</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrMeanEstab.2</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrMeanEstab.3</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrMeanEstab.4</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrMeanEstab.5</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrMeanEstab.6</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrMeanEstab.7</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrMeanEstab.8</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrMeanEstab.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="19">0</V>
+ <V i="20">0</V>
+ <V i="21">0</V>
+ <CV i="22">
+ <SN>ERAB.NbrHoInc.1</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrHoInc.2</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrHoInc.3</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrHoInc.4</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrHoInc.5</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrHoInc.6</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrHoInc.7</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrHoInc.8</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrHoInc.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="23">0</V>
+ <CV i="24">
+ <SN>ERAB.NbrAttEstab.1</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrAttEstab.2</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrAttEstab.3</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrAttEstab.4</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrAttEstab.5</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrAttEstab.6</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrAttEstab.7</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrAttEstab.8</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrAttEstab.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="25">0</V>
+ <CV i="26">
+ <SN>ERAB.NbrSuccEstab.1</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrSuccEstab.2</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrSuccEstab.3</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrSuccEstab.4</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrSuccEstab.5</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrSuccEstab.6</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrSuccEstab.7</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrSuccEstab.8</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrSuccEstab.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="27">0</V>
+ <V i="28">0</V>
+ <V i="29">0</V>
+ <V i="30">0</V>
+ <CV i="31">
+ <SN>ERAB.NbrFailEstab.InvalidQoSCombinations</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrFailEstab.MultipleERABIDInstances</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrFailEstab.NotSupportedQCIValue</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrFailEstab.S1IntraSystemHandoverTriggered</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrFailEstab.S1InterSystemHandoverTriggered</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrFailEstab.X2HandoverTriggered</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="32">0</V>
+ <V i="33">0</V>
+ <CV i="34">
+ <SN>ERAB.NbrReqRelEnb.CauseRADIORESOURCESNOTAVAILABLE</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.CauseREDUCELOADINSERVINGCELL</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.CauseFAILUREINTHERADIOINTERFACEPROCEDURE</SN>
+ <SV>135</SV>
+ <SN>ERAB.NbrReqRelEnb.CauseRELEASEDUETOEUTRANGENERATEDREASONS</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.CauseRADIOCONNECTIONWITHUELOST</SN>
+ <SV>65537</SV>
+ <SN>ERAB.NbrReqRelEnb.CauseOAMINTERVENTION</SN>
+ <SV>0</SV>
+ </CV>
+ <CV i="35">
+ <SN>ERAB.NbrReqRelEnb.1</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.2</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.3</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.4</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.5</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.6</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.7</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.8</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="36">0</V>
+ <CV i="37">
+ <SN>ERAB.NbrReqRelEnb.Normal.1</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.Normal.2</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.Normal.3</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.Normal.4</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.Normal.5</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.Normal.6</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.Normal.7</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.Normal.8</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrReqRelEnb.Normal.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="38">0</V>
+ <CV i="39">
+ <SN>ERAB.HoFail.1</SN>
+ <SV>0</SV>
+ <SN>ERAB.HoFail.2</SN>
+ <SV>0</SV>
+ <SN>ERAB.HoFail.3</SN>
+ <SV>0</SV>
+ <SN>ERAB.HoFail.4</SN>
+ <SV>0</SV>
+ <SN>ERAB.HoFail.5</SN>
+ <SV>0</SV>
+ <SN>ERAB.HoFail.6</SN>
+ <SV>0</SV>
+ <SN>ERAB.HoFail.7</SN>
+ <SV>0</SV>
+ <SN>ERAB.HoFail.8</SN>
+ <SV>0</SV>
+ <SN>ERAB.HoFail.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="40">0</V>
+ <CV i="41">
+ <SN>ERAB.NbrLeft.1</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrLeft.2</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrLeft.3</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrLeft.4</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrLeft.5</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrLeft.6</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrLeft.7</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrLeft.8</SN>
+ <SV>0</SV>
+ <SN>ERAB.NbrLeft.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="42">0</V>
+ <V i="43">0</V>
+ <V i="44">0</V>
+ <V i="45">0</V>
+ <V i="46">0</V>
+ <V i="47">0</V>
+ <V i="48">0</V>
+ <V i="49">0</V>
+ <V i="50">0</V>
+ <V i="51">0</V>
+ <V i="52">0</V>
+ <CV i="53">
+ <SN>CONTEXT.AttRelEnb.Cause1</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="54">0</V>
+ <V i="55">0</V>
+ <V i="56">0</V>
+ <V i="57">0</V>
+ <V i="58">0</V>
+ <V i="59">0</V>
+ <V i="60">0</V>
+ <V i="61">0</V>
+ <V i="62">0</V>
+ <V i="63">0</V>
+ <V i="64">0</V>
+ <V i="65">0</V>
+ <V i="66">0</V>
+ <V i="67">0</V>
+ <V i="68">0</V>
+ <V i="69">0</V>
+ <V i="70">0</V>
+ <V i="71">0</V>
+ <V i="72">0</V>
+ <V i="73">0</V>
+ <V i="74">0</V>
+ <V i="75">0</V>
+ <V i="76">0</V>
+ <V i="77">0</V>
+ <V i="78">0</V>
+ <V i="79">0</V>
+ <CV i="80">
+ <SN>HO.FailOut.1</SN>
+ <SV>0</SV>
+ <SN>HO.FailOut.2</SN>
+ <SV>0</SV>
+ <SN>HO.FailOut.3</SN>
+ <SV>0</SV>
+ <SN>HO.FailOut.4</SN>
+ <SV>0</SV>
+ <SN>HO.FailOut.5</SN>
+ <SV>0</SV>
+ <SN>HO.FailOut.6</SN>
+ <SV>0</SV>
+ <SN>HO.FailOut.7</SN>
+ <SV>0</SV>
+ <SN>HO.FailOut.8</SN>
+ <SV>0</SV>
+ <SN>HO.FailOut.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="81">0</V>
+ <V i="82">0</V>
+ <V i="83">0</V>
+ <CV i="84">
+ <SN>IRATHO.FailPrepOutGeran.Cause1</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="85">0</V>
+ <V i="86">0</V>
+ <V i="87">0</V>
+ <V i="88">0</V>
+ <V i="89">0</V>
+ <CV i="90">
+ <SN>IRATHO.FailPrepOutUtran.Cause1</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="91">0</V>
+ <V i="92">0</V>
+ <V i="93">0</V>
+ <V i="94">1000</V>
+ <CV i="95">
+ <SN>PDCP.UpOctUl.1</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctUl.2</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctUl.3</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctUl.4</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctUl.5</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctUl.6</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctUl.7</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctUl.8</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctUl.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="96">500</V>
+ <CV i="97">
+ <SN>PDCP.UpOctDl.1</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctDl.2</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctDl.3</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctDl.4</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctDl.5</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctDl.6</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctDl.7</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctDl.8</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpOctDl.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="98">0</V>
+ <V i="99">0</V>
+ <V i="100">0</V>
+ <CV i="101">
+ <SN>PDCP.NbrPktUl.1</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktUl.2</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktUl.3</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktUl.4</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktUl.5</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktUl.6</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktUl.7</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktUl.8</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktUl.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="102">0</V>
+ <CV i="103">
+ <SN>PDCP.NbrPktLossUl.1</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossUl.2</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossUl.3</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossUl.4</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossUl.5</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossUl.6</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossUl.7</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossUl.8</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossUl.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="104">0</V>
+ <CV i="105">
+ <SN>PDCP.NbrPktDl.1</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktDl.2</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktDl.3</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktDl.4</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktDl.5</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktDl.6</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktDl.7</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktDl.8</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktDl.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="106">0</V>
+ <CV i="107">
+ <SN>PDCP.NbrPktLossDl.1</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossDl.2</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossDl.3</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossDl.4</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossDl.5</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossDl.6</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossDl.7</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossDl.8</SN>
+ <SV>0</SV>
+ <SN>PDCP.NbrPktLossDl.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="108">0</V>
+ <CV i="109">
+ <SN>PDCP.UpPktDelayDl.1</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDelayDl.2</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDelayDl.3</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDelayDl.4</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDelayDl.5</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDelayDl.6</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDelayDl.7</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDelayDl.8</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDelayDl.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="110">0</V>
+ <CV i="111">
+ <SN>PDCP.UpPktDiscardDl.1</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDiscardDl.2</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDiscardDl.3</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDiscardDl.4</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDiscardDl.5</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDiscardDl.6</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDiscardDl.7</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDiscardDl.8</SN>
+ <SV>0</SV>
+ <SN>PDCP.UpPktDiscardDl.9</SN>
+ <SV>0</SV>
+ </CV>
+ <V i="112">0</V>
+ <V i="113">0</V>
+ <V i="114">0</V>
+ <V i="115">0</V>
+ <V i="116">0</V>
+ <V i="117">100</V>
+ <V i="118">100</V>
+ <V i="119">12000</V>
+ <V i="120">48002</V>
+ <V i="121">0</V>
+ <V i="122">0</V>
+ <V i="123">2</V>
+ <V i="124">0</V>
+ <V i="125">0</V>
+ <V i="126">0</V>
+ <V i="127">0</V>
+ <V i="128">0</V>
+ <V i="129">470</V>
+ <V i="130">0</V>
+ <V i="131">0</V>
+ <V i="132">0</V>
+ <V i="133">0</V>
+ <V i="134">0</V>
+ <V i="135">0</V>
+ <V i="136">2</V>
+ <V i="137">7456</V>
+ <V i="138">0</V>
+ <V i="139">0</V>
+ <V i="140">0</V>
+ <V i="141">0</V>
+ <V i="142">0</V>
+ <V i="143">0</V>
+ <V i="144">0</V>
+ <V i="145">0</V>
+ <V i="146">0</V>
+ <V i="147">0</V>
+ <V i="148">0</V>
+ <V i="149">7456</V>
+ <V i="150">0</V>
+ <V i="151">0</V>
+ <V i="152">0</V>
+ <V i="153">7456</V>
+ <V i="154">0</V>
+ <V i="155">0</V>
+ <V i="156">0</V>
+ <V i="157">0</V>
+ <V i="158">0</V>
+ <V i="159">0</V>
+ <V i="160">0</V>
+ <V i="161">0</V>
+ <V i="162">0</V>
+ <V i="163">0</V>
+ <V i="164">0</V>
+ <V i="165">0</V>
+ <V i="166">0</V>
+ <V i="167">0</V>
+ <V i="168">0</V>
+ <V i="169">0</V>
+ <V i="170">0</V>
+ <V i="171">0</V>
+ <V i="172">0</V>
+ <V i="173">0</V>
+ <V i="174">0</V>
+ <V i="175">0</V>
+ <V i="176">0</V>
+ <CV i="177">
+ <SN>PHY.ULMaxNL.1</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.2</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.3</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.4</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.5</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.6</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.7</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.8</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.9</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.10</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.11</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.12</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.13</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.14</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.15</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.16</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.17</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.18</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.19</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.20</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.21</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.22</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.23</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.24</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.25</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.26</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.27</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.28</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.29</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.30</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.31</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.32</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.33</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.34</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.35</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.36</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.37</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.38</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.39</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.40</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.41</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.42</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.43</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.44</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.45</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.46</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.47</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.48</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.49</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.50</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.51</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.52</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.53</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.54</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.55</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.56</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.57</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.58</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.59</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.60</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.61</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.62</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.63</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.64</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.65</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.66</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.67</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.68</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.69</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.70</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.71</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.72</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.73</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.74</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.75</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.76</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.77</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.78</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.79</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.80</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.81</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.82</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.83</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.84</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.85</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.86</SN>
+ <SV>-91</SV>
+ <SN>PHY.ULMaxNL.87</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.88</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.89</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.90</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.91</SN>
+ <SV>-90</SV>
+ <SN>PHY.ULMaxNL.92</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.93</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.94</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.95</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.96</SN>
+ <SV>-94</SV>
+ <SN>PHY.ULMaxNL.97</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.98</SN>
+ <SV>-93</SV>
+ <SN>PHY.ULMaxNL.99</SN>
+ <SV>-92</SV>
+ <SN>PHY.ULMaxNL.100</SN>
+ <SV>-93</SV>
+ </CV>
+ <CV i="178">
+ <SN>PHY.ULMeanNL.1</SN>
+ <SV>0</SV>
+ <SN>PHY.ULMeanNL.2</SN>
+ <SV>0</SV>
+ <SN>PHY.ULMeanNL.3</SN>
+ <SV>0</SV>
+ <SN>PHY.ULMeanNL.4</SN>
+ <SV>0</SV>
+ <SN>PHY.ULMeanNL.5</SN>
+ <SV>0</SV>
+ <SN>PHY.ULMeanNL.6</SN>
+ <SV>0</SV>
+ <SN>PHY.ULMeanNL.7</SN>
+ <SV>0</SV>
+ <SN>PHY.ULMeanNL.8</SN>
+ <SV>0</SV>
+ <SN>PHY.ULMeanNL.9</SN>
+ <SV>0</SV>
+ <SN>PHY.ULMeanNL.10</SN>
+ <SV>0</SV>
+ <SN>PHY.ULMeanNL.11</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.12</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.13</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.14</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.15</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.16</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.17</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.18</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.19</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.20</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.21</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.22</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.23</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.24</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.25</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.26</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.27</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.28</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.29</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.30</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.31</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.32</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.33</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.34</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.35</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.36</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.37</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.38</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.39</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.40</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.41</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.42</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.43</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.44</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.45</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.46</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.47</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.48</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.49</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.50</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.51</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.52</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.53</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.54</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.55</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.56</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.57</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.58</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.59</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.60</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.61</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.62</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.63</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.64</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.65</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.66</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.67</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.68</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.69</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.70</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.71</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.72</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.73</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.74</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.75</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.76</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.77</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.78</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.79</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.80</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.81</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.82</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.83</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.84</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.85</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.86</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.87</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.88</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.89</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.90</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.91</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.92</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.93</SN>
+ <SV>-104</SV>
+ <SN>PHY.ULMeanNL.94</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.95</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.96</SN>
+ <SV>-100</SV>
+ <SN>PHY.ULMeanNL.97</SN>
+ <SV>-103</SV>
+ <SN>PHY.ULMeanNL.98</SN>
+ <SV>-101</SV>
+ <SN>PHY.ULMeanNL.99</SN>
+ <SV>-102</SV>
+ <SN>PHY.ULMeanNL.100</SN>
+ <SV>-104</SV>
+ </CV>
+ <V i="179">-7</V>
+ <V i="180">-10</V>
+ <V i="181">0</V>
+ <V i="182">0</V>
+ </Pm>
+ </PmData>
+ </Measurements>
+ <Measurements>
+ <ObjectType>ManagedElement</ObjectType>
+ <PmName>
+ <N i="1">EQPT.MeanMeLoad</N>
+ <N i="2">EQPT.MaxMeLoad</N>
+ </PmName>
+ <PmData>
+ <Pm Dn="CMCC,SubNetwork=1,ManagedElement=1" UserLabel="BJ01">
+ <V i="1">0</V>
+ <V i="2">0</V>
+ </Pm>
+ </PmData>
+ </Measurements>
+ <Measurements>
+ <ObjectType>SctpAssoc</ObjectType>
+ <PmName>
+ <N i="1">SIG.SctpCongestionDuration</N>
+ <N i="2">SIG.SctpUnavailableDuration</N>
+ </PmName>
+ <PmData>
+ <Pm Dn="CMCC,SubNetwork=1,ManagedElement=1,SctpAssoc=1" UserLabel="BJ01">
+ <V i="1">0</V>
+ <V i="2">0</V>
+ </Pm>
+ </PmData>
+ </Measurements>
+</PmFile>
+
+
+
+<!--For reference, the XML schema for PM file uploads is:-->
+
+<!--<?xml version="1.0" encoding="UTF-8"?>-->
+<!-- 2011 rel. 2 sp1 (FoodHacker) use XMLSpy vCZTECH edit (http://www.altova.com) by -->
+<!--Telecom Network Management,Performance Management-->
+<!--Northbound Interface-->
+<!--Unified Performance Measurements File Format V1.0.0-->
+<!--Drafted by Li Jian, CMCC-->
+<!--Last Modified on 2012.02.08-->
+<!--
+<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" elementFormDefault="qualified" attributeFormDefault="unqualified" version="1.0.0">
+ <xs:element name="PmFile">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element name="FileHeader">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element name="InfoModelReferenced" type="xs:string"/>
+ <xs:element name="DnPrefix" type="xs:string"/>
+ <xs:element name="SenderName" type="xs:string"/>
+ <xs:element name="VendorName" type="xs:string"/>
+ <xs:element name="JobId" type="xs:string" nillable="true" minOccurs="1"/>
+ <xs:element name="BeginTime" type="xs:dateTime"/>
+ <xs:element name="EndTime" type="xs:dateTime"/>
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
+ <xs:element name="Measurements" minOccurs="0" maxOccurs="unbounded">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element name="ObjectType" type="xs:string"/>
+ <xs:element name="PmName">
+ <xs:complexType>
+ <xs:sequence minOccurs="0" maxOccurs="unbounded">
+ <xs:element name="N">
+ <xs:complexType>
+ <xs:simpleContent>
+ <xs:extension base="xs:string">
+ <xs:attribute name="i" type="xs:integer" use="required"/>
+ </xs:extension>
+ </xs:simpleContent>
+ </xs:complexType>
+ </xs:element>
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
+ <xs:element name="PmData">
+ <xs:complexType>
+ <xs:sequence minOccurs="0" maxOccurs="unbounded">
+ <xs:element name="Pm">
+ <xs:complexType>
+ <xs:choice minOccurs="0" maxOccurs="unbounded">
+ <xs:element name="V">
+ <xs:complexType>
+ <xs:simpleContent>
+ <xs:extension base="xs:string">
+ <xs:attribute name="i" type="xs:integer" use="required"/>
+ </xs:extension>
+ </xs:simpleContent>
+ </xs:complexType>
+ </xs:element>
+ <xs:element name="CV">
+ <xs:complexType>
+ <xs:sequence minOccurs="0" maxOccurs="unbounded">
+ <xs:element name="SN" type="xs:string"/>
+ <xs:element name="SV" type="xs:string"/>
+ </xs:sequence>
+ <xs:attribute name="i" type="xs:integer" use="required"/>
+ </xs:complexType>
+ </xs:element>
+ </xs:choice>
+ <xs:attribute name="Dn" type="xs:string" use="required"/>
+ <xs:attribute name="UserLabel" type="xs:string" use="required"/>
+ </xs:complexType>
+ </xs:element>
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
+</xs:schema>
+-->
diff --git a/tests/stats_manager_tests.py b/tests/stats_manager_tests.py
new file mode 100644
index 0000000..87d0177
--- /dev/null
+++ b/tests/stats_manager_tests.py
@@ -0,0 +1,126 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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 unittest import TestCase, mock
+from xml.etree import ElementTree
+
+import pkg_resources
+from enodebd import metrics
+from data_models.data_model_parameters import ParameterName
+from devices.device_utils import EnodebDeviceName
+from state_machines.enb_acs_manager import StateMachineManager
+from stats_manager import StatsManager
+from tests.test_utils.config_builder import EnodebConfigBuilder
+from tests.test_utils.enb_acs_builder import (
+ EnodebAcsStateMachineBuilder,
+)
+
+
+class StatsManagerTest(TestCase):
+ """
+ Tests for eNodeB statistics manager
+ """
+
+ def setUp(self) -> None:
+ service = EnodebConfigBuilder.get_service_config()
+ self.enb_acs_manager = StateMachineManager(service)
+ self.mgr = StatsManager(self.enb_acs_manager)
+ self.is_clear_stats_called = False
+
+ def tearDown(self):
+ self.mgr = None
+
+ def test_check_rf_tx(self):
+ """ Check that stats are cleared when transmit is disabled on eNB """
+ handler = EnodebAcsStateMachineBuilder \
+ .build_acs_state_machine(EnodebDeviceName.BAICELLS)
+ with mock.patch(
+ 'magma.enodebd.devices.baicells.BaicellsHandler.is_enodeb_connected',
+ return_value=True,
+ ):
+ handler.device_cfg.set_parameter(ParameterName.RF_TX_STATUS, True)
+ handler.device_cfg.set_parameter(
+ ParameterName.SERIAL_NUMBER,
+ '123454',
+ )
+ with mock.patch(
+ 'magma.enodebd.stats_manager.StatsManager'
+ '._clear_stats',
+ ) as func:
+ self.mgr._check_rf_tx_for_handler(handler)
+ func.assert_not_called()
+ handler.device_cfg.set_parameter(
+ ParameterName.RF_TX_STATUS,
+ False,
+ )
+ self.mgr._check_rf_tx_for_handler(handler)
+ func.assert_any_call()
+
+ def test_parse_stats(self):
+ """ Test that example statistics from eNodeB can be parsed, and metrics
+ updated """
+ # Example performance metrics structure, sent by eNodeB
+ pm_file_example = pkg_resources.resource_string(
+ __name__,
+ 'pm_file_example.xml',
+ )
+
+ root = ElementTree.fromstring(pm_file_example)
+ self.mgr._parse_pm_xml('1234', root)
+
+ # Check that metrics were correctly populated
+ # See '<V i="5">123</V>' in pm_file_example
+ rrc_estab_attempts = metrics.STAT_RRC_ESTAB_ATT.collect()
+ self.assertEqual(rrc_estab_attempts[0].samples[0][2], 123)
+ # See '<V i="7">99</V>' in pm_file_example
+ rrc_estab_successes = metrics.STAT_RRC_ESTAB_SUCC.collect()
+ self.assertEqual(rrc_estab_successes[0].samples[0][2], 99)
+ # See '<SV>654</SV>' in pm_file_example
+ rrc_reestab_att_reconf_fail = \
+ metrics.STAT_RRC_REESTAB_ATT_RECONF_FAIL.collect()
+ self.assertEqual(rrc_reestab_att_reconf_fail[0].samples[0][2], 654)
+ # See '<SV>65537</SV>' in pm_file_example
+ erab_rel_req_radio_conn_lost = \
+ metrics.STAT_ERAB_REL_REQ_RADIO_CONN_LOST.collect()
+ self.assertEqual(erab_rel_req_radio_conn_lost[0].samples[0][2], 65537)
+
+ pdcp_user_plane_bytes_ul = \
+ metrics.STAT_PDCP_USER_PLANE_BYTES_UL.collect()
+ pdcp_user_plane_bytes_dl = \
+ metrics.STAT_PDCP_USER_PLANE_BYTES_DL.collect()
+ self.assertEqual(pdcp_user_plane_bytes_ul[0].samples[0][1], {'enodeb': '1234'})
+ self.assertEqual(pdcp_user_plane_bytes_dl[0].samples[0][1], {'enodeb': '1234'})
+ self.assertEqual(pdcp_user_plane_bytes_ul[0].samples[0][2], 1000)
+ self.assertEqual(pdcp_user_plane_bytes_dl[0].samples[0][2], 500)
+
+ def test_clear_stats(self):
+ """
+ Check that stats of PMPM_FILE_TO_METRIC_MAP is cleared successfully
+ """
+ # Example performance metrics structure, sent by eNodeB
+ pm_file_example = pkg_resources.resource_string(
+ __name__,
+ 'pm_file_example.xml',
+ )
+
+ root = ElementTree.fromstring(pm_file_example)
+ self.mgr._parse_pm_xml('1234', root)
+
+ # Check that metrics were correctly populated
+ rrc_estab_attempts = metrics.STAT_RRC_ESTAB_ATT.collect()
+ self.assertEqual(rrc_estab_attempts[0].samples[0][2], 123)
+
+ self.mgr._clear_stats()
+ rrc_estab_attempts = metrics.STAT_RRC_ESTAB_ATT.collect()
+ # After clearing stats collection of metric should report 0
+ self.assertEqual(rrc_estab_attempts[0].samples[0][2], 0)
diff --git a/tests/test_utils/config_builder.py b/tests/test_utils/config_builder.py
new file mode 100644
index 0000000..f662b9f
--- /dev/null
+++ b/tests/test_utils/config_builder.py
@@ -0,0 +1,123 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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 lte.protos.mconfig import mconfigs_pb2
+from devices.device_utils import EnodebDeviceName
+
+
+class EnodebConfigBuilder:
+ @classmethod
+ def get_mconfig(
+ cls,
+ device: EnodebDeviceName = EnodebDeviceName.BAICELLS,
+ ) -> mconfigs_pb2.EnodebD:
+ mconfig = mconfigs_pb2.EnodebD()
+ mconfig.bandwidth_mhz = 20
+ # This earfcndl is actually unused, remove later
+ mconfig.earfcndl = 44490
+ mconfig.log_level = 1
+ mconfig.plmnid_list = "00101"
+ mconfig.pci = 260
+ mconfig.allow_enodeb_transmit = False
+ mconfig.tac = 1
+ if device is EnodebDeviceName.BAICELLS_QAFB:
+ # fdd config
+ mconfig.fdd_config.earfcndl = 9211
+ elif device is EnodebDeviceName.CAVIUM:
+ # fdd config
+ mconfig.fdd_config.earfcndl = 2405
+ else:
+ # tdd config
+ mconfig.tdd_config.earfcndl = 39150
+ mconfig.tdd_config.subframe_assignment = 2
+ mconfig.tdd_config.special_subframe_pattern = 7
+
+ return mconfig
+
+ @classmethod
+ def get_multi_enb_mconfig(
+ cls,
+ ) -> mconfigs_pb2.EnodebD:
+ mconfig = mconfigs_pb2.EnodebD()
+ mconfig.bandwidth_mhz = 20
+ mconfig.special_subframe_pattern = 7
+ # This earfcndl is actually unused, remove later
+ mconfig.earfcndl = 44490
+ mconfig.log_level = 1
+ mconfig.plmnid_list = "00101"
+ mconfig.pci = 260
+ mconfig.allow_enodeb_transmit = False
+ mconfig.subframe_assignment = 2
+ mconfig.tac = 1
+
+ # tdd config, unused because of multi-enb config
+ mconfig.tdd_config.earfcndl = 39150
+ mconfig.tdd_config.subframe_assignment = 2
+ mconfig.tdd_config.special_subframe_pattern = 7
+
+ id1 = '120200002618AGP0003'
+ #enb_conf_1 = mconfigs_pb2.EnodebD.EnodebConfig()
+ mconfig.enb_configs_by_serial[id1]\
+ .earfcndl = 39151
+ mconfig.enb_configs_by_serial[id1]\
+ .subframe_assignment = 2
+ mconfig.enb_configs_by_serial[id1]\
+ .special_subframe_pattern = 7
+ mconfig.enb_configs_by_serial[id1]\
+ .pci = 259
+ mconfig.enb_configs_by_serial[id1]\
+ .bandwidth_mhz = 20
+ mconfig.enb_configs_by_serial[id1] \
+ .tac = 1
+ mconfig.enb_configs_by_serial[id1] \
+ .cell_id = 0
+ mconfig.enb_configs_by_serial[id1]\
+ .transmit_enabled = True
+ mconfig.enb_configs_by_serial[id1]\
+ .device_class = 'Baicells Band 40'
+
+ id2 = '120200002618AGP0004'
+ #enb_conf_2 = mconfigs_pb2.EnodebD.EnodebConfig()
+ mconfig.enb_configs_by_serial[id2]\
+ .earfcndl = 39151
+ mconfig.enb_configs_by_serial[id2]\
+ .subframe_assignment = 2
+ mconfig.enb_configs_by_serial[id2]\
+ .special_subframe_pattern = 7
+ mconfig.enb_configs_by_serial[id2]\
+ .pci = 261
+ mconfig.enb_configs_by_serial[id2] \
+ .bandwidth_mhz = 20
+ mconfig.enb_configs_by_serial[id2] \
+ .tac = 1
+ mconfig.enb_configs_by_serial[id2] \
+ .cell_id = 0
+ mconfig.enb_configs_by_serial[id2]\
+ .transmit_enabled = True
+ mconfig.enb_configs_by_serial[id2]\
+ .device_class = 'Baicells Band 40'
+
+ return mconfig
+
+ @classmethod
+ def get_service_config(cls):
+ return {
+ "tr069": {
+ "interface": "eth1",
+ "port": 48080,
+ "perf_mgmt_port": 8081,
+ "public_ip": "192.88.99.142",
+ },
+ "reboot_enodeb_on_mme_disconnected": True,
+ "s1_interface": "eth1",
+ }
diff --git a/tests/test_utils/enb_acs_builder.py b/tests/test_utils/enb_acs_builder.py
new file mode 100644
index 0000000..75f55be
--- /dev/null
+++ b/tests/test_utils/enb_acs_builder.py
@@ -0,0 +1,92 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+import asyncio
+from typing import Dict
+from unittest import mock
+
+from lte.protos.mconfig import mconfigs_pb2
+from common.service import MagmaService
+from devices.device_map import get_device_handler_from_name
+from devices.device_utils import EnodebDeviceName
+from state_machines.enb_acs import EnodebAcsStateMachine
+from state_machines.enb_acs_manager import StateMachineManager
+from tests.test_utils.config_builder import EnodebConfigBuilder
+
+
+class EnodebAcsStateMachineBuilder:
+ @classmethod
+ def build_acs_manager(
+ cls,
+ device: EnodebDeviceName = EnodebDeviceName.BAICELLS,
+ ) -> StateMachineManager:
+ service = cls.build_magma_service(device)
+ return StateMachineManager(service)
+
+ @classmethod
+ def build_multi_enb_acs_manager(
+ cls,
+ ) -> StateMachineManager:
+ service = cls.build_multi_enb_magma_service()
+ return StateMachineManager(service)
+
+ @classmethod
+ def build_multi_enb_acs_state_machine(
+ cls,
+ device: EnodebDeviceName = EnodebDeviceName.BAICELLS,
+ ) -> EnodebAcsStateMachine:
+ # Build the state_machine
+ service = cls.build_multi_enb_magma_service()
+ handler_class = get_device_handler_from_name(device)
+ acs_state_machine = handler_class(service)
+ return acs_state_machine
+
+ @classmethod
+ def build_acs_state_machine(
+ cls,
+ device: EnodebDeviceName = EnodebDeviceName.BAICELLS,
+ ) -> EnodebAcsStateMachine:
+ # Build the state_machine
+ service = cls.build_magma_service(device)
+ handler_class = get_device_handler_from_name(device)
+ acs_state_machine = handler_class(service)
+ return acs_state_machine
+
+ @classmethod
+ def build_magma_service(
+ cls,
+ device: EnodebDeviceName = EnodebDeviceName.BAICELLS,
+ mconfig: mconfigs_pb2.EnodebD = None,
+ service_config: Dict = None,
+ ) -> MagmaService:
+ event_loop = asyncio.get_event_loop()
+ if not mconfig:
+ mconfig = EnodebConfigBuilder.get_mconfig(device)
+ if not service_config:
+ service_config = EnodebConfigBuilder.get_service_config()
+ with mock.patch('magma.common.service.MagmaService') as MockService:
+ MockService.config = service_config
+ MockService.mconfig = mconfig
+ MockService.loop = event_loop
+ return MockService
+
+ @classmethod
+ def build_multi_enb_magma_service(cls) -> MagmaService:
+ event_loop = asyncio.get_event_loop()
+ mconfig = EnodebConfigBuilder.get_multi_enb_mconfig()
+ service_config = EnodebConfigBuilder.get_service_config()
+ with mock.patch('magma.common.service.MagmaService') as MockService:
+ MockService.config = service_config
+ MockService.mconfig = mconfig
+ MockService.loop = event_loop
+ return MockService
diff --git a/tests/test_utils/enodeb_handler.py b/tests/test_utils/enodeb_handler.py
new file mode 100644
index 0000000..ef2b11f
--- /dev/null
+++ b/tests/test_utils/enodeb_handler.py
@@ -0,0 +1,39 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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 unittest import TestCase, mock
+
+import magma.enodebd.tests.test_utils.mock_functions as enb_mock
+
+
+class EnodebHandlerTestCase(TestCase):
+ """
+ Sets up test class with a set of patches needed for eNodeB handlers
+ """
+
+ def setUp(self):
+ self.patches = {
+ enb_mock.GET_IP_FROM_IF_PATH:
+ mock.Mock(side_effect=enb_mock.mock_get_ip_from_if),
+ enb_mock.LOAD_SERVICE_MCONFIG_PATH:
+ mock.Mock(
+ side_effect=enb_mock.mock_load_service_mconfig_as_json,
+ ),
+ }
+ self.applied_patches = [
+ mock.patch(patch, data) for patch, data in
+ self.patches.items()
+ ]
+ for patch in self.applied_patches:
+ patch.start()
+ self.addCleanup(mock.patch.stopall)
diff --git a/tests/test_utils/mock_functions.py b/tests/test_utils/mock_functions.py
new file mode 100644
index 0000000..59fb0c7
--- /dev/null
+++ b/tests/test_utils/mock_functions.py
@@ -0,0 +1,31 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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 typing import Any
+
+GET_IP_FROM_IF_PATH = \
+ 'magma.enodebd.device_config.configuration_init.get_ip_from_if'
+
+LOAD_SERVICE_MCONFIG_PATH = \
+ 'magma.enodebd.device_config.configuration_init.load_service_mconfig_as_json'
+
+
+def mock_get_ip_from_if(
+ _iface_name: str,
+ _preference: Any = None,
+) -> str:
+ return '192.168.60.142'
+
+
+def mock_load_service_mconfig_as_json(_service_name: str) -> Any:
+ return {}
diff --git a/tests/test_utils/spyne_builder.py b/tests/test_utils/spyne_builder.py
new file mode 100644
index 0000000..0717aad
--- /dev/null
+++ b/tests/test_utils/spyne_builder.py
@@ -0,0 +1,26 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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 unittest import mock
+
+from spyne.server.wsgi import WsgiMethodContext
+
+
+def get_spyne_context_with_ip(
+ req_ip: str = "192.168.60.145",
+) -> WsgiMethodContext:
+ with mock.patch('spyne.server.wsgi.WsgiApplication') as MockTransport:
+ MockTransport.req_env = {"REMOTE_ADDR": req_ip}
+ with mock.patch('spyne.server.wsgi.WsgiMethodContext') as MockContext:
+ MockContext.transport = MockTransport
+ return MockContext
diff --git a/tests/test_utils/tr069_msg_builder.py b/tests/test_utils/tr069_msg_builder.py
new file mode 100644
index 0000000..3dc0f64
--- /dev/null
+++ b/tests/test_utils/tr069_msg_builder.py
@@ -0,0 +1,1000 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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 typing import Any, List, Optional
+
+from tr069 import models
+
+
+class Tr069MessageBuilder:
+ @classmethod
+ def get_parameter_value_struct(
+ cls,
+ name: str,
+ val_type: str,
+ data: Any,
+ ) -> models.ParameterValueStruct:
+ param_value = models.ParameterValueStruct()
+ param_value.Name = name
+ value = models.anySimpleType()
+ value.type = val_type
+ value.Data = data
+ param_value.Value = value
+ return param_value
+
+ @classmethod
+ def get_fault(cls) -> models.Fault:
+ msg = models.Fault()
+ msg.FaultCode = 0
+ msg.FaultString = 'Some sort of fault'
+ return msg
+
+ @classmethod
+ def get_reboot_inform(cls) -> models.Inform:
+ msg = cls.get_inform()
+ events = []
+
+ event_boot = models.EventStruct()
+ event_boot.EventCode = '1 BOOT'
+ events.append(event_boot)
+
+ event_reboot = models.EventStruct()
+ event_reboot.EventCode = 'M Reboot'
+ events.append(event_reboot)
+
+ msg.Event.EventStruct = events
+ return msg
+
+ @classmethod
+ def get_qafb_inform(
+ cls,
+ oui: str = '48BF74',
+ sw_version: str = 'BaiBS_QAFB_1.6.4',
+ enb_serial: str = '1202000181186TB0006',
+ event_codes: Optional[List[str]] = None,
+ ) -> models.Inform:
+ if event_codes is None:
+ event_codes = []
+ msg = models.Inform()
+
+ # DeviceId
+ device_id = models.DeviceIdStruct()
+ device_id.Manufacturer = 'Unused'
+ device_id.OUI = oui
+ device_id.ProductClass = 'Unused'
+ device_id.SerialNumber = enb_serial
+ msg.DeviceId = device_id
+
+ # Event
+ msg.Event = models.EventList()
+ event_list = []
+ for code in event_codes:
+ event = models.EventStruct()
+ event.EventCode = code
+ event.CommandKey = ''
+ event_list.append(event)
+ msg.Event.EventStruct = event_list
+
+ # ParameterList
+ val_list = []
+ val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.DeviceInfo.HardwareVersion',
+ val_type='string',
+ data='VER.C',
+ ),
+ )
+ val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.DeviceInfo.ManufacturerOUI',
+ val_type='string',
+ data=oui,
+ ),
+ )
+ val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.DeviceInfo.SoftwareVersion',
+ val_type='string',
+ data=sw_version,
+ ),
+ )
+ val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.DeviceInfo.SerialNumber',
+ val_type='string',
+ data=enb_serial,
+ ),
+ )
+ val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.ManagementServer.ConnectionRequestURL',
+ val_type='string',
+ data='http://192.168.60.248:7547/25dbc91d31276f0cb03391160531ecae',
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = val_list
+
+ return msg
+
+ pass
+
+ @classmethod
+ def get_inform(
+ cls,
+ oui: str = '48BF74',
+ sw_version: str = 'BaiBS_RTS_3.1.6',
+ enb_serial: str = '120200002618AGP0003',
+ event_codes: Optional[List[str]] = None,
+ ) -> models.Inform:
+ if event_codes is None:
+ event_codes = []
+ msg = models.Inform()
+
+ # DeviceId
+ device_id = models.DeviceIdStruct()
+ device_id.Manufacturer = 'Unused'
+ device_id.OUI = oui
+ device_id.ProductClass = 'Unused'
+ device_id.SerialNumber = enb_serial
+ msg.DeviceId = device_id
+
+ # Event
+ msg.Event = models.EventList()
+ event_list = []
+ for code in event_codes:
+ event = models.EventStruct()
+ event.EventCode = code
+ event.CommandKey = ''
+ event_list.append(event)
+ msg.Event.EventStruct = event_list
+
+ # ParameterList
+ val_list = []
+ val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.DeviceInfo.HardwareVersion',
+ val_type='string',
+ data='VER.C',
+ ),
+ )
+ val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.DeviceInfo.ManufacturerOUI',
+ val_type='string',
+ data=oui,
+ ),
+ )
+ val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.DeviceInfo.SoftwareVersion',
+ val_type='string',
+ data=sw_version,
+ ),
+ )
+ val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.DeviceInfo.SerialNumber',
+ val_type='string',
+ data=enb_serial,
+ ),
+ )
+ val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.ManagementServer.ConnectionRequestURL',
+ val_type='string',
+ data='http://192.168.60.248:7547/25dbc91d31276f0cb03391160531ecae',
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = val_list
+
+ return msg
+
+ @classmethod
+ def get_qafb_read_only_param_values_response(
+ cls,
+ ) -> models.GetParameterValuesResponse:
+ msg = models.GetParameterValuesResponse()
+ param_val_list = []
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.1.LTE.X_QUALCOMM_FAPControl.OpState',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.1.LTE.X_QUALCOMM_FAPControl.OpState',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.1.LTE.X_QUALCOMM_FAPControl.OpState',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.FAP.GPS.latitude',
+ val_type='int',
+ data='0',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.FAP.GPS.longitude',
+ val_type='int',
+ data='0',
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = param_val_list
+ return msg
+
+ @classmethod
+ def get_read_only_param_values_response(
+ cls,
+ ) -> models.GetParameterValuesResponse:
+ msg = models.GetParameterValuesResponse()
+ param_val_list = []
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.OpState',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.RFTxStatus',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.DeviceInfo.X_BAICELLS_COM_GPS_Status',
+ val_type='boolean',
+ data='0',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.DeviceInfo.X_BAICELLS_COM_1588_Status',
+ val_type='boolean',
+ data='0',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.DeviceInfo.X_BAICELLS_COM_MME_Status',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.FAP.GPS.LockedLatitude',
+ val_type='int',
+ data='0',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.FAP.GPS.LockedLongitude',
+ val_type='int',
+ data='0',
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = param_val_list
+ return msg
+
+ @classmethod
+ def get_cavium_param_values_response(
+ cls,
+ admin_state: bool = False,
+ earfcndl: int = 2405,
+ num_plmns: int = 0,
+ ) -> models.GetParameterValuesResponse:
+ msg = models.GetParameterValuesResponse()
+ param_val_list = []
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.DLBandwidth',
+ val_type='string',
+ data='20',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.FreqBandIndicator',
+ val_type='string',
+ data='5',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.ManagementServer.PeriodicInformInterval',
+ val_type='int',
+ data='5',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.CellRestriction.CellReservedForOperatorUse',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.ULBandwidth',
+ val_type='string',
+ data='n100',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.Common.CellIdentity',
+ val_type='int',
+ data='138777000',
+ ),
+ )
+ # MME IP
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.Gateway.S1SigLinkServerList',
+ val_type='string',
+ data='"192.168.60.142"',
+ ),
+ )
+ # perf mgmt enable
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.PerfMgmt.Config.1.Enable',
+ val_type='boolean',
+ data='true',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.CellRestriction.CellBarred',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.PerfMgmt.Config.1.PeriodicUploadInterval',
+ val_type='int',
+ data='600',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.AdminState',
+ val_type='boolean',
+ data=admin_state,
+ ),
+ )
+ # Perf mgmt upload url
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.PerfMgmt.Config.1.URL',
+ val_type='string',
+ data='http://192.168.60.142:8081/',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.TAC',
+ val_type='int',
+ data='1',
+ ),
+ )
+ # PCI
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.PhyCellID',
+ val_type='int',
+ data='260',
+ ),
+ )
+ # MME port
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.Gateway.S1SigLinkPort',
+ val_type='int',
+ data='36412',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.IPsec.Enable',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.EARFCNDL',
+ val_type='int',
+ data='2405',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.EARFCNUL',
+ val_type='int',
+ data='20405',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.Capabilities.LTE.DuplexMode',
+ val_type='string',
+ data='FDDMode',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.Capabilities.LTE.BandsSupported',
+ val_type='string',
+ data='5',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.ManagementServer.PeriodicInformEnable',
+ val_type='int',
+ data='5',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNListNumberOfEntries',
+ val_type='int',
+ data=str(num_plmns),
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = param_val_list
+ return msg
+
+ @classmethod
+ def get_regular_param_values_response(
+ cls,
+ admin_state: bool = False,
+ earfcndl: int = 39250,
+ exclude_num_plmns: bool = False,
+ ) -> models.GetParameterValuesResponse:
+ msg = models.GetParameterValuesResponse()
+ param_val_list = []
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.DLBandwidth',
+ val_type='string',
+ data='n100',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.FreqBandIndicator',
+ val_type='string',
+ data='40',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.ManagementServer.PeriodicInformInterval',
+ val_type='int',
+ data='5',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.CellRestriction.CellReservedForOperatorUse',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.ULBandwidth',
+ val_type='string',
+ data='20',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.X_BAICELLS_COM_LTE.EARFCNDLInUse',
+ val_type='string',
+ data=earfcndl,
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.PHY.TDDFrame.SpecialSubframePatterns',
+ val_type='int',
+ data='7',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.Common.CellIdentity',
+ val_type='int',
+ data='138777000',
+ ),
+ )
+ # MME IP
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.Gateway.S1SigLinkServerList',
+ val_type='string',
+ data='"192.168.60.142"',
+ ),
+ )
+ if not exclude_num_plmns:
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNListNumberOfEntries',
+ val_type='int',
+ data='1',
+ ),
+ )
+ # perf mgmt enable
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.FAP.PerfMgmt.Config.1.Enable',
+ val_type='boolean',
+ data='true',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.CellRestriction.CellBarred',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.FAP.PerfMgmt.Config.1.PeriodicUploadInterval',
+ val_type='int',
+ data='300',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.AdminState',
+ val_type='boolean',
+ data=admin_state,
+ ),
+ )
+ # Local gateway enable
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.DeviceInfo.X_BAICELLS_COM_LTE_LGW_Switch',
+ val_type='boolean',
+ data='0',
+ ),
+ )
+ # Perf mgmt upload url
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.FAP.PerfMgmt.Config.1.URL',
+ val_type='string',
+ data='http://192.168.60.142:8081/',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.TAC',
+ val_type='int',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.Gateway.X_BAICELLS_COM_MmePool.Enable',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.PHY.TDDFrame.SubFrameAssignment',
+ val_type='int',
+ data='2',
+ ),
+ )
+ # PCI
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.RAN.RF.PhyCellID',
+ val_type='int',
+ data='260',
+ ),
+ )
+ # MME port
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.FAPControl.LTE.Gateway.S1SigLinkPort',
+ val_type='int',
+ data='36412',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.Ipsec.IPSEC_ENABLE',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.X_BAICELLS_COM_LTE.EARFCNULInUse',
+ val_type='int',
+ data='39150',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.Capabilities.LTE.DuplexMode',
+ val_type='string',
+ data='TDDMode',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.Capabilities.LTE.BandsSupported',
+ val_type='string',
+ data='40',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.ManagementServer.PeriodicInformEnable',
+ val_type='int',
+ data='5',
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = param_val_list
+ return msg
+
+ @classmethod
+ def get_qafb_regular_param_values_response(
+ cls,
+ admin_state: bool = False,
+ earfcndl: int = 39250,
+ ) -> models.GetParameterValuesResponse:
+ msg = models.GetParameterValuesResponse()
+ param_val_list = []
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.RAN.RF.DLBandwidth',
+ val_type='string',
+ data='20',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.RAN.RF.FreqBandIndicator',
+ val_type='string',
+ data='40',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.ManagementServer.PeriodicInformInterval',
+ val_type='int',
+ data='5',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.RAN.CellRestriction.CellReservedForOperatorUse',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.RAN.RF.ULBandwidth',
+ val_type='string',
+ data='20',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.RAN.RF.ULBandwidth',
+ val_type='int',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.X_BAICELLS_COM_LTE.EARFCNDLInUse',
+ val_type='string',
+ data=earfcndl,
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.RAN.PHY.TDDFrame.SpecialSubframePatterns',
+ val_type='int',
+ data='7',
+ ),
+ )
+ # MME IP
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.FAPControl.LTE.Gateway.S1SigLinkServerList',
+ val_type='string',
+ data='"192.168.60.142"',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.EPC.PLMNListNumberOfEntries',
+ val_type='int',
+ data='1',
+ ),
+ )
+ # perf mgmt enable
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.FAP.PerfMgmt.Config.1.Enable',
+ val_type='boolean',
+ data='true',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.RAN.CellRestriction.CellBarred',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.FAP.PerfMgmt.Config.1.PeriodicUploadInterval',
+ val_type='int',
+ data='300',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.FAPControl.LTE.AdminState',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ # Local gateway enable
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.DeviceInfo.X_BAICELLS_COM_LTE_LGW_Switch',
+ val_type='boolean',
+ data='0',
+ ),
+ )
+ # Perf mgmt upload url
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.FAP.PerfMgmt.Config.1.URL',
+ val_type='string',
+ data='http://192.168.60.142:8081/',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.EPC.TAC',
+ val_type='int',
+ data='1',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.FAPControl.LTE.Gateway.X_BAICELLS_COM_MmePool.Enable',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.RAN.PHY.TDDFrame.SubFrameAssignment',
+ val_type='int',
+ data='2',
+ ),
+ )
+ # PCI
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.RAN.RF.PhyCellID',
+ val_type='int',
+ data='260',
+ ),
+ )
+ # MME port
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.FAPControl.LTE.Gateway.S1SigLinkPort',
+ val_type='int',
+ data='36412',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='boardconf.ipsec.ipsecConfig.onBoot',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.X_BAICELLS_COM_LTE.EARFCNULInUse',
+ val_type='int',
+ data='9212',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='boardconf.status.eepromInfo.div_multiple',
+ val_type='string',
+ data='02',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='boardconf.status.eepromInfo.work_mode',
+ val_type='string',
+ data='1C000400',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.ManagementServer.PeriodicInformEnable',
+ val_type='int',
+ data='5',
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = param_val_list
+ return msg
+
+ @classmethod
+ def get_cavium_object_param_values_response(
+ cls,
+ num_plmns: int,
+ ) -> models.GetParameterValuesResponse:
+ msg = models.GetParameterValuesResponse()
+ param_val_list = []
+ for i in range(1, num_plmns + 1):
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.%d.IsPrimary' % i,
+ val_type='boolean',
+ data='true',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.%d.CellReservedForOperatorUse' % i,
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.%d.PLMNID' % i,
+ val_type='string',
+ data='00101',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.%d.Enable' % i,
+ val_type='boolean',
+ data='true',
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = param_val_list
+ return msg
+
+ @classmethod
+ def get_object_param_values_response(
+ cls,
+ ) -> models.GetParameterValuesResponse:
+ msg = models.GetParameterValuesResponse()
+ param_val_list = []
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.IsPrimary',
+ val_type='boolean',
+ data='true',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.CellReservedForOperatorUse',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.PLMNID',
+ val_type='string',
+ data='00101',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='Device.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.Enable',
+ val_type='boolean',
+ data='true',
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = param_val_list
+ return msg
+
+ @classmethod
+ def get_qafb_object_param_values_response(
+ cls,
+ ) -> models.GetParameterValuesResponse:
+ msg = models.GetParameterValuesResponse()
+ param_val_list = []
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.IsPrimary',
+ val_type='boolean',
+ data='true',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.CellReservedForOperatorUse',
+ val_type='boolean',
+ data='false',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.PLMNID',
+ val_type='string',
+ data='00101',
+ ),
+ )
+ param_val_list.append(
+ cls.get_parameter_value_struct(
+ name='InternetGatewayDevice.Services.FAPService.1.CellConfig.LTE.EPC.PLMNList.1.Enable',
+ val_type='boolean',
+ data='true',
+ ),
+ )
+ msg.ParameterList = models.ParameterValueList()
+ msg.ParameterList.ParameterValueStruct = param_val_list
+ return msg
+
+ @classmethod
+ def get_reboot_response(cls) -> models.RebootResponse:
+ return models.RebootResponse()
diff --git a/tests/timer_tests.py b/tests/timer_tests.py
new file mode 100644
index 0000000..ad00232
--- /dev/null
+++ b/tests/timer_tests.py
@@ -0,0 +1,26 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+from unittest import TestCase
+
+from state_machines.timer import StateMachineTimer
+
+
+class StateMachineTimerTests(TestCase):
+ def test_is_done(self):
+ timer_a = StateMachineTimer(0)
+ self.assertTrue(timer_a.is_done(), 'Timer should be done')
+
+ timer_b = StateMachineTimer(600)
+ self.assertFalse(timer_b.is_done(), 'Timer should not be done')
diff --git a/tests/tr069_tests.py b/tests/tr069_tests.py
new file mode 100644
index 0000000..649fcb6
--- /dev/null
+++ b/tests/tr069_tests.py
@@ -0,0 +1,886 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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 datetime import datetime, timedelta, timezone
+from unittest import TestCase, mock
+from unittest.mock import Mock, patch
+
+import lxml.etree as ET
+from tests.test_utils.enb_acs_builder import (
+ EnodebAcsStateMachineBuilder,
+)
+from tr069 import models
+from tr069.rpc_methods import AutoConfigServer
+from tr069.spyne_mods import Tr069Application, Tr069Soap11
+from spyne import MethodContext
+from spyne.server import ServerBase
+
+
+class Tr069Test(TestCase):
+ """ Tests for the TR-069 server """
+ acs_to_cpe_queue = None
+ cpe_to_acs_queue = None
+
+ def setUp(self):
+ # Set up the ACS
+ self.enb_acs_manager = EnodebAcsStateMachineBuilder.build_acs_manager()
+ self.handler = EnodebAcsStateMachineBuilder.build_acs_state_machine()
+ AutoConfigServer.set_state_machine_manager(self.enb_acs_manager)
+
+ def side_effect(*args, **_kwargs):
+ msg = args[1]
+ return msg
+
+ self.p = patch.object(
+ AutoConfigServer, '_handle_tr069_message',
+ Mock(side_effect=side_effect),
+ )
+ self.p.start()
+
+ self.app = Tr069Application(
+ [AutoConfigServer],
+ models.CWMP_NS,
+ in_protocol=Tr069Soap11(validator='soft'),
+ out_protocol=Tr069Soap11(),
+ )
+
+ def tearDown(self):
+ self.p.stop()
+ self.handler = None
+
+ def _get_mconfig(self):
+ return {
+ "@type": "type.googleapis.com/magma.mconfig.EnodebD",
+ "bandwidthMhz": 20,
+ "specialSubframePattern": 7,
+ "earfcndl": 44490,
+ "logLevel": "INFO",
+ "plmnidList": "00101",
+ "pci": 260,
+ "allowEnodebTransmit": False,
+ "subframeAssignment": 2,
+ "tac": 1,
+ },
+
+ def _get_service_config(self):
+ return {
+ "tr069": {
+ "interface": "eth1",
+ "port": 48080,
+ "perf_mgmt_port": 8081,
+ "public_ip": "192.88.99.142",
+ },
+ "reboot_enodeb_on_mme_disconnected": True,
+ "s1_interface": "eth1",
+ }
+
+ def test_acs_manager_exception(self):
+ """
+ Test that an unexpected exception from the ACS SM manager will result
+ in an empty response.
+ """
+ self.enb_acs_manager.handle_tr069_message = mock.MagicMock(
+ side_effect=Exception('mock exception'),
+ )
+ # stop the patcher because we want to use the above MagicMock
+ self.p.stop()
+ server = ServerBase(self.app)
+
+ ctx = MethodContext(server, MethodContext.SERVER)
+ ctx.in_string = [b'']
+ ctx, = server.generate_contexts(ctx)
+
+ server.get_in_object(ctx)
+ self.assertIsNone(ctx.in_error)
+
+ server.get_out_object(ctx)
+ self.assertIsNone(ctx.out_error)
+
+ server.get_out_string(ctx)
+ self.assertEqual(b''.join(ctx.out_string), b'')
+
+ # start the patcher otherwise the p.stop() in tearDown will complain
+ self.p.start()
+
+ def test_parse_inform(self):
+ """
+ Test that example Inform RPC call can be parsed correctly
+ """
+ # Example TR-069 CPE->ACS RPC call. Copied from:
+ # http://djuro82.blogspot.com/2011/05/tr-069-cpe-provisioning.html
+ cpe_string = b'''
+ <soapenv:Envelope soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
+ <soapenv:Header>
+ <cwmp:ID soapenv:mustUnderstand="1">0_THOM_TR69_ID</cwmp:ID>
+ </soapenv:Header>
+ <soapenv:Body>
+ <cwmp:Inform>
+ <DeviceId>
+ <Manufacturer>THOMSON</Manufacturer>
+ <OUI>00147F</OUI>
+ <ProductClass>SpeedTouch 780</ProductClass>
+ <SerialNumber>CP0611JTLNW</SerialNumber>
+ </DeviceId>
+ <Event soap:arrayType="cwmp:EventStruct[04]">
+ <EventStruct>
+ <EventCode>0 BOOTSTRAP</EventCode>
+ <CommandKey></CommandKey>
+ </EventStruct>
+ <EventStruct>
+ <EventCode>1 BOOT</EventCode>
+ <CommandKey></CommandKey>
+ </EventStruct>
+ <EventStruct>
+ <EventCode>2 PERIODIC</EventCode>
+ <CommandKey></CommandKey>
+ </EventStruct>
+ <EventStruct>
+ <EventCode>4 VALUE CHANGE</EventCode>
+ <CommandKey></CommandKey>
+ </EventStruct>
+ </Event>
+ <MaxEnvelopes>2</MaxEnvelopes>
+ <CurrentTime>1970-01-01T00:01:09Z</CurrentTime>
+ <RetryCount>05</RetryCount>
+ <ParameterList soap:arrayType="cwmp:ParameterValueStruct[12]">
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.DeviceSummary</Name>
+ <Value xsi:type="xsd:string">
+ InternetGatewayDevice:1.1[] (Baseline:1, EthernetLAN:1, ADSLWAN:1, Bridging:1, Time:1, WiFiLAN:1)</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.DeviceInfo.SpecVersion</Name>
+ <Value xsi:type="xsd:string">1.1</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.DeviceInfo.HardwareVersion</Name>
+ <Value xsi:type="xsd:string">BANT-R</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.DeviceInfo.SoftwareVersion</Name>
+ <Value xsi:type="xsd:string">6.2.35.0</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.DeviceInfo.ProvisioningCode</Name>
+ <Value xsi:type="xsd:string"></Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.DeviceInfo.VendorConfigFile.1.Name</Name>
+ <Value xsi:type="xsd:string">MyCompanyName</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.DeviceInfo.VendorConfigFile.1.Version</Name>
+ <Value xsi:type="xsd:string"></Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.DeviceInfo.VendorConfigFile.1.Date</Name>
+ <Value xsi:type="xsd:dateTime">0001-01-01T00:00:00</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.DeviceInfo.VendorConfigFile.1.Description</Name>
+ <Value xsi:type="xsd:string">MyCompanyName</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.ManagementServer.ConnectionRequestURL</Name>
+ <Value xsi:type="xsd:string">http://10.127.129.205:51005/</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.ManagementServer.ParameterKey</Name>
+ <Value xsi:type="xsd:string"></Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.WANDevice.1.WANConnectionDevice.1.WANIPConnection.1.ExternalIPAddress</Name>
+ <Value xsi:type="xsd:string">10.127.129.205</Value>
+ </ParameterValueStruct>
+ </ParameterList>
+ </cwmp:Inform>
+ </soapenv:Body>
+ </soapenv:Envelope>
+ '''
+
+ server = ServerBase(self.app)
+
+ ctx = MethodContext(server, MethodContext.SERVER)
+ ctx.in_string = [cpe_string]
+ ctx, = server.generate_contexts(ctx)
+
+ if ctx.in_error is not None:
+ print('In error: %s' % ctx.in_error)
+ self.assertEqual(ctx.in_error, None)
+
+ server.get_in_object(ctx)
+
+ self.assertEqual(ctx.in_object.DeviceId.OUI, '00147F')
+ self.assertEqual(
+ ctx.in_object.Event.EventStruct[0].EventCode, '0 BOOTSTRAP',
+ )
+ self.assertEqual(
+ ctx.in_object.Event.EventStruct[2].EventCode, '2 PERIODIC',
+ )
+ self.assertEqual(ctx.in_object.MaxEnvelopes, 2)
+ self.assertEqual(
+ ctx.in_object.ParameterList.ParameterValueStruct[1].Name,
+ 'InternetGatewayDevice.DeviceInfo.SpecVersion',
+ )
+ self.assertEqual(
+ str(ctx.in_object.ParameterList.ParameterValueStruct[1].Value), '1.1',
+ )
+
+ def test_parse_inform_cavium(self):
+ """
+ Test that example Inform RPC call can be parsed correctly from OC-LTE
+ """
+ cpe_string = b'''<?xml version="1.0" encoding="UTF-8"?>
+ <SOAP-ENV:Envelope xmlns:SOAP-ENV="http://schemas.xmlsoap.org/soap/envelope/" xmlns:SOAP-ENC="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
+ <SOAP-ENV:Header>
+ <cwmp:ID SOAP-ENV:mustUnderstand="1">CPE_1002</cwmp:ID>
+ </SOAP-ENV:Header>
+ <SOAP-ENV:Body SOAP-ENV:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">
+ <cwmp:Inform>
+ <DeviceId>
+ <Manufacturer>Cavium, Inc.</Manufacturer>
+ <OUI>000FB7</OUI>
+ <ProductClass>Cavium eNB</ProductClass>
+ <SerialNumber>10.18.104.79</SerialNumber>
+ </DeviceId>
+ <Event xsi:type="SOAP-ENC:Array" SOAP-ENC:arrayType="cwmp:EventStruct[1]">
+ <EventStruct>
+ <EventCode>0 BOOTSTRAP</EventCode>
+ <CommandKey></CommandKey>
+ </EventStruct>
+ </Event>
+ <MaxEnvelopes>1</MaxEnvelopes>
+ <CurrentTime>1970-01-02T00:01:05.021239+00:00</CurrentTime>
+ <RetryCount>2</RetryCount>
+ <ParameterList xsi:type="SOAP-ENC:Array" SOAP-ENC:arrayType="cwmp:ParameterValueStruct[15]">
+ <ParameterValueStruct>
+ <Name>Device.DeviceInfo.HardwareVersion</Name>
+ <Value xsi:type="xsd:string">1.0</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.DeviceInfo.SoftwareVersion</Name>
+ <Value xsi:type="xsd:string">1.0</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.DeviceInfo.AdditionalHardwareVersion</Name>
+ <Value xsi:type="xsd:string">1.0</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.DeviceInfo.AdditionalSoftwareVersion</Name>
+ <Value xsi:type="xsd:string">1.0</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.DeviceInfo.ProvisioningCode</Name>
+ <Value xsi:type="xsd:string">Cavium</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.ManagementServer.ParameterKey</Name>
+ <Value xsi:type="xsd:string"></Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.ManagementServer.ConnectionRequestURL</Name>
+ <Value xsi:type="xsd:string">http://192.88.99.253:8084/bucrhzjd</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.ManagementServer.UDPConnectionRequestAddress</Name>
+ <Value xsi:type="xsd:string"></Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.ManagementServer.NATDetected</Name>
+ <Value xsi:type="xsd:boolean">0</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.IP.Diagnostics.UDPEchoConfig.PacketsReceived</Name>
+ <Value xsi:type="xsd:unsignedInt">0</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.IP.Diagnostics.UDPEchoConfig.PacketsResponded</Name>
+ <Value xsi:type="xsd:unsignedInt">0</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.IP.Diagnostics.UDPEchoConfig.BytesReceived</Name>
+ <Value xsi:type="xsd:unsignedInt">0</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.IP.Diagnostics.UDPEchoConfig.BytesResponded</Name>
+ <Value xsi:type="xsd:unsignedInt">0</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.IP.Diagnostics.UDPEchoConfig.TimeFirstPacketReceived</Name>
+ <Value xsi:type="xsd:dateTime">1969-12-31T16:00:00.000000+00:00</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>Device.IP.Diagnostics.UDPEchoConfig.TimeLastPacketReceived</Name>
+ <Value xsi:type="xsd:dateTime">1969-12-31T16:00:00.000000+00:00</Value>
+ </ParameterValueStruct>
+ </ParameterList>
+ </cwmp:Inform>
+ </SOAP-ENV:Body>
+ </SOAP-ENV:Envelope>
+ '''
+
+ server = ServerBase(self.app)
+
+ ctx = MethodContext(server, MethodContext.SERVER)
+ ctx.in_string = [cpe_string]
+ ctx, = server.generate_contexts(ctx)
+
+ if ctx.in_error is not None:
+ print('In error: %s' % ctx.in_error)
+ self.assertEqual(ctx.in_error, None)
+
+ server.get_in_object(ctx)
+
+ self.assertEqual(ctx.in_object.DeviceId.OUI, '000FB7')
+ self.assertEqual(
+ ctx.in_object.Event.EventStruct[0].EventCode, '0 BOOTSTRAP',
+ )
+ self.assertEqual(ctx.in_object.MaxEnvelopes, 1)
+ self.assertEqual(
+ ctx.in_object.ParameterList.ParameterValueStruct[1].Name,
+ 'Device.DeviceInfo.SoftwareVersion',
+ )
+ self.assertEqual(
+ str(ctx.in_object.ParameterList.ParameterValueStruct[1].Value), '1.0',
+ )
+
+ def test_handle_transfer_complete(self):
+ """
+ Test that example TransferComplete RPC call can be parsed correctly, and
+ response is correctly generated.
+ """
+ # Example TransferComplete CPE->ACS RPC request/response.
+ # Manually created.
+ cpe_string = b'''
+ <soapenv:Envelope soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
+ <soapenv:Header>
+ <cwmp:ID soapenv:mustUnderstand="1">1234</cwmp:ID>
+ </soapenv:Header>
+ <soapenv:Body>
+ <cwmp:TransferComplete>
+ <CommandKey>Downloading stuff</CommandKey>
+ <FaultStruct>
+ <FaultCode>0</FaultCode>
+ <FaultString></FaultString>
+ </FaultStruct>
+ <StartTime>2016-11-30T10:16:29Z</StartTime>
+ <CompleteTime>2016-11-30T10:17:05Z</CompleteTime>
+ </cwmp:TransferComplete>
+ </soapenv:Body>
+ </soapenv:Envelope>
+ '''
+ expected_acs_string = b'''
+ <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
+ <soapenv:Header>
+ <cwmp:ID soapenv:mustUnderstand="1">1234</cwmp:ID>
+ </soapenv:Header>
+ <soapenv:Body>
+ <cwmp:TransferCompleteResponse>
+ </cwmp:TransferCompleteResponse>
+ </soapenv:Body>
+ </soapenv:Envelope>
+ '''
+
+ self.p.stop()
+ self.p.start()
+
+ server = ServerBase(self.app)
+
+ ctx = MethodContext(server, MethodContext.SERVER)
+ ctx.in_string = [cpe_string]
+ ctx, = server.generate_contexts(ctx)
+
+ if ctx.in_error is not None:
+ print('In error: %s' % ctx.in_error)
+ self.assertEqual(ctx.in_error, None)
+
+ server.get_in_object(ctx)
+ self.assertEqual(ctx.in_error, None)
+
+ server.get_out_object(ctx)
+ self.assertEqual(ctx.out_error, None)
+
+ output_msg = ctx.out_object[0]
+ self.assertEqual(type(output_msg), models.TransferComplete)
+ self.assertEqual(output_msg.CommandKey, 'Downloading stuff')
+ self.assertEqual(output_msg.FaultStruct.FaultCode, 0)
+ self.assertEqual(output_msg.FaultStruct.FaultString, '')
+ self.assertEqual(
+ output_msg.StartTime,
+ datetime(
+ 2016, 11, 30, 10, 16, 29,
+ tzinfo=timezone(timedelta(0)),
+ ),
+ )
+ self.assertEqual(
+ output_msg.CompleteTime,
+ datetime(
+ 2016, 11, 30, 10, 17, 5,
+ tzinfo=timezone(timedelta(0)),
+ ),
+ )
+
+ server.get_out_string(ctx)
+ self.assertEqual(ctx.out_error, None)
+
+ xml_tree = XmlTree()
+ match = xml_tree.xml_compare(
+ xml_tree.convert_string_to_tree(b''.join(ctx.out_string)),
+ xml_tree.convert_string_to_tree(expected_acs_string),
+ )
+ self.assertTrue(match)
+
+ def test_parse_empty_http(self):
+ """
+ Test that empty HTTP message gets correctly mapped to 'EmptyHttp'
+ function call
+ """
+ cpe_string = b''
+
+ server = ServerBase(self.app)
+
+ ctx = MethodContext(server, MethodContext.SERVER)
+ ctx.in_string = [cpe_string]
+ ctx, = server.generate_contexts(ctx)
+
+ if ctx.in_error is not None:
+ print('In error: %s' % ctx.in_error)
+
+ self.assertEqual(ctx.in_error, None)
+ self.assertEqual(ctx.function, AutoConfigServer.empty_http)
+
+ def test_generate_empty_http(self):
+ """
+ Test that empty HTTP message is generated when setting output message
+ name to 'EmptyHttp'
+ """
+ cpe_string = b''
+
+ server = ServerBase(self.app)
+
+ ctx = MethodContext(server, MethodContext.SERVER)
+ ctx.in_string = [cpe_string]
+ ctx, = server.generate_contexts(ctx)
+
+ server.get_in_object(ctx)
+ if ctx.in_error is not None:
+ raise ctx.in_error
+
+ server.get_out_object(ctx)
+ if ctx.out_error is not None:
+ raise ctx.out_error
+
+ ctx.descriptor.out_message.Attributes.sub_name = 'EmptyHttp'
+ ctx.out_object = [models.AcsToCpeRequests()]
+
+ server.get_out_string(ctx)
+
+ self.assertEqual(b''.join(ctx.out_string), b'')
+
+ def test_generate_get_parameter_values_string(self):
+ """
+ Test that correct string is generated for SetParameterValues ACS->CPE
+ request
+ """
+ # Example ACS->CPE RPC call. Copied from:
+ # http://djuro82.blogspot.com/2011/05/tr-069-cpe-provisioning.html
+ # Following edits made:
+ # - Change header ID value from 'null0' to 'null', to match magma
+ # default ID
+ expected_acs_string = b'''
+ <soapenv:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:cwmp="urn:dslforum-org:cwmp-1-0" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+ <soapenv:Header>
+ <cwmp:ID soapenv:mustUnderstand="1">null</cwmp:ID>
+ </soapenv:Header>
+ <soapenv:Body>
+ <cwmp:GetParameterValues>
+ <ParameterNames soap:arrayType="xsd:string[1]">
+ <string>foo</string>
+ </ParameterNames>
+ </cwmp:GetParameterValues>
+ </soapenv:Body>
+ </soapenv:Envelope>
+ '''
+
+ names = ['foo']
+ request = models.GetParameterValues()
+ request.ParameterNames = models.ParameterNames()
+ request.ParameterNames.arrayType = 'xsd:string[%d]' \
+ % len(names)
+ request.ParameterNames.string = []
+ for name in names:
+ request.ParameterNames.string.append(name)
+
+ request.ParameterKey = 'null'
+
+ def side_effect(*args, **_kwargs):
+ ctx = args[0]
+ ctx.out_header = models.ID(mustUnderstand='1')
+ ctx.out_header.Data = 'null'
+ ctx.descriptor.out_message.Attributes.sub_name = \
+ request.__class__.__name__
+ return AutoConfigServer._generate_acs_to_cpe_request_copy(request)
+
+ self.p.stop()
+ self.p = patch.object(
+ AutoConfigServer, '_handle_tr069_message',
+ side_effect=side_effect,
+ )
+ self.p.start()
+
+ server = ServerBase(self.app)
+
+ ctx = MethodContext(server, MethodContext.SERVER)
+ ctx.in_string = [b'']
+ ctx, = server.generate_contexts(ctx)
+
+ server.get_in_object(ctx)
+ if ctx.in_error is not None:
+ raise ctx.in_error
+
+ server.get_out_object(ctx)
+ if ctx.out_error is not None:
+ raise ctx.out_error
+
+ server.get_out_string(ctx)
+
+ xml_tree = XmlTree()
+ match = xml_tree.xml_compare(
+ xml_tree.convert_string_to_tree(b''.join(ctx.out_string)),
+ xml_tree.convert_string_to_tree(expected_acs_string),
+ )
+ self.assertTrue(match)
+
+ def test_generate_set_parameter_values_string(self):
+ """
+ Test that correct string is generated for SetParameterValues ACS->CPE
+ request
+ """
+ # Example ACS->CPE RPC call. Copied from:
+ # http://djuro82.blogspot.com/2011/05/tr-069-cpe-provisioning.html
+ # Following edits made:
+ # - Change header ID value from 'null0' to 'null', to match magma
+ # default ID
+ expected_acs_string = b'''
+ <soapenv:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:cwmp="urn:dslforum-org:cwmp-1-0" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+ <soapenv:Header>
+ <cwmp:ID soapenv:mustUnderstand="1">null</cwmp:ID>
+ </soapenv:Header>
+ <soapenv:Body>
+ <cwmp:SetParameterValues>
+ <ParameterList soap:arrayType="cwmp:ParameterValueStruct[4]">
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.ManagementServer.PeriodicInformEnable</Name>
+ <Value xsi:type="xsd:boolean">1</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.ManagementServer.ConnectionRequestUsername</Name>
+ <Value xsi:type="xsd:string">00147F-SpeedTouch780-CP0611JTLNW</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.ManagementServer.ConnectionRequestPassword</Name>
+ <Value xsi:type="xsd:string">98ff55fb377bf724c625f60dec448646</Value>
+ </ParameterValueStruct>
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.ManagementServer.PeriodicInformInterval</Name>
+ <Value xsi:type="xsd:unsignedInt">60</Value>
+ </ParameterValueStruct>
+ </ParameterList>
+ <ParameterKey xsi:type="xsd:string">SetParameter1</ParameterKey>
+ </cwmp:SetParameterValues>
+ </soapenv:Body>
+ </soapenv:Envelope>
+ '''
+
+ request = models.SetParameterValues()
+
+ request.ParameterList = \
+ models.ParameterValueList(arrayType='cwmp:ParameterValueStruct[4]')
+ request.ParameterList.ParameterValueStruct = []
+
+ param = models.ParameterValueStruct()
+ param.Name = 'InternetGatewayDevice.ManagementServer.PeriodicInformEnable'
+ param.Value = models.anySimpleType(type='xsd:boolean')
+ param.Value.Data = '1'
+ request.ParameterList.ParameterValueStruct.append(param)
+
+ param = models.ParameterValueStruct()
+ param.Name = 'InternetGatewayDevice.ManagementServer.ConnectionRequestUsername'
+ param.Value = models.anySimpleType(type='xsd:string')
+ param.Value.Data = '00147F-SpeedTouch780-CP0611JTLNW'
+ request.ParameterList.ParameterValueStruct.append(param)
+
+ param = models.ParameterValueStruct()
+ param.Name = 'InternetGatewayDevice.ManagementServer.ConnectionRequestPassword'
+ param.Value = models.anySimpleType(type='xsd:string')
+ param.Value.Data = '98ff55fb377bf724c625f60dec448646'
+ request.ParameterList.ParameterValueStruct.append(param)
+
+ param = models.ParameterValueStruct()
+ param.Name = 'InternetGatewayDevice.ManagementServer.PeriodicInformInterval'
+ param.Value = models.anySimpleType(type='xsd:unsignedInt')
+ param.Value.Data = '60'
+ request.ParameterList.ParameterValueStruct.append(param)
+
+ request.ParameterKey = models.ParameterKeyType()
+ request.ParameterKey.type = 'xsd:string'
+ request.ParameterKey.Data = 'SetParameter1'
+
+ def side_effect(*args, **_kwargs):
+ ctx = args[0]
+ ctx.out_header = models.ID(mustUnderstand='1')
+ ctx.out_header.Data = 'null'
+ ctx.descriptor.out_message.Attributes.sub_name = request.__class__.__name__
+ return request
+
+ self.p.stop()
+ self.p = patch.object(
+ AutoConfigServer, '_handle_tr069_message',
+ Mock(side_effect=side_effect),
+ )
+ self.p.start()
+
+ server = ServerBase(self.app)
+
+ ctx = MethodContext(server, MethodContext.SERVER)
+ ctx.in_string = [b'']
+ ctx, = server.generate_contexts(ctx)
+
+ server.get_in_object(ctx)
+ if ctx.in_error is not None:
+ raise ctx.in_error
+
+ server.get_out_object(ctx)
+ if ctx.out_error is not None:
+ raise ctx.out_error
+
+ server.get_out_string(ctx)
+
+ xml_tree = XmlTree()
+ NS_SOAP11_ENC = 'soap11enc'
+ NS_SOAP11_ENV = 'soap11env'
+ xml_str = b''.join(ctx.out_string)
+ # Get the namespaces and validate the soap enc and env prefix are right
+ nsmap = xml_tree.get_ns(xml_str)
+ self.assertTrue(NS_SOAP11_ENC in nsmap.keys())
+ self.assertTrue(NS_SOAP11_ENV in nsmap.keys())
+
+ match = xml_tree.xml_compare(
+ xml_tree.convert_string_to_tree(xml_str),
+ xml_tree.convert_string_to_tree(expected_acs_string),
+ )
+ self.assertTrue(match)
+
+ def test_parse_fault_response(self):
+ """ Tests that a fault response from CPE is correctly parsed. """
+ # Example CPE->ACS fault response. Copied from:
+ # http://djuro82.blogspot.com/2011/05/tr-069-cpe-provisioning.html
+ cpe_string = b'''
+ <soapenv:Envelope soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
+ <soapenv:Header>
+ <cwmp:ID soapenv:mustUnderstand="1">1031422463</cwmp:ID>
+ </soapenv:Header>
+ <soapenv:Body>
+ <soapenv:Fault>
+ <faultcode>Client</faultcode>
+ <faultstring>CWMP fault</faultstring>
+ <detail>
+ <cwmp:Fault>
+ <FaultCode>9003</FaultCode>
+ <FaultString>Invalid arguments</FaultString>
+ <SetParameterValuesFault>
+ <ParameterName>InternetGatewayDevice.WANDevice.1.WANConnectionDevice.3.WANPPPConnection.1.Password</ParameterName>
+ <FaultCode>9003</FaultCode>
+ <FaultString>Invalid arguments</FaultString>
+ </SetParameterValuesFault>
+ <SetParameterValuesFault>
+ <ParameterName>InternetGatewayDevice.WANDevice.1.WANConnectionDevice.3.WANPPPConnection.1.Username</ParameterName>
+ <FaultCode>9003</FaultCode>
+ <FaultString>Invalid arguments</FaultString>
+ </SetParameterValuesFault>
+ </cwmp:Fault>
+ </detail>
+ </soapenv:Fault>
+ </soapenv:Body>
+ </soapenv:Envelope>
+ '''
+ server = ServerBase(self.app)
+
+ ctx = MethodContext(server, MethodContext.SERVER)
+ ctx.in_string = [cpe_string]
+ ctx, = server.generate_contexts(ctx)
+ server.get_in_object(ctx)
+ if ctx.in_error is not None:
+ raise ctx.in_error
+
+ # Calls function to receive and process message
+ server.get_out_object(ctx)
+
+ output_msg = ctx.out_object[0]
+ self.assertEqual(type(output_msg), models.Fault)
+ self.assertEqual(output_msg.FaultCode, 9003)
+ self.assertEqual(output_msg.FaultString, 'Invalid arguments')
+ self.assertEqual(
+ output_msg.SetParameterValuesFault[1].ParameterName,
+ 'InternetGatewayDevice.WANDevice.1.WANConnectionDevice.3.WANPPPConnection.1.Username',
+ )
+ self.assertEqual(output_msg.SetParameterValuesFault[1].FaultCode, 9003)
+ self.assertEqual(
+ output_msg.SetParameterValuesFault[1].FaultString,
+ 'Invalid arguments',
+ )
+
+ def test_parse_hex_values(self):
+ """
+ Test that non-utf-8 hex values can be parsed without error
+ """
+ # Example TR-069 CPE->ACS RPC call. Copied from:
+ # http://djuro82.blogspot.com/2011/05/tr-069-cpe-provisioning.html
+ cpe_string = b'''
+ <soapenv:Envelope soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
+ <soapenv:Header>
+ <cwmp:ID soapenv:mustUnderstand="1">0_THOM_TR69_ID</cwmp:ID>
+ </soapenv:Header>
+ <soapenv:Body>
+ <cwmp:Inform>
+ <DeviceId>
+ <Manufacturer>THOMSON</Manufacturer>
+ <OUI>00147F</OUI>
+ <ProductClass>SpeedTouch 780</ProductClass>
+ <SerialNumber>CP0611JTLNW</SerialNumber>
+ </DeviceId>
+ <Event soap:arrayType="cwmp:EventStruct[04]">
+ <EventStruct>
+ <EventCode>0 BOOTSTRAP</EventCode>
+ <CommandKey></CommandKey>
+ </EventStruct>
+ <EventStruct>
+ <EventCode>1 BOOT</EventCode>
+ <CommandKey></CommandKey>
+ </EventStruct>
+ <EventStruct>
+ <EventCode>2 PERIODIC</EventCode>
+ <CommandKey></CommandKey>
+ </EventStruct>
+ <EventStruct>
+ <EventCode>4 VALUE CHANGE</EventCode>
+ <CommandKey></CommandKey>
+ </EventStruct>
+ </Event>
+ <MaxEnvelopes>2</MaxEnvelopes>
+ <CurrentTime>1970-01-01T00:01:09Z</CurrentTime>
+ <RetryCount>05</RetryCount>
+ <ParameterList soap:arrayType="cwmp:ParameterValueStruct[12]">
+ <ParameterValueStruct>
+ <Name>InternetGatewayDevice.DeviceSummary</Name>
+ <Value xsi:type="xsd:string">
+ \xff\xff\xff\xff\xff</Value>
+ </ParameterValueStruct>
+ </ParameterList>
+ </cwmp:Inform>
+ </soapenv:Body>
+ </soapenv:Envelope>
+ '''
+
+ server = ServerBase(self.app)
+
+ ctx = MethodContext(server, MethodContext.SERVER)
+ ctx.in_string = [cpe_string]
+ ctx, = server.generate_contexts(ctx)
+
+ if ctx.in_error is not None:
+ print('In error: %s' % ctx.in_error)
+ self.assertEqual(ctx.in_error, None)
+
+ server.get_in_object(ctx)
+
+
+class XmlTree():
+
+ @staticmethod
+ def convert_string_to_tree(xmlString):
+
+ return ET.fromstring(xmlString)
+
+ @staticmethod
+ def get_ns(xmlString):
+ return ET.fromstring(xmlString).nsmap
+
+ def xml_compare(self, x1, x2, excludes=None):
+ """
+ Compares two xml etrees
+ :param x1: the first tree
+ :param x2: the second tree
+ :param excludes: list of string of attributes to exclude from comparison
+ :return:
+ True if both files match
+ """
+ excludes = [] if excludes is None else excludes
+
+ if x1.tag != x2.tag:
+ print('Tags do not match: %s and %s' % (x1.tag, x2.tag))
+ return False
+ for name, value in x1.attrib.items():
+ if name not in excludes:
+ if x2.attrib.get(name) != value:
+ print(
+ 'Attributes do not match: %s=%r, %s=%r'
+ % (name, value, name, x2.attrib.get(name)),
+ )
+ return False
+ for name in x2.attrib.keys():
+ if name not in excludes:
+ if name not in x1.attrib:
+ print(
+ 'x2 has an attribute x1 is missing: %s'
+ % name,
+ )
+ return False
+ if not self.text_compare(x1.text, x2.text):
+ print('text: %r != %r' % (x1.text, x2.text))
+ return False
+ if not self.text_compare(x1.tail, x2.tail):
+ print('tail: %r != %r' % (x1.tail, x2.tail))
+ return False
+ cl1 = x1.getchildren()
+ cl2 = x2.getchildren()
+ if len(cl1) != len(cl2):
+ print(
+ 'children length differs, %i != %i'
+ % (len(cl1), len(cl2)),
+ )
+ return False
+ i = 0
+ for c1, c2 in zip(cl1, cl2):
+ i += 1
+ if c1.tag not in excludes:
+ if not self.xml_compare(c1, c2, excludes):
+ print(
+ 'children %i do not match: %s'
+ % (i, c1.tag),
+ )
+ return False
+ return True
+
+ def text_compare(self, t1, t2):
+ """
+ Compare two text strings
+ :param t1: text one
+ :param t2: text two
+ :return:
+ True if a match
+ """
+ if not t1 and not t2:
+ return True
+ if t1 == '*' or t2 == '*':
+ return True
+ return (t1 or '').strip() == (t2 or '').strip()
diff --git a/tests/transform_for_enb_tests.py b/tests/transform_for_enb_tests.py
new file mode 100644
index 0000000..bcc19fa
--- /dev/null
+++ b/tests/transform_for_enb_tests.py
@@ -0,0 +1,35 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+from unittest import TestCase
+
+from data_models.transform_for_enb import bandwidth
+
+
+class TransformForMagmaTests(TestCase):
+ def test_bandwidth(self) -> None:
+ inp = 1.4
+ out = bandwidth(inp)
+ expected = 'n6'
+ self.assertEqual(out, expected, 'Should work with a float')
+
+ inp = 20
+ out = bandwidth(inp)
+ expected = 'n100'
+ self.assertEqual(out, expected, 'Should work with an int')
+
+ inp = 10
+ out = bandwidth(inp)
+ expected = 'n50'
+ self.assertEqual(out, expected, 'Should work with int 10')
diff --git a/tests/transform_for_magma_tests.py b/tests/transform_for_magma_tests.py
new file mode 100644
index 0000000..4b76838
--- /dev/null
+++ b/tests/transform_for_magma_tests.py
@@ -0,0 +1,56 @@
+"""
+Copyright 2020 The Magma Authors.
+
+This source code is licensed under the BSD-style license found in the
+LICENSE file in the root directory of this source tree.
+
+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.
+"""
+
+# pylint: disable=protected-access
+from unittest import TestCase
+
+from data_models.transform_for_magma import bandwidth, gps_tr181
+from exceptions import ConfigurationError
+
+
+class TransformForMagmaTests(TestCase):
+ def test_gps_tr181(self) -> None:
+ # Negative longitude
+ inp = '-122150583'
+ out = gps_tr181(inp)
+ expected = '-122.150583'
+ self.assertEqual(out, expected, 'Should convert negative longitude')
+
+ inp = '122150583'
+ out = gps_tr181(inp)
+ expected = '122.150583'
+ self.assertEqual(out, expected, 'Should convert positive longitude')
+
+ inp = '0'
+ out = gps_tr181(inp)
+ expected = '0.0'
+ self.assertEqual(out, expected, 'Should leave zero as zero')
+
+ def test_bandwidth(self) -> None:
+ inp = 'n6'
+ out = bandwidth(inp)
+ expected = 1.4
+ self.assertEqual(out, expected, 'Should convert RBs')
+
+ inp = 1.4
+ out = bandwidth(inp)
+ expected = 1.4
+ self.assertEqual(out, expected, 'Should accept MHz')
+
+ with self.assertRaises(ConfigurationError):
+ inp = 'asdf'
+ bandwidth(inp)
+
+ with self.assertRaises(ConfigurationError):
+ inp = 1234
+ bandwidth(inp)