text: # SPDX-FileCopyrightText: 2020 The Magma Authors.
  # SPDX-FileCopyrightText: 2022 Open Networking Foundation <support@opennetworking.org>
  #
  # SPDX-License-Identifier: BSD-3-Clause
  
  # 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())
  