ADTRAN upstreamed automation
Change-Id: I47f8ae527c4f8a5d3b106bbb7d8392bf79964431
diff --git a/voltha/adapters/adtran_olt/adtran_device_handler.py b/voltha/adapters/adtran_olt/adtran_device_handler.py
index 6460df4..5a3598f 100644
--- a/voltha/adapters/adtran_olt/adtran_device_handler.py
+++ b/voltha/adapters/adtran_olt/adtran_device_handler.py
@@ -867,7 +867,7 @@
# Drop registration for ONU detection
# self.adapter_agent.unregister_for_onu_detect_state(self.device.id)
self._suspend_heartbeat()
-
+
# Update the operational status to UNKNOWN
device.oper_status = OperStatus.UNKNOWN
device.connect_status = ConnectStatus.UNREACHABLE
diff --git a/voltha/adapters/adtran_olt/flow/evc_map.py b/voltha/adapters/adtran_olt/flow/evc_map.py
index d1d9385..8dc601a 100644
--- a/voltha/adapters/adtran_olt/flow/evc_map.py
+++ b/voltha/adapters/adtran_olt/flow/evc_map.py
@@ -359,7 +359,7 @@
results = yield self._handler.netconf_client.edit_config(map_xml)
self._installed = results.ok
self._needs_update = results.ok
- self._status_message = '' if results.ok else results.error
+ self.status = '' if results.ok else results.error
if results.ok:
self._existing_acls.update(work_acls)
diff --git a/voltha/adapters/adtran_olt/flow/flow_entry.py b/voltha/adapters/adtran_olt/flow/flow_entry.py
index f031064..e129b5b 100644
--- a/voltha/adapters/adtran_olt/flow/flow_entry.py
+++ b/voltha/adapters/adtran_olt/flow/flow_entry.py
@@ -413,10 +413,8 @@
if self._flow_direction in FlowEntry.downstream_flow_types:
status = self._apply_downstream_mods()
-
elif self._flow_direction in FlowEntry.upstream_flow_types:
status = self._apply_upstream_mods()
-
else:
# TODO: Need to code this - Perhaps this is an NNI_PON for Multicast support?
log.error('unsupported-flow-direction')
@@ -545,10 +543,10 @@
def _decode_flow_direction(self):
# Determine direction of the flow
def port_type(port_number):
- if port_number in self._handler.northbound_ports:
+ if port_number in self.handler.northbound_ports:
return FlowEntry.PortType.NNI
- elif port_number in self._handler.southbound_ports:
+ elif port_number in self.handler.southbound_ports:
return FlowEntry.PortType.PON
elif port_number <= OFPP_MAX:
diff --git a/voltha/adapters/adtran_olt/resources/adtran_olt_resource_manager.py b/voltha/adapters/adtran_olt/resources/adtran_olt_resource_manager.py
index 93dfd8a..b95c99d 100644
--- a/voltha/adapters/adtran_olt/resources/adtran_olt_resource_manager.py
+++ b/voltha/adapters/adtran_olt/resources/adtran_olt_resource_manager.py
@@ -122,37 +122,6 @@
return alloc_id
- def get_gemport_id(self, pon_intf_onu_id, num_of_id=1):
- # TODO: Remove this if never used
- # Derive the pon_intf and onu_id from the pon_intf_onu_id tuple
- pon_intf = pon_intf_onu_id[0]
- onu_id = pon_intf_onu_id[1]
- uni_id = pon_intf_onu_id[2]
- assert False, 'unused function'
-
- gemport_id_list = self.resource_managers[pon_intf].get_current_gemport_ids_for_onu(
- pon_intf_onu_id)
- if gemport_id_list and len(gemport_id_list) > 0:
- return gemport_id_list
-
- gemport_id_list = self.resource_mgrs[pon_intf].get_resource_id(
- pon_intf_id=pon_intf,
- resource_type=PONResourceManager.GEMPORT_ID,
- num_of_id=num_of_id
- )
-
- if gemport_id_list and len(gemport_id_list) == 0:
- self.log.error("no-gemport-id-available")
- return None
-
- # update the resource map on KV store with the list of gemport_id
- # allocated for the pon_intf_onu_id tuple
- self.resource_managers[pon_intf].update_gemport_ids_for_onu(pon_intf_onu_id,
- gemport_id_list)
-
- self.update_gemports_ponport_to_onu_map_on_kv_store(gemport_id_list,
- pon_intf, onu_id, uni_id)
- return gemport_id_list
def free_pon_resources_for_onu(self, pon_intf_id_onu_id):
""" Typically called on ONU delete """
diff --git a/voltha/adapters/adtran_olt/test/flow/test_evc_map.py b/voltha/adapters/adtran_olt/test/flow/test_evc_map.py
index d1bb531..35ffb37 100644
--- a/voltha/adapters/adtran_olt/test/flow/test_evc_map.py
+++ b/voltha/adapters/adtran_olt/test/flow/test_evc_map.py
@@ -19,7 +19,7 @@
-## This section test the proberties of the class EVCMap
+## This section test the properties of the class EVCMap
def test_EVCMap_properties():
flow = MagicMock()
diff --git a/voltha/adapters/adtran_olt/test/xpon/test_traffic_descriptor.py b/voltha/adapters/adtran_olt/test/xpon/test_traffic_descriptor.py
new file mode 100644
index 0000000..d607a81
--- /dev/null
+++ b/voltha/adapters/adtran_olt/test/xpon/test_traffic_descriptor.py
@@ -0,0 +1,78 @@
+# Copyright 2017-present Adtran, Inc.
+#
+# 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 voltha.adapters.adtran_olt.xpon.traffic_descriptor import TrafficDescriptor
+
+import pytest
+
+f_bw = 100000000 # fixed_bandwidth
+a_bw = 95000000 # assured_bandwidth
+m_bw = 95000000 # maximum_bandwidth
+AbE = 1 # Additional Bandwidth Eligibility
+
+
+@pytest.fixture(scope='module')
+def td():
+ td = TrafficDescriptor(f_bw, a_bw, m_bw, AbE)
+ return td
+
+
+def test_init(td):
+ assert td.fixed_bandwidth == f_bw
+ assert td.assured_bandwidth == a_bw
+ assert td.maximum_bandwidth == m_bw
+ assert td.additional_bandwidth_eligibility == AbE
+
+
+def test_str(td):
+ assert str(td) == "TrafficDescriptor: {}/{}/{}".format(f_bw, a_bw, m_bw)
+
+
+@pytest.mark.parametrize("input_value, expected_value",
+ [(TrafficDescriptor.AdditionalBwEligibility.NON_ASSURED_SHARING, "non-assured-sharing"),
+ (TrafficDescriptor.AdditionalBwEligibility.BEST_EFFORT_SHARING, "best-effort-sharing"),
+ (TrafficDescriptor.AdditionalBwEligibility.NONE, "none")])
+def test_to_string(td, input_value, expected_value):
+
+ test_value = td.AdditionalBwEligibility.to_string(input_value)
+
+ assert test_value == expected_value
+
+
+@pytest.mark.parametrize("input_value, expected_value",
+ [(0, TrafficDescriptor.AdditionalBwEligibility.NONE),
+ (1, TrafficDescriptor.AdditionalBwEligibility.BEST_EFFORT_SHARING),
+ (2, TrafficDescriptor.AdditionalBwEligibility.NON_ASSURED_SHARING)])
+def test_from_value(td, input_value, expected_value):
+
+ test_value = td.AdditionalBwEligibility.from_value(input_value)
+
+ assert test_value == expected_value
+
+
+def test_to_dict(td):
+
+ dict_val = {
+ 'fixed-bandwidth': f_bw,
+ 'assured-bandwidth': a_bw,
+ 'maximum-bandwidth': m_bw,
+ 'additional-bandwidth-eligibility': td.AdditionalBwEligibility.to_string(AbE)
+
+ }
+
+ test_dict = td.to_dict()
+
+ assert dict_val == test_dict
+
+
diff --git a/voltha/adapters/adtran_onu/.coveragerc b/voltha/adapters/adtran_onu/.coveragerc
index aa8118f..7149b98 100644
--- a/voltha/adapters/adtran_onu/.coveragerc
+++ b/voltha/adapters/adtran_onu/.coveragerc
@@ -5,4 +5,4 @@
[report]
-[html]
+[html]
\ No newline at end of file
diff --git a/voltha/adapters/adtran_onu/.pylintrc b/voltha/adapters/adtran_onu/.pylintrc
index 41cf299..b29798a 100644
--- a/voltha/adapters/adtran_onu/.pylintrc
+++ b/voltha/adapters/adtran_onu/.pylintrc
@@ -548,4 +548,4 @@
# Exceptions that will emit a warning when being caught. Defaults to
# "Exception"
-overgeneral-exceptions=Exception
+overgeneral-exceptions=Exception
\ No newline at end of file
diff --git a/voltha/adapters/adtran_onu/test/test_adtran_onu_handler.py b/voltha/adapters/adtran_onu/test/test_adtran_onu_handler.py
index 38def80..780f58a 100644
--- a/voltha/adapters/adtran_onu/test/test_adtran_onu_handler.py
+++ b/voltha/adapters/adtran_onu/test/test_adtran_onu_handler.py
@@ -16,7 +16,10 @@
from mock import Mock
from mock import MagicMock
from mock import patch
+import json
+import ast
+from twisted.internet.defer import Deferred
from voltha.adapters.adtran_onu.adtran_onu_handler import AdtranOnuHandler
from voltha.adapters.adtran_onu.adtran_onu import AdtranOnuAdapter
@@ -26,6 +29,7 @@
from voltha.protos.voltha_pb2 import SelfTestResponse
from voltha.adapters.adtran_onu.test.resources.technology_profile import tech_profile_json
from voltha.adapters.adtran_onu.pon_port import PonPort
+from voltha.adapters.adtran_onu.adtran_onu_handler import _STARTUP_RETRY_WAIT
@pytest.fixture()
@@ -143,3 +147,48 @@
onu_handler.update_pm_config(device_info, pm_config)
onu_handler.pm_metrics.update.assert_called_with(pm_config)
+
+@pytest.mark.parametrize("success_case", [True, False])
+def test_load_and_configure_tech_profile(onu_handler, success_case):
+ with patch('voltha.adapters.adtran_onu.adtran_onu_handler.AdtnTpServiceSpecificTask'),\
+ patch('voltha.adapters.adtran_onu.adtran_onu_handler.AdtranOnuHandler.openomci') as openomci, \
+ patch('voltha.adapters.adtran_onu.adtran_onu_handler.reactor.callLater') as onu_handler_reactor:
+ uni_id, tp_path = 1, '/255/XPON/1024'
+ tp = json.dumps(tech_profile_json)
+ onu_handler.kv_client = {tp_path: tp}
+ onu_handler._do_tech_profile_configuration = MagicMock()
+ d = Deferred()
+ openomci.onu_omci_device.task_runner.queue_task.return_value = d
+ onu_handler.load_and_configure_tech_profile(uni_id, tp_path)
+ if success_case:
+ d.callback('success')
+ assert onu_handler._tech_profile_download_done[uni_id][tp_path] is True
+ else:
+ d.errback(Exception('test'))
+ onu_handler_reactor.assert_called_with(_STARTUP_RETRY_WAIT,
+ onu_handler.load_and_configure_tech_profile, uni_id, tp_path)
+
+ assert len(onu_handler._tp_service_specific_task[uni_id]) == 0
+ onu_handler._do_tech_profile_configuration.assert_called_with(uni_id, ast.literal_eval(tp), 255)
+
+
+def test_load_and_configure_tech_profile_handles_exceptions(onu_handler):
+ onu_handler.kv_client = Mock(side_effect=Exception())
+ uni_id, tp_path = 1, '/255/XPON/1024'
+ onu_handler.load_and_configure_tech_profile(uni_id, tp_path)
+ assert onu_handler._tech_profile_download_done[uni_id][tp_path] is False
+
+
+def test_load_and_configure_tech_profile_where_tech_profile_already_loaded(onu_handler):
+ uni_id, tp_path = 1, '/255/XPON/1024'
+ onu_handler._tech_profile_download_done[uni_id] = dict()
+ onu_handler._tech_profile_download_done[uni_id][tp_path] = True
+ onu_handler._do_tech_profile_configuration = MagicMock()
+ onu_handler.load_and_configure_tech_profile(uni_id, tp_path)
+ onu_handler._do_tech_profile_configuration.assert_not_called()
+
+
+def test_rx_inter_adapter_message_throws_not_implemented_error(onu_handler):
+ with pytest.raises(TypeError):
+ onu_handler.rx_inter_adapter_message('test')
+
diff --git a/voltha/adapters/adtran_onu/test/test_onu_gem_port.py b/voltha/adapters/adtran_onu/test/test_onu_gem_port.py
index dd9df77..9fef2f3 100644
--- a/voltha/adapters/adtran_onu/test/test_onu_gem_port.py
+++ b/voltha/adapters/adtran_onu/test/test_onu_gem_port.py
@@ -12,11 +12,567 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from voltha.adapters.adtran_olt.xpon.olt_gem_port import OltGemPort
+from voltha.adapters.adtran_onu.onu_gem_port import OnuGemPort
import pytest
from mock import patch, MagicMock
-import mock
-import json
import pytest_twisted
+GEMID = 11
+ALLOCID = 21
+TECHPROFILEID = 31
+PONID = 41
+ONUID = 51
+UNIID = 61
+ENTITYID = 71
+ENCRYPTION = False
+MULTICAST = 'multicast'
+TRAFFICCLASS = 'traffic_class'
+ISMOCK = 'ismock'
+MOCK_HANDLER_DEVICE_ID = "mock handler's device_id"
+UPSTREAM = 1
+DOWNSTREAM = 2
+BIDIRECTIONAL = 3
+TCONTENTITYID = 500
+IEEEENTITYID = 525
+GALENETENTITYID = 550
+
+GEM_DATA = {
+ 'gemport-id': 100,
+ 'encryption': 'encryption',
+ 'uni-id': 200,
+ 'direction': BIDIRECTIONAL,
+ 'pbit-map': '0b01010101',
+ 'priority-q': 300,
+ 'max-q-size': 400,
+ 'weight': 500,
+ 'discard-config': {'max-probability' : 'max prob', 'max-threshold' : 'max thresh', 'min-threshold' : 'min thresh'},
+ 'discard-policy': 'WTailDrop',
+ 'scheduling-policy': 'StrictPriority'
+ }
+
+GEM_DATA_BAD_UNI_ID = {
+ 'gemport-id': 100,
+ 'encryption': 'encryption',
+ # 'uni-id': 200,
+ 'direction': BIDIRECTIONAL,
+ 'pbit-map': '0b01010101',
+ 'priority-q': 300,
+ 'max-q-size': 400,
+ 'weight': 500,
+ 'discard-config': {'max-probability' : 'max prob', 'max-threshold' : 'max thresh', 'min-threshold' : 'min thresh'},
+ 'discard-policy': 'WTailDrop',
+ 'scheduling-policy': 'StrictPriority'
+ }
+
+
+def ogp_full(mock_handler):
+ ogp_obj = OnuGemPort(mock_handler, GEM_DATA, ALLOCID, TECHPROFILEID, UNIID, ENTITYID, MULTICAST, TRAFFICCLASS, ISMOCK)
+ return ogp_obj
+
+
+def ogp_defaults(mock_handler):
+ ogp_obj = OnuGemPort(mock_handler, GEM_DATA, ALLOCID, TECHPROFILEID, UNIID, ENTITYID)
+ return ogp_obj
+
+
+def ogp_exception(mock_handler):
+ try:
+ ogp_obj = OnuGemPort(mock_handler, GEM_DATA_BAD_UNI_ID, ALLOCID, TECHPROFILEID, UNIID, ENTITYID)
+ except:
+ raise
+
+ return ogp_obj
+
+
+@pytest.fixture()
+def ogp():
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ ogp_obj = OnuGemPort.create(mock_handler, GEM_DATA, ALLOCID, TECHPROFILEID, UNIID, ENTITYID)
+
+ return ogp_obj
+
+
+def test_onu_gem_port_init_values_missing():
+ """
+ verify __init__ fails when no values are specified
+ """
+
+ with pytest.raises(Exception):
+ OnuGemPort()
+
+
+@patch('voltha.adapters.adtran_onu.onu_gem_port.structlog.get_logger')
+def test_onu_gem_port_init_values(patched_get_logger):
+ """
+ verify __init__ values are set properly
+ """
+
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ ogp = ogp_full(mock_handler)
+
+ patched_get_logger.assert_called_once_with(device_id=MOCK_HANDLER_DEVICE_ID, gem_id=GEM_DATA.get('gemport-id'))
+ assert ogp.gem_id == GEM_DATA.get('gemport-id')
+ assert ogp._alloc_id == ALLOCID
+ # assert ogp_full.tech_profile_id == TECHPROFILEID
+ assert ogp.tech_profile_id is None # TODO: code says default may change to a property
+ assert ogp.encryption == GEM_DATA.get('encryption') # uses a getter so no need to inspect internal to object
+ assert ogp.multicast == MULTICAST
+ assert ogp.traffic_class == TRAFFICCLASS
+ assert ogp._handler == mock_handler
+ assert ogp._is_mock == ISMOCK
+
+ assert ogp._gem_data == GEM_DATA
+ assert ogp._entity_id == ENTITYID
+ assert ogp.entity_id == ENTITYID # tests getter
+ assert ogp._tcont_entity_id is None
+ assert ogp._interworking is False
+ assert ogp.uni_id == GEM_DATA['uni-id']
+ assert ogp.direction == GEM_DATA.get('direction')
+ assert ogp._pbit_map == GEM_DATA.get('pbit-map')[2:]
+ assert ogp.pbit_map == GEM_DATA.get('pbit-map')[2:] # tests getter
+ assert ogp.priority_q == GEM_DATA.get('priority-q')
+ assert ogp._max_q_size == GEM_DATA.get('max-q-size')
+ assert ogp.max_q_size == GEM_DATA.get('max-q-size') # tests getter
+ assert ogp.weight == GEM_DATA.get('weight')
+ assert ogp._discard_config == GEM_DATA.get('discard-config')
+ assert ogp.discard_config == GEM_DATA.get('discard-config') # tests getter
+ assert ogp._discard_policy == GEM_DATA.get('discard-policy')
+ assert ogp.discard_policy == GEM_DATA.get('discard-policy') # tests getter
+ assert ogp._scheduling_policy == GEM_DATA.get('scheduling-policy')
+ assert ogp.scheduling_policy == GEM_DATA.get('scheduling-policy') # tests getter
+
+
+@patch('voltha.adapters.adtran_onu.onu_gem_port.structlog.get_logger')
+def test_onu_gem_port_create(patched_get_logger):
+ """
+ verify call to create().
+
+ Equivalent to creating OnuGemPort with defaults so a 'defaults' test is not needed.
+ """
+
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ ogp = OnuGemPort.create(mock_handler, GEM_DATA, ALLOCID, TECHPROFILEID, UNIID, ENTITYID)
+
+ patched_get_logger.assert_called_once_with(device_id=MOCK_HANDLER_DEVICE_ID, gem_id=GEM_DATA.get('gemport-id'))
+ assert ogp.gem_id == GEM_DATA.get('gemport-id')
+ assert ogp._alloc_id == ALLOCID
+ # assert ogp_full.tech_profile_id == TECHPROFILEID
+ assert ogp.tech_profile_id is None # TODO: code says default may change to a property
+ assert ogp.encryption == GEM_DATA.get('encryption') # uses a getter
+ assert ogp.multicast is False
+ assert ogp.traffic_class is None
+ assert ogp._handler == mock_handler
+ assert ogp._is_mock is False
+
+ assert ogp._gem_data == GEM_DATA
+ assert ogp._entity_id == ENTITYID
+ assert ogp.entity_id == ENTITYID
+ assert ogp._tcont_entity_id is None
+ assert ogp._interworking is False
+ assert ogp.uni_id == GEM_DATA['uni-id']
+ assert ogp.direction == GEM_DATA.get('direction')
+ assert ogp._pbit_map == GEM_DATA.get('pbit-map')[2:]
+ assert ogp.pbit_map == GEM_DATA.get('pbit-map')[2:]
+ assert ogp.priority_q == GEM_DATA.get('priority-q')
+ assert ogp._max_q_size == GEM_DATA.get('max-q-size')
+ assert ogp.max_q_size == GEM_DATA.get('max-q-size')
+ assert ogp.weight == GEM_DATA.get('weight')
+ assert ogp._discard_config == GEM_DATA.get('discard-config')
+ assert ogp.discard_config == GEM_DATA.get('discard-config')
+ assert ogp._discard_policy == GEM_DATA.get('discard-policy')
+ assert ogp.discard_policy == GEM_DATA.get('discard-policy')
+ assert ogp._scheduling_policy == GEM_DATA.get('scheduling-policy')
+ assert ogp.scheduling_policy == GEM_DATA.get('scheduling-policy')
+
+
+@patch('voltha.adapters.adtran_onu.onu_gem_port.structlog.get_logger')
+def test_onu_gem_port_init_values_force_exception(patched_get_logger):
+ """
+ verify __init__ values are set properly
+ """
+
+ _ = patched_get_logger
+
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ with pytest.raises(Exception) as caught_ex:
+ ogp_exception(mock_handler)
+
+ # verify the raise
+ assert caught_ex.typename == 'KeyError'
+ assert caught_ex.value.message == 'uni-id'
+
+
+def test_onu_gem_port_setter_exceptions_max_q_size(ogp):
+ """
+ verify setters (and getters) for max_q_size exceptions not covered in other tests
+ """
+ ogp.max_q_size = 123
+ assert ogp.max_q_size == 123
+
+ ogp.max_q_size = 'auto'
+ assert ogp.max_q_size == 'auto'
+
+ with pytest.raises(Exception) as caught_ex:
+ ogp.max_q_size = 'Doh!'
+ assert caught_ex.typename == 'AssertionError'
+
+ with pytest.raises(Exception) as caught_ex:
+ ogp.max_q_size = 1.23
+ assert caught_ex.typename == 'AssertionError'
+
+
+def test_onu_gem_port_setter_exceptions_pbit_map(ogp):
+ """
+ verify setter for pbit_map exceptions not covered in other tests
+ """
+ with pytest.raises(Exception) as caught_ex:
+ ogp.pbit_map = 123
+ assert caught_ex.typename == 'AssertionError'
+
+ with pytest.raises(Exception) as caught_ex:
+ ogp.pbit_map = '0b001001001'
+ assert caught_ex.typename == 'AssertionError'
+
+ with pytest.raises(Exception) as caught_ex:
+ ogp.pbit_map = '0b20100101'
+ assert caught_ex.typename == 'Exception'
+ assert caught_ex.value.message == 'pbit_map-not-binary-string-0b20100101'
+
+
+@pytest_twisted.inlineCallbacks
+def test_add_to_hardware_isMock():
+ """
+ verify call to add hardware when isMock = True
+ """
+ _isMock = True
+
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ ogp = OnuGemPort(mock_handler, GEM_DATA, ALLOCID, TECHPROFILEID, UNIID, ENTITYID, MULTICAST, TRAFFICCLASS,
+ _isMock)
+
+ result = yield ogp.add_to_hardware('fake', 'fake', 'fake', 'fake')
+ assert result == 'mock'
+
+
+@patch('voltha.adapters.adtran_onu.onu_gem_port.GemInterworkingTpFrame')
+@patch('voltha.adapters.adtran_onu.onu_gem_port.GemPortNetworkCtpFrame')
+@patch('voltha.adapters.adtran_onu.onu_gem_port.structlog.get_logger')
+@pytest_twisted.inlineCallbacks
+def test_add_to_hardware(patched_get_logger, ctp_class, tp_class):
+ """
+ verify call to add hardware when isMock = False
+ """
+ ctp_class.return_value.create.return_value = 'CtpFrame Create Success!'
+ tp_class.return_value.create.return_value = 'TpFrame Create Success!'
+
+ _ = patched_get_logger
+
+ # create mock for handler
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ # CREATE the class object to be tested, send it the mock handler
+ ogp = ogp_defaults(mock_handler)
+
+ # PREPARE to call the add_hardware method
+
+ # prepare nested 'fields' result for omci.send
+ class MockResults(object):
+ def __init__(self, expected_output):
+ self.fields = expected_output
+
+ mock_send_result = MockResults({'omci_message': MockResults({'success_code': 0,
+ 'parameter_error_attributes_mask': 1234})})
+
+ # create mock for omci
+ mock_omci = MagicMock()
+ mock_omci.send.return_value = mock_send_result # omci.send() will return the nested 'fields' structure
+
+ # Test values before in_hardware
+ assert ogp.in_hardware is False
+
+ # make the call to add_to_hardware
+ result = yield ogp.add_to_hardware(mock_omci, TCONTENTITYID, IEEEENTITYID, GALENETENTITYID)
+
+ # Test values after in_hardware
+ assert ogp.in_hardware is True
+
+ # VERIFY Results
+
+ assert result == mock_send_result
+
+ assert ogp.log.debug.call_count == 3
+
+ ogp.log.debug.assert_any_call('add-to-hardware',
+ gem_id=GEM_DATA.get('gemport-id'),
+ gem_entity_id=ENTITYID,
+ tcont_entity_id=TCONTENTITYID,
+ ieee_mapper_service_profile_entity_id=IEEEENTITYID,
+ gal_enet_profile_entity_id=GALENETENTITYID)
+
+ ctp_class.assert_called_once_with(ENTITYID,
+ port_id=GEM_DATA.get('gemport-id'),
+ tcont_id=TCONTENTITYID,
+ direction='bi-directional',
+ upstream_tm=0x8000)
+
+ mock_omci.send.assert_any_call('CtpFrame Create Success!')
+
+ # the following validates the log entry and the mock_send_result from omci.send() after GemPortNetworkCtpFrame
+ ogp.log.debug.assert_any_call('create-gem-port-network-ctp', status=0, error_mask=1234)
+
+ assert ogp._tcont_entity_id == TCONTENTITYID
+
+ tp_class.assert_called_once_with(ENTITYID,
+ gem_port_network_ctp_pointer=ENTITYID,
+ interworking_option=5,
+ service_profile_pointer=IEEEENTITYID,
+ interworking_tp_pointer=0x0,
+ pptp_counter=1,
+ gal_profile_pointer=GALENETENTITYID,
+ attributes={'gal_loopback_configuration': 0})
+
+ mock_omci.send.assert_any_call('TpFrame Create Success!')
+
+ # the following validates the log entry and the mock_send_result from omci.send() after GemInterworkingTpFrame
+ ogp.log.debug.assert_any_call('create-gem-interworking-tp', status=0, error_mask=1234)
+
+ assert ogp._interworking is True
+
+
+@patch('voltha.adapters.adtran_onu.onu_gem_port.structlog.get_logger')
+@pytest_twisted.inlineCallbacks
+def test_add_to_hardware_exceptions_bad_tcont(patched_get_logger):
+ """
+ verify add hardware errors and exception
+
+ Ctp Section - Tests call to add_to_hardware with a bad tcont_entity_id
+ """
+
+ _ = patched_get_logger
+
+ # create mock for handler
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ # CREATE the class object to be tested, send it the mock handler
+ ogp = ogp_defaults(mock_handler)
+
+ # PREPARE to call the add_hardware method
+
+ # prepare nested 'fields' result for omci.send
+ class MockResults(object):
+ def __init__(self, expected_output):
+ self.fields = expected_output
+
+ mock_send_result = MockResults({'omci_message': MockResults({'success_code': 0,
+ 'parameter_error_attributes_mask': 1234})})
+
+ # create mock for omci
+ mock_omci = MagicMock()
+ mock_omci.send.return_value = mock_send_result # omci.send() will return the nested 'fields' structure
+
+ # create problem
+ ogp._tcont_entity_id = 999
+
+ with pytest.raises(Exception) as caught_ex:
+ yield ogp.add_to_hardware(mock_omci, TCONTENTITYID, IEEEENTITYID, GALENETENTITYID)
+
+ # verify the raise
+ assert caught_ex.typename == 'KeyError'
+ assert caught_ex.value.message == 'GEM Port already assigned to TCONT: 999'
+
+ # undo problem
+ ogp._tcont_entity_id = None
+
+
+@patch('voltha.adapters.adtran_onu.onu_gem_port.structlog.get_logger')
+@pytest_twisted.inlineCallbacks
+def test_add_to_hardware_exceptions_mcast_not_supported(patched_get_logger):
+ """
+ verify add hardware errors and exception cases
+
+ CtpFrame section - assert when MULTICAST.
+
+ """
+ _ = patched_get_logger
+
+ # create mock for handler
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ # CREATE the class object to be tested, send it the mock handler
+ ogp = ogp_defaults(mock_handler)
+
+ # PREPARE to call the add_hardware method
+
+ # prepare nested 'fields' result for omci.send
+ class MockResults(object):
+ def __init__(self, expected_output):
+ self.fields = expected_output
+
+ mock_send_result = MockResults({'omci_message': MockResults({'success_code': 0,
+ 'parameter_error_attributes_mask': 1234})})
+
+ # create mock for omci
+ mock_omci = MagicMock()
+ mock_omci.send.return_value = mock_send_result # omci.send() will return the nested 'fields' structure
+
+ ogp.multicast = 999
+ with pytest.raises(Exception) as caught_ex:
+ yield ogp.add_to_hardware(mock_omci, TCONTENTITYID, IEEEENTITYID, GALENETENTITYID)
+
+ # verify the raise
+ assert caught_ex.typename == 'AssertionError'
+ assert caught_ex.value.message == 'MCAST is not supported yet'
+
+ # Do not try to validate the 'except' in the try/except because it uses 'assert' which confuses the test
+
+
+@patch('voltha.adapters.adtran_onu.onu_gem_port.GemInterworkingTpFrame')
+@patch('voltha.adapters.adtran_onu.onu_gem_port.GemPortNetworkCtpFrame')
+@patch('voltha.adapters.adtran_onu.onu_gem_port.structlog.get_logger')
+@pytest_twisted.inlineCallbacks
+def test_add_to_hardware_exceptions_gem_port_failed_and_try_except(patched_get_logger, ctp_class, tp_class):
+ """
+ verify add hardware exception
+
+ Ctp section - Tests Reason Code not equal to Success, GEM Port creation FAILED
+
+ AND tests try/except logic of the CtpFrame section
+ """
+ ctp_class.return_value.create.return_value = 'CtpFrame Create Success!'
+ tp_class.return_value.create.return_value = 'TpFrame Create Success!'
+
+ _ = patched_get_logger
+
+ # create mock for handler
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ # CREATE the class object to be tested, send it the mock handler
+ ogp = ogp_defaults(mock_handler)
+
+ # PREPARE to call the add_hardware method
+
+ # prepare nested 'fields' result for omci.send
+ class MockResults(object):
+ def __init__(self, expected_output):
+ self.fields = expected_output
+
+ mock_send_result = MockResults({'omci_message': MockResults({'success_code': 0,
+ 'parameter_error_attributes_mask': 1234})})
+
+ # create mock for omci
+ mock_omci = MagicMock()
+ mock_omci.send.return_value = mock_send_result # omci.send() will return the nested 'fields' structure
+
+
+ # create problem
+ mock_send_result = MockResults({'omci_message': MockResults({'success_code': 2,
+ 'parameter_error_attributes_mask': 1234})})
+ mock_omci.send.return_value = mock_send_result
+
+ with pytest.raises(Exception) as caught_ex:
+ yield ogp.add_to_hardware(mock_omci, TCONTENTITYID, IEEEENTITYID, GALENETENTITYID)
+
+ # verify the raise
+ assert caught_ex.typename == 'Exception'
+ assert caught_ex.value.message == 'GEM Port create failed with status: 2'
+
+ # TODO - add more error and exception tests starting a onu_gem_port line 208 (if not self._interworking:)
+
+
+@pytest_twisted.inlineCallbacks
+def test_remove_from_hardware_isMock():
+ """
+ verify call to remove hardware when isMock = True
+ """
+ _isMock = True
+
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ ogp = OnuGemPort(mock_handler, GEM_DATA, ALLOCID, TECHPROFILEID, UNIID, ENTITYID, MULTICAST, TRAFFICCLASS,
+ _isMock)
+
+ result = yield ogp.remove_from_hardware('fake')
+ assert result == 'mock'
+
+
+@patch('voltha.adapters.adtran_onu.onu_gem_port.GemInterworkingTpFrame')
+@patch('voltha.adapters.adtran_onu.onu_gem_port.GemPortNetworkCtpFrame')
+@patch('voltha.adapters.adtran_onu.onu_gem_port.structlog.get_logger')
+@pytest_twisted.inlineCallbacks
+def test_remove_from_hardware(patched_get_logger, ctp_class, tp_class):
+ """
+ verify call to remove hardware when isMock = False
+ """
+
+ ctp_class.return_value.delete.return_value = 'CtpFrame Delete Success!'
+ tp_class.return_value.delete.return_value = 'TpFrame Delete Success!'
+
+ _ = patched_get_logger
+
+ # create mock for handler
+ mock_handler = MagicMock()
+ mock_handler.device_id = MOCK_HANDLER_DEVICE_ID
+
+ # CREATE the class object to be tested, send it the mock handler
+ ogp = ogp_defaults(mock_handler)
+
+ # adjust some class instance attributes so remove_hardware will run
+ ogp._interworking = True
+ ogp._tcont_entity_id = 1 # not None
+
+ # PREPARE to call the remove_hardware method
+
+ # prepare nested 'fields' result for omci.send
+ class MockResults(object):
+ def __init__(self, expected_output):
+ self.fields = expected_output
+
+ mock_send_result = MockResults({'omci_message': MockResults({'success_code': 0})})
+
+ # create mock for omci
+ mock_omci = MagicMock()
+ mock_omci.send.return_value = mock_send_result # omci.send() will return the nested 'fields' structure
+
+ # make the call to remove_from_hardware
+ result = yield ogp.remove_from_hardware(mock_omci)
+
+ # VERIFY Results
+ assert result == mock_send_result
+
+ assert ogp.log.debug.call_count == 3
+
+ ogp.log.debug.assert_any_call('remove-from-hardware', gem_id=GEM_DATA.get('gemport-id'))
+
+ ctp_class.assert_called_once_with(ENTITYID)
+
+ mock_omci.send.assert_any_call('CtpFrame Delete Success!')
+
+ # the following validates the log entry and the mock_send_result from omci.send() after GemPortNetworkCtpFrame
+ ogp.log.debug.assert_any_call('delete-gem-port-network-ctp', status=0)
+
+ assert ogp._tcont_entity_id is None
+
+ tp_class.assert_called_once_with(ENTITYID)
+
+ mock_omci.send.assert_any_call('TpFrame Delete Success!')
+
+ # the following validates the log entry and the mock_send_result from omci.send() after GemInterworkingTpFrame
+ ogp.log.debug.assert_any_call('delete-gem-interworking-tp', status=0)
+
+ assert ogp._interworking is False
diff --git a/voltha/adapters/adtran_onu/test/test_pon_port.py b/voltha/adapters/adtran_onu/test/test_pon_port.py
new file mode 100644
index 0000000..2da7a3b
--- /dev/null
+++ b/voltha/adapters/adtran_onu/test/test_pon_port.py
@@ -0,0 +1,67 @@
+# Copyright 2017-present Adtran, Inc.
+#
+# 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 voltha.adapters.adtran_onu.pon_port import PonPort
+from mock import MagicMock
+import pytest
+
+## Test class PonPort init settings ###############
+def test_PonPort_inits():
+ handler = MagicMock()
+ handler.device_id = 100
+ portnum = 1
+ testponport = PonPort(handler, portnum)
+
+ assert testponport._enabled is False
+ assert testponport._valid is True
+ assert testponport._handler is handler
+ assert testponport._deferred is None
+ assert testponport._port is None
+ assert testponport._port_number == 1
+ assert testponport._entity_id is None
+ assert testponport._next_entity_id == PonPort.MIN_GEM_ENTITY_ID
+
+
+
+
+## Test PonPort staticmethod #########
+def test_create():
+ handler = MagicMock()
+ handler.device_id = 200
+ port_no = 2
+ testcreate = PonPort.create(handler, port_no)
+
+ assert isinstance(testcreate, PonPort)
+ assert testcreate._handler is handler
+ assert testcreate._port_number is port_no
+
+
+
+
+
+## Test PonPort @property #########
+def test_PonPort_properties():
+ handler = MagicMock()
+ handler.device_id = 300
+ port_no = 3
+ testprop1 = PonPort(handler, port_no)
+
+ assert testprop1.enabled is False
+ assert testprop1.port_number == 3
+ assert testprop1.entity_id is None
+ assert testprop1.next_gem_entity_id == PonPort.MIN_GEM_ENTITY_ID
+ assert testprop1.tconts == {}
+ assert testprop1.gem_ports == {}
+
+