| Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 1 | # |
| 2 | # Copyright 2017 the original author or authors. |
| 3 | # |
| 4 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | # you may not use this file except in compliance with the License. |
| 6 | # You may obtain a copy of the License at |
| 7 | # |
| 8 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | # |
| 10 | # Unless required by applicable law or agreed to in writing, software |
| 11 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | # See the License for the specific language governing permissions and |
| 14 | # limitations under the License. |
| 15 | # |
| 16 | |
| 17 | from mock_adapter_agent import MockProxyAddress, MockDevice |
| 18 | from pyvoltha.adapters.extensions.omci.omci_cc import * |
| 19 | from pyvoltha.adapters.extensions.omci.omci_entities import entity_id_to_class_map |
| 20 | |
| 21 | |
| 22 | class MockOnuHandler(MockDevice): |
| 23 | """ |
| 24 | Minimal class to handle ONU needs in OpenOMCI testing |
| 25 | |
| 26 | So that we do not have to duplicate the IAdapter functionality, just |
| 27 | the handler, the OLT and ONU handlers are derived from a mock Device |
| 28 | base class so that we can access the _devices map and get either a |
| 29 | Device to play with (like the real thing) or the associated handler |
| 30 | """ |
| 31 | def __init__(self, adapter_agent, parent_id, device_id, pon_id, onu_id): |
| 32 | |
| 33 | self.proxy_address = MockProxyAddress(parent_id, pon_id, onu_id) |
| 34 | super(MockOnuHandler, self).__init__(device_id, self.proxy_address) |
| 35 | |
| 36 | self.device_id = device_id |
| 37 | self.device = self |
| 38 | self._adapter_agent = adapter_agent |
| 39 | |
| 40 | self.onu_mock = None |
| 41 | self.omci_cc = OMCI_CC(adapter_agent, device_id, me_map=entity_id_to_class_map) |
| 42 | |
| 43 | # Items that you can change to perform various test failures |
| 44 | |
| 45 | self._enabled = True |
| 46 | |
| 47 | def tearDown(self): |
| 48 | """Test case cleanup""" |
| 49 | if self.onu_mock is not None: |
| 50 | self.onu_mock.tearDown() |
| 51 | self.onu_mock = None |
| 52 | |
| 53 | @property |
| 54 | def enabled(self): |
| 55 | return self._enabled |
| 56 | |
| 57 | @enabled.setter |
| 58 | def enabled(self, value): |
| 59 | if self._enabled != value: |
| 60 | self._enabled = value |
| 61 | olt = self._adapter_agent.get_device(self.proxy_address.device_id) |
| 62 | if olt is not None and self.proxy_address.channel_id in olt.enabled_pons: |
| 63 | if self._enabled: |
| 64 | olt.activated_onus.add(self.serial_number) |
| 65 | else: |
| 66 | olt.activated_onus.discard(self.serial_number) |
| 67 | |
| 68 | # Begin minimal set of needed IAdapter interfaces |
| 69 | |
| 70 | # TODO: Implement minimal functionality |
| 71 | |
| 72 | def send_proxied_message(self, proxy_address, msg): |
| 73 | assert False, 'OpenOMCI will implement this for the MOCK ONU' |
| 74 | |
| 75 | def receive_proxied_message(self, _, msg): |
| 76 | # Rx of OMCI message from MOCK OLT |
| 77 | |
| 78 | if self.omci_cc is not None and self.enabled: |
| 79 | self.omci_cc.receive_message(msg.decode('hex')) |