blob: 43b27d9962feef331979895853e075c9a8a9b776 [file] [log] [blame]
#
# 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 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': long(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/O)
'maximum_queue_size': int(222) # Only on OLT but read-only
}
attributes_onu = {
'related_port': int(5678) # IntField (R/O)
}
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()