| # |
| # 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 Pre-provisioning Test Case module |
| """ |
| |
| import time |
| import testCaseUtils |
| import logging |
| |
| |
| class Preprovisioning(object): |
| |
| """ |
| This class implements voltha pre-provisioning test |
| """ |
| |
| def __init__(self): |
| self.dirs = dict() |
| self.dirs['log'] = None |
| self.dirs['root'] = None |
| self.dirs['voltha'] = None |
| |
| self.__oltIpAddress = None |
| self.__oltPort = None |
| self.__oltType = None |
| self.__onuType = None |
| self.__onuCount = None |
| self.__fields = [] |
| self.__oltDeviceId = None |
| |
| def p_set_log_dirs(self, log_dir): |
| testCaseUtils.config_dirs(self, log_dir) |
| |
| def p_configure(self, olt_ip_address, olt_port, olt_type, onu_type, onu_count): |
| self.__oltIpAddress = olt_ip_address |
| self.__oltPort = olt_port |
| self.__oltType = olt_type |
| self.__onuType = onu_type |
| self.__onuCount = onu_count |
| |
| def preprovision_olt(self): |
| logging.info('Do PROVISIONING') |
| testCaseUtils.send_command_to_voltha_cli(testCaseUtils.get_dir(self, 'log'), |
| 'voltha_preprovision_olt.log', 'preprovision_olt -t %s -H %s:%s' % |
| (self.__oltType, self.__oltIpAddress, self.__oltPort)) |
| time.sleep(5) |
| |
| def status_should_be_success_after_preprovision_command(self): |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, 'success', 'voltha_preprovision_olt.log') |
| assert statusLines, 'Preprovision Olt command should have returned success but did not' |
| |
| def query_devices_before_enabling(self): |
| testCaseUtils.send_command_to_voltha_cli(testCaseUtils.get_dir(self, 'log'), |
| 'voltha_devices_before_enable.log', 'devices') |
| testCaseUtils.print_log_file(self, 'voltha_devices_before_enable.log') |
| time.sleep(5) |
| |
| def check_olt_fields_before_enabling(self): |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, self.__oltType, 'voltha_devices_before_enable.log') |
| assert statusLines, 'No Olt listed under devices' |
| self.__fields = testCaseUtils.parse_fields(statusLines, '|') |
| self.__oltDeviceId = self.__fields[1].strip() |
| logging.debug("OLT device id = %s" % self.__oltDeviceId) |
| adminState = self.__fields[3].strip() |
| assert adminState == 'PREPROVISIONED', 'Admin State not PREPROVISIONED' |
| hostPort = self.__fields[4].strip() |
| assert hostPort, 'hostPort field is empty' |
| hostPortFields = hostPort.split(":") |
| assert hostPortFields[0].strip() == self.__oltIpAddress or hostPortFields[1] == str(self.__oltPort), \ |
| 'Olt IP or Port does not match' |
| |
| def check_states(self, dev_type): |
| result = True |
| stateMatchCount = 0 |
| for field in self.__fields: |
| field_no_space = field.strip() |
| if field_no_space == 'ENABLED' or field_no_space == 'ACTIVE' or field_no_space == 'DISCOVERED' or field_no_space == 'REACHABLE': |
| stateMatchCount += 1 |
| assert stateMatchCount == 3, 'State of %s is not ENABLED, ACTIVE or DISCOVERED and REACHABLE' % dev_type |
| return result |
| |
| def check_olt_fields_after_enabling(self): |
| 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, '|') |
| assert self.check_states(self.__oltType), 'States of %s does match expected' % self.__oltType |
| hostPortCount = False |
| for field in self.__fields: |
| if field.strip() == self.__oltIpAddress + ':' + str(self.__oltPort): |
| hostPortCount = True |
| assert hostPortCount, 'hostPort field is empty or Olt IP and/or Port does not match' |
| |
| def check_onu_fields_after_enabling(self): |
| 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() |
| lenLines = len(lines) |
| assert lenLines == self.__onuCount, ' Discovered onu(s) does not match, ONU Count was %d' % lenLines |
| for line in lines: |
| self.__fields = testCaseUtils.parse_fields(line, '|') |
| assert (self.check_states(self.__onuType) is True), 'States of %s does match expected' % self.__onuType |
| |
| def enable(self): |
| logging.info('Enable %s OLT device' % self.__oltDeviceId) |
| testCaseUtils.send_command_to_voltha_cli(testCaseUtils.get_dir(self, 'log'), |
| 'voltha_enable.log', 'enable ' + self.__oltDeviceId) |
| |
| def status_should_be_success_after_enable_command(self): |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, 'success', 'voltha_enable.log') |
| assert statusLines, 'Enable command should have returned success but did not' |
| |
| def query_devices_after_enabling(self): |
| testCaseUtils.send_command_to_voltha_cli(testCaseUtils.get_dir(self, 'log'), |
| 'voltha_devices_after_enable.log', 'devices') |
| testCaseUtils.print_log_file(self, 'voltha_devices_after_enable.log') |
| |
| def proceed(self): |
| logging.info('Pre-provisioning hold') |
| onuOnline = 0 |
| if self.__oltType == 'ponsim_olt': |
| portType = 'PON_OLT' |
| while onuOnline < self.__onuCount: |
| testCaseUtils.send_command_to_voltha_cli(testCaseUtils.get_dir(self, 'log'), 'voltha_olt_device.log', 'device ' + |
| self.__oltDeviceId, 'voltha_olt_ports.log', 'ports') |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, portType, 'voltha_olt_ports.log') |
| lines = statusLines.splitlines() |
| onuOnline = len(lines) |
| time.sleep(5) |
| logging.info('All ONUs now Online!') |
| elif self.__oltType == 'openolt': |
| while onuOnline < self.__onuCount: |
| testCaseUtils.send_command_to_onos_cli(testCaseUtils.get_dir(self, 'log'), |
| 'voltha_onu_auth.log', 'aaa-users') |
| statusLines = testCaseUtils.get_fields_from_grep_command(self, 'AUTHORIZED', 'voltha_onu_auth.log') |
| lines = statusLines.splitlines() |
| onuOnline = len(lines) |
| time.sleep(5) |
| logging.info('All ONUs now AUTHORIZED') |
| |
| |
| def run_test(olt_ip_address, olt_port, olt_type, onu_type, onu_count, log_dir): |
| preprovisioning = Preprovisioning() |
| preprovisioning.p_set_log_dirs(log_dir) |
| preprovisioning.p_configure(olt_ip_address, olt_port, olt_type, onu_type, onu_count) |
| preprovisioning.preprovision_olt() |
| preprovisioning.status_should_be_success_after_preprovision_command() |
| preprovisioning.query_devices_before_enabling() |
| preprovisioning.check_olt_fields_before_enabling() |
| preprovisioning.enable() |
| preprovisioning.status_should_be_success_after_enable_command() |
| preprovisioning.proceed() |
| preprovisioning.query_devices_after_enabling() |
| preprovisioning.check_olt_fields_after_enabling() |
| preprovisioning.check_onu_fields_after_enabling() |