test/unit/extensions/omci/test_omci_configuration.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 hashlib import md5
from unittest import TestCase, main
from nose.tools import 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.onu_configuration import OMCCVersion
from pyvoltha.adapters.extensions.omci.omci_defs import *
from pyvoltha.adapters.extensions.omci.omci_entities import OntG, Ont2G, Cardholder, \
    CircuitPack, SoftwareImage, AniG, UniG
from pyvoltha.common.utils.asleep import asleep
from pyvoltha.adapters.extensions.omci.database.mib_db_dict import MibDbVolatileDict
from datetime import datetime

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


class TestOmciConfiguration(TestCase):
    """
    Test the OMCI read-only Configuration library methods
    """
    def setUp(self):
        self.adapter_agent = MockAdapterAgent()

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

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

    def tearDown(self):
        if self.omci_agent is not None:
            self.omci_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)
        # Add device to OpenOMCI
        self.onu_device = self.omci_agent.add_device(DEFAULT_ONU_DEVICE_ID,
                                                     self.adapter_agent)

        # Allow timeout trigger support while in disabled state for mib sync
        # to make tests run cleanly while profiling.
        self.onu_device.mib_synchronizer.machine.add_transition('timeout', 'disabled', 'disabled')

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

    def _stuff_database(self, entries):
        """
        Stuff the MIB database with some entries that we will use during tests
        """
        database = self.onu_device.mib_synchronizer._database

        # Stuff a value into last in sync. This makes it look like
        # the ONU has been in in-sync at least once.
        self.onu_device.mib_synchronizer.last_mib_db_sync = datetime.utcnow()

        # Entry is a tuple of (class_id, instance_id, {attributes})
        for entry in entries:
            database.set(DEFAULT_ONU_DEVICE_ID, entry[0], entry[1], entry[2])

    def test_OMCCVersion(self):
        for key, value in OMCCVersion.__members__.items():
            self.assertEqual(OMCCVersion.to_enum(OMCCVersion[key].value), value)

        self.assertEqual(OMCCVersion.to_enum(-1), OMCCVersion.Unknown)

    @deferred(timeout=50000)
    def test_defaults(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.omci_agent.device_ids()), 1)

        @raises(AssertionError)
        def do_my_tests(_results):
            config = self.onu_device.configuration
            # Should raise assertion if never been synchronized
            config.version

        # No capabilities available until started
        self.assertIsNone(self.onu_device.configuration)

        # 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
        self.onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(do_my_tests, self.not_called)
        return d

    @deferred(timeout=5)
    def test_in_sync_but_empty(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.omci_agent.device_ids()), 1)

        def stuff_db(_results):
            self._stuff_database([])

        def do_my_tests(_results):
            config = self.onu_device.configuration

            # On no Class ID for requested property, None should be
            # returned
            self.assertIsNone(config.version)
            self.assertIsNone(config.traffic_management_option)
            self.assertIsNone(config.onu_survival_time)
            self.assertIsNone(config.equipment_id)
            self.assertIsNone(config.omcc_version)
            self.assertIsNone(config.vendor_product_code)
            self.assertIsNone(config.total_priority_queues)
            self.assertIsNone(config.total_traffic_schedulers)
            self.assertIsNone(config.total_gem_ports)
            self.assertIsNone(config.uptime)
            self.assertIsNone(config.connectivity_capability)
            self.assertIsNone(config.qos_configuration_flexibility)
            self.assertIsNone(config.priority_queue_scale_factor)
            self.assertIsNone(config.cardholder_entities)
            self.assertIsNone(config.circuitpack_entities)
            self.assertIsNone(config.software_images)
            self.assertIsNone(config.ani_g_entities)
            self.assertIsNone(config.uni_g_entities)

        # No capabilities available until started
        self.assertIsNone(self.onu_device.configuration)

        # Yield context so that MIB Database callLater runs.
        self.onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(stuff_db, self.not_called)
        d.addCallbacks(do_my_tests, self.not_called)
        return d

    @deferred(timeout=5)
    def test_in_sync_with_ont_g_values(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.omci_agent.device_ids()), 1)

        version = 'abcDEF'
        tm_opt = 2
        onu_survival = 123

        def stuff_db(_results):
            self._stuff_database([
                (OntG.class_id, 0, {'version': version,
                                    'traffic_management_options': tm_opt,
                                    'ont_survival_time': onu_survival
                                    })])

        def do_my_tests(_results):
            config = self.onu_device.configuration

            # On no Class ID for requested property, None should be
            # returned
            self.assertEqual(config.version, version)
            self.assertEqual(config.traffic_management_option, tm_opt)
            self.assertEqual(config.onu_survival_time, onu_survival)

        # No capabilities available until started
        self.assertIsNone(self.onu_device.configuration)

        # Yield context so that MIB Database callLater runs.
        self.onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(stuff_db, self.not_called)
        d.addCallbacks(do_my_tests, self.not_called)
        return d

    @deferred(timeout=5)
    def test_in_sync_with_ont_2g_values(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.omci_agent.device_ids()), 1)

        equip_id = 'br-549'
        omcc_ver = OMCCVersion.G_988_2012
        vend_code = 0x1234
        queues = 64
        scheds = 8
        gem_ports = 24
        uptime = 12345
        conn_capp = 0x00aa
        qos_flex = 0x001b
        queue_scale = 1

        def stuff_db(_results):
            self._stuff_database([
                (Ont2G.class_id, 0, {'equipment_id': equip_id,
                                     'omcc_version': omcc_ver.value,
                                     'vendor_product_code': vend_code,
                                     'total_priority_queue_number': queues,
                                     'total_traffic_scheduler_number': scheds,
                                     'total_gem_port_id_number': gem_ports,
                                     'sys_uptime': uptime,
                                     'connectivity_capability': conn_capp,
                                     'qos_configuration_flexibility': qos_flex,
                                     'priority_queue_scale_factor': queue_scale
                                     })])

        def do_my_tests(_results):
            config = self.onu_device.configuration

            self.assertEqual(config.equipment_id, equip_id)
            self.assertEqual(config.omcc_version, omcc_ver)
            self.assertEqual(config.vendor_product_code, vend_code)
            self.assertEqual(config.total_priority_queues, queues)
            self.assertEqual(config.total_traffic_schedulers, scheds)
            self.assertEqual(config.total_gem_ports, gem_ports)
            self.assertEqual(config.uptime, uptime)
            self.assertEqual(config.connectivity_capability, conn_capp)
            self.assertEqual(config.qos_configuration_flexibility, qos_flex)
            self.assertEqual(config.priority_queue_scale_factor, queue_scale)

        # No capabilities available until started
        self.assertIsNone(self.onu_device.configuration)

        # Yield context so that MIB Database callLater runs.
        self.onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(stuff_db, self.not_called)
        d.addCallbacks(do_my_tests, self.not_called)
        return d

    @deferred(timeout=5)
    def test_in_sync_with_cardholder_values(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.omci_agent.device_ids()), 1)

        ch_entity = 0x102
        unit_type = 255
        clie_code = 'abc123'
        prot_ptr = 0

        def stuff_db(_results):
            self._stuff_database([
            (Cardholder.class_id, ch_entity, {'actual_plugin_unit_type': unit_type,
                                              'actual_equipment_id': clie_code,
                                              'protection_profile_pointer': prot_ptr,
                                              })])

        def do_my_tests(_results):
            config = self.onu_device.configuration

            cardholder = config.cardholder_entities
            self.assertTrue(isinstance(cardholder, dict))
            self.assertEqual(len(cardholder), 1)
            self.assertEqual(cardholder[ch_entity]['entity-id'], ch_entity)
            self.assertEqual(cardholder[ch_entity]['is-single-piece'], ch_entity >= 256)
            self.assertEqual(cardholder[ch_entity]['slot-number'], ch_entity & 0xFF)
            self.assertEqual(cardholder[ch_entity]['actual-plug-in-type'], unit_type)
            self.assertEqual(cardholder[ch_entity]['actual-equipment-id'], clie_code)
            self.assertEqual(cardholder[ch_entity]['protection-profile-ptr'], prot_ptr)

        # No capabilities available until started
        self.assertIsNone(self.onu_device.configuration)

        # Yield context so that MIB Database callLater runs.
        self.onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(stuff_db, self.not_called)
        d.addCallbacks(do_my_tests, self.not_called)
        return d

    @deferred(timeout=5)
    def test_in_sync_with_circuitpack_values(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.omci_agent.device_ids()), 1)

        cp_entity = 0x100
        num_ports = 1
        serial_num = 'ABCD01234'
        cp_version = '1234ABCD'
        vendor_id = 'AB-9876'
        tconts = 2
        pqueues = 64
        sched_count = 8

        def stuff_db(_results):
            self._stuff_database([
                (CircuitPack.class_id, cp_entity, {'number_of_ports': num_ports,
                                                   'serial_number': serial_num,
                                                   'version': cp_version,
                                                   'vendor_id': vendor_id,
                                                   'total_tcont_buffer_number': tconts,
                                                   'total_priority_queue_number': pqueues,
                                                   'total_traffic_scheduler_number': sched_count,
                                                   })])

        def do_my_tests(_results):
            config = self.onu_device.configuration

            circuitpack = config.circuitpack_entities
            self.assertTrue(isinstance(circuitpack, dict))
            self.assertEqual(len(circuitpack), 1)
            self.assertEqual(circuitpack[cp_entity]['entity-id'], cp_entity)
            self.assertEqual(circuitpack[cp_entity]['number-of-ports'], num_ports)
            self.assertEqual(circuitpack[cp_entity]['serial-number'], serial_num)
            self.assertEqual(circuitpack[cp_entity]['version'], cp_version)
            self.assertEqual(circuitpack[cp_entity]['vendor-id'], vendor_id)
            self.assertEqual(circuitpack[cp_entity]['total-tcont-count'], tconts)
            self.assertEqual(circuitpack[cp_entity]['total-priority-queue-count'], pqueues)
            self.assertEqual(circuitpack[cp_entity]['total-traffic-sched-count'], sched_count)

        # No capabilities available until started
        self.assertIsNone(self.onu_device.configuration)

        # Yield context so that MIB Database callLater runs.
        self.onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(stuff_db, self.not_called)
        d.addCallbacks(do_my_tests, self.not_called)
        return d

    @deferred(timeout=5)
    def test_in_sync_with_software_values(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.omci_agent.device_ids()), 1)

        sw_entity = 0x200
        sw_version = 'Beta-0.0.2'
        sw_hash = md5("just_a_test").hexdigest()
        prod_code = 'MySoftware'
        sw_active = True
        sw_committed = True
        sw_valid = True

        def stuff_db(_results):
            self._stuff_database([
                (SoftwareImage.class_id, sw_entity, {'version': sw_version,
                                                     'is_committed': sw_committed,
                                                     'is_active': sw_active,
                                                     'is_valid': sw_valid,
                                                     'product_code': prod_code,
                                                     'image_hash': sw_hash,
                                                     })])

        def do_my_tests(_results):
            config = self.onu_device.configuration

            images = config.software_images
            self.assertTrue(isinstance(images, list))
            self.assertEqual(len(images), 1)
            self.assertEqual(images[0].name, 'running-revision' if sw_active else 'candidate-revision')
            self.assertEqual(images[0].version, sw_version)
            self.assertEqual(images[0].is_active, 1 if sw_active else 0)
            self.assertEqual(images[0].is_committed, 1 if sw_committed else 0)
            self.assertEqual(images[0].is_valid,  1 if sw_valid else 0)
            self.assertEqual(images[0].hash, sw_hash)

        # No capabilities available until started
        self.assertIsNone(self.onu_device.configuration)

        # Yield context so that MIB Database callLater runs.
        self.onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(stuff_db, self.not_called)
        d.addCallbacks(do_my_tests, self.not_called)
        return d

    @deferred(timeout=5)
    def test_in_sync_with_ani_g_values(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.omci_agent.device_ids()), 1)

        entity_id = 0x0106
        tconts = 4
        dba_report = 4

        def stuff_db(_results):
            self._stuff_database([
                (AniG.class_id, entity_id, {'total_tcont_number': tconts,
                                            'piggyback_dba_reporting': dba_report
                                            })
            ])

        def do_my_tests(_results):
            config = self.onu_device.configuration

            anig = config.ani_g_entities
            self.assertTrue(isinstance(anig, dict))
            self.assertEqual(len(anig), 1)

            self.assertEqual(anig[entity_id]['entity-id'], entity_id)
            self.assertEqual(anig[entity_id]['slot-number'], (entity_id >> 8) & 0xff)
            self.assertEqual(anig[entity_id]['port-number'], entity_id & 0xff)
            self.assertEqual(anig[entity_id]['total-tcont-count'], tconts)
            self.assertEqual(anig[entity_id]['piggyback-dba-reporting'], dba_report)

        # No capabilities available until started
        self.assertIsNone(self.onu_device.configuration)

        # Yield context so that MIB Database callLater runs.
        self.onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(stuff_db, self.not_called)
        d.addCallbacks(do_my_tests, self.not_called)
        return d

    @deferred(timeout=5)
    def test_in_sync_with_uni_g_values(self):
        self.setup_one_of_each()
        self.assertEqual(len(self.omci_agent.device_ids()), 1)

        entity_id = 0x4321
        mgmt_cap = 0

        def stuff_db(_results):
            self._stuff_database([
                (UniG.class_id, entity_id, {'management_capability': mgmt_cap})
            ])

        def do_my_tests(_results):
            config = self.onu_device.configuration

            unig = config.uni_g_entities
            self.assertTrue(isinstance(unig, dict))
            self.assertEqual(len(unig), 1)

            self.assertEqual(unig[entity_id]['entity-id'], entity_id)
            self.assertEqual(unig[entity_id]['management-capability'], mgmt_cap)

        # No capabilities available until started
        self.assertIsNone(self.onu_device.configuration)

        # Yield context so that MIB Database callLater runs.
        self.onu_device.start()
        d = asleep(0.2)
        d.addCallbacks(stuff_db, self.not_called)
        d.addCallbacks(do_my_tests, self.not_called)
        return d


if __name__ == '__main__':
    main()