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