| # |
| # 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. |
| # |
| |
| """ |
| vOLT-HA Discovery Test Case module |
| """ |
| |
| import testCaseUtils |
| import logging |
| |
| |
| class Discovery(object): |
| |
| """ |
| This class implements voltha discovery test |
| """ |
| |
| def __init__(self): |
| self.dirs = dict() |
| self.dirs['log'] = None |
| self.dirs['root'] = None |
| self.dirs['voltha'] = None |
| |
| self.__logicalDeviceType = None |
| self.__oltType = None |
| self.__onuType = None |
| self.__fields = [] |
| self.__logicalDeviceId = None |
| self.__oltDeviceId = None |
| self.__onuDeviceIds = [] |
| self.__peers = None |
| |
| def d_set_log_dirs(self, log_dir): |
| testCaseUtils.config_dirs(self, log_dir) |
| |
| def d_configure(self, logical_device_type, olt_type, onu_type): |
| self.__logicalDeviceType = logical_device_type |
| self.__oltType = olt_type |
| self.__onuType = onu_type |
| |
| def logical_device(self): |
| logging.info('Logical Device Info') |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, self.__logicalDeviceType, 'voltha_devices_after_enable.log') |
| assert statusLines, 'No Logical Devices listed under devices' |
| self.__fields = testCaseUtils.parse_fields(statusLines) |
| self.__logicalDeviceId = self.__fields[4].strip() |
| testCaseUtils.send_command_to_voltha_cli(testCaseUtils.get_dir(self, 'log'), |
| 'voltha_logical_device.log', 'logical_device ' + self.__logicalDeviceId, |
| 'voltha_logical_device_ports.log', 'ports', 'voltha_logical_device_flows.log', 'flows') |
| testCaseUtils.print_log_file(self, 'voltha_logical_device_ports.log') |
| testCaseUtils.print_log_file(self, 'voltha_logical_device_flows.log') |
| |
| def olt_discovery(self): |
| logging.info('Olt Discovery') |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, self.__oltType, 'voltha_devices_after_enable.log') |
| assert statusLines, 'No Olt listed under devices' |
| self.__fields = testCaseUtils.parse_fields(statusLines) |
| self.__oltDeviceId = self.__fields[1].strip() |
| testCaseUtils.send_command_to_voltha_cli(testCaseUtils.get_dir(self, 'log'), |
| 'voltha_olt_device.log', 'device ' + self.__oltDeviceId, 'voltha_olt_ports.log', |
| 'ports', 'voltha_olt_flows.log', 'flows') |
| testCaseUtils.print_log_file(self, 'voltha_olt_ports.log') |
| testCaseUtils.print_log_file(self, 'voltha_olt_flows.log') |
| |
| def onu_discovery(self): |
| logging.info('Onu Discovery') |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, self.__onuType, 'voltha_devices_after_enable.log') |
| assert statusLines, 'No Onu listed under devices' |
| lines = statusLines.splitlines() |
| for line in lines: |
| self.__fields = testCaseUtils.parse_fields(line) |
| onuDeviceId = self.__fields[1].strip() |
| self.__onuDeviceIds.append(onuDeviceId) |
| testCaseUtils.send_command_to_voltha_cli(testCaseUtils.get_dir(self, 'log'), |
| 'voltha_onu_device_' + str(self.__onuDeviceIds.index(onuDeviceId)) + '.log', |
| 'device ' + onuDeviceId, 'voltha_onu_ports_' + |
| str(self.__onuDeviceIds.index(onuDeviceId)) + '.log', 'ports', 'voltha_onu_flows_' + |
| str(self.__onuDeviceIds.index(onuDeviceId)) + '.log', 'flows') |
| testCaseUtils.print_log_file(self, 'voltha_onu_ports_' + str(self.__onuDeviceIds.index(onuDeviceId)) + '.log') |
| testCaseUtils.print_log_file(self, 'voltha_onu_flows_' + str(self.__onuDeviceIds.index(onuDeviceId)) + '.log') |
| |
| def olt_ports_should_be_enabled_and_active(self): |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, self.__oltDeviceId, 'voltha_olt_ports.log') |
| assert statusLines, 'No Olt device listed under ports' |
| lines = statusLines.splitlines() |
| for line in lines: |
| self.__fields = testCaseUtils.parse_fields(line) |
| assert (self.check_states(self.__oltDeviceId) is True), 'States of %s does match expected ' % self.__oltDeviceId |
| portType = self.__fields[3].strip() |
| assert (portType == 'ETHERNET_NNI' or portType == 'PON_OLT'),\ |
| 'Port type for %s does not match expected ETHERNET_NNI or PON_OLT' % self.__oltDeviceId |
| if portType == 'PON_OLT': |
| self.__peers = self.__fields[7].strip() |
| peerFields = self.__peers.split(',') |
| peerDevices = peerFields[1::2] |
| for peerDevice in peerDevices: |
| deviceFields = peerDevice.split(':') |
| deviceId = deviceFields[1].replace("'", "").replace('u', '').rstrip("}]").strip() |
| assert deviceId in self.__onuDeviceIds, 'ONU Device %s not found as Peer' % deviceId |
| |
| def onu_ports_should_be_enabled_and_active(self): |
| for onuDeviceId in self.__onuDeviceIds: |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, onuDeviceId, 'voltha_onu_ports_' + |
| str(self.__onuDeviceIds.index(onuDeviceId)) + '.log') |
| assert statusLines, 'No Onu device listed under ports' |
| lines = statusLines.splitlines() |
| for line in lines: |
| self.__fields = testCaseUtils.parse_fields(line) |
| assert (self.check_states(onuDeviceId) is True), 'States of %s does match expected ' % onuDeviceId |
| portType = self.__fields[3].strip() |
| assert (portType == 'ETHERNET_UNI' or portType == 'PON_ONU'),\ |
| 'Port type for %s does not match expected ETHERNET_UNI or PON_ONU' % onuDeviceId |
| if portType == 'PON_ONU': |
| self.__peers = self.__fields[7].strip() |
| peerFields = self.__peers.split(',') |
| peerDevice = peerFields[1] |
| deviceFields = peerDevice.split(':') |
| deviceId = deviceFields[1].replace("'", "").replace('u', '').rstrip("}]").strip() |
| assert deviceId == self.__oltDeviceId, 'OLT Device %s not found as Peer' % deviceId |
| |
| def check_states(self, device_id): |
| result = True |
| adminState = self.__fields[4].strip() |
| assert adminState == 'ENABLED', 'Admin State of %s not ENABLED' % device_id |
| operatorStatus = self.__fields[5].strip() |
| assert operatorStatus == 'ACTIVE', 'Operator Status of %s not ACTIVE' % device_id |
| return result |
| |
| def olt_should_have_at_least_one_flow(self): |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, 'Flows', 'voltha_olt_flows.log') |
| assert statusLines, 'No Olt flows under device %s' % self.__oltDeviceId |
| before, flows, numFlows = statusLines.partition('Flows') |
| plainNumber = numFlows.strip().strip('():') |
| if plainNumber.isdigit(): |
| assert int(plainNumber) > 0, 'Zero number of flows for Olt %s' % self.__oltDeviceId |
| |
| def onu_should_have_at_least_one_flow(self): |
| for onuDeviceId in self.__onuDeviceIds: |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, 'Flows', 'voltha_onu_flows_' + |
| str(self.__onuDeviceIds.index(onuDeviceId)) + '.log') |
| assert statusLines, 'No Onu flows under device %s' % onuDeviceId |
| before, flows, numFlows = statusLines.partition('Flows') |
| plainNumber = numFlows.strip().strip('():') |
| if plainNumber.isdigit(): |
| assert int(plainNumber) > 0, 'Zero number of flows for Onu %s' % onuDeviceId |
| |
| |
| def run_test(logical_device_type, olt_type, onu_type, log_dir): |
| discovery = Discovery() |
| discovery.d_set_log_dirs(log_dir) |
| discovery.d_configure(logical_device_type, olt_type, onu_type) |
| discovery.olt_discovery() |
| discovery.onu_discovery() |
| discovery.logical_device() |
| discovery.olt_ports_should_be_enabled_and_active() |
| discovery.onu_ports_should_be_enabled_and_active() |
| discovery.olt_should_have_at_least_one_flow() |
| discovery.onu_should_have_at_least_one_flow() |