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