blob: a7f8a7f2af1ff549d7fd83df02e58249c326154f [file] [log] [blame]
#
# 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.__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):
self.__oltIpAddress = olt_ip_address
self.__oltPort = olt_port
self.__oltType = olt_type
self.__onuType = onu_type
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 ponsim_olt -H %s:%s' %
(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
adminState = self.__fields[7].strip()
assert adminState == 'ENABLED', 'Admin State of %s not ENABLED' % dev_type
operatorStatus = self.__fields[8].strip()
assert operatorStatus == 'ACTIVE', 'Operator Status of %s not ACTIVE' % dev_type
connectStatus = self.__fields[9].strip()
assert connectStatus == 'REACHABLE', 'Connect Status of %s not 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
hostPort = self.__fields[11].strip()
assert hostPort, 'hostPort field is empty'
hostPortFields = hostPort.split(":")
assert hostPortFields[0] == self.__oltIpAddress or hostPortFields[1] == str(self.__oltPort), \
'Olt IP 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 == 1, 'Fixed single onu 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 run_test(olt_ip_address, olt_port, olt_type, onu_type, log_dir):
preprovisioning = Preprovisioning()
preprovisioning.p_set_log_dirs(log_dir)
preprovisioning.p_configure(olt_ip_address, olt_port, olt_type, onu_type)
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.query_devices_after_enabling()
preprovisioning.check_olt_fields_after_enabling()
preprovisioning.check_onu_fields_after_enabling()