tests/enb_acs_manager_tests.py

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