test/unit/extensions/omci/test_onu_device_entry.py

#
# Copyright 2018 the original author or authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# 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 __future__ import absolute_import
from unittest import TestCase, main
from nose.tools import assert_raises
from nose.twistedtools import deferred
from copy import deepcopy
from .mock.mock_adapter_agent import MockAdapterAgent, MockCore
from .mock.mock_onu_handler import MockOnuHandler
from .mock.mock_olt_handler import MockOltHandler
from .mock.mock_onu import MockOnu
from pyvoltha.adapters.extensions.omci.openomci_agent import OpenOMCIAgent, OpenOmciAgentDefaults
from pyvoltha.adapters.extensions.omci.omci_defs import *
from pyvoltha.common.utils.asleep import asleep
from pyvoltha.adapters.extensions.omci.database.mib_db_api import DEVICE_ID_KEY, CLASS_ID_KEY, CREATED_KEY, \
    MODIFIED_KEY, MDS_KEY, LAST_SYNC_KEY, VERSION_KEY, DatabaseStateError
from pyvoltha.adapters.extensions.omci.database.mib_db_dict import MibDbVolatileDict
from six.moves import range


DEFAULT_OLT_DEVICE_ID = 'default_olt_mock'
DEFAULT_ONU_DEVICE_ID = 'default_onu_mock'
DEFAULT_PON_ID = 0
DEFAULT_ONU_ID = 0
DEFAULT_ONU_SN = 'TEST00000001'

OP = EntityOperations
RC = ReasonCodes


def chunk(indexable, chunk_size):
    for i in range(0, len(indexable), chunk_size):
        yield indexable[i:i + chunk_size]


def hex2raw(hex_string):
    return ''.join(chr(int(byte, 16)) for byte in chunk(hex_string, 2))


class TestOnuDeviceEntry(TestCase):
    """
    Test the ONU Device Entry methods
    """
    def setUp(self):
        self.adapter_agent = MockAdapterAgent()

        custom = deepcopy(OpenOmciAgentDefaults)
        custom['mib-synchronizer']['database'] = MibDbVolatileDict

        self.agent = OpenOMCIAgent(MockCore, support_classes=custom)
        self.agent.start()

    def tearDown(self):
        if self.agent is not None:
            self.agent.stop()

        if self.adapter_agent is not None:
            self.adapter_agent.tearDown()

    def setup_mock_olt(self, device_id=DEFAULT_OLT_DEVICE_ID):
        handler = MockOltHandler(self.adapter_agent, device_id)
        self.adapter_agent.add_device(handler.device)
        return handler

    def setup_mock_onu(self, parent_id=DEFAULT_OLT_DEVICE_ID,
                       device_id=DEFAULT_ONU_DEVICE_ID,
                       pon_id=DEFAULT_PON_ID,
                       onu_id=DEFAULT_ONU_ID,
                       serial_no=DEFAULT_ONU_SN):
        handler = MockOnuHandler(self.adapter_agent, parent_id, device_id, pon_id, onu_id)
        handler.serial_number = serial_no
        onu = MockOnu(serial_no, self.adapter_agent, handler.device_id) \
            if serial_no is not None else None
        handler.onu_mock = onu
        return handler

    def setup_one_of_each(self):
        # Most tests will use at lease one or more OLT and ONU
        self.olt_handler = self.setup_mock_olt()
        self.onu_handler = self.setup_mock_onu(parent_id=self.olt_handler.device_id)
        self.onu_device = self.onu_handler.onu_mock

        self.adapter_agent.add_child_device(self.olt_handler.device,
                                            self.onu_handler.device)

    def test_add_remove_device(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.agent.device_ids()), 0)

        onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
                                           self.adapter_agent)
        self.assertIsNotNone(onu_device)
        self.assertEqual(len(self.agent.device_ids()), 1)
        self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)

        # No MIB if not started
        assert_raises(KeyError, onu_device.query_mib)

        self.agent.remove_device(DEFAULT_ONU_DEVICE_ID)
        self.assertEqual(len(self.agent.device_ids()), 1)

    def test_delete_device(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.agent.device_ids()), 0)

        onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
                                           self.adapter_agent)
        self.assertIsNotNone(onu_device)
        self.assertEqual(len(self.agent.device_ids()), 1)
        self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)
        # Can delete if it was not started
        onu_device.delete()
        self.assertEqual(len(self.agent.device_ids()), 0)

        ##########################################
        # Delete of ONU device okay if it is started
        onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
                                           self.adapter_agent)
        self.assertIsNotNone(onu_device)
        self.assertEqual(len(self.agent.device_ids()), 1)
        self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)

        # Start it and then delete it
        onu_device.start()
        onu_device.delete()
        self.assertEqual(len(self.agent.device_ids()), 0)

    @deferred(timeout=5)
    def test_mib_query_fails_if_dev_not_started(self):
        self.setup_one_of_each()

        onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
                                           self.adapter_agent)
        self.assertIsNotNone(onu_device)
        self.assertEqual(len(self.agent.device_ids()), 1)
        self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)

        def not_called(_reason):
            assert False, 'Should never be called'

        def check_status(_results):
            # Device not yet started. Query should fail with KeyError since
            # ONU is not in database yet
            assert_raises(KeyError, onu_device.query_mib)

        # Yield context so that MIB Database callLater runs. This is a waiting
        # Async task from when the OpenOMCIAgent was started.
        d = asleep(0.2)
        d.addCallbacks(check_status, not_called)

        return d

    @deferred(timeout=5)
    def test_mib_query_ok_if_dev_started(self):
        self.setup_one_of_each()

        onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
                                           self.adapter_agent)
        self.assertIsNotNone(onu_device)
        self.assertEqual(len(self.agent.device_ids()), 1)
        self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)

        def not_called(_reason):
            onu_device.stop()
            assert False, 'Should never be called'

        def check_status(_results):
            # Device started. Query will succeed but nothing should be populated
            # but the most basic items

            results = onu_device.query_mib()
            self.assertTrue(isinstance(results, dict))
            self.assertEqual(results.get(DEVICE_ID_KEY), DEFAULT_ONU_DEVICE_ID)

            self.assertIsNotNone(results.get(VERSION_KEY))
            self.assertIsNotNone(results.get(CREATED_KEY))
            self.assertIsNone(results.get(MODIFIED_KEY))        # Created! but not yet modified

            self.assertEqual(results.get(MDS_KEY), 0)
            self.assertIsNone(results.get(LAST_SYNC_KEY))

            self.assertIsNone(results.get(CLASS_ID_KEY))

            # Stopping still allows a query.  Note you just delete a device
            # to clean up any associated databases
            onu_device.stop()
            results = onu_device.query_mib()
            self.assertTrue(isinstance(results, dict))

        # Yield context so that MIB Database callLater runs. This is a waiting
        # Async task from when the OpenOMCIAgent was started. But also start the
        # device so that it's queued async state machines can run as well
        onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(check_status, not_called)

        return d

    @deferred(timeout=5)
    def test_delete_scrubs_mib(self):
        self.setup_one_of_each()

        onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
                                           self.adapter_agent)
        self.assertIsNotNone(onu_device)
        self.assertEqual(len(self.agent.device_ids()), 1)
        self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)

        def not_called(_reason):
            onu_device.stop()
            assert False, 'Should never be called'

        def check_status(_results):
            # Device started. Query will succeed but nothing should be populated
            # but the most basic items

            results = onu_device.query_mib()
            self.assertTrue(isinstance(results, dict))
            self.assertEqual(results.get(DEVICE_ID_KEY), DEFAULT_ONU_DEVICE_ID)

            # Delete should wipe out any MIB data. Note that a delete of a started
            # or stopped ONU device is allowed.  In this case we are deleting a
            # started ONU Device

            onu_device.delete()
            assert_raises(Exception, onu_device.query_mib)
            # TODO: When capabilities are supported, make sure capabilities get cleared as well

        # Yield context so that MIB Database callLater runs. This is a waiting
        # Async task from when the OpenOMCIAgent was started. But also start the
        # device so that it's queued async state machines can run as well
        onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(check_status, not_called)

        return d

    # TODO: Test pub/sub interface if possible
    # TODO: Test custom/vendor-specific ME support
    # TODO: Test override of various state machines or OMCI tasks if possible


if __name__ == '__main__':
    main()