Init commit for standalone enodebd
Change-Id: I88eeef5135dd7ba8551ddd9fb6a0695f5325337b
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)