| # |
| # 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 __future__ import absolute_import |
| import structlog |
| import six |
| # 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()) |
| if self.root is None: |
| from nose import SkipTest |
| msg = "TODO: Unit tests involving the kv-store are not yet supported by pyvoltha" |
| raise SkipTest(msg) |
| |
| 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 six.itervalues(self._devices): |
| 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 six.itervalues(self._devices) |
| 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) |