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/Makefile b/test/Makefile
new file mode 100644
index 0000000..0d4d081
--- /dev/null
+++ b/test/Makefile
@@ -0,0 +1,27 @@
+#
+# 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.
+#
+.PHONY: utests
+
+utest:
+ @ nosetests utests
+
+COVERAGE_OPTS=--with-xcoverage --with-xunit --cover-package=voltha,common,ofagent --cover-html\
+ --cover-html-dir=tmp/cover
+
+utest-with-coverage:
+ @ nosetests $(COVERAGE_OPTS) utests
+
+# end file
diff --git a/test/__init__.py b/test/__init__.py
new file mode 100644
index 0000000..cfcdc97
--- /dev/null
+++ b/test/__init__.py
@@ -0,0 +1,15 @@
+#
+# 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.
+#
\ No newline at end of file
diff --git a/test/unit/README.md b/test/unit/README.md
new file mode 100644
index 0000000..f85cfd4
--- /dev/null
+++ b/test/unit/README.md
@@ -0,0 +1,17 @@
+## VOLTHA TESTS
+
+Currently only unit tests are supported by the PyVoltha package
+* **Unit Tests**
+ * These tests exercise the smallest testable parts of the code. They
+ are designed to be fully automated and can be executed by a build
+ machine (e.g. Jenkins).
+
+This document focuses on running the unit tests only.
+
+##### Running the utests
+* **Triggering all the utests as a batch run**: Unit tests under voltha can be run as follows:
+```
+cd /pyvoltha/
+. ./env.sh
+make utest
+```
diff --git a/test/unit/__init__.py b/test/unit/__init__.py
new file mode 100644
index 0000000..b0fb0b2
--- /dev/null
+++ b/test/unit/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2017-present Open Networking Foundation
+#
+# 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.
diff --git a/test/unit/common/__init__.py b/test/unit/common/__init__.py
new file mode 100644
index 0000000..b0fb0b2
--- /dev/null
+++ b/test/unit/common/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2017-present Open Networking Foundation
+#
+# 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.
diff --git a/test/unit/common/test_event_bus.py b/test/unit/common/test_event_bus.py
new file mode 100644
index 0000000..be325e8
--- /dev/null
+++ b/test/unit/common/test_event_bus.py
@@ -0,0 +1,220 @@
+#
+# Copyright 2016 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.
+#
+import re
+
+from mock import Mock
+from mock import call
+from twisted.internet.defer import DeferredQueue, inlineCallbacks
+from twisted.trial.unittest import TestCase
+
+from pyvoltha.common.event_bus import EventBusClient, EventBus
+
+
+class TestEventBus(TestCase):
+
+ def test_subscribe(self):
+
+ ebc = EventBusClient()
+ sub = ebc.subscribe('news', lambda msg, topic: None)
+ self.assertEqual(len(ebc.list_subscribers()), 1)
+ self.assertEqual(len(ebc.list_subscribers('news')), 1)
+ self.assertEqual(len(ebc.list_subscribers('other')), 0)
+
+ def test_unsubscribe(self):
+
+ ebc = EventBusClient(EventBus())
+ sub = ebc.subscribe('news', lambda msg, topic: None)
+ ebc.unsubscribe(sub)
+ self.assertEqual(ebc.list_subscribers(), [])
+ self.assertEqual(ebc.list_subscribers('news'), [])
+
+ def test_simple_publish(self):
+
+ ebc = EventBusClient(EventBus())
+
+ mock = Mock()
+ ebc.subscribe('news', mock)
+
+ ebc.publish('news', 'message')
+
+ self.assertEqual(mock.call_count, 1)
+ mock.assert_called_with('news', 'message')
+
+ def test_topic_filtering(self):
+
+ ebc = EventBusClient(EventBus())
+
+ mock = Mock()
+ ebc.subscribe('news', mock)
+
+ ebc.publish('news', 'msg1')
+ ebc.publish('alerts', 'msg2')
+ ebc.publish('logs', 'msg3')
+
+ self.assertEqual(mock.call_count, 1)
+ mock.assert_called_with('news', 'msg1')
+
+ def test_multiple_subscribers(self):
+
+ ebc = EventBusClient(EventBus())
+
+ mock1 = Mock()
+ ebc.subscribe('news', mock1)
+
+ mock2 = Mock()
+ ebc.subscribe('alerts', mock2)
+
+ mock3 = Mock()
+ ebc.subscribe('logs', mock3)
+
+ mock4 = Mock()
+ ebc.subscribe('logs', mock4)
+
+ ebc.publish('news', 'msg1')
+ ebc.publish('alerts', 'msg2')
+ ebc.publish('logs', 'msg3')
+
+ self.assertEqual(mock1.call_count, 1)
+ mock1.assert_called_with('news', 'msg1')
+
+ self.assertEqual(mock2.call_count, 1)
+ mock2.assert_called_with('alerts', 'msg2')
+
+ self.assertEqual(mock3.call_count, 1)
+ mock3.assert_called_with('logs', 'msg3')
+
+ self.assertEqual(mock4.call_count, 1)
+ mock4.assert_called_with('logs', 'msg3')
+
+ def test_predicates(self):
+
+ ebc = EventBusClient(EventBus())
+
+ get_foos = Mock()
+ ebc.subscribe('', get_foos, lambda msg: msg.startswith('foo'))
+
+ get_bars = Mock()
+ ebc.subscribe('', get_bars, lambda msg: msg.endswith('bar'))
+
+ get_all = Mock()
+ ebc.subscribe('', get_all)
+
+ get_none = Mock()
+ ebc.subscribe('', get_none, lambda msg: msg.find('zoo') >= 0)
+
+ errored = Mock()
+ ebc.subscribe('', errored, lambda msg: 1/0)
+
+ ebc.publish('', 'foo')
+ ebc.publish('', 'foobar')
+ ebc.publish('', 'bar')
+
+ c = call
+
+ self.assertEqual(get_foos.call_count, 2)
+ get_foos.assert_has_calls([c('', 'foo'), c('', 'foobar')])
+
+ self.assertEqual(get_bars.call_count, 2)
+ get_bars.assert_has_calls([c('', 'foobar'), c('', 'bar')])
+
+ self.assertEqual(get_all.call_count, 3)
+ get_all.assert_has_calls([c('', 'foo'), c('', 'foobar'), c('', 'bar')])
+
+ get_none.assert_not_called()
+
+ errored.assert_not_called()
+
+ def test_wildcard_topic(self):
+
+ ebc = EventBusClient(EventBus())
+ subs = []
+
+ wildcard_sub = Mock()
+ subs.append(ebc.subscribe(re.compile(r'.*'), wildcard_sub))
+
+ prefix_sub = Mock()
+ subs.append(ebc.subscribe(re.compile(r'ham.*'), prefix_sub))
+
+ contains_sub = Mock()
+ subs.append(ebc.subscribe(re.compile(r'.*burg.*'), contains_sub))
+
+ ebc.publish('news', 1)
+ ebc.publish('hamsters', 2)
+ ebc.publish('hamburgers', 3)
+ ebc.publish('nonsense', 4)
+
+ c = call
+
+ self.assertEqual(wildcard_sub.call_count, 4)
+ wildcard_sub.assert_has_calls([
+ c('news', 1),
+ c('hamsters', 2),
+ c('hamburgers', 3),
+ c('nonsense', 4)])
+
+ self.assertEqual(prefix_sub.call_count, 2)
+ prefix_sub.assert_has_calls([
+ c('hamsters', 2),
+ c('hamburgers', 3)])
+
+ self.assertEqual(contains_sub.call_count, 1)
+ contains_sub.assert_has_calls([c('hamburgers', 3)])
+
+ for sub in subs:
+ ebc.unsubscribe(sub)
+
+ self.assertEqual(ebc.list_subscribers(), [])
+
+ @inlineCallbacks
+ def test_deferred_queue_receiver(self):
+
+ ebc = EventBus()
+
+ queue = DeferredQueue()
+
+ ebc.subscribe('', lambda _, msg: queue.put(msg))
+
+ for i in xrange(10):
+ ebc.publish('', i)
+
+ self.assertEqual(len(queue.pending), 10)
+ for i in xrange(10):
+ msg = yield queue.get()
+ self.assertEqual(msg, i)
+ self.assertEqual(len(queue.pending), 0)
+
+ def test_subscribers_that_unsubscribe_when_called(self):
+ # VOL-943 bug fix check
+ ebc = EventBusClient(EventBus())
+
+ class UnsubscribeWhenCalled(object):
+ def __init__(self):
+ self.subscription = ebc.subscribe('news', self.unsubscribe)
+ self.called = False
+
+ def unsubscribe(self, _topic, _msg):
+ self.called = True
+ ebc.unsubscribe(self.subscription)
+
+ ebc1 = UnsubscribeWhenCalled()
+ ebc2 = UnsubscribeWhenCalled()
+ ebc3 = UnsubscribeWhenCalled()
+
+ ebc.publish('news', 'msg1')
+
+ self.assertTrue(ebc1.called)
+ self.assertTrue(ebc2.called)
+ self.assertTrue(ebc3.called)
diff --git a/test/unit/common/utils/__init__.py b/test/unit/common/utils/__init__.py
new file mode 100644
index 0000000..b0fb0b2
--- /dev/null
+++ b/test/unit/common/utils/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2017-present Open Networking Foundation
+#
+# 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.
diff --git a/test/unit/common/utils/test_bpf.py b/test/unit/common/utils/test_bpf.py
new file mode 100644
index 0000000..21a80da
--- /dev/null
+++ b/test/unit/common/utils/test_bpf.py
@@ -0,0 +1,48 @@
+# Copyright 2017-present Open Networking Foundation
+#
+# 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 scapy.layers.l2 import Ether, Dot1Q
+
+from pyvoltha.adapters.common.frameio.frameio import BpfProgramFilter
+
+
+class TestBpf(TestCase):
+
+ def test_bpf1(self):
+ vid = 4090
+ pcp = 7
+ frame_match = 'ether[14:2] = 0x{:01x}{:03x}'.format(pcp << 1, vid)
+ filter = BpfProgramFilter(frame_match)
+ self.assertTrue(filter(str(Ether()/Dot1Q(prio=pcp, vlan=vid))))
+ self.assertFalse(filter(str(Ether()/Dot1Q(prio=pcp, vlan=4000))))
+
+ def test_bpf2(self):
+ vid1 = 4090
+ pcp1 = 7
+ frame_match_case1 = 'ether[14:2] = 0x{:01x}{:03x}'.format(
+ pcp1 << 1, vid1)
+
+ vid2 = 4000
+ frame_match_case2 = '(ether[14:2] & 0xfff) = 0x{:03x}'.format(vid2)
+
+ filter = BpfProgramFilter('{} or {}'.format(
+ frame_match_case1, frame_match_case2))
+ self.assertTrue(filter(str(Ether()/Dot1Q(prio=pcp1, vlan=vid1))))
+ self.assertTrue(filter(str(Ether()/Dot1Q(vlan=vid2))))
+ self.assertFalse(filter(str(Ether()/Dot1Q(vlan=4001))))
+
+
+if __name__ == '__main__':
+ main()
diff --git a/test/unit/common/utils/test_indexpool.py b/test/unit/common/utils/test_indexpool.py
new file mode 100644
index 0000000..7eb1050
--- /dev/null
+++ b/test/unit/common/utils/test_indexpool.py
@@ -0,0 +1,48 @@
+# Copyright 2017-present Open Networking Foundation
+#
+# 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 pyvoltha.common.utils.indexpool import IndexPool
+
+class TestIndexPool(TestCase):
+ pool = IndexPool(8, 0)
+ def test_01_get_next(self):
+ self.assertEqual(self.pool.indices.bin, '00000000')
+ for i in range(8):
+ self.assertEqual(self.pool.get_next(), i)
+ #to check if there's any bit left after using all 8 bits
+ self.assertIsNone(self.pool.get_next())
+
+ def test_02_pre_allocate(self):
+ _pool2 = IndexPool(8, 0)
+ self.assertEqual(_pool2.indices.bin, '00000000')
+ _pool2.pre_allocate((0,1,2,))
+ self.assertEqual(_pool2.indices.bin, '11100000')
+
+ def test_03_release(self):
+ self.pool.release(5)
+ self.assertEqual(self.pool.indices.bin, '11111011')
+ self.pool.release(10)
+ self.assertEqual(self.pool.indices.bin, '11111011')
+ self.pool.release(0)
+ self.assertEqual(self.pool.indices.bin, '01111011')
+
+ def test_04_check_offset(self):
+ _offset = 5
+ self.pool = IndexPool(8, _offset)
+ for i in range(8):
+ self.assertEqual(self.pool.get_next(), _offset + i)
+
+
+if __name__ == '__main__':
+ main()
\ No newline at end of file
diff --git a/test/unit/common/utils/test_ordered_weakvalue_dict.py b/test/unit/common/utils/test_ordered_weakvalue_dict.py
new file mode 100644
index 0000000..a8ce47c
--- /dev/null
+++ b/test/unit/common/utils/test_ordered_weakvalue_dict.py
@@ -0,0 +1,51 @@
+# Copyright 2017-present Open Networking Foundation
+#
+# 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 pyvoltha.common.utils.ordered_weakvalue_dict import OrderedWeakValueDict
+
+
+class O(object):
+ def __init__(self, a):
+ self.a = a
+
+
+class TestOrderedWeakValueDict(TestCase):
+
+ def test_standard_behavior(self):
+ holder = dict() # a real dict so we can control which object real ref
+ def mk(k):
+ o = O(k)
+ holder[k] = o
+ return o
+ o = OrderedWeakValueDict((k, mk(k)) for k in xrange(10))
+ self.assertEqual(len(o), 10)
+ self.assertEqual(o[3].a, 3)
+ o[3] = mk(-3)
+ self.assertEqual(o[3].a, -3)
+ del o[3]
+ self.assertEqual(len(o), 9)
+ o[100] = mk(100)
+ self.assertEqual(len(o), 10)
+ self.assertEqual(o.keys(), [0, 1, 2, 4, 5, 6, 7, 8, 9, 100])
+
+ # remove a few items from the holder, they should be gone from o too
+ del holder[1]
+ del holder[5]
+ del holder[6]
+
+ self.assertEqual(o.keys(), [0, 2, 4, 7, 8, 9, 100])
+ self.assertEqual([v.a for v in o.values()], [0, 2, 4, 7, 8, 9, 100])
+
+
diff --git a/test/unit/extensions/__init__.py b/test/unit/extensions/__init__.py
new file mode 100644
index 0000000..b0fb0b2
--- /dev/null
+++ b/test/unit/extensions/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2017-present Open Networking Foundation
+#
+# 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.
diff --git a/test/unit/extensions/omci/__init__.py b/test/unit/extensions/omci/__init__.py
new file mode 100644
index 0000000..b0fb0b2
--- /dev/null
+++ b/test/unit/extensions/omci/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2017-present Open Networking Foundation
+#
+# 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.
diff --git a/test/unit/extensions/omci/mock/__init__.py b/test/unit/extensions/omci/mock/__init__.py
new file mode 100644
index 0000000..2792694
--- /dev/null
+++ b/test/unit/extensions/omci/mock/__init__.py
@@ -0,0 +1,24 @@
+#
+# 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 nose.twistedtools import threaded_reactor, stop_reactor
+
+
+def setup_module():
+ threaded_reactor()
+
+
+def teardown_module():
+ stop_reactor()
diff --git a/test/unit/extensions/omci/mock/mock_adapter_agent.py b/test/unit/extensions/omci/mock/mock_adapter_agent.py
new file mode 100644
index 0000000..866eb67
--- /dev/null
+++ b/test/unit/extensions/omci/mock/mock_adapter_agent.py
@@ -0,0 +1,165 @@
+#
+# 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.
+#
+# import binascii
+import structlog
+# from twisted.internet.defer import Deferred
+# from voltha.core.config.config_root import ConfigRoot
+# from pyvoltha.protos.voltha_pb2 import VolthaInstance
+# from pyvoltha.adapters.extensions.omci.omci_frame import OmciFrame
+
+class MockProxyAddress(object):
+ def __init__(self, device_id, pon_id, onu_id):
+ self.device_id = device_id # Device ID of proxy (OLT)
+ self.onu_id = onu_id
+ self.onu_session_id = onu_id
+
+ self.channel_group_id = pon_id # close enough for mock
+ self.channel_id = pon_id
+ self.channel_termination = pon_id
+
+
+class MockDevice(object):
+ def __init__(self, device_id, proxy_address=None, serial_number=None):
+ from pyvoltha.adapters.extensions.omci.omci_entities import entity_id_to_class_map
+ self.id = device_id
+ self.parent_id = None
+ self.proxy_address = proxy_address
+ self.serial_number = serial_number
+ self.me_map = entity_id_to_class_map
+
+
+class MockCore(object):
+ def __init__(self):
+ self.root = None # ConfigRoot(VolthaInstance())
+
+ def get_proxy(self, path):
+ return self.root.get_proxy(path)
+
+
+class MockAdapterAgent(object):
+ """
+ Minimal class to handle adapter-agent needs in OpenOMCI. It can be
+ used by a mock OLT or ONU.
+
+ So that we do not have to duplicate the IAdapter functionality, just
+ the handler, the OLT and ONU handlers are derived from a mock Device
+ base class so that we can access the _devices map and get either a
+ Device to play with (like the real thing) or the associated handler
+ """
+ def __init__(self, d=None):
+ self.log = structlog.get_logger()
+ self._devices = dict() # device-id -> mock device
+ self.core = MockCore()
+ self.deferred = d
+ self.timeout_the_message = False
+
+ @property
+ def send_omci_defer(self):
+ return self.deferred
+
+ @send_omci_defer.setter
+ def send_omci_defer(self, value):
+ self.deferred = value
+
+ @property
+ def name(self):
+ return "cig_mock_ont"
+
+ def tearDown(self):
+ """Test case cleanup"""
+ for device in self._devices.itervalues():
+ device.tearDown()
+ self._devices.clear()
+
+ def add_device(self, device):
+ self._devices[device.id] = device
+
+ def add_child_device(self, parent_device, child_device):
+ # Set parent
+ child_device.parent_id = parent_device.id
+
+ # Add ONU serial number if PON and ONU enabled
+
+ if (child_device.enabled and
+ child_device.serial_number is not None and
+ child_device.proxy_address.channel_id in parent_device.enabled_pons):
+ parent_device.activated_onus.add(child_device.serial_number)
+
+ self.add_device(child_device)
+
+ def get_device(self, device_id):
+ return self._devices[device_id]
+
+ def get_child_device(self, parent_device_id, **kwargs):
+ onu_id = kwargs.pop('onu_id', None)
+ pon_id = kwargs.pop('pon_id', None)
+ if onu_id is None and pon_id is None:
+ return None
+
+ # Get all child devices with the same parent ID
+ children_ids = set(d.id for d in self._devices.itervalues()
+ if d.parent_id == parent_device_id)
+
+ # Loop through all the child devices with this parent ID
+ for child_id in children_ids:
+ device = self.get_device(child_id)
+
+ # Does this child device match the passed in ONU ID?
+ found_onu_id = False
+ if onu_id is not None:
+ if device.proxy_address.onu_id == onu_id:
+ found_onu_id = True
+
+ # Does this child device match the passed in SERIAL NUMBER?
+ found_pon_id = False
+ if pon_id is not None:
+ if device.proxy_address.channel_id == pon_id:
+ found_pon_id = True
+ # Match ONU ID and PON ID
+ if onu_id is not None and pon_id is not None:
+ found = found_onu_id & found_pon_id
+ # Otherwise ONU ID or PON ID
+ else:
+ found = found_onu_id | found_pon_id
+
+ # Return the matched child device
+ if found:
+ return device
+
+ return None
+
+ def send_proxied_message(self, proxy_address, msg):
+ # Look up ONU handler and forward the message
+ self.log.debug("--> send_proxied_message", message=msg)
+
+ # if proxy_address is None:
+ if self.deferred is not None and not self.timeout_the_message:
+ self.deferred.callback(msg)
+ # return None
+
+ # olt_handler = self.get_device(proxy_address.device_id)
+
+ # if olt_handler is not None:
+ # olt_handler.send_proxied_message(proxy_address, msg)
+
+ def receive_proxied_message(self, proxy_address, msg):
+ # Look up ONU handler and forward the message
+
+ onu_handler = self.get_child_device(proxy_address.device_id,
+ onu_id=proxy_address.onu_id,
+ pon_id=proxy_address.channel_id)
+ if onu_handler is not None:
+ onu_handler.receive_proxied_message(proxy_address, msg)
diff --git a/test/unit/extensions/omci/mock/mock_olt_handler.py b/test/unit/extensions/omci/mock/mock_olt_handler.py
new file mode 100644
index 0000000..142dbd8
--- /dev/null
+++ b/test/unit/extensions/omci/mock/mock_olt_handler.py
@@ -0,0 +1,108 @@
+#
+# 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.
+#
+
+import sys
+from mock_adapter_agent import MockDevice
+from nose.twistedtools import reactor
+
+
+class MockOltHandler(MockDevice):
+ """
+ VERY Minimal class to handle OLT needs in OpenOMCI testing
+
+ So that we do not have to duplicate the IAdapter functionality, just
+ the handler, the OLT and ONU handlers are derived from a mock Device
+ base class so that we can access the _devices map and get either a
+ Device to play with (like the real thing) or the associated handler
+ """
+ def __init__(self, adapter_agent, device_id):
+ super(MockOltHandler, self).__init__(device_id)
+
+ self.device_id = device_id
+ self.device = self
+ self._adapter_agent = adapter_agent
+ self._num_tx = 0
+
+ ####################################################################
+ # NOTE: The following can be manipulated in your test case to modify the behaviour
+ # of this mock.
+ #
+ # Note that activated ONUs are added during adapter add_child_device
+ # if the ONU handler associated is 'enabled'
+
+ self.enabled = True # OLT is enabled/active
+ self.activated_onus = set() # Activated ONU serial numbers
+ self.enabled_pons = range(0, 16) # Enabled PONs
+ self.max_tx = sys.maxint # Fail after this many tx requests
+ self.latency = 0.0 # OMCI response latency (keep small)
+
+ # TODO: Implement minimal functionality
+
+ # TODO: Implement minimal functionality
+
+ def tearDown(self):
+ """Test case cleanup"""
+ pass
+
+ # Begin minimal set of needed IAdapter interfaces
+
+ def send_proxied_message(self, proxy_address, msg):
+ """Check various enabled flags and status and send if okay"""
+
+ if not self.enabled:
+ return None
+
+ pon_id = proxy_address.channel_id
+
+ if pon_id not in self.enabled_pons:
+ return None
+
+ # Look up ONU device ID.
+ onu_id = proxy_address.onu_id
+ onu_handler = self._adapter_agent.get_child_device(proxy_address.device_id,
+ pon_id=pon_id,
+ onu_id=onu_id)
+
+ if onu_handler is None or not onu_handler.enabled:
+ return None
+
+ onu_mock = onu_handler.onu_mock
+ if onu_mock is None or onu_mock.serial_number not in self.activated_onus:
+ return None
+
+ # And Tx success (silent discard for OMCI timeout testing)
+ if self._num_tx >= self.max_tx:
+ return None
+ self._num_tx += 1
+
+ response = onu_mock.rx_omci_frame(msg)
+
+ # Make async and add any requested latency. Bound it to less
+ # than 5 seconds since this is a unit test that need to be
+ # somewhat responsive
+
+ assert 0.0 <= self.latency <= 5, 'Best practice is latency <= 5 seconds'
+ if response is not None:
+ reactor.callLater(self.latency, self._deliver_proxy_message, proxy_address, response)
+
+ def _deliver_proxy_message(self, proxy_address, response):
+ from common.frameio.frameio import hexify
+ self._adapter_agent.receive_proxied_message(proxy_address,
+ hexify(str(response)))
+
+ def receive_proxied_message(self, _, __):
+ assert False, 'This is never called on the OLT side of proxy messaging'
+
diff --git a/test/unit/extensions/omci/mock/mock_onu.py b/test/unit/extensions/omci/mock/mock_onu.py
new file mode 100644
index 0000000..e63c5cd
--- /dev/null
+++ b/test/unit/extensions/omci/mock/mock_onu.py
@@ -0,0 +1,283 @@
+#
+# 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 pyvoltha.adapters.extensions.omci.omci_frame import OmciFrame
+from pyvoltha.adapters.extensions.omci.omci_defs import *
+from pyvoltha.adapters.extensions.omci.omci_entities import *
+from pyvoltha.adapters.extensions.omci.omci_messages import *
+
+# abbreviations
+OP = EntityOperations
+RC = ReasonCodes
+
+
+class MockOnu(object):
+ """
+ Minimal class that acts line an ONU. The Mock OLT handler will call into this
+ object with OMCI frames that it will respond to appropriately
+ """
+ def __init__(self, serial_number, adapter_agent, handler_id):
+ self.serial_number = serial_number
+ self._adapter_agent = adapter_agent # TODO: Remove any unused attributes
+ self._handler_id = handler_id
+ self.mib_data_sync = 0 # Assume at reboot!
+
+ # NOTE: when creating response frames, use the basic method of constructing
+ # these frames as the encoding created is unit-tested elsewhere
+ self._omci_response = {
+ OP.Get.value: {
+ CircuitPack.class_id: {
+ 257: OmciFrame(transaction_id=0, # Will get replaced
+ message_type=OmciGetResponse.message_id,
+ omci_message=OmciGetResponse(
+ entity_class=CircuitPack.class_id,
+ entity_id=0,
+ success_code=RC.Success.value,
+ attributes_mask=CircuitPack.mask_for('number_of_ports'),
+ data=OmciMaskedData('value',
+ entity_class=CircuitPack.class_id,
+ attributes_mask=CircuitPack.mask_for('number_of_ports'))
+ ))
+ },
+ # Additional OMCI GET request responses here if needed
+ },
+ OP.GetNext.value: {},
+ OP.Create.value: {
+ # TODO: Create some OMCI CREATE request responses here.
+
+ # def send_create_gal_ethernet_profile(self,
+ # entity_id,
+ # max_gem_payload_size):
+ # frame = OmciFrame(
+ # transaction_id=self.get_tx_id(),
+ # message_type=OmciCreate.message_id,
+ # omci_message=OmciCreate(
+ # entity_class=GalEthernetProfile.class_id,
+ # entity_id=entity_id,
+ # data=dict(
+ # max_gem_payload_size=max_gem_payload_size
+ # )
+ # )
+ # )
+ # self.send_omci_message(frame)
+ },
+ OP.Set.value: {
+ # TODO: Create some OMCI SET request responses here.
+
+ # def send_set_admin_state(self,
+ # entity_id,
+ # admin_state):
+ # data = dict(
+ # administrative_state=admin_state
+ # )
+ # frame = OmciFrame(
+ # transaction_id=self.get_tx_id(),
+ # message_type=OmciSet.message_id,
+ # omci_message=OmciSet(
+ # entity_class=OntG.class_id,
+ # entity_id=entity_id,
+ # attributes_mask=OntG.mask_for(*data.keys()),
+ # data=data
+ # )
+ # )
+ # self.send_omci_message(frame)
+
+ },
+ OP.Delete.value: {
+ # TODO: Create some OMCI DELETE responses here.
+ },
+ OP.MibReset.value: {
+ OntData.class_id: {
+ 0: OmciFrame(transaction_id=0, # Will get replaced
+ message_type=OmciMibResetResponse.message_id,
+ omci_message=OmciMibResetResponse(
+ entity_class=OntData.class_id,
+ entity_id=0,
+ success_code=RC.Success.value
+ ))
+ }
+ },
+ OP.MibUpload.value: {
+ OntData.class_id: {
+ 0: OmciFrame(transaction_id=0, # Will get replaced
+ message_type=OmciMibUploadResponse.message_id,
+ omci_message=OmciMibUploadResponse(
+ entity_class=OntData.class_id,
+ entity_id=0,
+ number_of_commands=3 # Should match list size for MibUploadNext below
+ ))
+ }
+ },
+ # OP.MibUploadNext.value: {
+ # OntData.class_id: {
+ # 0: [
+ # OmciFrame(transaction_id=0,
+ # message_type=OmciMibUploadNextResponse.message_id,
+ # omci_message=OmciMibUploadNextResponse(
+ # entity_class=OntData.class_id,
+ # entity_id=0,
+ # object_entity_id=0, # TODO: Pick one
+ # object_attributes_mask=0, # TODO: Pick one
+ # object_data=None # TODO: Pick one
+ # )),
+ # OmciFrame(transaction_id=0,
+ # message_type=OmciMibUploadNextResponse.message_id,
+ # omci_message=OmciMibUploadNextResponse(
+ # entity_class=OntData.class_id,
+ # entity_id=0,
+ # object_entity_id=0, # TODO: Pick one
+ # object_attributes_mask=0, # TODO: Pick one
+ # object_data=None # TODO: Pick one
+ # )),
+ # OmciFrame(transaction_id=0,
+ # message_type=OmciMibUploadNextResponse.message_id,
+ # omci_message=OmciMibUploadNextResponse(
+ # entity_class=OntData.class_id,
+ # entity_id=0,
+ # object_entity_id=0, # TODO: Pick one
+ # object_attributes_mask=0, # TODO: Pick one
+ # object_data=None # TODO: Pick one
+ # )),
+ # ]
+ # }
+ # },
+ OP.Reboot.value: {
+ OntData.class_id: {
+ 0: OmciFrame(transaction_id=0, # Will get replaced
+ message_type=OmciRebootResponse.message_id,
+ omci_message=OmciRebootResponse(
+ entity_class=OntG.class_id,
+ entity_id=0,
+ success_code=RC.Success.value
+ ))
+ }
+ },
+ }
+ # TODO: Support Autonomous ONU messages as well
+
+ def tearDown(self):
+ """Test case cleanup"""
+ pass
+
+ def _request_to_response_type(self, message_type):
+ return {
+ OP.Create.value: OmciCreateResponse,
+ OP.Delete.value: OmciDeleteResponse,
+ OP.Set.value: OmciSetResponse,
+ OP.Get.value: OmciGetResponse,
+ OP.GetNext.value: OmciGetNextResponse,
+ OP.MibUpload.value: OmciMibUploadResponse,
+ OP.MibUploadNext.value: OmciMibUploadNextResponse,
+ OP.MibReset.value: OmciMibResetResponse,
+ OP.Reboot.value: OmciRebootResponse,
+ }.get(message_type & 0x1F, None)
+
+ def rx_omci_frame(self, msg):
+ try:
+ frame = OmciFrame(msg.decode('hex'))
+ response = None
+ response_type = self._request_to_response_type(frame.fields['message_type'])
+ transaction_id = frame.fields['transaction_id']
+
+ omci_message = frame.fields.get('omci_message')
+
+ class_id = omci_message.fields.get('entity_class') \
+ if omci_message is not None else None
+ instance_id = omci_message.fields.get('entity_id') \
+ if omci_message is not None else None
+
+ # Look up hardcode responses based on class and instance ID. If found
+ # return the response, otherwise send back an error
+
+ if response_type is None:
+ status = RC.ProcessingError.value
+ elif class_id is None:
+ status = RC.UnknownEntity.value
+ elif instance_id is None:
+ status = RC.UnknownInstance.value
+ else:
+ status = RC.Success.value
+ try:
+ response_id = response_type.message_id & 0x1f
+ response = self._omci_response[response_id][class_id][instance_id]
+
+ if response_id == OP.MibUploadNext.value:
+ # Special case. Need to get requested entry
+ assert isinstance(response, list)
+ pass
+ pass
+ pass
+ pass
+
+ if isinstance(omci_message, OmciGetNext):
+ response = response[omci_message.fields['command_sequence_number']]
+
+ if isinstance(response, dict):
+ if response['failures'] > 0:
+ response['failures'] -= 1
+ return None
+ else: response = response['frame']
+
+ response.fields['transaction_id'] = transaction_id
+ if 'success_code' in response.fields['omci_message'].fields:
+ response.fields['omci_message'].fields['success_code'] = status
+
+ if status == RC.Success.value:
+ if response_type.message_id in [OmciCreateResponse.message_id,
+ OmciDeleteResponse.message_id,
+ OmciSetResponse.message_id]:
+ self.mib_data_sync += 1
+ if self.mib_data_sync > 255:
+ self.mib_data_sync = 1
+ elif response_type.message_id == OmciMibResetResponse.message_id:
+ self.mib_data_sync = 0
+
+ except KeyError as e:
+ bad_key = e.args[0]
+ if bad_key == class_id:
+ status = RC.UnknownEntity.value
+ elif bad_key == instance_id:
+ status = RC.UnknownInstance.value
+ else:
+ status = RC.ProcessingError.value
+
+ if status != RC.Success.value and \
+ response_type not in [OmciMibUploadResponse,
+ OmciMibUploadNextResponse]:
+ response = OmciFrame(transaction_id=transaction_id,
+ message_type=response_type.message_id,
+ omci_message=response_type(
+ entity_class=class_id,
+ entity_id=instance_id,
+ success_code=status
+ ))
+ return response
+
+ except Exception as e:
+ pass
+
+ @property
+ def proxy_address(self, device_id='1'):
+ if self._proxy_address is None:
+ self._proxy_address = Device.ProxyAddress(
+ device_id=device_id,
+ channel_group_id=1,
+ channel_id=1,
+ channel_termination="XGSPON",
+ onu_id=20,
+ onu_session_id=1)
+
+ return self._proxy_address
+
diff --git a/test/unit/extensions/omci/mock/mock_onu_handler.py b/test/unit/extensions/omci/mock/mock_onu_handler.py
new file mode 100644
index 0000000..9ebe1f6
--- /dev/null
+++ b/test/unit/extensions/omci/mock/mock_onu_handler.py
@@ -0,0 +1,79 @@
+#
+# 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 mock_adapter_agent import MockProxyAddress, MockDevice
+from pyvoltha.adapters.extensions.omci.omci_cc import *
+from pyvoltha.adapters.extensions.omci.omci_entities import entity_id_to_class_map
+
+
+class MockOnuHandler(MockDevice):
+ """
+ Minimal class to handle ONU needs in OpenOMCI testing
+
+ So that we do not have to duplicate the IAdapter functionality, just
+ the handler, the OLT and ONU handlers are derived from a mock Device
+ base class so that we can access the _devices map and get either a
+ Device to play with (like the real thing) or the associated handler
+ """
+ def __init__(self, adapter_agent, parent_id, device_id, pon_id, onu_id):
+
+ self.proxy_address = MockProxyAddress(parent_id, pon_id, onu_id)
+ super(MockOnuHandler, self).__init__(device_id, self.proxy_address)
+
+ self.device_id = device_id
+ self.device = self
+ self._adapter_agent = adapter_agent
+
+ self.onu_mock = None
+ self.omci_cc = OMCI_CC(adapter_agent, device_id, me_map=entity_id_to_class_map)
+
+ # Items that you can change to perform various test failures
+
+ self._enabled = True
+
+ def tearDown(self):
+ """Test case cleanup"""
+ if self.onu_mock is not None:
+ self.onu_mock.tearDown()
+ self.onu_mock = None
+
+ @property
+ def enabled(self):
+ return self._enabled
+
+ @enabled.setter
+ def enabled(self, value):
+ if self._enabled != value:
+ self._enabled = value
+ olt = self._adapter_agent.get_device(self.proxy_address.device_id)
+ if olt is not None and self.proxy_address.channel_id in olt.enabled_pons:
+ if self._enabled:
+ olt.activated_onus.add(self.serial_number)
+ else:
+ olt.activated_onus.discard(self.serial_number)
+
+ # Begin minimal set of needed IAdapter interfaces
+
+ # TODO: Implement minimal functionality
+
+ def send_proxied_message(self, proxy_address, msg):
+ assert False, 'OpenOMCI will implement this for the MOCK ONU'
+
+ def receive_proxied_message(self, _, msg):
+ # Rx of OMCI message from MOCK OLT
+
+ if self.omci_cc is not None and self.enabled:
+ self.omci_cc.receive_message(msg.decode('hex'))
diff --git a/test/unit/extensions/omci/mock/mock_task.py b/test/unit/extensions/omci/mock/mock_task.py
new file mode 100644
index 0000000..aad0c60
--- /dev/null
+++ b/test/unit/extensions/omci/mock/mock_task.py
@@ -0,0 +1,94 @@
+#
+# 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 pyvoltha.adapters.extensions.omci.tasks.task import Task
+from pyvoltha.common.utils.asleep import asleep
+from twisted.internet.defer import inlineCallbacks, failure
+from twisted.internet import reactor
+
+
+class SimpleTask(Task):
+ def __init__(self, omci_agent, device_id,
+ exclusive=True,
+ success=True,
+ delay=0,
+ value=None,
+ priority=Task.DEFAULT_PRIORITY,
+ watchdog_timeout=Task.DEFAULT_WATCHDOG_SECS):
+ """
+ Class initialization
+
+ :param omci_agent: (OmciAdapterAgent) OMCI Adapter agent
+ :param device_id: (str) ONU Device ID
+ :param exclusive: (bool) True if the task should run by itself
+ :param success: (bool) True if the task should complete successfully
+ :param delay: (int/float) Time it takes the task to complete
+ :param priority (int) Priority of the task
+ :param watchdog_timeout (int or float) Watchdog timeout after task start
+ :param value: (various) The value (string, int, ...) to return if successful
+ or an Exception to send to the errBack if 'success'
+ is False
+ """
+ super(SimpleTask, self).__init__('Simple Mock Task',
+ omci_agent,
+ device_id,
+ exclusive=exclusive,
+ priority=priority,
+ watchdog_timeout=watchdog_timeout)
+ self._delay = delay
+ self._success = success
+ self._value = value
+ self._local_deferred = None
+
+ def cancel_deferred(self):
+ super(SimpleTask, self).cancel_deferred()
+
+ d, self._local_deferred = self._local_deferred, None
+ try:
+ if d is not None and not d.called:
+ d.cancel()
+ except:
+ pass
+
+ def start(self):
+ """
+ Start MIB Synchronization tasks
+ """
+ super(SimpleTask, self).start()
+ self._local_deferred = reactor.callLater(0, self.perform_task)
+
+ def stop(self):
+ """
+ Shutdown MIB Synchronization tasks
+ """
+ self.cancel_deferred()
+ super(SimpleTask, self).stop()
+
+ @inlineCallbacks
+ def perform_task(self):
+ """
+ Get the 'mib_data_sync' attribute of the ONU
+ """
+ try:
+ if self._delay > 0:
+ yield asleep(self._delay)
+
+ if self._success:
+ self.deferred.callback(self._value)
+
+ self.deferred.errback(failure.Failure(self._value))
+
+ except Exception as e:
+ self.deferred.errback(failure.Failure(e))
diff --git a/test/unit/extensions/omci/test_image_agent.py b/test/unit/extensions/omci/test_image_agent.py
new file mode 100644
index 0000000..a801d0a
--- /dev/null
+++ b/test/unit/extensions/omci/test_image_agent.py
@@ -0,0 +1,295 @@
+#
+# 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.
+#
+
+import binascii
+import structlog
+from unittest import TestCase, TestSuite, skip
+from pyvoltha.adapters.extensions.omci.openomci_agent import OpenOMCIAgent
+from pyvoltha.adapters.extensions.omci.omci_entities import SoftwareImage
+from pyvoltha.adapters.extensions.omci.omci_frame import OmciFrame
+from pyvoltha.adapters.extensions.omci.omci_messages import \
+ OmciStartSoftwareDownload, OmciStartSoftwareDownloadResponse, \
+ OmciEndSoftwareDownload, OmciEndSoftwareDownloadResponse, \
+ OmciDownloadSection, OmciDownloadSectionLast, OmciDownloadSectionResponse, \
+ OmciActivateImage, OmciActivateImageResponse, \
+ OmciCommitImage, OmciCommitImageResponse
+from pyvoltha.protos.voltha_pb2 import ImageDownload
+from pyvoltha.protos.device_pb2 import Device
+
+from tests.utests.voltha.extensions.omci.mock.mock_adapter_agent import MockAdapterAgent, MockCore
+from twisted.internet import reactor
+from twisted.internet.defer import Deferred
+from twisted.internet.epollreactor import EPollReactor
+from time import sleep
+
+class TestOmciDownload(TestCase):
+ # def __init__(self, device_id='1', image_file='/home/lcui/work/tmp/v_change.txt', **kwargs):
+ # self.device_id = device_id
+ # self.image_file = image_file
+ # super(TestOmciDownload, self).__init__(**kwargs)
+ sw_dwld_resp = {
+ 'tid': '0001',
+ 'mid': '33',
+ 'did': '0A',
+ 'entity_class': '0007',
+ 'entity_id' : '0000',
+ 'reason' : '0000',
+ 'window_size' : '001F',
+ 'inst_num' : '0001',
+ 'inst_id' : '0000',
+ 'trailer' : '00000028',
+ 'mic' : '00000000'
+ }
+
+ # sw_dwld_resp = '0001330A000700000000001f010000'
+
+ ### Test Functions ###
+ def sim_receive_start_sw_download_resp(self, tid, eid, r=0):
+ msg = OmciFrame(
+ transaction_id=tid,
+ message_type=OmciStartSoftwareDownloadResponse.message_id,
+ omci_message=OmciStartSoftwareDownloadResponse(
+ entity_class=0x7,
+ entity_id=eid,
+ result=r,
+ window_size=0x1F,
+ image_number=1,
+ instance_id=eid
+ )
+ )
+ self.device.omci_cc.receive_message(msg)
+
+ def sim_receive_download_section_resp(self, tid, eid, section, r=0):
+ msg = OmciFrame(
+ transaction_id=tid,
+ message_type=OmciDownloadSectionResponse.message_id,
+ omci_message=OmciDownloadSectionResponse(
+ entity_class=0x7,
+ entity_id=eid,
+ result=r,
+ section_number=section
+ )
+ )
+ self.device.omci_cc.receive_message(msg)
+
+ def sim_receive_end_sw_download_resp(self, tid, eid, r=0):
+ msg = OmciFrame(
+ transaction_id=tid,
+ message_type=OmciEndSoftwareDownloadResponse.message_id,
+ omci_message=OmciEndSoftwareDownloadResponse(
+ entity_class=0x7,
+ entity_id=eid,
+ result=r,
+ image_number=0x1,
+ instance_id=eid,
+ result0=0x0
+ )
+ )
+ self.device.omci_cc.receive_message(msg)
+
+ def sim_receive_activate_image_resp(self, tid, eid, r=0):
+ msg = OmciFrame(
+ transaction_id=tid,
+ message_type=OmciActivateImageResponse.message_id,
+ omci_message=OmciActivateImageResponse(
+ entity_class=0x7,
+ entity_id=eid,
+ result = r
+ ))
+ self.device.omci_cc.receive_message(msg)
+
+ def sim_receive_commit_image_resp(self, tid, eid, r=0):
+ msg = OmciFrame(
+ transaction_id=tid,
+ message_type=OmciCommitImageResponse.message_id,
+ omci_message=OmciCommitImageResponse(
+ entity_class=0x7,
+ entity_id=eid,
+ result = r
+ ))
+ self.device.omci_cc.receive_message(msg)
+
+ def cb_after_send_omci(self, msg):
+ self.log.debug("cb_after_send_omci")
+ dmsg = OmciFrame(binascii.unhexlify(msg))
+ tid = dmsg.fields['transaction_id']
+ mid = dmsg.fields['message_type']
+ dmsg_body = dmsg.fields['omci_message']
+ eid = dmsg_body.fields['entity_id']
+
+ # print("%X" % dmsg.fields['transaction_id'])
+ # print("%X" % dmsg.fields['message_type'])
+ # print("%X" % OmciActivateImage.message_id)
+ # print("%X" % dmsg_body.fields['entity_id'])
+
+ if mid == OmciStartSoftwareDownload.message_id:
+ self.log.debug("response start download")
+ self.reactor.callLater(0, self.sim_receive_start_sw_download_resp, tid, eid)
+ elif mid == OmciEndSoftwareDownload.message_id:
+ self.log.debug("response end download")
+ if self._end_image_busy_try > 0:
+ self.reactor.callLater(0, self.sim_receive_end_sw_download_resp, tid, eid, r=6)
+ self._end_image_busy_try -= 1
+ else:
+ self.reactor.callLater(0, self.sim_receive_end_sw_download_resp, tid, eid)
+ elif mid == OmciDownloadSection.message_id:
+ self.log.debug("receive download section, not respond")
+ elif mid == OmciDownloadSectionLast.message_id:
+ self.log.debug("response download last section")
+ self.reactor.callLater(0, self.sim_receive_download_section_resp, tid, eid,
+ section=dmsg_body.fields["section_number"])
+ elif mid == OmciActivateImage.message_id:
+ self.log.debug("response activate image")
+ if self._act_image_busy_try > 0:
+ self.reactor.callLater(0, self.sim_receive_activate_image_resp, tid, eid, r=6)
+ self._act_image_busy_try -= 1
+ else:
+ self.reactor.callLater(0, self.sim_receive_activate_image_resp, tid, eid)
+ self.reactor.callLater(2, self.device.image_agent.onu_bootup)
+ elif mid == OmciCommitImage.message_id:
+ self.log.debug("response commit image")
+ self.reactor.callLater(0, self.sim_receive_commit_image_resp, tid, eid)
+ else:
+ self.log.debug("Unsupported message type", message_type=mid)
+
+ self.defer = Deferred()
+ self.defer.addCallback(self.cb_after_send_omci)
+ self.adapter_agent.send_omci_defer = self.defer
+
+ def setUp(self):
+ self.log = structlog.get_logger()
+ self.log.debug("do setup")
+ self.device_id = '1'
+ self._image_dnld = ImageDownload()
+ self._image_dnld.id = '1'
+ self._image_dnld.name = 'switchd_1012'
+ # self._image_dnld.name = 'xgsont_4.4.4.006.img'
+ self._image_dnld.url = 'http://192.168.100.222:9090/load/4.4.4.006.img'
+ self._image_dnld.crc = 0
+ self._image_dnld.local_dir = '/home/lcui/work/tmp'
+ self._image_dnld.state = ImageDownload.DOWNLOAD_SUCCEEDED # ImageDownload.DOWNLOAD_UNKNOWN
+ self._end_image_busy_try = 2
+ self._act_image_busy_try = 0
+ # self.image_file = '/home/lcui/work/tmp/v_change.txt'
+ self.reactor = EPollReactor()
+ self.defer = Deferred()
+ self.adapter_agent = MockAdapterAgent(self.defer)
+ self.defer.addCallback(self.cb_after_send_omci)
+ pb2_dev = Device(id='1')
+ self.adapter_agent.add_device(pb2_dev)
+ self.core = self.adapter_agent.core
+ self.omci_agent = OpenOMCIAgent(self.core, clock=self.reactor)
+ self.device = self.omci_agent.add_device(self.device_id, self.adapter_agent)
+ self.omci_agent.start()
+ self.omci_agent.database.add('1')
+ self.omci_agent.database.set('1', SoftwareImage.class_id, 0, {"is_committed": 1, "is_active": 1, "is_valid": 1})
+ self.omci_agent.database.set('1', SoftwareImage.class_id, 1, {"is_committed": 0, "is_active": 0, "is_valid": 1})
+
+ def tearDown(self):
+ self.log.debug("Test is Done")
+ self.omci_agent.database.remove('1')
+ self.device = None
+
+ def stop(self):
+ self.reactor.stop()
+ self.log.debug("stopped");
+
+ def get_omci_msg(self, *args, **kargs):
+ m = ''
+ for s in args:
+ m += s
+ m = m.ljust(80, '0')
+ return m + kargs['trailer'] + kargs['mic']
+
+ def sim_receive_sw_download_resp2(self):
+ r = self.get_omci_msg(self.sw_dwld_resp['tid'], self.sw_dwld_resp['mid'],
+ self.sw_dwld_resp['did'], self.sw_dwld_resp['entity_class'],
+ self.sw_dwld_resp['entity_id'], self.sw_dwld_resp['reason'],
+ self.sw_dwld_resp['window_size'], self.sw_dwld_resp['inst_num'], self.sw_dwld_resp['inst_id'],
+ trailer=self.sw_dwld_resp['trailer'], mic=self.sw_dwld_resp['mic'])
+ data = binascii.unhexlify(r)
+ #msg = OmciFrame(data)
+ #print(msg.fields['transaction_id'])
+ #print(msg.fields['omci'])
+ self.device.omci_cc.receive_message(data)
+
+ def sw_action_success(self, instance_id, device_id):
+ self.log.debug("Action Success", device_id=device_id, entity_id=instance_id)
+ self.reactor.callLater(0, self.onu_do_activate)
+
+ def sw_action2_success(self, instance_id, device_id):
+ self.log.debug("Action2 Success", device_id=device_id, entity_id=instance_id)
+
+ def sw_action_fail(self, fail, device_id):
+ self.log.debug("Finally Failed", device_id=device_id)
+ self.log.debug(fail)
+
+ # def test_onu_do_activate(self):
+ def onu_do_activate(self):
+ self.log.debug("do test_onu_do_activate")
+ self.defer = self.device.do_onu_image_activate(self._image_dnld.name)
+ self.defer.addCallbacks(self.sw_action2_success, self.sw_action_fail, callbackArgs=(self.device_id,), errbackArgs=(self.device_id,))
+ self.reactor.callLater(100, self.stop)
+ # self.reactor.run()
+
+ @skip("for Jenkins Verification")
+ def test_onu_do_software_upgrade(self):
+ self.log.debug("do test_onu_do_software_upgrade", download=self._image_dnld)
+ dr = self.omci_agent.database.query('1', SoftwareImage.class_id, 0, "is_committed")
+ self.defer = self.device.do_onu_software_download(self._image_dnld)
+ self.defer.addCallbacks(self.sw_action_success, self.sw_action_fail, callbackArgs=(self.device_id,), errbackArgs=(self.device_id,))
+ # self.defer.addCallbacks(self.sw_action_success, self.sw_action_fail) #, errbackArgs=(self.device_id,))
+ # self.reactor.callLater(1, self.sim_receive_start_sw_download_resp)
+ # self.reactor.callLater(12, self.stop)
+ self.reactor.run()
+
+ @skip("Not used")
+ def test_omci_message(self):
+ self.log.debug("do test_omci_message")
+ r = self.get_omci_msg(self.sw_dwld_resp['tid'], self.sw_dwld_resp['mid'],
+ self.sw_dwld_resp['did'], self.sw_dwld_resp['entity_class'],
+ self.sw_dwld_resp['entity_id'], self.sw_dwld_resp['reason'],
+ self.sw_dwld_resp['window_size'], self.sw_dwld_resp['inst_num'], self.sw_dwld_resp['inst_id'],
+ trailer=self.sw_dwld_resp['trailer'], mic=self.sw_dwld_resp['mic'])
+ data = binascii.unhexlify(r)
+ msg = OmciFrame(data)
+ self.log.debug(binascii.hexlify(str(msg)))
+ # print("%04X" % msg.fields['transaction_id'])
+ # print("%02X" % msg.fields['message_type'])
+ # print("%02X" % msg.fields['omci'])
+ # print("%X" % msg.fields['omci_message'])
+
+ @skip("Not used")
+ def test_omci_message2(self):
+ self.log.debug("do test_omci_message2")
+ msg = OmciFrame(
+ transaction_id=0x0001,
+ message_type=OmciStartSoftwareDownloadResponse.message_id,
+ omci_message=OmciStartSoftwareDownloadResponse(
+ entity_class=0x7,
+ entity_id=0x0,
+ result=0x0,
+ window_size=0x1F,
+ image_number=1,
+ instance_id=0
+ )
+ )
+ self.log.debug(binascii.hexlify(str(msg)))
+
+this_suite = TestSuite()
+# this_suite.addTest(TestOmciDownload('test_onu_do_software_upgrade'))
+# this_suite.addTest(TestOmciDownload('test_onu_do_activate'))
+
diff --git a/test/unit/extensions/omci/test_me_frame.py b/test/unit/extensions/omci/test_me_frame.py
new file mode 100644
index 0000000..74e2b78
--- /dev/null
+++ b/test/unit/extensions/omci/test_me_frame.py
@@ -0,0 +1,317 @@
+#
+# 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 TestCase, main
+from nose.tools import assert_raises
+from pyvoltha.adapters.extensions.omci.me_frame import *
+from pyvoltha.adapters.extensions.omci.omci_me import *
+from pyvoltha.adapters.extensions.omci.omci import *
+
+
+def hexify(buffer):
+ """Return a hexadecimal string encoding of input buffer"""
+ return ''.join('%02x' % ord(c) for c in buffer)
+
+
+class TestSelectMeFrameGeneration(TestCase):
+
+ def assertGeneratedFrameEquals(self, frame, ref):
+ assert isinstance(frame, Packet)
+ serialized_hexified_frame = hexify(str(frame)).upper()
+ ref = ref.upper()
+ if serialized_hexified_frame != ref:
+ self.fail('Mismatch:\nReference:\n{}\nGenerated (bad):\n{}'.format(
+ ref, serialized_hexified_frame
+ ))
+
+ def test_mib_reset_message_serialization(self):
+ ref = '00004F0A000200000000000000000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OntDataFrame().mib_reset()
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_gal_ethernet_profile(self):
+ ref = '0000440A011000010030000000000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = GalEthernetProfileFrame(1, max_gem_payload_size=48).create()
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_tcont_1(self):
+ ref = '0000480A010680008000040000000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+
+ frame = TcontFrame(0x8000, alloc_id=0x400).set()
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_tcont_2(self):
+ ref = '0000480A010680018000040100000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+
+ frame = TcontFrame(0x8001, alloc_id=0x401).set()
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_8021p_mapper_service_profile(self):
+ ref = '0000440A00828000ffffffffffffffff' \
+ 'ffffffffffffffffffff000000000000' \
+ '000000000000000000000028'
+ frame = Ieee8021pMapperServiceProfileFrame(0x8000).create()
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_mac_bridge_service_profile(self):
+ ref = '0000440A002D02010001008000140002' \
+ '000f0001000000000000000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ spanning_tree_ind=False,
+ learning_ind=True,
+ priority=0x8000,
+ max_age=20 * 256,
+ hello_time=2 * 256,
+ forward_delay=15 * 256,
+ unknown_mac_address_discard=True
+ )
+ frame = MacBridgeServiceProfileFrame(0x201, attributes=data).create()
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_gem_port_network_ctp(self):
+ ref = '0000440A010C01000400800003010000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+
+ data = dict(
+ port_id=0x400,
+ tcont_pointer=0x8000,
+ direction=3,
+ traffic_management_pointer_upstream=0x100
+ )
+ frame = GemPortNetworkCtpFrame(0x100, attributes=data).create()
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ # Also test direction as a string parameter
+ frame = GemPortNetworkCtpFrame(0x100, port_id=0x400,
+ tcont_id=0x8000,
+ direction='bi-directional',
+ upstream_tm=0x100).create()
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_gem_inteworking_tp(self):
+ ref = '0000440A010A80010100058000000000' \
+ '01000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = GemInterworkingTpFrame(0x8001,
+ gem_port_network_ctp_pointer=0x100,
+ interworking_option=5,
+ service_profile_pointer=0x8000,
+ interworking_tp_pointer=0x0,
+ gal_profile_pointer=0x1).create()
+
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_8021p_mapper_service_profile(self):
+ ref = '0000480A008280007F80800100000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ ptrs = [0x8001, 0, 0, 0, 0, 0, 0, 0]
+ frame = Ieee8021pMapperServiceProfileFrame(0x8000,
+ interwork_tp_pointers=ptrs).set()
+
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ ptrs = [0x8001, 0]
+ frame = Ieee8021pMapperServiceProfileFrame(0x8000,
+ interwork_tp_pointers=ptrs).set()
+
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_mac_bridge_port_configuration_data(self):
+ ref = '0000440A002F21010201020380000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+
+ frame = MacBridgePortConfigurationDataFrame(0x2101,
+ bridge_id_pointer=0x201,
+ port_num=2,
+ tp_type=3,
+ tp_pointer=0x8000).create()
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_vlan_tagging_filter_data(self):
+ ref = '0000440A005421010400000000000000' \
+ '00000000000000000000000000000000' \
+ '100100000000000000000028'
+ frame = VlanTaggingFilterDataFrame(0x2101,
+ vlan_tcis=[0x400],
+ forward_operation=0x10).create()
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_extended_vlan_tagging_operation_configuration_data(self):
+ ref = '0000440A00AB02020A04010000000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ association_type=10,
+ associated_me_pointer=0x401
+ )
+ frame = \
+ ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
+ attributes=data)\
+ .create()
+
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_extended_vlan_tagging_operation_configuration_data(self):
+ ref = '0000480A00AB02023800810081000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ input_tpid=0x8100,
+ output_tpid=0x8100,
+ downstream_mode=0, # inverse of upstream
+ )
+ frame = \
+ ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
+ attributes=data)\
+ .set()
+
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_extended_vlan_tagging_1(self):
+ ref = '0000480A00AB02020400f00000008200' \
+ '5000402f000000082004000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ received_frame_vlan_tagging_operation_table=\
+ 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
+ )
+ )
+ frame = \
+ ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
+ attributes=data)\
+ .set()
+
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_extended_vlan_tagging_2(self):
+ ref = '0000480A00AB02020400F00000008200' \
+ 'd000402f00000008200c000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ received_frame_vlan_tagging_operation_table=
+ VlanTaggingOperation(
+ filter_outer_priority=15,
+ filter_inner_priority=8,
+ filter_inner_vid=1025,
+ 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=1025,
+ treatment_inner_tpid_de=4
+ )
+ )
+
+ frame = \
+ ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
+ attributes=data)\
+ .set()
+
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_mac_bridge_port_configuration_data2(self):
+ ref = '0000440A002F02010201010b04010000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ bridge_id_pointer=0x201,
+ encapsulation_methods=0,
+ port_num=1,
+ port_priority=0,
+ port_path_cost=0,
+ port_spanning_tree_in=0,
+ lan_fcs_ind=0,
+ tp_type=11,
+ tp_pointer=0x401,
+ mac_learning_depth=0
+ )
+ frame = MacBridgePortConfigurationDataFrame(0x201,
+ attributes=data).create()
+
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_pptp_ethernet_uni_frame(self):
+ ref = '0000480A000B020109000005EE000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ administrative_state=0, # 0 - Unlock
+ max_frame_size=1518 # two-octet field
+ )
+ frame = PptpEthernetUniFrame(0x201,
+ attributes=data).set()
+
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_constraint_errors(self):
+ self.assertTrue(True) # TODO Also test some attribute constraint failures
+
+ def test_mib_upload_next(self):
+ # Test for VOL-649 error. SCAPY was only originally coded for a 'get'
+ # action (8-bit MIB Data Sync value) but MIB Upload Next commands have
+ # a 16-bit field.
+ #
+ # 255 and less always worked
+ OntDataFrame(sequence_number=0).mib_upload_next()
+ OntDataFrame(sequence_number=255).mib_upload_next()
+ # But not 256+
+ OntDataFrame(sequence_number=256).mib_upload_next()
+ OntDataFrame(sequence_number=1000).mib_upload_next()
+ OntDataFrame(sequence_number=0xFFFE).mib_upload_next()
+
+ # Also test the optional arguments for the other actions
+ OntDataFrame().get()
+ OntDataFrame(mib_data_sync=4).set()
+ OntDataFrame().mib_reset()
+ OntDataFrame().mib_upload()
+ # OntDataFrame(ignore_arc=True).get_all_alarms() Not yet coded
+ # OntDataFrame(ignore_arc=False).get_all_alarms() Not yet coded
+
+ # Range/type checks
+ assert_raises(ValueError, OntDataFrame, mib_data_sync=-1)
+ assert_raises(ValueError, OntDataFrame, mib_data_sync=256)
+ assert_raises(ValueError, OntDataFrame, sequence_number=-1)
+ assert_raises(ValueError, OntDataFrame, sequence_number=0x10000)
+ assert_raises(TypeError, OntDataFrame, ignore_arc=123)
+
+
+if __name__ == '__main__':
+ main()
+
diff --git a/test/unit/extensions/omci/test_mib_db_dict.py b/test/unit/extensions/omci/test_mib_db_dict.py
new file mode 100644
index 0000000..2f7f26d
--- /dev/null
+++ b/test/unit/extensions/omci/test_mib_db_dict.py
@@ -0,0 +1,521 @@
+#
+# 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.database.mib_db_dict import *
+from pyvoltha.adapters.extensions.omci.database.mib_db_api import MODIFIED_KEY, CREATED_KEY,\
+ DEVICE_ID_KEY, MDS_KEY, LAST_SYNC_KEY
+from mock.mock_adapter_agent import MockAdapterAgent, MockDevice
+from nose.tools import raises, assert_raises
+import time
+
+_DEVICE_ID = 'br-549'
+
+
+class TestOmciMibDbDict(TestCase):
+
+ def setUp(self):
+ self.adapter_agent = MockAdapterAgent()
+ self.adapter_agent.add_device(MockDevice(_DEVICE_ID)) # For Entity class lookups
+ self.db = MibDbVolatileDict(self.adapter_agent)
+
+ def tearDown(self):
+ self.db.stop()
+
+ def test_start_stop(self):
+ # Simple start stop
+ self.assertFalse(self.db.active)
+ self.db.start()
+ self.assertTrue(self.db.active)
+ self.db.stop()
+ self.assertFalse(self.db.active)
+
+ # Start after start still okay
+ self.db.start()
+ self.db.start()
+ self.assertTrue(self.db.active)
+
+ self.db.stop()
+ self.db.stop()
+ self.assertFalse(self.db.active)
+
+ @raises(DatabaseStateError)
+ def test_bad_state_add(self):
+ self.db.add(_DEVICE_ID)
+
+ @raises(DatabaseStateError)
+ def test_bad_state_remove(self):
+ self.db.remove(_DEVICE_ID)
+
+ @raises(DatabaseStateError)
+ def test_bad_state_query_1(self):
+ self.db.query(_DEVICE_ID, 0)
+
+ @raises(DatabaseStateError)
+ def test_bad_state_query_2(self):
+ self.db.query(_DEVICE_ID, 0, 0)
+
+ @raises(DatabaseStateError)
+ def test_bad_state_query_3(self):
+ self.db.query(_DEVICE_ID, 0, 0, 'test')
+
+ @raises(DatabaseStateError)
+ def test_bad_state_set(self):
+ self.db.set(_DEVICE_ID, 0, 0, {'test': 123})
+
+ @raises(DatabaseStateError)
+ def test_bad_state_delete(self):
+ self.db.delete(_DEVICE_ID, 0, 0)
+
+ @raises(KeyError)
+ def test_no_device_query(self):
+ self.db.start()
+ self.db.query(_DEVICE_ID)
+
+ def test_no_device_last_sync(self):
+ self.db.start()
+ # Returns None, not a KeyError
+ value = self.db.get_last_sync(_DEVICE_ID)
+ self.assertIsNone(value)
+
+ def test_no_device_mds(self):
+ self.db.start()
+ # Returns None, not a KeyError
+ value = self.db.get_mib_data_sync(_DEVICE_ID)
+ self.assertIsNone(value)
+
+ @raises(KeyError)
+ def test_no_device_save_last_sync(self):
+ self.db.start()
+ self.db.save_last_sync(_DEVICE_ID, datetime.utcnow())
+
+ @raises(KeyError)
+ def test_no_device_save_mds(self):
+ self.db.start()
+ self.db.save_mib_data_sync(_DEVICE_ID, 123)
+
+ def test_param_types(self):
+ self.db.start()
+ assert_raises(TypeError, self.db.add, 123)
+ assert_raises(TypeError, self.db.remove, 123)
+ assert_raises(TypeError, self.db.query, 123)
+
+ assert_raises(TypeError, self.db.get_mib_data_sync, 123)
+ assert_raises(TypeError, self.db.save_mib_data_sync, 123, 0)
+ assert_raises(TypeError, self.db.save_mib_data_sync, _DEVICE_ID, 'zero')
+
+ assert_raises(TypeError, self.db.get_last_sync, 123)
+ assert_raises(TypeError, self.db.save_last_sync, 123, datetime.utcnow())
+ assert_raises(TypeError, self.db.save_last_sync, _DEVICE_ID, 'bad-date')
+
+ assert_raises(TypeError, self.db.set, 123, 0, 0, {'test': 0})
+ assert_raises(TypeError, self.db.set, None, 0, 0, {'test': 0})
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, None, 0, {'test': 0})
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, 0, None, {'test': 0})
+ assert_raises(TypeError, self.db.set, _DEVICE_ID, 0, 0, None)
+ assert_raises(TypeError, self.db.set, _DEVICE_ID, 0, 0, 'not-a-dict')
+
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, -1, 0, {'test': 0})
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, 0x10000, 0, {'test': 0})
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, 0, -1, {'test': 0})
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, 0, 0x10000, {'test': 0})
+
+ assert_raises(TypeError, self.db.delete, 123, 0, 0)
+ assert_raises(ValueError, self.db.delete, _DEVICE_ID, -1, 0)
+ assert_raises(ValueError, self.db.delete, _DEVICE_ID, 0x10000, 0)
+ assert_raises(ValueError, self.db.delete, _DEVICE_ID, 0, -1)
+ assert_raises(ValueError, self.db.delete, _DEVICE_ID, 0, 0x10000)
+
+ def test_add_remove_device(self):
+ self.db.start()
+
+ # Remove of non-existent device is not an error
+ assert_raises(KeyError, self.db.query, _DEVICE_ID)
+ self.db.remove(_DEVICE_ID)
+
+ start_time = datetime.utcnow()
+ self.db.add(_DEVICE_ID)
+ dev_data = self.db.query(_DEVICE_ID)
+ end_time = datetime.utcnow()
+
+ self.assertEqual(dev_data[DEVICE_ID_KEY], _DEVICE_ID)
+ self.assertEquals(dev_data[MDS_KEY], 0)
+ self.assertIsNone(dev_data[LAST_SYNC_KEY])
+ self.assertEqual(dev_data[VERSION_KEY], MibDbVolatileDict.CURRENT_VERSION)
+
+ # Remove it
+ self.db.remove(_DEVICE_ID)
+ assert_raises(KeyError, self.db.query, _DEVICE_ID)
+
+ # Remove of non-existant dev okay
+ self.db.remove(_DEVICE_ID +'abcd')
+
+ # Overwrite tests
+ self.db.add(_DEVICE_ID)
+ assert_raises(KeyError, self.db.add, _DEVICE_ID)
+ self.db.add(_DEVICE_ID, overwrite=True) # This is okay
+
+ def test_mib_data_sync(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+ self.assertEquals(self.db.get_mib_data_sync(_DEVICE_ID), 0)
+
+ self.db.save_mib_data_sync(_DEVICE_ID, 100)
+ self.assertEqual(self.db.get_mib_data_sync(_DEVICE_ID), 100)
+
+ assert_raises(ValueError, self.db.save_mib_data_sync, _DEVICE_ID, -1)
+ assert_raises(ValueError, self.db.save_mib_data_sync, _DEVICE_ID, 256)
+
+ def test_last_sync(self):
+ self.db.start()
+ self.assertIsNone(self.db.get_last_sync(_DEVICE_ID))
+
+ self.db.add(_DEVICE_ID)
+ self.assertIsNone(self.db.get_last_sync(_DEVICE_ID))
+
+ now = datetime.utcnow()
+
+ self.db.save_last_sync(_DEVICE_ID, now)
+ self.assertEqual(self.db.get_last_sync(_DEVICE_ID), now)
+
+ assert_raises(TypeError, self.db.save_last_sync, _DEVICE_ID, 'hello')
+
+ def test_set_and_query(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID) # Base device DB created here
+ time.sleep(0.1)
+
+ class_id = OntG.class_id
+ inst_id = 0
+ attributes = {'vendor_id': 'ABCD'}
+
+ start_time = datetime.utcnow()
+ set_occurred = self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ self.assertTrue(set_occurred)
+ end_time = datetime.utcnow()
+
+ dev_data = self.db.query(_DEVICE_ID)
+ self.assertEqual(dev_data[DEVICE_ID_KEY], _DEVICE_ID)
+
+ dev_classes = [v for k, v in dev_data.items() if isinstance(k, int)]
+
+ self.assertEqual(len(dev_classes), 1)
+ class_data = dev_classes[0]
+
+ self.assertEqual(class_data[CLASS_ID_KEY], class_id)
+
+ class_insts = [v for k, v in class_data.items() if isinstance(k, int)]
+
+ self.assertEqual(len(class_insts), 1)
+ inst_data = class_insts[0]
+
+ self.assertEqual(inst_data[INSTANCE_ID_KEY], inst_id)
+ self.assertGreaterEqual(inst_data[MODIFIED_KEY], start_time)
+ self.assertLessEqual(inst_data[MODIFIED_KEY], end_time)
+ self.assertLessEqual(inst_data[CREATED_KEY], inst_data[MODIFIED_KEY])
+
+ inst_attributes = inst_data[ATTRIBUTES_KEY]
+ self.assertEqual(len(inst_attributes), 1)
+
+ self.assertTrue('vendor_id' in inst_attributes)
+ self.assertEqual(inst_attributes['vendor_id'], attributes['vendor_id'])
+
+ ########################################
+ # Query with device and class. Should be same as from full device query
+ cls_2_data = self.db.query(_DEVICE_ID, class_id)
+
+ self.assertEqual(class_data[CLASS_ID_KEY], cls_2_data[CLASS_ID_KEY])
+
+ cl2_insts = {k:v for k, v in cls_2_data.items() if isinstance(k, int)}
+ self.assertEqual(len(cl2_insts), len(class_insts))
+
+ # Bad class id query
+ cls_no_data = self.db.query(_DEVICE_ID, class_id + 1)
+ self.assertTrue(isinstance(cls_no_data, dict))
+ self.assertEqual(len(cls_no_data), 0)
+
+ ########################################
+ # Query with device, class, instance
+ inst_2_data = self.db.query(_DEVICE_ID, class_id, inst_id)
+
+ self.assertEqual(inst_data[INSTANCE_ID_KEY], inst_2_data[INSTANCE_ID_KEY])
+ self.assertEqual(inst_data[MODIFIED_KEY], inst_2_data[MODIFIED_KEY])
+ self.assertEqual(inst_data[CREATED_KEY], inst_2_data[CREATED_KEY])
+
+ inst2_attr = inst_2_data[ATTRIBUTES_KEY]
+ self.assertEqual(len(inst2_attr), len(inst_attributes))
+
+ # Bad instance id query
+ inst_no_data = self.db.query(_DEVICE_ID, class_id, inst_id + 100)
+ self.assertTrue(isinstance(inst_no_data, dict))
+ self.assertEqual(len(inst_no_data), 0)
+
+ ########################################
+ # Attribute queries
+ attr_2_data = self.db.query(_DEVICE_ID, class_id, inst_id, 'vendor_id')
+ self.assertEqual(attr_2_data['vendor_id'], attributes['vendor_id'])
+
+ attr_3_data = self.db.query(_DEVICE_ID, class_id, inst_id, ['vendor_id'])
+ self.assertEqual(attr_3_data['vendor_id'], attributes['vendor_id'])
+
+ attr_4_data = self.db.query(_DEVICE_ID, class_id, inst_id, {'vendor_id'})
+ self.assertEqual(attr_4_data['vendor_id'], attributes['vendor_id'])
+
+ attr_no_data = self.db.query(_DEVICE_ID, class_id, inst_id, 'no_such_thing')
+ self.assertTrue(isinstance(attr_no_data, dict))
+ self.assertEqual(len(attr_no_data), 0)
+
+ # Set to same value does not change modified data. The modified is
+ # at the instance level
+
+ class_id = OntG.class_id
+ inst_id = 0
+ attributes = {'vendor_id': 'ABCD'}
+ set_occurred = self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ self.assertFalse(set_occurred)
+
+ inst_3_data = self.db.query(_DEVICE_ID, class_id, inst_id)
+ self.assertEqual(inst_data[MODIFIED_KEY], inst_3_data[MODIFIED_KEY])
+ self.assertEqual(inst_data[CREATED_KEY], inst_3_data[CREATED_KEY])
+
+ # But set to new value does
+ time.sleep(0.1)
+ attributes = {'vendor_id': 'WXYZ'}
+ set_occurred = self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ self.assertTrue(set_occurred)
+
+ inst_4_data = self.db.query(_DEVICE_ID, class_id, inst_id)
+ self.assertLess(inst_3_data[MODIFIED_KEY], inst_4_data[MODIFIED_KEY])
+ self.assertEqual(inst_3_data[CREATED_KEY], inst_4_data[CREATED_KEY])
+
+ def test_delete_instances(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+ create_time = datetime.utcnow()
+
+ class_id = GalEthernetProfile.class_id
+ inst_id_1 = 0x100
+ inst_id_2 = 0x200
+ attributes = {'max_gem_payload_size': 1500}
+
+ self.db.set(_DEVICE_ID, class_id, inst_id_1, attributes)
+ self.db.set(_DEVICE_ID, class_id, inst_id_2, attributes)
+ set_time = datetime.utcnow()
+ time.sleep(0.1)
+
+ dev_data = self.db.query(_DEVICE_ID)
+ cls_data = self.db.query(_DEVICE_ID, class_id)
+ inst_data = {k: v for k, v in cls_data.items() if isinstance(k, int)}
+ self.assertEqual(len(inst_data), 2)
+
+ self.assertLessEqual(dev_data[CREATED_KEY], create_time)
+ self.assertLessEqual(self.db.created, create_time)
+
+ # Delete one instance
+ time.sleep(0.1)
+ del_time = datetime.utcnow()
+ result = self.db.delete(_DEVICE_ID, class_id, inst_id_1)
+ self.assertTrue(result) # True returned if a del actually happened
+
+ dev_data = self.db.query(_DEVICE_ID)
+ cls_data = self.db.query(_DEVICE_ID, class_id)
+ inst_data = {k: v for k, v in cls_data.items() if isinstance(k, int)}
+ self.assertEqual(len(inst_data), 1)
+
+ self.assertLessEqual(dev_data[CREATED_KEY], create_time)
+ self.assertLessEqual(self.db.created, create_time)
+
+ # Delete remaining instance
+ time.sleep(0.1)
+ result = self.db.delete(_DEVICE_ID, class_id, inst_id_2)
+ self.assertTrue(result) # True returned if a del actually happened
+
+ dev_data = self.db.query(_DEVICE_ID)
+ cls_data = {k: v for k, v in dev_data.items() if isinstance(k, int)}
+ self.assertEqual(len(cls_data), 0)
+ self.assertLessEqual(dev_data[CREATED_KEY], create_time)
+
+ # Delete returns false if not instance
+ self.assertFalse(self.db.delete(_DEVICE_ID, class_id, inst_id_1))
+ self.assertFalse(self.db.delete(_DEVICE_ID, class_id, inst_id_2))
+
+ def test_on_mib_reset_listener(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+ time.sleep(0.1)
+
+ class_id = OntG.class_id
+ inst_id = 0
+ attributes = {'vendor_id': 'ABCD'}
+
+ set_time = datetime.utcnow()
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+
+ time.sleep(0.1)
+ self.db.on_mib_reset(_DEVICE_ID)
+
+ dev_data = self.db.query(_DEVICE_ID)
+ self.assertEqual(dev_data[DEVICE_ID_KEY], _DEVICE_ID)
+ self.assertLessEqual(dev_data[CREATED_KEY], set_time)
+ self.assertLessEqual(self.db.created, set_time)
+
+ self.assertFalse(any(isinstance(cls, int) for cls in dev_data.iterkeys()))
+
+ def test_str_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ class_id = OltG.class_id
+ inst_id = 0
+ attributes = {
+ 'olt_vendor_id': 'ABCD', # StrFixedLenField(4)
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], basestring) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_mac_address_ip_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ 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.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], basestring) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_byte_and_short_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ class_id = UniG.class_id
+ inst_id = 0
+ attributes = {
+ 'administrative_state': int(1), # ByteField
+ 'non_omci_management_identifier': int(12345) # IPField
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_int_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ class_id = PriorityQueueG.class_id
+ inst_id = 0
+ attributes = {
+ 'related_port': int(1234567) # IntField
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_long_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ class_id = PriorityQueueG.class_id
+ inst_id = 0
+ attributes = {
+ 'packet_drop_queue_thresholds': int(0x1234) # LongField
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_bit_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ class_id = OntG.class_id
+ inst_id = 0
+ attributes = {
+ 'extended_tc_layer_options': long(0x1234), # BitField(16)
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_list_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ 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.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_complex_json_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ 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.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ table_as_dict = json.loads(table_data.to_json())
+
+ self.assertTrue(all(isinstance(data['received_frame_vlan_tagging_operation_table'][0].fields[k],
+ type(attributes['received_frame_vlan_tagging_operation_table'].fields[k]))
+ for k in attributes['received_frame_vlan_tagging_operation_table'].fields.keys()))
+ self.assertTrue(all(data['received_frame_vlan_tagging_operation_table'][0].fields[k] ==
+ attributes['received_frame_vlan_tagging_operation_table'].fields[k]
+ for k in attributes['received_frame_vlan_tagging_operation_table'].fields.keys()))
+ self.assertTrue(all(data['received_frame_vlan_tagging_operation_table'][0].fields[k] == table_as_dict[k]
+ for k in table_as_dict.keys()))
+
+
+if __name__ == '__main__':
+ main()
diff --git a/test/unit/extensions/omci/test_mib_db_ext.py b/test/unit/extensions/omci/test_mib_db_ext.py
new file mode 100644
index 0000000..925e81f
--- /dev/null
+++ b/test/unit/extensions/omci/test_mib_db_ext.py
@@ -0,0 +1,537 @@
+#
+# 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.database.mib_db_ext import *
+from pyvoltha.adapters.extensions.omci.database.mib_db_api import MODIFIED_KEY, CREATED_KEY,\
+ DEVICE_ID_KEY, MDS_KEY, LAST_SYNC_KEY
+from pyvoltha.adapters.extensions.omci.omci_cc import UNKNOWN_CLASS_ATTRIBUTE_KEY
+from mock.mock_adapter_agent import MockAdapterAgent, MockDevice
+from nose.tools import raises, assert_raises
+import time
+
+_DEVICE_ID = 'br-549'
+
+
+class TestOmciMibDbExt(TestCase):
+
+ def setUp(self):
+ self.adapter_agent = MockAdapterAgent()
+ self.adapter_agent.add_device(MockDevice(_DEVICE_ID)) # For Entity class lookups
+ self.db = MibDbExternal(self.adapter_agent)
+
+ def tearDown(self):
+ self.db.stop()
+
+ def test_start_stop(self):
+ # Simple start stop
+ self.assertFalse(self.db.active)
+ self.db.start()
+ self.assertTrue(self.db.active)
+ self.db.stop()
+ self.assertFalse(self.db.active)
+
+ # Start after start still okay
+ self.db.start()
+ self.db.start()
+ self.assertTrue(self.db.active)
+
+ self.db.stop()
+ self.db.stop()
+ self.assertFalse(self.db.active)
+
+ @raises(DatabaseStateError)
+ def test_bad_state_add(self):
+ self.db.add(_DEVICE_ID)
+
+ @raises(DatabaseStateError)
+ def test_bad_state_remove(self):
+ self.db.remove(_DEVICE_ID)
+
+ @raises(DatabaseStateError)
+ def test_bad_state_query_1(self):
+ self.db.query(_DEVICE_ID, 0)
+
+ @raises(DatabaseStateError)
+ def test_bad_state_query_2(self):
+ self.db.query(_DEVICE_ID, 0, 0)
+
+ @raises(DatabaseStateError)
+ def test_bad_state_query_3(self):
+ self.db.query(_DEVICE_ID, 0, 0, 'test')
+
+ @raises(DatabaseStateError)
+ def test_bad_state_set(self):
+ self.db.set(_DEVICE_ID, 0, 0, {'test': 123})
+
+ @raises(DatabaseStateError)
+ def test_bad_state_delete(self):
+ self.db.delete(_DEVICE_ID, 0, 0)
+
+ @raises(KeyError)
+ def test_no_device_query(self):
+ self.db.start()
+ self.db.query(_DEVICE_ID)
+
+ def test_no_device_last_sync(self):
+ self.db.start()
+ # Returns None, not a KeyError
+ value = self.db.get_last_sync(_DEVICE_ID)
+ self.assertIsNone(value)
+
+ def test_no_device_mds(self):
+ self.db.start()
+ # Returns None, not a KeyError
+ value = self.db.get_mib_data_sync(_DEVICE_ID)
+ self.assertIsNone(value)
+
+ @raises(KeyError)
+ def test_no_device_save_last_sync(self):
+ self.db.start()
+ self.db.save_last_sync(_DEVICE_ID, datetime.utcnow())
+
+ @raises(KeyError)
+ def test_no_device_save_mds(self):
+ self.db.start()
+ self.db.save_mib_data_sync(_DEVICE_ID, 123)
+
+ def test_param_types(self):
+ self.db.start()
+ assert_raises(TypeError, self.db.add, 123)
+ assert_raises(TypeError, self.db.remove, 123)
+ assert_raises(TypeError, self.db.query, 123)
+
+ assert_raises(TypeError, self.db.get_mib_data_sync, 123)
+ assert_raises(TypeError, self.db.save_mib_data_sync, 123, 0)
+ assert_raises(TypeError, self.db.save_mib_data_sync, _DEVICE_ID, 'zero')
+
+ assert_raises(TypeError, self.db.get_last_sync, 123)
+ assert_raises(TypeError, self.db.save_last_sync, 123, datetime.utcnow())
+ assert_raises(TypeError, self.db.save_last_sync, _DEVICE_ID, 'bad-date')
+
+ assert_raises(TypeError, self.db.set, 123, 0, 0, {'test': 0})
+ assert_raises(TypeError, self.db.set, None, 0, 0, {'test': 0})
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, None, 0, {'test': 0})
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, 0, None, {'test': 0})
+ assert_raises(TypeError, self.db.set, _DEVICE_ID, 0, 0, None)
+ assert_raises(TypeError, self.db.set, _DEVICE_ID, 0, 0, 'not-a-dict')
+
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, -1, 0, {'test': 0})
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, 0x10000, 0, {'test': 0})
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, 0, -1, {'test': 0})
+ assert_raises(ValueError, self.db.set, _DEVICE_ID, 0, 0x10000, {'test': 0})
+
+ assert_raises(TypeError, self.db.delete, 123, 0, 0)
+ assert_raises(ValueError, self.db.delete, _DEVICE_ID, -1, 0)
+ assert_raises(ValueError, self.db.delete, _DEVICE_ID, 0x10000, 0)
+ assert_raises(ValueError, self.db.delete, _DEVICE_ID, 0, -1)
+ assert_raises(ValueError, self.db.delete, _DEVICE_ID, 0, 0x10000)
+
+ def test_add_remove_device(self):
+ self.db.start()
+
+ # Remove of non-existent device is not an error
+ assert_raises(KeyError, self.db.query, _DEVICE_ID)
+ self.db.remove(_DEVICE_ID)
+
+ start_time = datetime.utcnow()
+ self.db.add(_DEVICE_ID)
+ dev_data = self.db.query(_DEVICE_ID)
+
+ self.assertEqual(dev_data[DEVICE_ID_KEY], _DEVICE_ID)
+ self.assertEquals(dev_data[MDS_KEY], 0)
+ self.assertIsNone(dev_data[LAST_SYNC_KEY])
+ self.assertEqual(dev_data[VERSION_KEY], MibDbExternal.CURRENT_VERSION)
+
+ self.assertGreaterEqual(self.db.created, start_time)
+
+ # Remove it
+ self.db.remove(_DEVICE_ID)
+ assert_raises(KeyError, self.db.query, _DEVICE_ID)
+
+ # Remove of non-existant dev okay
+ self.db.remove(_DEVICE_ID +'abcd')
+
+ # Overwrite tests
+ self.db.add(_DEVICE_ID)
+ assert_raises(KeyError, self.db.add, _DEVICE_ID)
+ self.db.add(_DEVICE_ID, overwrite=True) # This is okay
+
+ def test_mib_data_sync(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+ self.assertEquals(self.db.get_mib_data_sync(_DEVICE_ID), 0)
+
+ self.db.save_mib_data_sync(_DEVICE_ID, 100)
+ self.assertEqual(self.db.get_mib_data_sync(_DEVICE_ID), 100)
+
+ assert_raises(ValueError, self.db.save_mib_data_sync, _DEVICE_ID, -1)
+ assert_raises(ValueError, self.db.save_mib_data_sync, _DEVICE_ID, 256)
+
+ def test_last_sync(self):
+ self.db.start()
+ self.assertIsNone(self.db.get_last_sync(_DEVICE_ID))
+
+ self.db.add(_DEVICE_ID)
+ self.assertIsNone(self.db.get_last_sync(_DEVICE_ID))
+
+ now = datetime.utcnow()
+
+ self.db.save_last_sync(_DEVICE_ID, now)
+ self.assertEqual(self.db.get_last_sync(_DEVICE_ID), now)
+
+ assert_raises(TypeError, self.db.save_last_sync, _DEVICE_ID, 'hello')
+
+ def test_set_and_query(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID) # Base device DB created here
+ time.sleep(0.1)
+
+ class_id = OntG.class_id
+ inst_id = 0
+ attributes = {'vendor_id': 'ABCD'}
+
+ start_time = datetime.utcnow()
+ set_occurred = self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ self.assertTrue(set_occurred)
+ end_time = datetime.utcnow()
+
+ dev_data = self.db.query(_DEVICE_ID)
+ self.assertEqual(dev_data[DEVICE_ID_KEY], _DEVICE_ID)
+
+ dev_classes = [v for k, v in dev_data.items() if isinstance(k, int)]
+
+ self.assertEqual(len(dev_classes), 1)
+ class_data = dev_classes[0]
+
+ self.assertEqual(class_data[CLASS_ID_KEY], class_id)
+
+ class_insts = [v for k, v in class_data.items() if isinstance(k, int)]
+
+ self.assertEqual(len(class_insts), 1)
+ inst_data = class_insts[0]
+
+ self.assertEqual(inst_data[INSTANCE_ID_KEY], inst_id)
+ self.assertGreaterEqual(inst_data[MODIFIED_KEY], start_time)
+ self.assertLessEqual(inst_data[MODIFIED_KEY], end_time)
+ self.assertLessEqual(inst_data[CREATED_KEY], inst_data[MODIFIED_KEY])
+
+ inst_attributes = inst_data[ATTRIBUTES_KEY]
+ self.assertEqual(len(inst_attributes), 1)
+
+ self.assertTrue('vendor_id' in inst_attributes)
+ self.assertEqual(inst_attributes['vendor_id'], attributes['vendor_id'])
+
+ ########################################
+ # Query with device and class. Should be same as from full device query
+ cls_2_data = self.db.query(_DEVICE_ID, class_id)
+
+ self.assertEqual(class_data[CLASS_ID_KEY], cls_2_data[CLASS_ID_KEY])
+
+ cl2_insts = {k:v for k, v in cls_2_data.items() if isinstance(k, int)}
+ self.assertEqual(len(cl2_insts), len(class_insts))
+
+ # Bad class id query
+ cls_no_data = self.db.query(_DEVICE_ID, class_id + 1)
+ self.assertTrue(isinstance(cls_no_data, dict))
+ self.assertEqual(len(cls_no_data), 0)
+
+ ########################################
+ # Query with device, class, instance
+ inst_2_data = self.db.query(_DEVICE_ID, class_id, inst_id)
+
+ self.assertEqual(inst_data[INSTANCE_ID_KEY], inst_2_data[INSTANCE_ID_KEY])
+ self.assertEqual(inst_data[MODIFIED_KEY], inst_2_data[MODIFIED_KEY])
+ self.assertEqual(inst_data[CREATED_KEY], inst_2_data[CREATED_KEY])
+
+ inst2_attr = inst_2_data[ATTRIBUTES_KEY]
+ self.assertEqual(len(inst2_attr), len(inst_attributes))
+
+ # Bad instance id query
+ inst_no_data = self.db.query(_DEVICE_ID, class_id, inst_id + 100)
+ self.assertTrue(isinstance(inst_no_data, dict))
+ self.assertEqual(len(inst_no_data), 0)
+
+ ########################################
+ # Attribute queries
+ attr_2_data = self.db.query(_DEVICE_ID, class_id, inst_id, 'vendor_id')
+ self.assertEqual(attr_2_data['vendor_id'], attributes['vendor_id'])
+
+ attr_3_data = self.db.query(_DEVICE_ID, class_id, inst_id, ['vendor_id'])
+ self.assertEqual(attr_3_data['vendor_id'], attributes['vendor_id'])
+
+ attr_4_data = self.db.query(_DEVICE_ID, class_id, inst_id, {'vendor_id'})
+ self.assertEqual(attr_4_data['vendor_id'], attributes['vendor_id'])
+
+ attr_no_data = self.db.query(_DEVICE_ID, class_id, inst_id, 'no_such_thing')
+ self.assertTrue(isinstance(attr_no_data, dict))
+ self.assertEqual(len(attr_no_data), 0)
+
+ # Set to same value does not change modified data. The modified is
+ # at the instance level
+
+ class_id = OntG.class_id
+ inst_id = 0
+ attributes = {'vendor_id': 'ABCD'}
+ set_occurred = self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ self.assertFalse(set_occurred)
+
+ inst_3_data = self.db.query(_DEVICE_ID, class_id, inst_id)
+ self.assertEqual(inst_data[MODIFIED_KEY], inst_3_data[MODIFIED_KEY])
+ self.assertEqual(inst_data[CREATED_KEY], inst_3_data[CREATED_KEY])
+
+ # But set to new value does
+ time.sleep(0.1)
+ attributes = {'vendor_id': 'WXYZ'}
+ set_occurred = self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ self.assertTrue(set_occurred)
+
+ inst_4_data = self.db.query(_DEVICE_ID, class_id, inst_id)
+ self.assertLess(inst_3_data[MODIFIED_KEY], inst_4_data[MODIFIED_KEY])
+ self.assertEqual(inst_3_data[CREATED_KEY], inst_4_data[CREATED_KEY])
+
+ def test_delete_instances(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+ create_time = datetime.utcnow()
+
+ class_id = GalEthernetProfile.class_id
+ inst_id_1 = 0x100
+ inst_id_2 = 0x200
+ attributes = {'max_gem_payload_size': 1500}
+
+ self.db.set(_DEVICE_ID, class_id, inst_id_1, attributes)
+ self.db.set(_DEVICE_ID, class_id, inst_id_2, attributes)
+ time.sleep(0.1)
+
+ dev_data = self.db.query(_DEVICE_ID)
+ cls_data = self.db.query(_DEVICE_ID, class_id)
+ inst_data = {k: v for k, v in cls_data.items() if isinstance(k, int)}
+ self.assertEqual(len(inst_data), 2)
+
+ self.assertLessEqual(dev_data[CREATED_KEY], create_time)
+ self.assertLessEqual(self.db.created, create_time)
+
+ # Delete one instance
+ time.sleep(0.1)
+ result = self.db.delete(_DEVICE_ID, class_id, inst_id_1)
+ self.assertTrue(result) # True returned if a del actually happened
+
+ dev_data = self.db.query(_DEVICE_ID)
+ cls_data = self.db.query(_DEVICE_ID, class_id)
+ inst_data = {k: v for k, v in cls_data.items() if isinstance(k, int)}
+ self.assertEqual(len(inst_data), 1)
+
+ self.assertLessEqual(dev_data[CREATED_KEY], create_time)
+ self.assertLessEqual(self.db.created, create_time)
+
+ # Delete remaining instance
+ time.sleep(0.1)
+ result = self.db.delete(_DEVICE_ID, class_id, inst_id_2)
+ self.assertTrue(result) # True returned if a del actually happened
+
+ dev_data = self.db.query(_DEVICE_ID)
+ cls_data = {k: v for k, v in dev_data.items() if isinstance(k, int)}
+ self.assertEqual(len(cls_data), 0)
+ self.assertLessEqual(dev_data[CREATED_KEY], create_time)
+
+ # Delete returns false if not instance
+ self.assertFalse(self.db.delete(_DEVICE_ID, class_id, inst_id_1))
+ self.assertFalse(self.db.delete(_DEVICE_ID, class_id, inst_id_2))
+
+ def test_on_mib_reset_listener(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+ time.sleep(0.1)
+
+ class_id = OntG.class_id
+ inst_id = 0
+ attributes = {'vendor_id': 'ABCD'}
+
+ set_time = datetime.utcnow()
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+
+ time.sleep(0.1)
+ self.db.on_mib_reset(_DEVICE_ID)
+
+ dev_data = self.db.query(_DEVICE_ID)
+ self.assertEqual(dev_data[DEVICE_ID_KEY], _DEVICE_ID)
+ self.assertLessEqual(dev_data[CREATED_KEY], set_time)
+ self.assertLessEqual(self.db.created, set_time)
+
+ self.assertFalse(any(isinstance(cls, int) for cls in dev_data.iterkeys()))
+
+ def test_str_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ class_id = OltG.class_id
+ inst_id = 0
+ attributes = {
+ 'olt_vendor_id': 'ABCD', # StrFixedLenField(4)
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], basestring) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_mac_address_ip_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ 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.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], basestring) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_byte_and_short_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ class_id = UniG.class_id
+ inst_id = 0
+ attributes = {
+ 'administrative_state': int(1), # ByteField
+ 'non_omci_management_identifier': int(12345) # IPField
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_int_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ class_id = PriorityQueueG.class_id
+ inst_id = 0
+ attributes = {
+ 'related_port': int(1234567) # IntField
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_long_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ class_id = PriorityQueueG.class_id
+ inst_id = 0
+ attributes = {
+ 'packet_drop_queue_thresholds': int(0x1234) # LongField
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_bit_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ class_id = OntG.class_id
+ inst_id = 0
+ attributes = {
+ 'extended_tc_layer_options': long(0x1234), # BitField(16)
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_list_field_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ 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.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(all(isinstance(data[k], type(attributes[k])) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+ def test_complex_json_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ 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.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ table_as_dict = json.loads(table_data.to_json())
+
+ self.assertTrue(all(isinstance(data['received_frame_vlan_tagging_operation_table'][0].fields[k],
+ type(attributes['received_frame_vlan_tagging_operation_table'].fields[k]))
+ for k in attributes['received_frame_vlan_tagging_operation_table'].fields.keys()))
+ self.assertTrue(all(data['received_frame_vlan_tagging_operation_table'][0].fields[k] ==
+ attributes['received_frame_vlan_tagging_operation_table'].fields[k]
+ for k in attributes['received_frame_vlan_tagging_operation_table'].fields.keys()))
+ self.assertTrue(all(data['received_frame_vlan_tagging_operation_table'][0].fields[k] == table_as_dict[k]
+ for k in table_as_dict.keys()))
+
+ def test_unknown_me_serialization(self):
+ self.db.start()
+ self.db.add(_DEVICE_ID)
+
+ blob = '00010000000c0000000000000000000000000000000000000000'
+ class_id = 0xff78
+ inst_id = 0x101
+ attributes = {
+ UNKNOWN_CLASS_ATTRIBUTE_KEY: blob
+ }
+ self.db.set(_DEVICE_ID, class_id, inst_id, attributes)
+
+ data = self.db.query(_DEVICE_ID, class_id, inst_id, attributes.keys())
+ self.assertTrue(isinstance(UNKNOWN_CLASS_ATTRIBUTE_KEY, basestring))
+ self.assertTrue(all(isinstance(attributes[k], basestring) for k in attributes.keys()))
+ self.assertTrue(all(data[k] == attributes[k] for k in attributes.keys()))
+
+
+if __name__ == '__main__':
+ main()
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()
diff --git a/test/unit/extensions/omci/test_mib_sync.py b/test/unit/extensions/omci/test_mib_sync.py
new file mode 100644
index 0000000..10a1172
--- /dev/null
+++ b/test/unit/extensions/omci/test_mib_sync.py
@@ -0,0 +1,37 @@
+#
+# 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 TestCase, main
+from mock.mock_adapter_agent import MockAdapterAgent
+
+
+
+class TestMibSync(TestCase):
+ """
+ Test the MIB Synchronizer State Machine
+ """
+ def setUp(self):
+ self.adapter_agent = MockAdapterAgent()
+
+ def tearDown(self):
+ if self.adapter_agent is not None:
+ self.adapter_agent.tearDown()
+
+ # TODO: Add tests
+
+
+if __name__ == '__main__':
+ main()
diff --git a/test/unit/extensions/omci/test_mib_upload.py b/test/unit/extensions/omci/test_mib_upload.py
new file mode 100644
index 0000000..c372819
--- /dev/null
+++ b/test/unit/extensions/omci/test_mib_upload.py
@@ -0,0 +1,36 @@
+#
+# 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 TestCase, main
+from mock.mock_adapter_agent import MockAdapterAgent
+
+
+class TestMibUpload(TestCase):
+ """
+ Test the MIB Upload Task
+ """
+ def setUp(self):
+ self.adapter_agent = MockAdapterAgent()
+
+ def tearDown(self):
+ if self.adapter_agent is not None:
+ self.adapter_agent.tearDown()
+
+ # TODO: Add tests
+
+
+if __name__ == '__main__':
+ main()
diff --git a/test/unit/extensions/omci/test_omci.py b/test/unit/extensions/omci/test_omci.py
new file mode 100644
index 0000000..6df072b
--- /dev/null
+++ b/test/unit/extensions/omci/test_omci.py
@@ -0,0 +1,1162 @@
+#
+# 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 TestCase, main
+
+from pyvoltha.adapters.extensions.omci.omci import *
+
+
+def hexify(buffer):
+ """Return a hexadecimal string encoding of input buffer"""
+ return ''.join('%02x' % ord(c) for c in buffer)
+
+
+def chunk(indexable, chunk_size):
+ for i in range(0, len(indexable), chunk_size):
+ yield indexable[i:i + chunk_size]
+
+
+def hex2raw(hex_string):
+ return ''.join(chr(int(byte, 16)) for byte in chunk(hex_string, 2))
+
+
+class TestOmciFundamentals(TestCase):
+
+ def test_bitpos_from_mask(self):
+
+ f = lambda x: bitpos_from_mask(x)
+ self.assertEqual(f(0), [])
+ self.assertEqual(f(1), [0])
+ self.assertEqual(f(3), [0, 1])
+ self.assertEqual(f(255), [0, 1, 2, 3, 4, 5, 6, 7])
+ self.assertEqual(f(0x800), [11])
+ self.assertEqual(f(0x811), [0, 4, 11])
+
+ f = lambda x: bitpos_from_mask(x, 16, -1)
+ self.assertEqual(f(0), [])
+ self.assertEqual(f(1), [16])
+ self.assertEqual(f(0x800), [5])
+ self.assertEqual(f(0x801), [5, 16])
+
+
+ def test_attribute_indeices_from_mask(self):
+
+ f = EntityClass.attribute_indices_from_mask
+ self.assertEqual(f(0), [])
+ self.assertEqual(f(0x800), [5])
+ self.assertEqual(f(0xf000), [1, 2, 3, 4])
+ self.assertEqual(f(0xf804), [1, 2, 3, 4, 5, 14])
+
+ def test_entity_attribute_serialization(self):
+
+ e = CircuitPack(vendor_id='F')
+ self.assertEqual(e.serialize(), 'F\x00\x00\x00')
+
+ e = CircuitPack(vendor_id='FOOX')
+ self.assertEqual(e.serialize(), 'FOOX')
+
+ e = CircuitPack(vendor_id='FOOX', number_of_ports=16)
+ self.assertEqual(e.serialize(), '\x10FOOX')
+
+ def test_entity_attribute_serialization_mask_based(self):
+
+ e = CircuitPack(
+ number_of_ports=4,
+ serial_number='BCMX31323334', # serial number is 4 ascii + 4 hex. 8 octets on the wire
+ version='a1c12fba91de',
+ vendor_id='BCM',
+ total_tcont_buffer_number=128
+ )
+
+ # Full object
+ self.assertEqual(e.serialize(),
+ '\x04BCMX1234a1c12fba91de\x00\x00BCM\x00\x80')
+
+ # Explicit mask with valid values
+ self.assertEqual(e.serialize(0x800), 'BCM\x00')
+ self.assertEqual(e.serialize(0x6800), '\x04BCMX1234BCM\x00')
+
+ # Referring to an unfilled field is regarded as error
+ self.assertRaises(OmciUninitializedFieldError, e.serialize, 0xc00)
+
+ def test_omci_mask_value_gen(self):
+ cls = CircuitPack
+ self.assertEqual(cls.mask_for('vendor_id'), 0x800)
+ self.assertEqual(
+ cls.mask_for('vendor_id', 'bridged_or_ip_ind'), 0x900)
+
+ reference_get_request_hex = (
+ '00 00 49 0a'
+ '00 06 01 01'
+ '08 00 00 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 28'.replace(' ', '')
+ )
+ reference_get_request_raw = hex2raw(reference_get_request_hex)
+
+ reference_get_response_hex = (
+ '00 00 29 0a'
+ '00 06 01 01'
+ '00 08 00 50'
+ '4d 43 53 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 00'
+ '00 00 00 28'.replace(' ', '')
+ )
+ reference_get_response_raw = hex2raw(reference_get_response_hex)
+
+ def test_omci_frame_serialization(self):
+
+ frame = OmciFrame(
+ transaction_id=0,
+ message_type=OmciGet.message_id,
+ omci_message=OmciGet(
+ entity_class=CircuitPack.class_id,
+ entity_id=0x101,
+ attributes_mask=CircuitPack.mask_for('vendor_id')
+ )
+ )
+ self.assertEqual(hexify(str(frame)), self.reference_get_request_hex)
+
+ def test_omci_frame_deserialization_no_data(self):
+ frame = OmciFrame(self.reference_get_request_raw)
+ self.assertEqual(frame.transaction_id, 0)
+ self.assertEqual(frame.message_type, 0x49)
+ self.assertEqual(frame.omci, 10)
+ self.assertEqual(frame.omci_message.entity_class, 0x6)
+ self.assertEqual(frame.omci_message.entity_id, 0x101)
+ self.assertEqual(frame.omci_message.attributes_mask, 0x800)
+ self.assertEqual(frame.omci_trailer, 0x28)
+
+ def test_omci_frame_deserialization_with_data(self):
+ frame = OmciFrame(self.reference_get_response_raw)
+ self.assertEqual(frame.transaction_id, 0)
+ self.assertEqual(frame.message_type, 0x29)
+ self.assertEqual(frame.omci, 10)
+ self.assertEqual(frame.omci_message.success_code, 0x0)
+ self.assertEqual(frame.omci_message.entity_class, 0x6)
+ self.assertEqual(frame.omci_message.entity_id, 0x101)
+ self.assertEqual(frame.omci_message.attributes_mask, 0x800)
+ self.assertEqual(frame.omci_trailer, 0x28)
+
+ def test_entity_attribute_deserialization(self):
+ pass
+
+
+class TestSelectMessageGeneration(TestCase):
+
+ def assertGeneratedFrameEquals(self, frame, ref):
+ assert isinstance(frame, Packet)
+ serialized_hexified_frame = hexify(str(frame)).upper()
+ ref = ref.upper()
+ if serialized_hexified_frame != ref:
+ self.fail('Mismatch:\nReference:\n{}\nGenerated (bad):\n{}'.format(
+ ref, serialized_hexified_frame
+ ))
+
+ def test_mib_reset_message_serialization(self):
+ ref = '00014F0A000200000000000000000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=1,
+ message_type=OmciMibReset.message_id,
+ omci_message=OmciMibReset(
+ entity_class=OntData.class_id
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_gal_ethernet_profile(self):
+ ref = '0002440A011000010030000000000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=2,
+ message_type=OmciCreate.message_id,
+ omci_message=OmciCreate(
+ entity_class=GalEthernetProfile.class_id,
+ entity_id=1,
+ data=dict(
+ max_gem_payload_size=48
+ )
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_tcont_1(self):
+ ref = '0003480A010680008000040000000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ alloc_id=0x400
+ )
+ frame = OmciFrame(
+ transaction_id=3,
+ message_type=OmciSet.message_id,
+ omci_message=OmciSet(
+ entity_class=Tcont.class_id,
+ entity_id=0x8000,
+ attributes_mask=Tcont.mask_for(*data.keys()),
+ data=data
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_tcont_2(self):
+ ref = '0004480A010680018000040100000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ alloc_id=0x401
+ )
+ frame = OmciFrame(
+ transaction_id=4,
+ message_type=OmciSet.message_id,
+ omci_message=OmciSet(
+ entity_class=Tcont.class_id,
+ entity_id=0x8001,
+ attributes_mask=Tcont.mask_for(*data.keys()),
+ data=data
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_8021p_mapper_service_profile(self):
+ ref = '0007440A00828000ffffffffffffffff' \
+ 'ffffffffffffffffffff000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=7,
+ message_type=OmciCreate.message_id,
+ omci_message=OmciCreate(
+ entity_class=Ieee8021pMapperServiceProfile.class_id,
+ entity_id=0x8000,
+ data=dict(
+ tp_pointer=OmciNullPointer,
+ interwork_tp_pointer_for_p_bit_priority_0=OmciNullPointer,
+ )
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_mac_bridge_service_profile(self):
+ ref = '000B440A002D02010001008000140002' \
+ '000f0001000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=11,
+ message_type=OmciCreate.message_id,
+ omci_message=OmciCreate(
+ entity_class=MacBridgeServiceProfile.class_id,
+ entity_id=0x201,
+ data=dict(
+ spanning_tree_ind=False,
+ learning_ind=True,
+ priority=0x8000,
+ max_age=20 * 256,
+ hello_time=2 * 256,
+ forward_delay=15 * 256,
+ unknown_mac_address_discard=True
+ )
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_gem_port_network_ctp(self):
+ ref = '000C440A010C01000400800003010000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=12,
+ message_type=OmciCreate.message_id,
+ omci_message=OmciCreate(
+ entity_class=GemPortNetworkCtp.class_id,
+ entity_id=0x100,
+ data=dict(
+ port_id=0x400,
+ tcont_pointer=0x8000,
+ direction=3,
+ traffic_management_pointer_upstream=0x100
+ )
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_multicast_gem_interworking_tp(self):
+ ref = '0011440A011900060104000001000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=17,
+ message_type=OmciCreate.message_id,
+ omci_message=OmciCreate(
+ entity_class=MulticastGemInterworkingTp.class_id,
+ entity_id=0x6,
+ data=dict(
+ gem_port_network_ctp_pointer=0x104,
+ interworking_option=0,
+ service_profile_pointer=0x1,
+ )
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_gem_inteworking_tp(self):
+ ref = '0012440A010A80010100058000000000' \
+ '01000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=18,
+ message_type=OmciCreate.message_id,
+ omci_message=OmciCreate(
+ entity_class=GemInterworkingTp.class_id,
+ entity_id=0x8001,
+ data=dict(
+ gem_port_network_ctp_pointer=0x100,
+ interworking_option=5,
+ service_profile_pointer=0x8000,
+ interworking_tp_pointer=0x0,
+ gal_profile_pointer=0x1
+ )
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_8021p_mapper_service_profile(self):
+ ref = '0016480A008280004000800100000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ interwork_tp_pointer_for_p_bit_priority_0=0x8001
+ )
+ frame = OmciFrame(
+ transaction_id=22,
+ message_type=OmciSet.message_id,
+ omci_message=OmciSet(
+ entity_class=Ieee8021pMapperServiceProfile.class_id,
+ entity_id=0x8000,
+ attributes_mask=Ieee8021pMapperServiceProfile.mask_for(
+ *data.keys()),
+ data=data
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_mac_bridge_port_configuration_data(self):
+ ref = '001A440A002F21010201020380000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=26,
+ message_type=OmciCreate.message_id,
+ omci_message=OmciCreate(
+ entity_class=MacBridgePortConfigurationData.class_id,
+ entity_id=0x2101,
+ data=dict(
+ bridge_id_pointer=0x201,
+ port_num=2,
+ tp_type=3,
+ tp_pointer=0x8000
+ )
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_vlan_tagging_filter_data(self):
+ ref = '001F440A005421010400000000000000' \
+ '00000000000000000000000000000000' \
+ '100100000000000000000028'
+ vlan_filter_list = [0] * 12
+ vlan_filter_list[0] = 0x0400
+
+ frame = OmciFrame(
+ transaction_id=31,
+ message_type=OmciCreate.message_id,
+ omci_message=OmciCreate(
+ entity_class=VlanTaggingFilterData.class_id,
+ entity_id=0x2101,
+ data=dict(
+ vlan_filter_list=vlan_filter_list,
+ forward_operation=0x10,
+ number_of_entries=1
+ )
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_extended_vlan_tagging_operation_configuration_data(self):
+ ref = '0023440A00AB02020A04010000000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=35,
+ message_type=OmciCreate.message_id,
+ omci_message=OmciCreate(
+ entity_class=
+ ExtendedVlanTaggingOperationConfigurationData.class_id,
+ entity_id=0x202,
+ data=dict(
+ association_type=10,
+ associated_me_pointer=0x401
+ )
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_extended_vlan_tagging_operation_configuration_data(self):
+ ref = '0024480A00AB02023800810081000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ input_tpid=0x8100,
+ output_tpid=0x8100,
+ downstream_mode=0, # inverse of upstream
+ )
+ frame = OmciFrame(
+ transaction_id=36,
+ message_type=OmciSet.message_id,
+ omci_message=OmciSet(
+ entity_class=\
+ ExtendedVlanTaggingOperationConfigurationData.class_id,
+ entity_id=0x202,
+ attributes_mask= \
+ ExtendedVlanTaggingOperationConfigurationData.mask_for(
+ *data.keys()),
+ data=data
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_extended_vlan_tagging_1(self):
+ ref = '0025480A00AB02020400f00000008200' \
+ '5000402f000000082004000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ received_frame_vlan_tagging_operation_table=\
+ 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
+ )
+ )
+ frame = OmciFrame(
+ transaction_id=37,
+ message_type=OmciSet.message_id,
+ omci_message=OmciSet(
+ entity_class=\
+ ExtendedVlanTaggingOperationConfigurationData.class_id,
+ entity_id=0x202,
+ attributes_mask= \
+ ExtendedVlanTaggingOperationConfigurationData.mask_for(
+ *data.keys()),
+ data=data
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_set_extended_vlan_tagging_2(self):
+ ref = '0026480A00AB02020400F00000008200' \
+ 'd000402f00000008200c000000000000' \
+ '000000000000000000000028'
+ data = dict(
+ received_frame_vlan_tagging_operation_table=
+ VlanTaggingOperation(
+ filter_outer_priority=15,
+ filter_inner_priority=8,
+ filter_inner_vid=1025,
+ 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=1025,
+ treatment_inner_tpid_de=4
+ )
+ )
+ frame = OmciFrame(
+ transaction_id=38,
+ message_type=OmciSet.message_id,
+ omci_message=OmciSet(
+ entity_class=
+ ExtendedVlanTaggingOperationConfigurationData.class_id,
+ entity_id=0x202,
+ attributes_mask=
+ ExtendedVlanTaggingOperationConfigurationData.mask_for(
+ *data.keys()),
+ data=data
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_create_mac_bridge_port_configuration_data2(self):
+ ref = '0029440A002F02010201010b04010000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=41,
+ message_type=OmciCreate.message_id,
+ omci_message=OmciCreate(
+ entity_class=MacBridgePortConfigurationData.class_id,
+ entity_id=0x201,
+ data=dict(
+ bridge_id_pointer=0x201,
+ encapsulation_methods=0,
+ port_num=1,
+ port_priority=0,
+ port_path_cost=0,
+ port_spanning_tree_in=0,
+ lan_fcs_ind=0,
+ tp_type=11,
+ tp_pointer=0x401,
+ mac_learning_depth=0
+ )
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+ frame2 = OmciFrame(hex2raw(ref))
+ self.assertEqual(frame2, frame)
+
+ def test_mib_upload(self):
+ ref = '00304D0A000200000000000000000000' \
+ '00000000000000000000000000000000' \
+ '000000000000000000000028'
+ frame = OmciFrame(
+ transaction_id=48,
+ message_type=OmciMibUpload.message_id,
+ omci_message=OmciMibUpload(
+ entity_class=OntData.class_id
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_parse_enh_security_avc(self):
+ refs = [
+ "0000110a014c0000008000202020202020202020202020202020202020202020"
+ "2020202020202020000000280be43cf4"
+ ]
+ for i, data in enumerate(refs):
+ frame = OmciFrame(hex2raw(data))
+ omci = frame.omci_message
+ # frame.show()
+
+ def test_parse_alarm_message(self):
+ refs = [
+ "0000100a00050101000000000000000000000000000000000000000000000000"
+ "0000000220000000000000280be43cf4"
+ ]
+ for i, data in enumerate(refs):
+ frame = OmciFrame(hex2raw(data))
+ omci = frame.omci_message
+ # frame.show()
+
+ def test_parse_results(self):
+ refs = [
+ "00001B0a014c0000008000202020202020202020202020202020202020202020"
+ "2020202020202020000000280be43cf4"
+ ]
+ for i, data in enumerate(refs):
+ frame = OmciFrame(hex2raw(data))
+ omci = frame.omci_message
+ # frame.show()
+
+ def test_parsing_mib_upload_next_responses(self):
+ refs = [
+ "00032e0a00020000000200008000000000000000000000000000000000000000"
+ "00000000000000000000002828ce00e2",
+ "00042e0a0002000000050101f0002f2f05202020202020202020202020202020"
+ "202020202000000000000028d4eb4bdf",
+ "00052e0a00020000000501010f80202020202020202020202020202020202020"
+ "2020000000000000000000282dbe4b44",
+ "00062e0a0002000000050104f000303001202020202020202020202020202020"
+ "202020202000000000000028ef1b035b",
+ "00072e0a00020000000501040f80202020202020202020202020202020202020"
+ "202000000000000000000028fec29135",
+ "00082e0a0002000000050180f000f8f801202020202020202020202020202020"
+ "202020202000000000000028fd4e0b07",
+ "00092e0a00020000000501800f80202020202020202020202020202020202020"
+ "2020000000000000000000283306b3c0",
+ "000a2e0a0002000000060101f0002f054252434d123456780000000000000000"
+ "00000000000c000000000028585c2083",
+ "000b2e0a00020000000601010f004252434d0000000000000000000000000000"
+ "0000000000000000000000284f0e82b9",
+ "000c2e0a000200000006010100f8202020202020202020202020202020202020"
+ "202000000000000000000028e68bdb63",
+ "000d2e0a00020000000601010004000000000000000000000000000000000000"
+ "00000000000000000000002857bc2730",
+ "000e2e0a0002000000060104f00030014252434d123456780000000000000000"
+ "00000000000c000000000028afe656f5",
+ "000f2e0a00020000000601040f004252434d0000000000000000000000000000"
+ "000000000000000000000028f8f6db74",
+ "00102e0a000200000006010400f8202020202020202020202020202020202020"
+ "202000000800000000000028064fc177",
+ "00112e0a00020000000601040004000000000000000000000000000000000000"
+ "0000000000000000000000285a5c0841",
+ "00122e0a0002000000060180f000f8014252434d123456780000000000000000"
+ "00000000000c0000000000286826eafe",
+ "00132e0a00020000000601800f004252434d0000000000000000000000000000"
+ "0000000000000000000000281c4b7033",
+ "00142e0a000200000006018000f8202020202020202020202020202020202020"
+ "202000084040000000000028ac144eb3",
+ "00152e0a00020000000601800004000000000000000000000000000000000000"
+ "0000000000000000000000280a81a9a7",
+ "00162e0a0002000000070000f0003530323247574f3236363230303301010100"
+ "0000000000000000000000287ea42d51",
+ "00172e0a0002000000070001f0003530323247574f3236363230303300000100"
+ "000000000000000000000028b17f567f",
+ "00182e0a0002000000830000c000202020202020202020202020202020202020"
+ "2020202020200000000000280e7eebaa",
+ "00192e0a00020000008300002000202020202020202020202020202000000000"
+ "000000000000000000000028a95c03b3",
+ "001a2e0a00020000008300001000000000000000000000000000000000000000"
+ "000000000000000000000028f30515a1",
+ "001b2e0a0002000000850000ffe0000000000000000000000000000000000000"
+ "000000000000000000000028764c18de",
+ "001c2e0a0002000000860001c00000001018aaaa000000000000000000000000"
+ "000000000000000000000028ea220ce0",
+ "001d2e0a00020000008600012000000000000000000000000000000000000000"
+ "000000000000000000000028fbdb571a",
+ "001e2e0a00020000008600011f80000000000000000000000000000000000000"
+ "000000000000000000000028c2682282",
+ "001f2e0a00020000008600010078000000000000000000000000000000000000"
+ "0000000000000000000000289c4809b1",
+ "00202e0a00020000008600010004000000000000000000000000000000000000"
+ "000000000000000000000028d174a7d6",
+ "00212e0a00020000008600010002000000000000000000000000000000000000"
+ "0000000000000000000000288f353976",
+ "00222e0a0002000001000000e0004252434d0000000000000000000000000000"
+ "4252434d123456780000002803bbceb6",
+ "00232e0a00020000010000001f80000000000000000000000000000000000000"
+ "0000000000000000000000281b9674db",
+ "00242e0a00020000010000000040000000000000000000000000000000000000"
+ "000000000000000000000028b1050b9b",
+ "00252e0a00020000010000000038000000000000000000000000000003000000"
+ "0000000000000000000000288266645e",
+ "00262e0a0002000001010000f80042564d344b3030425241303931352d303038"
+ "3300b3000001010000000028837d624f",
+ "00272e0a000200000101000007f8000000010020027c85630016000030000000"
+ "00000000000000000000002896c707e1",
+ "00282e0a0002000001068000e00000ff01010000000000000000000000000000"
+ "00000000000000000000002811acb324",
+ "00292e0a0002000001068001e00000ff01010000000000000000000000000000"
+ "00000000000000000000002823ad6aa9",
+ "002a2e0a0002000001068002e00000ff01010000000000000000000000000000"
+ "000000000000000000000028a290efd9",
+ "002b2e0a0002000001068003e00000ff01010000000000000000000000000000"
+ "000000000000000000000028af893357",
+ "002c2e0a0002000001068004e00000ff01010000000000000000000000000000"
+ "000000000000000000000028901141a3",
+ "002d2e0a0002000001068005e00000ff01010000000000000000000000000000"
+ "000000000000000000000028c4398bcc",
+ "002e2e0a0002000001068006e00000ff01010000000000000000000000000000"
+ "000000000000000000000028e60acd99",
+ "002f2e0a0002000001068007e00000ff01010000000000000000000000000000"
+ "0000000000000000000000284b5faf23",
+ "00302e0a0002000001078001ffff01000800300000050900000000ffff000000"
+ "008181000000000000000028bef89455",
+ "00312e0a0002000001080401f000000000000401000000000000000000000000"
+ "0000000000000000000000287dc5183d",
+ "00322e0a0002000001150401fff0000080008000000000040100000000010000"
+ "000000000000000000000028cc0a46a9",
+ "00332e0a0002000001150401000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000288c42acdd",
+ "00342e0a0002000001150402fff0000080008000000000040100010000010000"
+ "000000000000000000000028de9f625a",
+ "00352e0a0002000001150402000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000280587860b",
+ "00362e0a0002000001150403fff0000080008000000000040100020000010000"
+ "000000000000000000000028a49cc820",
+ "00372e0a0002000001150403000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028b4e4a2b9",
+ "00382e0a0002000001150404fff0000080008000000000040100030000010000"
+ "0000000000000000000000288233147b",
+ "00392e0a0002000001150404000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002881b706b0",
+ "003a2e0a0002000001150405fff0000080008000000000040100040000010000"
+ "000000000000000000000028be8efc9f",
+ "003b2e0a0002000001150405000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028d944804b",
+ "003c2e0a0002000001150406fff0000080008000000000040100050000010000"
+ "000000000000000000000028725c3864",
+ "003d2e0a0002000001150406000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000284e2d5cd2",
+ "003e2e0a0002000001150407fff0000080008000000000040100060000010000"
+ "000000000000000000000028464b03ba",
+ "003f2e0a0002000001150407000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000287006cfd0",
+ "00402e0a0002000001150408fff0000080008000000000040100070000010000"
+ "000000000000000000000028cd88ebeb",
+ "00412e0a0002000001150408000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000285a5905e2",
+ "00422e0a0002000001158000fff0000100010000000000800000000000010000"
+ "000000000000000000000028e61b19d1",
+ "00432e0a0002000001158000000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028b0cc5937",
+ "00442e0a0002000001158001fff0000100010000000000800000010000010000"
+ "0000000000000000000000285386bbf2",
+ "00452e0a0002000001158001000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028c06723ab",
+ "00462e0a0002000001158002fff0000100010000000000800000020000010000"
+ "000000000000000000000028ab49704a",
+ "00472e0a0002000001158002000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002857432f25",
+ "00482e0a0002000001158003fff0000100010000000000800000030000010000"
+ "000000000000000000000028b383c057",
+ "00492e0a0002000001158003000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028dca40d66",
+ "004a2e0a0002000001158004fff0000100010000000000800000040000010000"
+ "0000000000000000000000286b7ba0e2",
+ "004b2e0a0002000001158004000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028fd442363",
+ "004c2e0a0002000001158005fff0000100010000000000800000050000010000"
+ "0000000000000000000000280ee9a0b8",
+ "004d2e0a0002000001158005000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028bc1b9843",
+ "004e2e0a0002000001158006fff0000100010000000000800000060000010000"
+ "0000000000000000000000280c535114",
+ "004f2e0a0002000001158006000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002887032f2b",
+ "00502e0a0002000001158007fff0000100010000000000800000070000010000"
+ "000000000000000000000028a77d7f61",
+ "00512e0a0002000001158007000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002835e9f567",
+ "00522e0a0002000001158008fff0000100010000000000800100000000010000"
+ "000000000000000000000028ff4ca94b",
+ "00532e0a0002000001158008000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000281e2f1e33",
+ "00542e0a0002000001158009fff0000100010000000000800100010000010000"
+ "0000000000000000000000283c473db0",
+ "00552e0a0002000001158009000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002869f51dda",
+ "00562e0a000200000115800afff0000100010000000000800100020000010000"
+ "000000000000000000000028046b8feb",
+ "00572e0a000200000115800a000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002868b1495e",
+ "00582e0a000200000115800bfff0000100010000000000800100030000010000"
+ "0000000000000000000000282b927566",
+ "00592e0a000200000115800b000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028cd43de96",
+ "005a2e0a000200000115800cfff0000100010000000000800100040000010000"
+ "000000000000000000000028c49617dd",
+ "005b2e0a000200000115800c000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028fbbb972a",
+ "005c2e0a000200000115800dfff0000100010000000000800100050000010000"
+ "00000000000000000000002893d4c2b5",
+ "005d2e0a000200000115800d000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028dc9d97ca",
+ "005e2e0a000200000115800efff0000100010000000000800100060000010000"
+ "0000000000000000000000280e1ec245",
+ "005f2e0a000200000115800e000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028be3d56f1",
+ "00602e0a000200000115800ffff0000100010000000000800100070000010000"
+ "0000000000000000000000280c046099",
+ "00612e0a000200000115800f000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028d770e4ea",
+ "00622e0a0002000001158010fff0000100010000000000800200000000010000"
+ "0000000000000000000000281b449092",
+ "00632e0a0002000001158010000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000282b7a8604",
+ "00642e0a0002000001158011fff0000100010000000000800200010000010000"
+ "000000000000000000000028ad498068",
+ "00652e0a0002000001158011000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028a114b304",
+ "00662e0a0002000001158012fff0000100010000000000800200020000010000"
+ "000000000000000000000028c091715d",
+ "00672e0a0002000001158012000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028d4ab49e7",
+ "00682e0a0002000001158013fff0000100010000000000800200030000010000"
+ "000000000000000000000028e39dd5dd",
+ "00692e0a0002000001158013000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000288779ebf0",
+ "006a2e0a0002000001158014fff0000100010000000000800200040000010000"
+ "000000000000000000000028c47a741f",
+ "006b2e0a0002000001158014000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028ce765fcd",
+ "006c2e0a0002000001158015fff0000100010000000000800200050000010000"
+ "0000000000000000000000288f732591",
+ "006d2e0a0002000001158015000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028920b6f5e",
+ "006e2e0a0002000001158016fff0000100010000000000800200060000010000"
+ "000000000000000000000028f072e1c3",
+ "006f2e0a0002000001158016000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028b47ea00f",
+ "00702e0a0002000001158017fff0000100010000000000800200070000010000"
+ "00000000000000000000002813461627",
+ "00712e0a0002000001158017000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002809013378",
+ "00722e0a0002000001158018fff0000100010000000000800300000000010000"
+ "0000000000000000000000286168e200",
+ "00732e0a0002000001158018000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028eccc81f7",
+ "00742e0a0002000001158019fff0000100010000000000800300010000010000"
+ "00000000000000000000002855fe8072",
+ "00752e0a0002000001158019000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028c159c496",
+ "00762e0a000200000115801afff0000100010000000000800300020000010000"
+ "00000000000000000000002872652aca",
+ "00772e0a000200000115801a000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000283ba1c255",
+ "00782e0a000200000115801bfff0000100010000000000800300030000010000"
+ "0000000000000000000000286b2ecb95",
+ "00792e0a000200000115801b000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028441fbe05",
+ "007a2e0a000200000115801cfff0000100010000000000800300040000010000"
+ "000000000000000000000028f07ad5d8",
+ "007b2e0a000200000115801c000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028237d6a28",
+ "007c2e0a000200000115801dfff0000100010000000000800300050000010000"
+ "000000000000000000000028e47dfdca",
+ "007d2e0a000200000115801d000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000280ca941be",
+ "007e2e0a000200000115801efff0000100010000000000800300060000010000"
+ "0000000000000000000000283a1ef4d4",
+ "007f2e0a000200000115801e000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000289c905cd5",
+ "00802e0a000200000115801ffff0000100010000000000800300070000010000"
+ "000000000000000000000028384ae4c6",
+ "00812e0a000200000115801f000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028be87eb55",
+ "00822e0a0002000001158020fff0000100010000000000800400000000010000"
+ "000000000000000000000028f0412282",
+ "00832e0a0002000001158020000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028842ada0c",
+ "00842e0a0002000001158021fff0000100010000000000800400010000010000"
+ "000000000000000000000028a6eed1bc",
+ "00852e0a0002000001158021000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000280f3dd903",
+ "00862e0a0002000001158022fff0000100010000000000800400020000010000"
+ "000000000000000000000028474a0823",
+ "00872e0a0002000001158022000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028e00456b3",
+ "00882e0a0002000001158023fff0000100010000000000800400030000010000"
+ "00000000000000000000002851cbe1a6",
+ "00892e0a0002000001158023000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002869a99563",
+ "008a2e0a0002000001158024fff0000100010000000000800400040000010000"
+ "00000000000000000000002867705534",
+ "008b2e0a0002000001158024000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000286f9570c0",
+ "008c2e0a0002000001158025fff0000100010000000000800400050000010000"
+ "000000000000000000000028450ef70e",
+ "008d2e0a0002000001158025000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002847588afa",
+ "008e2e0a0002000001158026fff0000100010000000000800400060000010000"
+ "000000000000000000000028c8218600",
+ "008f2e0a0002000001158026000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028391a6ba7",
+ "00902e0a0002000001158027fff0000100010000000000800400070000010000"
+ "000000000000000000000028afc0878b",
+ "00912e0a0002000001158027000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002819130d66",
+ "00922e0a0002000001158028fff0000100010000000000800500000000010000"
+ "0000000000000000000000289afa4cf7",
+ "00932e0a0002000001158028000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002873a4e20b",
+ "00942e0a0002000001158029fff0000100010000000000800500010000010000"
+ "000000000000000000000028633debd9",
+ "00952e0a0002000001158029000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000280397eb28",
+ "00962e0a000200000115802afff0000100010000000000800500020000010000"
+ "0000000000000000000000280ed5ee7a",
+ "00972e0a000200000115802a000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028f886ba59",
+ "00982e0a000200000115802bfff0000100010000000000800500030000010000"
+ "00000000000000000000002888ff79b1",
+ "00992e0a000200000115802b000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002846baf278",
+ "009a2e0a000200000115802cfff0000100010000000000800500040000010000"
+ "0000000000000000000000281fd1e68f",
+ "009b2e0a000200000115802c000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028d99760f9",
+ "009c2e0a000200000115802dfff0000100010000000000800500050000010000"
+ "000000000000000000000028557aaf84",
+ "009d2e0a000200000115802d000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028064210fd",
+ "009e2e0a000200000115802efff0000100010000000000800500060000010000"
+ "0000000000000000000000285fd6c061",
+ "009f2e0a000200000115802e000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028299efbb5",
+ "00a02e0a000200000115802ffff0000100010000000000800500070000010000"
+ "00000000000000000000002834f127c4",
+ "00a12e0a000200000115802f000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028edd30591",
+ "00a22e0a0002000001158030fff0000100010000000000800600000000010000"
+ "000000000000000000000028183183f2",
+ "00a32e0a0002000001158030000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028a27e71f6",
+ "00a42e0a0002000001158031fff0000100010000000000800600010000010000"
+ "000000000000000000000028bd64dfc0",
+ "00a52e0a0002000001158031000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002839e2f37e",
+ "00a62e0a0002000001158032fff0000100010000000000800600020000010000"
+ "0000000000000000000000283e72282e",
+ "00a72e0a0002000001158032000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028cef19baa",
+ "00a82e0a0002000001158033fff0000100010000000000800600030000010000"
+ "0000000000000000000000281c1caf44",
+ "00a92e0a0002000001158033000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002814712e27",
+ "00aa2e0a0002000001158034fff0000100010000000000800600040000010000"
+ "000000000000000000000028f02a30a4",
+ "00ab2e0a0002000001158034000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028068fcbf5",
+ "00ac2e0a0002000001158035fff0000100010000000000800600050000010000"
+ "000000000000000000000028436bd783",
+ "00ad2e0a0002000001158035000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000288da3200f",
+ "00ae2e0a0002000001158036fff0000100010000000000800600060000010000"
+ "000000000000000000000028c26a02ca",
+ "00af2e0a0002000001158036000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028147a41ee",
+ "00b02e0a0002000001158037fff0000100010000000000800600070000010000"
+ "0000000000000000000000287c2bbec0",
+ "00b12e0a0002000001158037000f0200020002000200ffff0900000000000000"
+ "0000000000000000000000284c86c11f",
+ "00b22e0a0002000001158038fff0000100010000000000800700000000010000"
+ "00000000000000000000002895b94e06",
+ "00b32e0a0002000001158038000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028a2b34012",
+ "00b42e0a0002000001158039fff0000100010000000000800700010000010000"
+ "00000000000000000000002804b205a3",
+ "00b52e0a0002000001158039000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002886856d76",
+ "00b62e0a000200000115803afff0000100010000000000800700020000010000"
+ "0000000000000000000000282a22752c",
+ "00b72e0a000200000115803a000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028488e67db",
+ "00b82e0a000200000115803bfff0000100010000000000800700030000010000"
+ "000000000000000000000028a55f79ea",
+ "00b92e0a000200000115803b000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002842d77ba7",
+ "00ba2e0a000200000115803cfff0000100010000000000800700040000010000"
+ "000000000000000000000028da65268a",
+ "00bb2e0a000200000115803c000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028c58443ec",
+ "00bc2e0a000200000115803dfff0000100010000000000800700050000010000"
+ "000000000000000000000028997aca59",
+ "00bd2e0a000200000115803d000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028a2670b7d",
+ "00be2e0a000200000115803efff0000100010000000000800700060000010000"
+ "00000000000000000000002813e904cb",
+ "00bf2e0a000200000115803e000f0200020002000200ffff0900000000000000"
+ "000000000000000000000028c387a9e5",
+ "00c02e0a000200000115803ffff0000100010000000000800700070000010000"
+ "000000000000000000000028d556a6b2",
+ "00c12e0a000200000115803f000f0200020002000200ffff0900000000000000"
+ "00000000000000000000002868d9961a",
+ "00c22e0a0002000001168000f000800000000200000000000000000000000000"
+ "000000000000000000000028b69b53c1",
+ "00c32e0a0002000001168001f000800000000200000000000000000000000000"
+ "000000000000000000000028537705d4",
+ "00c42e0a0002000001168002f000800000000200000000000000000000000000"
+ "000000000000000000000028db171b7b",
+ "00c52e0a0002000001168003f000800000000200000000000000000000000000"
+ "000000000000000000000028f9b3fa54",
+ "00c62e0a0002000001168004f000800000000200000000000000000000000000"
+ "000000000000000000000028cdacda4e",
+ "00c72e0a0002000001168005f000800000000200000000000000000000000000"
+ "00000000000000000000002837133b6e",
+ "00c82e0a0002000001168006f000800000000200000000000000000000000000"
+ "000000000000000000000028d6447905",
+ "00c92e0a0002000001168007f000800000000200000000000000000000000000"
+ "000000000000000000000028021a3910",
+ "00ca2e0a0002000001168008f000800100000200000000000000000000000000"
+ "00000000000000000000002835d3cf43",
+ "00cb2e0a0002000001168009f000800100000200000000000000000000000000"
+ "00000000000000000000002887ad76fc",
+ "00cc2e0a000200000116800af000800100000200000000000000000000000000"
+ "00000000000000000000002895e3d838",
+ "00cd2e0a000200000116800bf000800100000200000000000000000000000000"
+ "000000000000000000000028a07489ac",
+ "00ce2e0a000200000116800cf000800100000200000000000000000000000000"
+ "0000000000000000000000285d08821d",
+ "00cf2e0a000200000116800df000800100000200000000000000000000000000"
+ "000000000000000000000028302249a4",
+ "00d02e0a000200000116800ef000800100000200000000000000000000000000"
+ "0000000000000000000000283966d3bc",
+ "00d12e0a000200000116800ff000800100000200000000000000000000000000"
+ "0000000000000000000000289519cdb5",
+ "00d22e0a0002000001168010f000800200000200000000000000000000000000"
+ "0000000000000000000000281bc99b7b",
+ "00d32e0a0002000001168011f000800200000200000000000000000000000000"
+ "000000000000000000000028e483b1a0",
+ "00d42e0a0002000001168012f000800200000200000000000000000000000000"
+ "0000000000000000000000286885d8bd",
+ "00d52e0a0002000001168013f000800200000200000000000000000000000000"
+ "000000000000000000000028cbe7afd8",
+ "00d62e0a0002000001168014f000800200000200000000000000000000000000"
+ "00000000000000000000002809009846",
+ "00d72e0a0002000001168015f000800200000200000000000000000000000000"
+ "0000000000000000000000285bee86c4",
+ "00d82e0a0002000001168016f000800200000200000000000000000000000000"
+ "0000000000000000000000281f25725c",
+ "00d92e0a0002000001168017f000800200000200000000000000000000000000"
+ "00000000000000000000002872e94fe1",
+ "00da2e0a0002000001168018f000800300000200000000000000000000000000"
+ "000000000000000000000028e39d572f",
+ "00db2e0a0002000001168019f000800300000200000000000000000000000000"
+ "0000000000000000000000281c9dcadd",
+ "00dc2e0a000200000116801af000800300000200000000000000000000000000"
+ "0000000000000000000000287c5b8405",
+ "00dd2e0a000200000116801bf000800300000200000000000000000000000000"
+ "00000000000000000000002826334420",
+ "00de2e0a000200000116801cf000800300000200000000000000000000000000"
+ "00000000000000000000002871ee1536",
+ "00df2e0a000200000116801df000800300000200000000000000000000000000"
+ "0000000000000000000000289dfeeeb9",
+ "00e02e0a000200000116801ef000800300000200000000000000000000000000"
+ "000000000000000000000028954d55b3",
+ "00e12e0a000200000116801ff000800300000200000000000000000000000000"
+ "000000000000000000000028930c564e",
+ "00e22e0a0002000001168020f000800400000200000000000000000000000000"
+ "000000000000000000000028b9cec3bf",
+ "00e32e0a0002000001168021f000800400000200000000000000000000000000"
+ "0000000000000000000000284263f268",
+ "00e42e0a0002000001168022f000800400000200000000000000000000000000"
+ "000000000000000000000028913e5219",
+ "00e52e0a0002000001168023f000800400000200000000000000000000000000"
+ "000000000000000000000028efe86fe1",
+ "00e62e0a0002000001168024f000800400000200000000000000000000000000"
+ "000000000000000000000028deb045df",
+ "00e72e0a0002000001168025f000800400000200000000000000000000000000"
+ "000000000000000000000028255bcd32",
+ "00e82e0a0002000001168026f000800400000200000000000000000000000000"
+ "000000000000000000000028355392ad",
+ "00e92e0a0002000001168027f000800400000200000000000000000000000000"
+ "000000000000000000000028404a6aca",
+ "00ea2e0a0002000001168028f000800500000200000000000000000000000000"
+ "0000000000000000000000281de78f94",
+ "00eb2e0a0002000001168029f000800500000200000000000000000000000000"
+ "000000000000000000000028501a3aae",
+ "00ec2e0a000200000116802af000800500000200000000000000000000000000"
+ "0000000000000000000000282947d976",
+ "00ed2e0a000200000116802bf000800500000200000000000000000000000000"
+ "000000000000000000000028095cfe0d",
+ "00ee2e0a000200000116802cf000800500000200000000000000000000000000"
+ "000000000000000000000028bbcfc27a",
+ "00ef2e0a000200000116802df000800500000200000000000000000000000000"
+ "000000000000000000000028dbb27396",
+ "00f02e0a000200000116802ef000800500000200000000000000000000000000"
+ "000000000000000000000028dbe9b225",
+ "00f12e0a000200000116802ff000800500000200000000000000000000000000"
+ "000000000000000000000028840c0b08",
+ "00f22e0a0002000001168030f000800600000200000000000000000000000000"
+ "0000000000000000000000287683e4f8",
+ "00f32e0a0002000001168031f000800600000200000000000000000000000000"
+ "00000000000000000000002844d131d1",
+ "00f42e0a0002000001168032f000800600000200000000000000000000000000"
+ "0000000000000000000000284d2c2c6d",
+ "00f52e0a0002000001168033f000800600000200000000000000000000000000"
+ "000000000000000000000028e89a166c",
+ "00f62e0a0002000001168034f000800600000200000000000000000000000000"
+ "0000000000000000000000280f47db8c",
+ "00f72e0a0002000001168035f000800600000200000000000000000000000000"
+ "0000000000000000000000283ede8b3e",
+ "00f82e0a0002000001168036f000800600000200000000000000000000000000"
+ "000000000000000000000028580547db",
+ "00f92e0a0002000001168037f000800600000200000000000000000000000000"
+ "000000000000000000000028d72a270e",
+ "00fa2e0a0002000001168038f000800700000200000000000000000000000000"
+ "000000000000000000000028c25ce712",
+ "00fb2e0a0002000001168039f000800700000200000000000000000000000000"
+ "000000000000000000000028b908637e",
+ "00fc2e0a000200000116803af000800700000200000000000000000000000000"
+ "0000000000000000000000285b66e6fa",
+ "00fd2e0a000200000116803bf000800700000200000000000000000000000000"
+ "00000000000000000000002855c10393",
+ "00fe2e0a000200000116803cf000800700000200000000000000000000000000"
+ "0000000000000000000000283e94c57d",
+ "00ff2e0a000200000116803df000800700000200000000000000000000000000"
+ "0000000000000000000000284347e7f0",
+ "01002e0a000200000116803ef000800700000200000000000000000000000000"
+ "000000000000000000000028be66429d",
+ "01012e0a000200000116803ff000800700000200000000000000000000000000"
+ "0000000000000000000000284f7db145",
+ "01022e0a0002000001490401c000000000000000000000000000000000000000"
+ "000000000000000000000028470aa043",
+ "01032e0a00020000014904012000000000000000000000000000000000000000"
+ "000000000000000000000028a6bc6e48",
+ "01042e0a00020000014904011800ffffffff0000000000000000000000000000"
+ "000000000000000000000028f747c739",
+ ]
+ mask = "%5s %9s %20s %9s %s"
+ print
+ print mask % ("seq", "class_id", "class", "instance", "attributes")
+ for i, data in enumerate(refs):
+ frame = OmciFrame(hex2raw(data))
+ omci = frame.omci_message
+ # frame.show()
+ print mask % (
+ str(i),
+ str(omci.object_entity_class),
+ entity_id_to_class_map[omci.object_entity_class].__name__,
+ '0x%x' % omci.object_entity_id,
+ '\n '.join(
+ '%s: %s' % (k, v) for k, v in omci.object_data.items())
+ )
+
+ def test_onu_reboot(self):
+ ref = '0016590a01000000000000000000000000000'\
+ '0000000000000000000000000000000000000'\
+ '00000000000028'
+
+ frame = OmciFrame(
+ transaction_id=22,
+ message_type=OmciReboot.message_id,
+ omci_message=OmciReboot(
+ entity_class=OntG.class_id,
+ entity_id=0
+ )
+ )
+ self.assertGeneratedFrameEquals(frame, ref)
+
+ def test_omci_entity_ids(self):
+ from pyvoltha.adapters.extensions.omci.omci_entities import entity_classes
+
+ # For Entity Classes that have a Managed Entity ID with Set-By-Create
+ # access, verify that the attribute name matches 'managed_entity_id'
+ #
+ # This is critical for the MIB Synchronizer state machine as it needs
+ # to backfill Set-By-Create attributes when it sees a Create response
+ # but it needs to ignore the 'managed_entity_id' attribute (by name).
+
+ for entity in entity_classes:
+ mei_attr = entity.attributes[0]
+ self.assertIsNotNone(mei_attr)
+ self.assertTrue(AA.SBC not in mei_attr.access or
+ mei_attr.field.name == 'managed_entity_id')
+
+
+if __name__ == '__main__':
+ main()
diff --git a/test/unit/extensions/omci/test_omci_cc.py b/test/unit/extensions/omci/test_omci_cc.py
new file mode 100644
index 0000000..7c1491d
--- /dev/null
+++ b/test/unit/extensions/omci/test_omci_cc.py
@@ -0,0 +1,1233 @@
+#
+# 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.
+#
+import binascii
+from common.frameio.frameio import hexify
+from twisted.python.failure import Failure
+from unittest import TestCase, main, skip
+from mock.mock_adapter_agent import MockAdapterAgent
+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.omci_defs import *
+from pyvoltha.adapters.extensions.omci.omci_frame import *
+from pyvoltha.adapters.extensions.omci.omci_entities import *
+from pyvoltha.adapters.extensions.omci.omci_me import ExtendedVlanTaggingOperationConfigurationDataFrame
+from pyvoltha.adapters.extensions.omci.omci_cc import OMCI_CC, UNKNOWN_CLASS_ATTRIBUTE_KEY,\
+ MAX_OMCI_REQUEST_AGE
+
+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
+
+successful = False
+error_reason = None
+
+
+def chunk(indexable, chunk_size):
+ for i in range(0, len(indexable), chunk_size):
+ yield indexable[i:i + chunk_size]
+
+
+def hex2raw(hex_string):
+ return ''.join(chr(int(byte, 16)) for byte in chunk(hex_string, 2))
+
+
+class TestOmciCc(TestCase):
+ """
+ Test the Open OMCI Communication channels
+
+ Note also added some testing of MockOnu behaviour since its behaviour during more
+ complicated unit/integration tests may be performed in the future.
+ """
+ def setUp(self, let_msg_timeout=False):
+ self.adapter_agent = MockAdapterAgent()
+
+ def tearDown(self):
+ 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, timeout_messages=False):
+ # 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.timeout_the_message = timeout_messages
+
+ self.adapter_agent.add_child_device(self.olt_handler.device,
+ self.onu_handler.device)
+
+ def _is_omci_frame(self, results, omci_msg_type):
+ assert isinstance(results, OmciFrame), 'Not OMCI Frame'
+ assert 'omci_message' in results.fields, 'Not OMCI Frame'
+ if omci_msg_type is not None:
+ assert isinstance(results.fields['omci_message'], omci_msg_type)
+ return results
+
+ def _check_status(self, results, value):
+ if value is not None: assert results is not None, 'unexpected emtpy message'
+ status = results.fields['omci_message'].fields['success_code']
+ assert status == value,\
+ 'Unexpected Status Code. Got {}, Expected: {}'.format(status, value)
+ return results
+
+ def _check_mib_sync(self, results, value):
+ assert self.onu_device.mib_data_sync == value, \
+ 'Unexpected MIB DATA Sync value. Got {}, Expected: {}'.format(
+ self.onu_device.mib_data_sync, value)
+ return results
+
+ def _check_stats(self, results, _, stat, expected):
+ snapshot = self._snapshot_stats()
+ assert snapshot[stat] == expected, \
+ 'Invalid statistic "{}". Got {}, Expected: {}'.format(stat,
+ snapshot[stat],
+ expected)
+ return results
+
+ def _check_value_equal(self, results, name, value, expected):
+ assert value == expected, \
+ 'Value "{}" not equal. Got {}, Expected: {}'.format(name, value,
+ expected)
+ return results
+
+ def _default_errback(self, failure):
+ from twisted.internet.defer import TimeoutError
+ assert isinstance(failure.type, type(TimeoutError))
+
+ def _snapshot_stats(self):
+ omci_cc = self.onu_handler.omci_cc
+ return {
+ 'tx_frames': omci_cc.tx_frames,
+ 'rx_frames': omci_cc.rx_frames,
+ 'rx_unknown_tid': omci_cc.rx_unknown_tid,
+ 'rx_onu_frames': omci_cc.rx_onu_frames,
+ 'rx_onu_discards': omci_cc.rx_onu_discards,
+ 'rx_timeouts': omci_cc.rx_timeouts,
+ 'rx_unknown_me': omci_cc.rx_unknown_me,
+ 'rx_late': omci_cc.rx_late,
+ 'tx_errors': omci_cc.tx_errors,
+ 'consecutive_errors': omci_cc.consecutive_errors,
+ 'reply_min': omci_cc.reply_min,
+ 'reply_max': omci_cc.reply_max,
+ 'reply_average': omci_cc.reply_average,
+ 'hp_tx_queue_len': omci_cc.hp_tx_queue_len,
+ 'lp_tx_queue_len': omci_cc.lp_tx_queue_len,
+ 'max_hp_tx_queue': omci_cc.max_hp_tx_queue,
+ 'max_lp_tx_queue': omci_cc._max_lp_tx_queue,
+ }
+
+ def test_default_init(self):
+ self.setup_one_of_each()
+ # Test default construction of OMCI_CC as well as
+ # various other parameter settings
+ omci_cc = self.onu_handler.omci_cc
+
+ # No device directly associated
+ self.assertIsNotNone(omci_cc._adapter_agent)
+ self.assertIsNone(omci_cc._proxy_address)
+
+ # No outstanding requests
+ self.assertEqual(len(omci_cc._pending[OMCI_CC.LOW_PRIORITY]), 0)
+ self.assertEqual(len(omci_cc._pending[OMCI_CC.HIGH_PRIORITY]), 0)
+
+ # No active requests
+ self.assertIsNone(omci_cc._tx_request[OMCI_CC.LOW_PRIORITY])
+ self.assertIsNone(omci_cc._tx_request[OMCI_CC.HIGH_PRIORITY])
+
+ # Flags/properties
+ self.assertFalse(omci_cc.enabled)
+
+ # Statistics
+ self.assertEqual(omci_cc.tx_frames, 0)
+ self.assertEqual(omci_cc.rx_frames, 0)
+ self.assertEqual(omci_cc.rx_unknown_tid, 0)
+ self.assertEqual(omci_cc.rx_onu_frames, 0)
+ self.assertEqual(omci_cc.rx_onu_discards, 0)
+ self.assertEqual(omci_cc.rx_unknown_me, 0)
+ self.assertEqual(omci_cc.rx_timeouts, 0)
+ self.assertEqual(omci_cc.rx_late, 0)
+ self.assertEqual(omci_cc.tx_errors, 0)
+ self.assertEqual(omci_cc.consecutive_errors, 0)
+ self.assertNotEquals(omci_cc.reply_min, 0.0)
+ self.assertEqual(omci_cc.reply_max, 0.0)
+ self.assertEqual(omci_cc.reply_average, 0.0)
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0.0)
+ self.assertEqual(omci_cc.max_hp_tx_queue, 0.0)
+ self.assertEqual(omci_cc._max_hp_tx_queue, 0.0)
+ self.assertEqual(omci_cc._max_lp_tx_queue, 0.0)
+
+ def test_enable_disable(self):
+ self.setup_one_of_each()
+
+ # Test enable property
+ omci_cc = self.onu_handler.omci_cc
+
+ # Initially disabled
+ self.assertFalse(omci_cc.enabled)
+ omci_cc.enabled = False
+ self.assertFalse(omci_cc.enabled)
+
+ omci_cc.enabled = True
+ self.assertTrue(omci_cc.enabled)
+ self.assertIsNotNone(omci_cc._proxy_address)
+ self.assertEqual(len(omci_cc._pending[OMCI_CC.LOW_PRIORITY]), 0)
+ self.assertEqual(len(omci_cc._pending[OMCI_CC.HIGH_PRIORITY]), 0)
+
+ omci_cc.enabled = True # Should be a NOP
+ self.assertTrue(omci_cc.enabled)
+ self.assertIsNotNone(omci_cc._proxy_address)
+ self.assertEqual(len(omci_cc._pending[OMCI_CC.LOW_PRIORITY]), 0)
+ self.assertEqual(len(omci_cc._pending[OMCI_CC.HIGH_PRIORITY]), 0)
+
+ omci_cc.enabled = False
+ self.assertFalse(omci_cc.enabled)
+ self.assertIsNone(omci_cc._proxy_address)
+
+ def test_rx_discard_if_disabled(self):
+ # ME without a known decoder
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = False
+ snapshot = self._snapshot_stats()
+
+ msg = '00fc2e0a00020000ff780000e00000010000000c' \
+ '0000000000000000000000000000000000000000' \
+ '00000028105a86ef'
+
+ omci_cc.receive_message(hex2raw(msg))
+
+ # Note: No counter increments
+ self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
+ self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
+ self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'])
+ self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
+
+ def test_message_send_get(self):
+ # Various tests of sending an OMCI message and it either
+ # getting a response or send catching some errors of
+ # importance
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+ mib_data_sync = self.onu_device.mib_data_sync
+
+ # GET
+ # d = omci_cc.send() # TODO: Implement
+ #
+ # d.addCallbacks(self._is_omci_frame, self._default_errback)
+ # d.addCallback(self._check_status, RC.Success.value)
+ # d.addCallback(self._check_mib_sync, mib_data_sync)
+ #
+ # d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
+ # d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
+ # d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
+ # d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
+ # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
+
+ # return d
+
+ def test_message_send_set(self):
+ # Various tests of sending an OMCI message and it either
+ # getting a response or send catching some errors of
+ # importance
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+ mib_data_sync = self.onu_device.mib_data_sync
+
+ # SET
+ # d = omci_cc.send() # TODO: Implement
+ #
+ # d.addCallbacks(self._is_omci_frame, self._default_errback)
+ # d.addCallback(self._check_status, RC.Success.value)
+ # d.addCallback(self._check_mib_sync, mib_data_sync + 1 if mib_data_sync < 255 else 1)
+ #
+ # d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
+ # d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
+ # d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
+ # d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
+ # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
+
+ # return d
+ #
+ # # Also test mib_data_sync rollover. 255 -> 1 (zero reserved)
+ #
+ # self.onu_device.mib_data_sync = 255
+ # # SET
+ # self.assertTrue(True) # TODO: Implement (copy previous one here)
+ # self.assertEqual(1, self.onu_device.mib_data_sync)
+
+ def test_message_send_create(self):
+ # Various tests of sending an OMCI message and it either
+ # getting a response or send catching some errors of
+ # importance
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+ mib_data_sync = self.onu_device.mib_data_sync
+
+ # Create
+ # d = omci_cc.send() # TODO: Implement
+ #
+ # d.addCallbacks(self._is_omci_frame, self._default_errback)
+ # d.addCallback(self._check_status, RC.Success.value)
+ # d.addCallback(self._check_mib_sync, mib_data_sync + 1 if mib_data_sync < 255 else 1)
+ #
+ # d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
+ # d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
+ # d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
+ # d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
+ # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
+
+ # return d
+
+ def test_message_send_delete(self):
+ # Various tests of sending an OMCI message and it either
+ # getting a response or send catching some errors of
+ # importance
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+ mib_data_sync = self.onu_device.mib_data_sync
+
+ # Delete
+ # d = omci_cc.send() # TODO: Implement
+ #
+ # d.addCallbacks(self._is_omci_frame, self._default_errback)
+ # d.addCallback(self._check_status, RC.Success.value)
+ # d.addCallback(self._check_mib_sync, mib_data_sync + 1 if mib_data_sync < 255 else 1)
+ #
+ # d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
+ # d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
+ # d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
+ # d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
+ # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
+
+ # return d
+
+ def test_message_send_mib_reset(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ self.onu_device.mib_data_sync = 10
+ snapshot = self._snapshot_stats()
+
+ # Successful MIB Reset
+ d = omci_cc.send_mib_reset(timeout=1.0)
+
+ d.addCallbacks(self._is_omci_frame, self._default_errback)
+ d.addCallback(self._check_status, RC.Success)
+ d.addCallback(self._check_mib_sync, 0)
+
+ d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
+ d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
+ d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
+ d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
+ d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
+ d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
+ d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
+ d.addCallback(self._check_stats, snapshot, 'rx_late', snapshot['rx_late'])
+ d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
+ d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
+ return d
+
+ def test_message_send_mib_upload(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+ mib_data_sync = self.onu_device.mib_data_sync
+
+ # MIB Upload
+ d = omci_cc.send_mib_upload(timeout=1.0)
+
+ d.addCallbacks(self._is_omci_frame, self._default_errback)
+ d.addCallback(self._check_status, RC.Success)
+ d.addCallback(self._check_mib_sync, mib_data_sync)
+
+ # TODO: MIB Upload Results specific tests here
+
+ d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
+ d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
+ d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
+ d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
+ d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
+ d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
+ d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
+ d.addCallback(self._check_stats, snapshot, 'rx_late', snapshot['rx_late'])
+ d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
+ d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
+ return d
+
+ def test_message_send_mib_upload_next(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+ mib_data_sync = self.onu_device.mib_data_sync
+
+ # # MIB Upload Next
+ # d = omci_cc.send_mib_upload_next(0, timeout=1.0)
+ #
+ # d.addCallbacks(self._is_omci_frame, self._default_errback)
+ # d.addCallback(self._check_status, RC.Success)
+ # d.addCallback(self._check_mib_sync, mib_data_sync)
+ #
+ # # TODO: MIB Upload Next Results specific tests here
+ #
+ # d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
+ # d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
+ # d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
+ # d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
+ # d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
+ # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
+ # return d
+
+ def test_message_send_no_timeout(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ self.onu_device.mib_data_sync = 10
+ snapshot = self._snapshot_stats()
+
+ d = omci_cc.send_mib_reset(timeout=0)
+ d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
+ d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'])
+ d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
+ return d
+
+ def test_message_send_bad_timeout(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ self.onu_device.mib_data_sync = 10
+ snapshot = self._snapshot_stats()
+
+ d = omci_cc.send_mib_reset(timeout=MAX_OMCI_REQUEST_AGE + 1)
+ d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'])
+ d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'])
+ d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'] + 1)
+ return d
+
+ def test_message_send_not_a_frame(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ self.onu_device.mib_data_sync = 10
+ snapshot = self._snapshot_stats()
+
+ d = omci_cc.send('hello world', timeout=1)
+ d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'])
+ d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'])
+ d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'] + 1)
+ return d
+
+ def test_message_send_reboot(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # ONU Reboot
+ d = omci_cc.send_reboot(timeout=1.0)
+
+ d.addCallbacks(self._is_omci_frame, self._default_errback)
+ d.addCallback(self._check_status, RC.Success)
+
+ d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
+ d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
+ d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
+ d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
+ d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
+ d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
+ d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
+ d.addCallback(self._check_stats, snapshot, 'rx_late', snapshot['rx_late'])
+ d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
+ d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
+ return d
+
+ def test_message_send_with_omci_disabled(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ self.assertFalse(omci_cc.enabled)
+
+ # Successful MIB Reset
+ d = omci_cc.send_mib_reset(timeout=1.0)
+
+ def success_is_bad(_results):
+ assert False, 'This test should throw a failure/error'
+
+ def fail_fast(_failure):
+ pass
+ return None
+
+ d.addCallbacks(success_is_bad, fail_fast)
+ return d
+
+ def test_message_send_get_with_latency(self):
+ # Various tests of sending an OMCI message and it either
+ # getting a response or send catching some errors of
+ # importance
+ self.setup_one_of_each()
+ self.olt_handler.latency = 0.500 # 1/2 second
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+
+ # Successful MIB Reset
+ d = omci_cc.send_mib_reset(timeout=1.0)
+
+ d.addCallbacks(self._is_omci_frame, self._default_errback)
+ d.addCallback(self._check_status, RC.Success)
+
+ def check_latency_values(_):
+ self.assertGreaterEqual(omci_cc.reply_min, self.olt_handler.latency)
+ self.assertGreaterEqual(omci_cc.reply_max, self.olt_handler.latency)
+ self.assertGreaterEqual(omci_cc.reply_average, self.olt_handler.latency)
+
+ d.addCallback(check_latency_values)
+ return d
+
+ def test_message_failures(self):
+ # Various tests of sending an OMCI message and it fails
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ self.assertEqual(omci_cc.tx_frames, 0)
+ self.assertEqual(omci_cc.rx_frames, 0)
+ self.assertEqual(omci_cc.rx_unknown_tid, 0)
+ self.assertEqual(omci_cc.rx_timeouts, 0)
+ self.assertEqual(omci_cc.rx_late, 0)
+ self.assertEqual(omci_cc.tx_errors, 0)
+
+ # # Class ID not found
+ # d = omci_cc.send_mib_reset(timeout=1.0)
+ # self.assertTrue(True) # TODO: Implement
+ # todo: Test non-zero consecutive errors
+ #
+ # # Instance ID not found
+ # d = omci_cc.send_mib_reset(timeout=1.0)
+ # self.assertTrue(True) # TODO: Implement
+ # todo: Test non-zero consecutive errors
+ #
+ # # PON is disabled
+ # d = omci_cc.send_mib_reset(timeout=1.0)
+ # self.assertTrue(True) # TODO: Implement
+ # todo: Test non-zero consecutive errors
+ #
+ # # ONU is disabled
+ # d = omci_cc.send_mib_reset(timeout=1.0)
+ # self.assertTrue(True) # TODO: Implement
+ # todo: Test non-zero consecutive errors
+ #
+ # # ONU is not activated
+ # d = omci_cc.send_mib_reset(timeout=1.0)
+ # self.assertTrue(True) # TODO: Implement
+ # todo: Test non-zero consecutive errors
+
+ # TODO: make OLT send back an unknown TID (
+
+ # todo: Test non-zero consecutive errors
+ # todo: Send a good frame
+ # todo: Test zero consecutive errors
+ # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
+
+ def test_rx_unknown_me(self):
+ # ME without a known decoder
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # This is the ID ------+
+ # v
+ msg = '00fc2e0a00020000ff780000e00000010000000c' \
+ '0000000000000000000000000000000000000000' \
+ '00000028'
+
+ omci_cc.receive_message(hex2raw(msg))
+
+ # Note: After successful frame decode, a lookup of the corresponding request by
+ # TID is performed. None should be found, so we should see the Rx Unknown TID
+ # increment.
+ self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
+ self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'] + 1)
+ self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'] + 1)
+ self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
+ self.assertEqual(omci_cc.consecutive_errors, 0)
+
+ def test_rx_decode_unknown_me(self):
+ # ME without a known decoder
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # This is a MIB Upload Next Response. Where we would probably first see an
+ # unknown Class ID
+ #
+ # This is the ID ------+
+ # v
+ msg = '00fc2e0a00020000ff780001e000'
+ blob = '00010000000c0000000000000000000000000000000000000000'
+ msg += blob + '00000028'
+
+ # Dig into the internal method so we can get the returned frame
+ frame = omci_cc._decode_unknown_me(hex2raw(msg))
+
+ self.assertEqual(frame.fields['transaction_id'], 0x00fc)
+ self.assertEqual(frame.fields['message_type'], 0x2e)
+
+ omci_fields = frame.fields['omci_message'].fields
+
+ self.assertEqual(omci_fields['entity_class'], 0x0002)
+ self.assertEqual(omci_fields['entity_id'], 0x00)
+ self.assertEqual(omci_fields['object_entity_class'], 0x0ff78)
+ self.assertEqual(omci_fields['object_entity_id'], 0x01)
+ self.assertEqual(omci_fields['object_attributes_mask'], 0xe000)
+
+ data_fields = omci_fields['object_data']
+
+ decoded_blob = data_fields.get(UNKNOWN_CLASS_ATTRIBUTE_KEY)
+ self.assertIsNotNone(decoded_blob)
+ self.assertEqual(decoded_blob, blob)
+
+ def test_flush(self):
+ # Test flush of autonomous ONU queues
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # TODO: add more
+ self.assertTrue(True) # TODO: Implement
+
+ def test_avc_rx(self):
+ # Test flush of autonomous ONU queues
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # TODO: add more
+ self.assertTrue(True) # TODO: Implement
+
+ def test_rx_discard_if_disabled(self):
+ # ME without a known decoder
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = False
+ snapshot = self._snapshot_stats()
+
+ msg = '00fc2e0a00020000ff780000e00000010000000c' \
+ '0000000000000000000000000000000000000000' \
+ '00000028105a86ef'
+
+ omci_cc.receive_message(hex2raw(msg))
+
+ # Note: No counter increments
+ self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
+ self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
+ self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'])
+ self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
+
+ def test_omci_alarm_decode(self):
+ """
+ This test covers an issue discovered in Sept 2018 (JIRA-1213). It was
+ an exception during frame decode.
+ """
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # Frame from the JIRA issue
+ msg = '0000100a000b0102800000000000000000000000' \
+ '0000000000000000000000000000000000000015' \
+ '000000282d3ae0a6'
+
+ _results = omci_cc.receive_message(hex2raw(msg))
+
+ self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
+ self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
+ self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'])
+ self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'] + 1)
+ self.assertEqual(omci_cc.rx_onu_discards, snapshot['rx_onu_discards'])
+
+ def test_omci_avc_decode(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # Frame from the JIRA issue
+ msg = '0000110a0007000080004d4c2d33363236000000' \
+ '0000000020202020202020202020202020202020' \
+ '00000028'
+
+ _results = omci_cc.receive_message(hex2raw(msg))
+
+ self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
+ self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
+ self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'])
+ self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'] + 1)
+ self.assertEqual(omci_cc.rx_onu_discards, snapshot['rx_onu_discards'])
+
+ def test_omci_unknown_onu_decode(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # Frame from the JIRA issue
+ msg = '0000190a0007000080004d4c2d33363236000000' \
+ '0000000020202020202020202020202020202020' \
+ '00000028'
+
+ _results = omci_cc.receive_message(hex2raw(msg))
+
+ self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
+ self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
+ self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'])
+ self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'] + 1)
+ self.assertEqual(omci_cc.rx_onu_discards, snapshot['rx_onu_discards'] + 1)
+
+ def test_omci_bad_frame_decode(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # Frame from the JIRA issue
+ msg = '0020190a0007000080004d4c2d33363236000000' \
+ '0000000000000028'
+
+ _results = omci_cc.receive_message(hex2raw(msg))
+ # NOTE: Currently do not increment any Rx Discard counters, just throw it away
+ self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'] + 1)
+ self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
+ self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'] + 1)
+ self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
+ self.assertEqual(omci_cc.rx_onu_discards, snapshot['rx_onu_discards'])
+
+ def test_rx_decode_onu_g(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ msg = '001e2e0a0002000001000000e000424657530000' \
+ '0000000000000000000000324246575300107496' \
+ '00000028e7fb4a91'
+
+ omci_cc.receive_message(hex2raw(msg))
+
+ # Note: No counter increments
+ self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'] + 1)
+ self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
+ self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'] + 1)
+ self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
+
+ def test_rx_decode_extvlantagging(self):
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ msg = '030a290a00ab0201000d00000000001031323334' \
+ '3536373839303132333435363738393031323334' \
+ '000000281166d283'
+
+ omci_cc.receive_message(hex2raw(msg))
+
+ self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'] + 1)
+ self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
+ self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'] + 1)
+ self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
+
+ def _check_vlan_tag_op(self, results, attr, expected):
+ omci_msg = results.fields['omci_message']
+ data = omci_msg.fields['data']
+ val = data[attr]
+ self.assertEqual(expected, val)
+ return results
+
+ @skip('for unknown omci failure')
+ #@deferred()
+ def test_rx_table_get_extvlantagging(self):
+ self.setup_one_of_each()
+
+ onu = self.onu_handler.onu_mock
+ entity_id = 1
+ vlan_tag_op1 = VlanTaggingOperation(
+ filter_outer_priority=15,
+ filter_outer_vid=4096,
+ filter_outer_tpid_de=2,
+ filter_inner_priority=15,
+ filter_inner_vid=4096,
+ filter_inner_tpid_de=0,
+ filter_ether_type=0,
+ treatment_tags_to_remove=0,
+ treatment_outer_priority=15,
+ treatment_outer_vid=1234,
+ treatment_outer_tpid_de=0,
+ treatment_inner_priority=0,
+ treatment_inner_vid=4091,
+ treatment_inner_tpid_de=4,
+ )
+ vlan_tag_op2 = VlanTaggingOperation(
+ filter_outer_priority=14,
+ filter_outer_vid=1234,
+ filter_outer_tpid_de=5,
+ filter_inner_priority=1,
+ filter_inner_vid=2345,
+ filter_inner_tpid_de=1,
+ filter_ether_type=0,
+ treatment_tags_to_remove=1,
+ treatment_outer_priority=15,
+ treatment_outer_vid=2222,
+ treatment_outer_tpid_de=1,
+ treatment_inner_priority=1,
+ treatment_inner_vid=3333,
+ treatment_inner_tpid_de=5,
+ )
+ vlan_tag_op3 = VlanTaggingOperation(
+ filter_outer_priority=13,
+ filter_outer_vid=55,
+ filter_outer_tpid_de=1,
+ filter_inner_priority=7,
+ filter_inner_vid=4567,
+ filter_inner_tpid_de=1,
+ filter_ether_type=0,
+ treatment_tags_to_remove=1,
+ treatment_outer_priority=2,
+ treatment_outer_vid=1111,
+ treatment_outer_tpid_de=1,
+ treatment_inner_priority=1,
+ treatment_inner_vid=3131,
+ treatment_inner_tpid_de=5,
+ )
+ tbl = [vlan_tag_op1, vlan_tag_op2, vlan_tag_op3]
+ tblstr = str(vlan_tag_op1) + str(vlan_tag_op2) + str(vlan_tag_op3)
+
+ onu._omci_response[OP.Get.value][ExtendedVlanTaggingOperationConfigurationData.class_id] = {
+ entity_id: OmciFrame(transaction_id=0,
+ message_type=OmciGetResponse.message_id,
+ omci_message=OmciGetResponse(
+ entity_class=ExtendedVlanTaggingOperationConfigurationData.class_id,
+ entity_id=1,
+ success_code=RC.Success.value,
+ attributes_mask=ExtendedVlanTaggingOperationConfigurationData.mask_for(
+ 'received_frame_vlan_tagging_operation_table'),
+ data={'received_frame_vlan_tagging_operation_table': 16 * len(tbl)}
+ ))
+ }
+
+ rsp1 = binascii.a2b_hex(hexify(tblstr[0:OmciTableField.PDU_SIZE]))
+ rsp2 = binascii.a2b_hex(hexify(tblstr[OmciTableField.PDU_SIZE:]))
+ onu._omci_response[OP.GetNext.value][ExtendedVlanTaggingOperationConfigurationData.class_id] = {
+ entity_id: {0: {'failures':2,
+ 'frame':OmciFrame(transaction_id=0,
+ message_type=OmciGetNextResponse.message_id,
+ omci_message=OmciGetNextResponse(
+ entity_class=ExtendedVlanTaggingOperationConfigurationData.class_id,
+ entity_id=1,
+ success_code=RC.Success.value,
+ attributes_mask=ExtendedVlanTaggingOperationConfigurationData.mask_for(
+ 'received_frame_vlan_tagging_operation_table'),
+ data={'received_frame_vlan_tagging_operation_table': rsp1
+ }
+ ))},
+ 1: OmciFrame(transaction_id=0,
+ message_type=OmciGetNextResponse.message_id,
+ omci_message=OmciGetNextResponse(
+ entity_class=ExtendedVlanTaggingOperationConfigurationData.class_id,
+ entity_id=1,
+ success_code=RC.Success.value,
+ attributes_mask=ExtendedVlanTaggingOperationConfigurationData.mask_for(
+ 'received_frame_vlan_tagging_operation_table'),
+ data={'received_frame_vlan_tagging_operation_table': rsp2
+ }
+ ))
+ }
+ }
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+
+ msg = ExtendedVlanTaggingOperationConfigurationDataFrame(
+ entity_id,
+ attributes={'received_frame_vlan_tagging_operation_table':True}
+ )
+
+ snapshot = self._snapshot_stats()
+
+ frame = msg.get()
+ d = omci_cc.send(frame, timeout=5.0)
+
+ d.addCallbacks(self._is_omci_frame, self._default_errback, [OmciGetResponse])
+ d.addCallback(self._check_status, RC.Success)
+
+ d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 5)
+ d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 3)
+ d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
+ d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
+ d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
+ d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
+ d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'] + 2)
+ d.addCallback(self._check_stats, snapshot, 'rx_late', snapshot['rx_late'])
+ d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
+ d.addCallback(self._check_stats, snapshot, 'consecutive_errors', 0)
+ d.addCallback(self._check_vlan_tag_op, 'received_frame_vlan_tagging_operation_table', tbl)
+
+ return d
+
+ ##################################################################
+ # Start of tests specific to new stop_and_wait changes
+ #
+ def test_message_send_low_priority(self):
+ # self.setup_one_of_each(timeout_messages=True)
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # MIB Upload
+ d = omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
+ d.addCallback(self._check_stats, snapshot, 'lp_tx_queue_len', snapshot['lp_tx_queue_len'])
+ d.addCallback(self._check_stats, snapshot, 'hp_tx_queue_len', snapshot['hp_tx_queue_len'])
+ d.addCallback(self._check_stats, snapshot, 'max_lp_tx_queue', snapshot['max_lp_tx_queue'] + 1)
+ d.addCallback(self._check_stats, snapshot, 'max_hp_tx_queue', snapshot['max_hp_tx_queue'])
+
+ # Flush to get ready for next test (one frame queued)
+ omci_cc.flush()
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ self.adapter_agent.timeout_the_message = True
+ omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
+ omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
+
+ self.assertEqual(omci_cc.lp_tx_queue_len, 1)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.max_lp_tx_queue, 1)
+ self.assertEqual(omci_cc.max_hp_tx_queue, 0)
+
+ # Flush to get ready for next test (two queued and new max)
+ omci_cc.flush()
+ omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
+ omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
+ omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
+
+ self.assertEqual(omci_cc.lp_tx_queue_len, 2)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.max_lp_tx_queue, 2)
+ self.assertEqual(omci_cc.max_hp_tx_queue, 0)
+
+ def test_message_send_high_priority(self):
+ # self.setup_one_of_each(timeout_messages=True)
+ self.setup_one_of_each()
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+ snapshot = self._snapshot_stats()
+
+ # MIB Upload
+ d = omci_cc.send_mib_upload(high_priority=True)
+ d.addCallback(self._check_stats, snapshot, 'lp_tx_queue_len', snapshot['lp_tx_queue_len'])
+ d.addCallback(self._check_stats, snapshot, 'hp_tx_queue_len', snapshot['hp_tx_queue_len'])
+ d.addCallback(self._check_stats, snapshot, 'max_lp_tx_queue', snapshot['max_lp_tx_queue'])
+ d.addCallback(self._check_stats, snapshot, 'max_hp_tx_queue', snapshot['max_hp_tx_queue'] + 1)
+
+ # Flush to get ready for next test (one frame queued)
+ omci_cc.flush()
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ self.adapter_agent.timeout_the_message = True
+ omci_cc.send_mib_upload(high_priority=True)
+ omci_cc.send_mib_upload(high_priority=True)
+
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 1)
+ self.assertEqual(omci_cc.max_lp_tx_queue, 0)
+ self.assertEqual(omci_cc.max_hp_tx_queue, 1)
+
+ # Flush to get ready for next test (two queued and new max)
+ omci_cc.flush()
+ omci_cc.send_mib_upload(high_priority=True)
+ omci_cc.send_mib_upload(high_priority=True)
+ omci_cc.send_mib_upload(high_priority=True)
+
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 2)
+ self.assertEqual(omci_cc.max_lp_tx_queue, 0)
+ self.assertEqual(omci_cc.max_hp_tx_queue, 2)
+
+ def test_message_send_and_cancel(self):
+ global error_reason
+ global successful
+ # Do not send messages to adapter_agent
+ self.setup_one_of_each(timeout_messages=True)
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+
+ def success(_results):
+ global successful
+ successful = True
+
+ def failure(reason):
+ global error_reason
+ error_reason = reason
+
+ def notCalled(reason):
+ assert isinstance(reason, Failure), 'Should not be called with success'
+
+ # Cancel one that is actively being sent
+ d = omci_cc.send_mib_upload(high_priority=False)
+ d.addCallbacks(success, failure)
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ d.cancel()
+ self.assertIsInstance(error_reason, Failure)
+ self.assertFalse(successful)
+ self.assertTrue(d.called)
+
+ self.assertEqual(omci_cc.max_lp_tx_queue, 1)
+ self.assertEqual(omci_cc.max_hp_tx_queue, 0)
+
+ # Flush to get ready for next test (one running, one queued, cancel the
+ # running one, so queued runs)
+ omci_cc.flush()
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ error_reason = None
+ d1 = omci_cc.send_mib_upload(high_priority=False)
+ d2 = omci_cc.send_mib_upload(high_priority=False)
+ d1.addCallbacks(success, failure)
+ d2.addCallbacks(notCalled, notCalled)
+ self.assertEqual(omci_cc.lp_tx_queue_len, 1)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ d1.cancel()
+ self.assertIsInstance(error_reason, Failure)
+ self.assertFalse(successful)
+ self.assertTrue(d1.called)
+ self.assertFalse(d2.called)
+
+ self.assertEqual(omci_cc.max_lp_tx_queue, 1)
+ self.assertEqual(omci_cc.max_hp_tx_queue, 0)
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ # Flush to get ready for next test (one running, one queued, cancel the queued one)
+
+ omci_cc.flush()
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ error_reason = None
+ d3 = omci_cc.send_mib_upload(timeout=55, high_priority=False)
+ d4 = omci_cc.send_mib_upload(timeout=55, high_priority=False)
+ d5 = omci_cc.send_mib_upload(timeout=55, high_priority=False)
+ d3.addCallbacks(notCalled, notCalled)
+ d4.addCallbacks(success, failure)
+ d5.addCallbacks(notCalled, notCalled)
+ self.assertEqual(omci_cc.lp_tx_queue_len, 2)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ d4.cancel()
+ self.assertIsInstance(error_reason, Failure)
+ self.assertFalse(successful)
+ self.assertFalse(d3.called)
+ self.assertTrue(d4.called)
+ self.assertFalse(d5.called)
+
+ def test_message_send_low_and_high_priority(self):
+ self.setup_one_of_each(timeout_messages=True)
+
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+
+ omci_cc.send_mib_reset(high_priority=False)
+ omci_cc.send_mib_reset(high_priority=True)
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ omci_cc.flush()
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ omci_cc.send_mib_reset(high_priority=False)
+ omci_cc.send_mib_reset(high_priority=True)
+ omci_cc.send_mib_reset(high_priority=False)
+ omci_cc.send_mib_reset(high_priority=True)
+ self.assertEqual(omci_cc.lp_tx_queue_len, 1)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 1)
+
+ def test_no_sw_download_and_mib_upload_at_same_time(self):
+ # Section B.2.3 of ITU G.988-2017 specifies that a MIB
+ # upload or software download at a given priority level
+ # is not allowed while a similar action in the other
+ # priority level is in progress. Relates to possible memory
+ # consumption/needs on the ONU.
+ #
+ # OMCI_CC only checks if the commands are currently in
+ # progress. ONU should reject messages if the upload/download
+ # is in progress (but not an active request is in progress).
+
+ self.setup_one_of_each(timeout_messages=True)
+ omci_cc = self.onu_handler.omci_cc
+ omci_cc.enabled = True
+
+ mib_upload_msgs = [omci_cc.send_mib_upload,
+ # omci_cc.send_mib_upload_next
+ ]
+ sw_download_msgs = [omci_cc.send_start_software_download,
+ # omci_cc.send_download_section,
+ # omci_cc.send_end_software_download
+ ]
+
+ for upload in mib_upload_msgs:
+ for download in sw_download_msgs:
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ upload(high_priority=False)
+ download(1, 1, 1, high_priority=True) # Should stall send-next 50mS
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 1)
+
+ omci_cc.flush()
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ upload(high_priority=True)
+ download(1, 1, 1, high_priority=False) # Should stall send-next 50mS
+ self.assertEqual(omci_cc.lp_tx_queue_len, 1)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ omci_cc.flush()
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ download(1, 1, 1, high_priority=False)
+ upload(high_priority=True) # Should stall send-next 50mS
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 1)
+
+ omci_cc.flush()
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ download(1, 1, 1, high_priority=True)
+ upload(high_priority=False) # Should stall send-next 50mS)
+ self.assertEqual(omci_cc.lp_tx_queue_len, 1)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ omci_cc.flush()
+ self.assertEqual(omci_cc.lp_tx_queue_len, 0)
+ self.assertEqual(omci_cc.hp_tx_queue_len, 0)
+
+ # Some more ideas for tests that we could add
+ # Send explicit tid that is not valid
+ # - Look at top of 'Send' method and test all the error conditions could may hit
+
+ # Send multiple and have the OLT proxy throw an exception. Should call errback and
+ # schedule remainder in queue to still tx.
+
+ # Send a frame and then inject a response and test the RX logic out, including late
+ # rx and retries by the OMCI_CC transmitter.
+
+
+if __name__ == '__main__':
+ main()
diff --git a/test/unit/extensions/omci/test_omci_configuration.py b/test/unit/extensions/omci/test_omci_configuration.py
new file mode 100644
index 0000000..1cdfd21
--- /dev/null
+++ b/test/unit/extensions/omci/test_omci_configuration.py
@@ -0,0 +1,484 @@
+#
+# 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()
+
diff --git a/test/unit/extensions/omci/test_onu_device_entry.py b/test/unit/extensions/omci/test_onu_device_entry.py
new file mode 100644
index 0000000..9e917f6
--- /dev/null
+++ b/test/unit/extensions/omci/test_onu_device_entry.py
@@ -0,0 +1,256 @@
+#
+# 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 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.omci_defs import *
+from pyvoltha.common.utils.asleep import asleep
+from pyvoltha.adapters.extensions.omci.database.mib_db_api import DEVICE_ID_KEY, CLASS_ID_KEY, CREATED_KEY, \
+ MODIFIED_KEY, MDS_KEY, LAST_SYNC_KEY, VERSION_KEY, DatabaseStateError
+from pyvoltha.adapters.extensions.omci.database.mib_db_dict import MibDbVolatileDict
+
+
+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
+
+
+def chunk(indexable, chunk_size):
+ for i in range(0, len(indexable), chunk_size):
+ yield indexable[i:i + chunk_size]
+
+
+def hex2raw(hex_string):
+ return ''.join(chr(int(byte, 16)) for byte in chunk(hex_string, 2))
+
+
+class TestOnuDeviceEntry(TestCase):
+ """
+ Test the ONU Device Entry methods
+ """
+ def setUp(self):
+ self.adapter_agent = MockAdapterAgent()
+
+ custom = deepcopy(OpenOmciAgentDefaults)
+ custom['mib-synchronizer']['database'] = MibDbVolatileDict
+
+ self.agent = OpenOMCIAgent(MockCore, support_classes=custom)
+ self.agent.start()
+
+ def tearDown(self):
+ if self.agent is not None:
+ self.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)
+
+ def test_add_remove_device(self):
+ self.setup_one_of_each()
+ self.assertEqual(len(self.agent.device_ids()), 0)
+
+ onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
+ self.adapter_agent)
+ self.assertIsNotNone(onu_device)
+ self.assertEqual(len(self.agent.device_ids()), 1)
+ self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)
+
+ # No MIB if not started
+ assert_raises(KeyError, onu_device.query_mib)
+
+ self.agent.remove_device(DEFAULT_ONU_DEVICE_ID)
+ self.assertEqual(len(self.agent.device_ids()), 1)
+
+ def test_delete_device(self):
+ self.setup_one_of_each()
+ self.assertEqual(len(self.agent.device_ids()), 0)
+
+ onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
+ self.adapter_agent)
+ self.assertIsNotNone(onu_device)
+ self.assertEqual(len(self.agent.device_ids()), 1)
+ self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)
+ # Can delete if it was not started
+ onu_device.delete()
+ self.assertEqual(len(self.agent.device_ids()), 0)
+
+ ##########################################
+ # Delete of ONU device okay if it is started
+ onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
+ self.adapter_agent)
+ self.assertIsNotNone(onu_device)
+ self.assertEqual(len(self.agent.device_ids()), 1)
+ self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)
+
+ # Start it and then delete it
+ onu_device.start()
+ onu_device.delete()
+ self.assertEqual(len(self.agent.device_ids()), 0)
+
+ @deferred(timeout=5)
+ def test_mib_query_fails_if_dev_not_started(self):
+ self.setup_one_of_each()
+
+ onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
+ self.adapter_agent)
+ self.assertIsNotNone(onu_device)
+ self.assertEqual(len(self.agent.device_ids()), 1)
+ self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)
+
+ def not_called(_reason):
+ assert False, 'Should never be called'
+
+ def check_status(_results):
+ # Device not yet started. Query should fail with KeyError since
+ # ONU is not in database yet
+ assert_raises(KeyError, onu_device.query_mib)
+
+ # Yield context so that MIB Database callLater runs. This is a waiting
+ # Async task from when the OpenOMCIAgent was started.
+ d = asleep(0.2)
+ d.addCallbacks(check_status, not_called)
+
+ return d
+
+ @deferred(timeout=5)
+ def test_mib_query_ok_if_dev_started(self):
+ self.setup_one_of_each()
+
+ onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
+ self.adapter_agent)
+ self.assertIsNotNone(onu_device)
+ self.assertEqual(len(self.agent.device_ids()), 1)
+ self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)
+
+ def not_called(_reason):
+ onu_device.stop()
+ assert False, 'Should never be called'
+
+ def check_status(_results):
+ # Device started. Query will succeed but nothing should be populated
+ # but the most basic items
+
+ results = onu_device.query_mib()
+ self.assertTrue(isinstance(results, dict))
+ self.assertEqual(results.get(DEVICE_ID_KEY), DEFAULT_ONU_DEVICE_ID)
+
+ self.assertIsNotNone(results.get(VERSION_KEY))
+ self.assertIsNotNone(results.get(CREATED_KEY))
+ self.assertIsNone(results.get(MODIFIED_KEY)) # Created! but not yet modified
+
+ self.assertEqual(results.get(MDS_KEY), 0)
+ self.assertIsNone(results.get(LAST_SYNC_KEY))
+
+ self.assertIsNone(results.get(CLASS_ID_KEY))
+
+ # Stopping still allows a query. Note you just delete a device
+ # to clean up any associated databases
+ onu_device.stop()
+ results = onu_device.query_mib()
+ self.assertTrue(isinstance(results, dict))
+
+ # 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
+ onu_device.start()
+ d = asleep(0.2)
+ d.addCallbacks(check_status, not_called)
+
+ return d
+
+ @deferred(timeout=5)
+ def test_delete_scrubs_mib(self):
+ self.setup_one_of_each()
+
+ onu_device = self.agent.add_device(DEFAULT_ONU_DEVICE_ID,
+ self.adapter_agent)
+ self.assertIsNotNone(onu_device)
+ self.assertEqual(len(self.agent.device_ids()), 1)
+ self.assertEqual(self.agent.get_device(DEFAULT_ONU_DEVICE_ID), onu_device)
+
+ def not_called(_reason):
+ onu_device.stop()
+ assert False, 'Should never be called'
+
+ def check_status(_results):
+ # Device started. Query will succeed but nothing should be populated
+ # but the most basic items
+
+ results = onu_device.query_mib()
+ self.assertTrue(isinstance(results, dict))
+ self.assertEqual(results.get(DEVICE_ID_KEY), DEFAULT_ONU_DEVICE_ID)
+
+ # Delete should wipe out any MIB data. Note that a delete of a started
+ # or stopped ONU device is allowed. In this case we are deleting a
+ # started ONU Device
+
+ onu_device.delete()
+ assert_raises(Exception, onu_device.query_mib)
+ # TODO: When capabilities are supported, make sure capabilities get cleared as well
+
+ # 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
+ onu_device.start()
+ d = asleep(0.2)
+ d.addCallbacks(check_status, not_called)
+
+ return d
+
+ # TODO: Test pub/sub interface if possible
+ # TODO: Test custom/vendor-specific ME support
+ # TODO: Test override of various state machines or OMCI tasks if possible
+
+
+if __name__ == '__main__':
+ main()
+
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()
+
diff --git a/test/unit/extensions/omci/test_task_runner.py b/test/unit/extensions/omci/test_task_runner.py
new file mode 100644
index 0000000..e35e151
--- /dev/null
+++ b/test/unit/extensions/omci/test_task_runner.py
@@ -0,0 +1,484 @@
+#
+# 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 TestCase, main
+from nose.tools import raises
+from twisted.internet import defer
+from twisted.internet.defer import inlineCallbacks, returnValue, CancelledError
+from mock.mock_task import SimpleTask
+from nose.twistedtools import deferred
+from pyvoltha.adapters.extensions.omci.tasks.task_runner import TaskRunner
+
+DEVICE_ID = 'omci-unit-tests'
+
+
+class TestTaskRunner(TestCase):
+ """
+ Test the Task Runner Object
+ """
+
+ def setUp(self):
+ # defer.setDebugging(True)
+ self.runner = TaskRunner(DEVICE_ID)
+
+ def tearDown(self):
+ r, self.runner = self.runner, None
+ r.stop()
+
+ def test_default_init(self):
+ self.assertFalse(self.runner.active)
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.successful_tasks_completed, 0)
+ self.assertEqual(self.runner.failed_tasks, 0)
+
+ def test_start_stop(self):
+ self.assertFalse(self.runner.active)
+
+ self.runner.start()
+ self.assertTrue(self.runner.active)
+
+ self.runner.stop()
+ self.assertFalse(self.runner.active)
+
+ def unexpected_error(self, _failure):
+ self.assertEqual('Should not be here, expected success', _failure)
+
+ def unexpected_success(self, _results):
+ self.assertEqual('Should not be here, expected a failure', _results)
+
+ def test_simple_task_init(self):
+ t = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=0,
+ success=True, value=0, delay=0)
+
+ self.assertEqual(t.priority, 0)
+ self.assertGreater(t.task_id, 0)
+ self.assertTrue(t.exclusive)
+ self.assertFalse(t.deferred.called)
+
+ def test_task_defaults_and_bound(self):
+ # Make sure no one has changed some declared defaults/max/min values
+ from pyvoltha.adapters.extensions.omci.tasks.task import Task
+ self.assertEqual(128, Task.DEFAULT_PRIORITY)
+ self.assertEqual(0, Task.MIN_PRIORITY)
+ self.assertEqual(255, Task.MAX_PRIORITY)
+ self.assertEqual(10, Task.DEFAULT_WATCHDOG_SECS)
+ self.assertEqual(3, Task.MIN_WATCHDOG_SECS)
+ self.assertEqual(60, Task.MAX_WATCHDOG_SECS)
+
+ @raises(AssertionError)
+ def test_task_priority_min(self):
+ from pyvoltha.adapters.extensions.omci.tasks.task import Task
+ _ = SimpleTask(None, DEVICE_ID, priority=Task.MIN_PRIORITY - 1)
+
+ @raises(AssertionError)
+ def test_task_priority_max(self):
+ from pyvoltha.adapters.extensions.omci.tasks.task import Task
+ _ = SimpleTask(None, DEVICE_ID, priority=Task.MAX_PRIORITY + 1)
+
+ @raises(AssertionError)
+ def test_task_watchdog_min(self):
+ from pyvoltha.adapters.extensions.omci.tasks.task import Task
+ _ = SimpleTask(None, DEVICE_ID, watchdog_timeout=Task.MIN_WATCHDOG_SECS - 0.000001)
+
+ @raises(AssertionError)
+ def test_task_watchdog_max(self):
+ from pyvoltha.adapters.extensions.omci.tasks.task import Task
+ _ = SimpleTask(None, DEVICE_ID, watchdog_timeout=Task.MAX_WATCHDOG_SECS + 0.000001)
+
+ @deferred(timeout=5)
+ def test_simple_success(self):
+ expected_result = 123
+
+ t = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=0,
+ success=True, value=expected_result, delay=0)
+
+ d = self.runner.queue_task(t)
+ self.assertEqual(self.runner.pending_tasks, 1)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.runner.start()
+
+ def check_results(results):
+ self.assertEqual(results, expected_result)
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.successful_tasks_completed, 1)
+ self.assertEqual(self.runner.failed_tasks, 0)
+ self.assertTrue(self.runner.active)
+ return results
+
+ d.addCallbacks(check_results, self.unexpected_error)
+ return d
+
+ @raises(Exception)
+ @deferred(timeout=5)
+ def test_simple_failure(self):
+ self.expected_failure = Exception('Testing a task failure')
+
+ t = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=0,
+ success=False, value=self.expected_failure,
+ delay=0)
+
+ d = self.runner.queue_task(t)
+ self.assertEqual(self.runner.pending_tasks, 1)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.runner.start()
+
+ def expected_failure(failure):
+ self.assertEqual(failure, self.expected_failure)
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.successful_tasks_completed, 0)
+ self.assertEqual(self.runner.failed_tasks, 1)
+ self.assertTrue(self.runner.active)
+ return failure
+
+ d.addCallbacks(self.unexpected_success, expected_failure)
+ return d
+
+ @deferred(timeout=5)
+ def test_priority(self):
+ self.last_value_set = 0
+
+ t1 = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=1,
+ success=True, value=1, delay=0)
+
+ t2 = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=2, # Should finish first
+ success=True, value=2, delay=0)
+
+ d1 = self.runner.queue_task(t1)
+ d2 = self.runner.queue_task(t2)
+
+ def set_last_value(results):
+ self.last_value_set = results
+
+ d1.addCallbacks(set_last_value, self.unexpected_error)
+ d2.addCallbacks(set_last_value, self.unexpected_error)
+
+ self.assertEqual(self.runner.pending_tasks, 2)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ d = defer.gatherResults([d1, d2], consumeErrors=True)
+
+ def check_results(_):
+ self.assertEqual(self.last_value_set, 1)
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.successful_tasks_completed, 2)
+
+ d.addCallbacks(check_results, self.unexpected_error)
+
+ self.runner.start()
+ return d
+
+ @inlineCallbacks
+ def check_that_t1_t2_running_and_last_is_not(self, results):
+ from pyvoltha.common.utils.asleep import asleep
+ yield asleep(0.1)
+
+ self.assertEqual(self.runner.pending_tasks, 1)
+ self.assertEqual(self.runner.running_tasks, 2)
+ self.assertEqual(self.runner.successful_tasks_completed, 1)
+
+ returnValue(results)
+
+ @deferred(timeout=10)
+ def test_concurrent(self):
+ blocker = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=10,
+ success=True, value=1, delay=0.5)
+
+ t1 = SimpleTask(None, DEVICE_ID,
+ exclusive=False, priority=9,
+ success=True, value=1, delay=2)
+
+ t2 = SimpleTask(None, DEVICE_ID,
+ exclusive=False, priority=9,
+ success=True, value=1, delay=2)
+
+ last = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=8,
+ success=True, value=1, delay=0)
+
+ d0 = self.runner.queue_task(blocker)
+ d0.addCallbacks(self.check_that_t1_t2_running_and_last_is_not,
+ self.unexpected_error)
+
+ d1 = self.runner.queue_task(t1)
+ d2 = self.runner.queue_task(t2)
+ d3 = self.runner.queue_task(last)
+
+ self.assertEqual(self.runner.pending_tasks, 4)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ d = defer.gatherResults([d0, d1, d2, d3], consumeErrors=True)
+
+ def check_final_results(_):
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.successful_tasks_completed, 4)
+ self.assertEqual(self.runner.failed_tasks, 0)
+
+ d.addCallbacks(check_final_results, self.unexpected_error)
+
+ self.runner.start()
+ return d
+
+ @raises(CancelledError)
+ @deferred(timeout=2)
+ def test_cancel_queued(self):
+ t = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=9,
+ success=True, value=1, delay=0)
+
+ d = self.runner.queue_task(t)
+ self.assertEqual(self.runner.pending_tasks, 1)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ self.runner.cancel_task(t.task_id)
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ return d
+
+ @raises(CancelledError)
+ @deferred(timeout=2)
+ def test_task_stop_queued(self):
+ t = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=9,
+ success=True, value=1, delay=0)
+
+ d = self.runner.queue_task(t)
+ self.assertEqual(self.runner.pending_tasks, 1)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ t.stop()
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ return d
+
+ def test_task_stop_not_queued(self):
+ t = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=9,
+ success=True, value=1, delay=0)
+
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ t.stop()
+ self.assertFalse(t.running)
+
+ @deferred(timeout=3)
+ def test_task_runner_cancel_running(self):
+ # Both task run in parallel but t1 will finish first and
+ # will request t2 to terminate by calling the TaskRunner's
+ # cancel task method
+ t1 = SimpleTask(None, DEVICE_ID,
+ exclusive=False, priority=9,
+ success=True, value=1, delay=0.5)
+ t2 = SimpleTask(None, DEVICE_ID,
+ exclusive=False, priority=9,
+ success=True, value=1, delay=200)
+
+ d1 = self.runner.queue_task(t1)
+ d2 = self.runner.queue_task(t2)
+
+ self.assertEqual(self.runner.pending_tasks, 2)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ def kill_task_t2(_, task_2_id):
+ # Called on successful completion of task t1
+ self.assertIsInstance(task_2_id, int)
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 1)
+
+ # Cancel task runner and t2 task ID
+ self.runner.cancel_task(task_2_id)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ d1.addCallbacks(kill_task_t2, self.unexpected_error,
+ callbackArgs=[t2.task_id])
+
+ def expected_error(failure):
+ self.assertTrue(isinstance(failure.value, CancelledError))
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.successful_tasks_completed, 1)
+ self.assertEqual(self.runner.failed_tasks, 1)
+
+ # T2 should not finish successfully, should get a cancel error
+ d2.addCallbacks(self.unexpected_success, expected_error)
+
+ # Run it
+ self.runner.start()
+ return defer.gatherResults([d1, d2], consumeErrors=True)
+
+ @deferred(timeout=3)
+ def test_task_stop_running(self):
+ # Run two tasks where T1 completes first and requests T2 to be
+ # canceled by calling T2's stop method
+
+ t1 = SimpleTask(None, DEVICE_ID,
+ exclusive=False, priority=9,
+ success=True, value=1, delay=0.5)
+ t2 = SimpleTask(None, DEVICE_ID,
+ exclusive=False, priority=9,
+ success=True, value=1, delay=200)
+
+ d1 = self.runner.queue_task(t1)
+ d2 = self.runner.queue_task(t2)
+
+ self.assertEqual(self.runner.pending_tasks, 2)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ def kill_task_t2(_, task_2):
+ # Called on successful completion of task t1
+ self.assertIsInstance(task_2, SimpleTask)
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 1)
+
+ # Cancel by telling the task to stop itself
+ task_2.stop()
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ d1.addCallbacks(kill_task_t2, self.unexpected_error,
+ callbackArgs=[t2])
+
+ def expected_error(failure):
+ self.assertTrue(isinstance(failure.value, CancelledError))
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.successful_tasks_completed, 1)
+ self.assertEqual(self.runner.failed_tasks, 1)
+
+ # T2 should not finish successfully, should get a cancel error
+ d2.addCallbacks(self.unexpected_success, expected_error)
+
+ # Run it
+ self.runner.start()
+ return defer.gatherResults([d1, d2], consumeErrors=True)
+
+ @deferred(timeout=3)
+ def test_task_cancel_not_queued(self):
+ t = SimpleTask(None, DEVICE_ID,
+ exclusive=True, priority=9,
+ success=True, value=1, delay=0)
+
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ def expected_error(failure):
+ self.assertTrue(isinstance(failure.value, CancelledError))
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.successful_tasks_completed, 0)
+ self.assertEqual(self.runner.failed_tasks, 0)
+ # self.fail(msg='made it here') # Uncomment to verify called
+
+ t.deferred.addCallbacks(self.unexpected_success, expected_error)
+
+ self.runner.start()
+ t.deferred.cancel()
+ self.assertFalse(t.running)
+ return t.deferred
+
+ @deferred(timeout=3)
+ def test_task_deferred_cancel_running(self):
+ # Run two tasks where T1 completes first and requests T2 to be
+ # canceled by doing a 'cancel' on T2's deferred
+
+ t1 = SimpleTask(None, DEVICE_ID,
+ exclusive=False, priority=9,
+ success=True, value=1, delay=0.5)
+ t2 = SimpleTask(None, DEVICE_ID,
+ exclusive=False, priority=9,
+ success=True, value=1, delay=200)
+
+ d1 = self.runner.queue_task(t1)
+ d2 = self.runner.queue_task(t2)
+
+ self.assertEqual(self.runner.pending_tasks, 2)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ def kill_task_t2(_, deferred_2):
+ # Called on successful completion of task t1
+ self.assertIsInstance(deferred_2, defer.Deferred)
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 1)
+
+ # Cancel the deferred for T2
+ deferred_2.cancel()
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ d1.addCallbacks(kill_task_t2, self.unexpected_error,
+ callbackArgs=[t2.deferred])
+
+ def expected_error(failure):
+ self.assertTrue(isinstance(failure.value, CancelledError))
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.successful_tasks_completed, 1)
+ self.assertEqual(self.runner.failed_tasks, 1)
+ # self.fail(msg='made it here') # Uncomment to verify called
+
+ # T2 should not finish successfully, should get a cancel error
+ d2.addCallbacks(self.unexpected_success, expected_error)
+
+ # Run it
+ self.runner.start()
+ return defer.gatherResults([d1, d2], consumeErrors=True)
+
+ @deferred(timeout=3)
+ def test_watchdog_timeout(self):
+ t = SimpleTask(None, DEVICE_ID, delay=2)
+
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.watchdog_timeouts, 0)
+
+ # Actual watchdog minimum is probably to long for an automated test, reach
+ # around and force ti to something smaller (kids, don't try this at home)
+
+ t._watchdog_timeout = 0.1
+ self.runner.queue_task(t)
+
+ self.assertEqual(self.runner.pending_tasks, 1)
+ self.assertEqual(self.runner.running_tasks, 0)
+
+ def expected_error(failure):
+ from pyvoltha.adapters.extensions.omci.tasks.task import WatchdogTimeoutFailure
+ self.assertTrue(isinstance(failure.value, WatchdogTimeoutFailure))
+ self.assertEqual(self.runner.pending_tasks, 0)
+ self.assertEqual(self.runner.running_tasks, 0)
+ self.assertEqual(self.runner.successful_tasks_completed, 0)
+ self.assertEqual(self.runner.failed_tasks, 1)
+ self.assertEqual(self.runner.watchdog_timeouts, 1)
+ self.assertEqual(self.runner.last_watchdog_failure_task, t.name)
+ # self.fail(msg='made it here') # Uncomment to verify called
+
+ t.deferred.addCallbacks(self.unexpected_success, expected_error)
+
+ # Run it
+ self.runner.start()
+ return t.deferred
+
+
+if __name__ == '__main__':
+ main()