VOL-1448: Initial checkin of pyvoltha repository
This is very early work and unit tests are not currently running.
Future versions of this code will remove the protobuf directory
and address any v2.0 API changes such as the key-value store API
used by various libraries in pyvoltha
- Added .gitreview config file
- Moved VERSION file to expected location and specified a dev version
so no git tags or PyPI publishing occurs until we are ready.
- Removed generated .desc protobuf files
Change-Id: Icaedc6a4d2cff87cd7d538d3610586d0f5a5db18
diff --git a/test/unit/extensions/omci/test_mib_resync_task.py b/test/unit/extensions/omci/test_mib_resync_task.py
new file mode 100644
index 0000000..43b27d9
--- /dev/null
+++ b/test/unit/extensions/omci/test_mib_resync_task.py
@@ -0,0 +1,372 @@
+#
+# 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()