test/unit/extensions/omci/test_mib_resync_task.py

#
# Copyright 2017 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 main, TestCase
from pyvoltha.adapters.extensions.omci.omci_entities import *
from pyvoltha.adapters.extensions.omci.tasks.mib_resync_task import MibResyncTask
from pyvoltha.adapters.extensions.omci.database.mib_db_dict import MibDbVolatileDict as OnuDB
from pyvoltha.adapters.extensions.omci.database.mib_db_ext import MibDbExternal as OltDB
from .mock.mock_adapter_agent import MockAdapterAgent, MockDevice

_DEVICE_ID = 'br-549'


class TestOmciMibResyncTask(TestCase):
    def setUp(self):
        self.adapter_agent = MockAdapterAgent()
        self.adapter_agent.add_device(MockDevice(_DEVICE_ID))  # For Entity class lookups

        self.onu_db = OnuDB(self.adapter_agent)
        self.olt_db = OltDB(self.adapter_agent)

        self.onu_db.start()
        self.olt_db.start()

        self.olt_db.add(_DEVICE_ID)
        self.onu_db.add(_DEVICE_ID)

        self.task = MibResyncTask(self.adapter_agent, _DEVICE_ID)

    def tearDown(self):
        self.onu_db.stop()
        self.olt_db.stop()

    def test_not_same_type_dbs(self):
        #
        # OLT DB is a copy of the 'external' DB, ONU is a volatile DB
        #
        self.assertNotEqual(type(self.olt_db), type(self.onu_db))

    def test_db_same_format_str_field_serialization(self):
        class_id = OltG.class_id
        inst_id = 0
        attributes = {
            'olt_vendor_id': 'ABCD',             # StrFixedLenField(4)
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)

    def test_db_same_format_mac_address_ip_field_serialization(self):
        class_id = IpHostConfigData.class_id
        inst_id = 0
        attributes = {
            'mac_address': '00:01:02:03:04:05',             # MACField
            'ip_address': '1.2.3.4',                        # IPField
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)

    def test_db_same_format_byte_and_short_field_serialization(self):
        class_id = UniG.class_id
        inst_id = 0
        attributes = {
            'administrative_state': int(1),                # ByteField
            'non_omci_management_identifier': int(12345)   # IPField
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)

    def test_db_same_format_int_field_serialization(self):
        class_id = PriorityQueueG.class_id
        inst_id = 0
        attributes = {
            'related_port': int(1234567)    # IntField
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)

    def test_db_same_format_long_field_serialization(self):
        class_id = PriorityQueueG.class_id
        inst_id = 0
        attributes = {
            'packet_drop_queue_thresholds': int(0x1234)        # LongField
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)

    def test_db_same_format_bit_field_serialization(self):
        class_id = OntG.class_id
        inst_id = 0
        attributes = {
            'extended_tc_layer_options': int(0x1234),        # BitField(16)
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)

    def test_db_same_format_list_field_serialization(self):
        class_id = VlanTaggingFilterData.class_id
        inst_id = 0
        vlan_filter_list = [0] * 12
        vlan_filter_list[0] = 0x1234

        attributes = {
            'vlan_filter_list': vlan_filter_list,        # FieldListField
            'forward_operation': 0,
            'number_of_entries': 1
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)

    def test_db_same_format_complex_json_serialization(self):
        class_id = ExtendedVlanTaggingOperationConfigurationData.class_id
        inst_id = 0x202
        table_data = VlanTaggingOperation(
            filter_outer_priority=15,
            filter_inner_priority=8,
            filter_inner_vid=1024,
            filter_inner_tpid_de=5,
            filter_ether_type=0,
            treatment_tags_to_remove=1,
            pad3=2,
            treatment_outer_priority=15,
            treatment_inner_priority=8,
            treatment_inner_vid=1024,
            treatment_inner_tpid_de=4
        )
        attributes = dict(
            received_frame_vlan_tagging_operation_table=table_data
        )
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)

    def test_on_olt_only(self):
        class_id = GemInterworkingTp.class_id
        inst_id = 0
        attributes = {
            'gal_loopback_configuration': int(1)
        }
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 1)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)
        self.assertEqual(olt_only, [(class_id, inst_id)])

        # Now a little more complex (extra instance on the OLT
        self.olt_db.set(_DEVICE_ID, class_id, inst_id + 1, attributes)
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 1)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)
        self.assertEqual(olt_only, [(class_id, inst_id + 1)])

    def test_on_onu_only(self):
        class_id = PriorityQueueG.class_id
        inst_id = 0
        attributes = {
            'related_port': int(1234567)    # IntField
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 1)
        self.assertEqual(len(attr_diffs), 0)
        self.assertEqual(onu_only, [(class_id, inst_id)])   # Test contents of what was returned

        # Now a little more complex (extra instance on the ONU
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes)
        self.onu_db.set(_DEVICE_ID, class_id, inst_id + 1, attributes)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 1)
        self.assertEqual(len(attr_diffs), 0)
        self.assertEqual(onu_only, [(class_id, inst_id + 1)])   # Test contents of what was returned

    def test_on_attr_different_value(self):
        class_id = PriorityQueueG.class_id
        inst_id = 0
        attributes_olt = {
            'weight': int(12)    # ByteField
        }
        attributes_onu = {
            'weight': int(34)    # ByteField
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes_onu)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes_olt)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 1)
        self.assertEqual(attr_diffs, [(class_id, inst_id, 'weight')])

    def test_ignore_read_only_attribute_differences(self):
        class_id = PriorityQueueG.class_id
        inst_id = 0
        attributes_olt = {
            'related_port': int(1234),      # IntField (R/W)
            'maximum_queue_size': int(222)  # Only on OLT but read-only
        }
        attributes_onu = {
            'related_port': int(1234)    # IntField (R/W)
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes_onu)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes_olt)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 0)

    def test_on_attr_more_on_olt(self):
        class_id = PriorityQueueG.class_id
        inst_id = 0
        attributes_olt = {
            'related_port': int(1234),       # IntField
            'back_pressure_time': int(1234)  # IntField
        }
        attributes_onu = {
            'related_port': int(1234)  # IntField
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes_onu)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes_olt)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 1)
        self.assertEqual(attr_diffs, [(class_id, inst_id, 'back_pressure_time')])

    def test_on_attr_more_on_onu(self):
        class_id = PriorityQueueG.class_id
        inst_id = 0
        attributes_olt = {
            'related_port': int(1234)  # IntField
        }
        attributes_onu = {
            'related_port': int(1234),       # IntField
            'back_pressure_time': int(5678)  # IntField
        }
        self.onu_db.set(_DEVICE_ID, class_id, inst_id, attributes_onu)
        self.olt_db.set(_DEVICE_ID, class_id, inst_id, attributes_olt)

        db_copy = self.olt_db.query(_DEVICE_ID)
        db_active = self.onu_db.query(_DEVICE_ID)

        olt_only, onu_only, attr_diffs = self.task.compare_mibs(db_copy, db_active)

        self.assertEqual(len(olt_only), 0)
        self.assertEqual(len(onu_only), 0)
        self.assertEqual(len(attr_diffs), 1)
        self.assertEqual(attr_diffs, [(class_id, inst_id, 'back_pressure_time')])


if __name__ == '__main__':
    main()