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_openomci_agent.py b/test/unit/extensions/omci/test_openomci_agent.py
new file mode 100644
index 0000000..cef505c
--- /dev/null
+++ b/test/unit/extensions/omci/test_openomci_agent.py
@@ -0,0 +1,91 @@
+#
+# 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 unittest import TestCase, main
+from nose.tools import assert_raises
+from copy import deepcopy
+from mock.mock_adapter_agent import MockAdapterAgent, MockCore
+from pyvoltha.adapters.extensions.omci.openomci_agent import OpenOMCIAgent, OpenOmciAgentDefaults
+from pyvoltha.adapters.extensions.omci.database.mib_db_ext import MibDbExternal
+from pyvoltha.adapters.extensions.omci.database.mib_db_dict import MibDbVolatileDict
+from pyvoltha.adapters.extensions.omci.state_machines.mib_sync import MibSynchronizer
+from pyvoltha.adapters.extensions.omci.tasks.mib_upload import MibUploadTask
+from pyvoltha.adapters.extensions.omci.tasks.get_mds_task import GetMdsTask
+from pyvoltha.adapters.extensions.omci.tasks.mib_resync_task import MibResyncTask
+from pyvoltha.adapters.extensions.omci.tasks.mib_reconcile_task import MibReconcileTask
+
+
+class TestOpenOmciAgent(TestCase):
+ """
+ Test the Open OMCI Agent
+ """
+ def setUp(self):
+ self.adapter_agent = MockAdapterAgent()
+
+ def tearDown(self):
+ if self.adapter_agent is not None:
+ self.adapter_agent.tearDown()
+
+ def test_omci_agent_defaults(self):
+ # Make sure someone does not check in bad default values
+
+ mib_sync = OpenOmciAgentDefaults.get('mib-synchronizer')
+
+ self.assertIsNotNone(mib_sync)
+ self.assertTrue(isinstance(mib_sync['state-machine'], type(MibSynchronizer)))
+ self.assertTrue(isinstance(mib_sync['database'], type(MibDbExternal)))
+
+ mib_sync_tasks = mib_sync.get('tasks')
+
+ self.assertIsNotNone(mib_sync_tasks)
+ self.assertTrue(isinstance(mib_sync_tasks['mib-upload'], type(MibUploadTask)))
+ self.assertTrue(isinstance(mib_sync_tasks['get-mds'], type(GetMdsTask)))
+ self.assertTrue(isinstance(mib_sync_tasks['mib-audit'], type(GetMdsTask)))
+ self.assertTrue(isinstance(mib_sync_tasks['mib-resync'], type(MibResyncTask)))
+ self.assertTrue(isinstance(mib_sync_tasks['mib-reconcile'], type(MibReconcileTask)))
+
+ # caps = OpenOmciAgentDefaults.get('onu-capabilities')
+ #
+ # self.assertIsNotNone(caps)
+
+ def test_omci_agent_default_init(self):
+ agent = OpenOMCIAgent(MockCore)
+
+ self.assertTrue(isinstance(agent.core, type(MockCore)))
+ self.assertTrue(isinstance(agent.database_class, type(MibDbExternal)))
+ self.assertEqual(len(agent.device_ids()), 0)
+ assert_raises(KeyError, agent.get_device, 'deadbeef')
+
+ def test_omci_agent_custom_mib_database(self):
+ custom = deepcopy(OpenOmciAgentDefaults)
+ custom['mib-synchronizer']['database'] = MibDbVolatileDict
+ agent = OpenOMCIAgent(MockCore, support_classes=custom)
+
+ self.assertTrue(isinstance(agent.core, type(MockCore)))
+ self.assertTrue(isinstance(agent.database_class, type(MibDbVolatileDict)))
+
+ def test_omci_agent_start_stop(self):
+ agent = OpenOMCIAgent(MockCore)
+
+ agent.start()
+ agent.start() # Should be a NOP, no side effects
+
+ agent.stop()
+ agent.stop() # Should be a NOP, no side effects
+
+
+if __name__ == '__main__':
+ main()
+