| # 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 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) |