blob: a8bbe3ef5539c66db538f61701739616d03a11e4 [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 module
"""
import time
import os
import commands
class preprovisioningTest(object):
"""
This class implements voltha pre-provisioning test
"""
def __init__(self):
self.__oltIpAddress = None
self.__oltPort = None
self.__logDir = None
self.__oltDeviceId = None
self.__statusLine = ""
self.__fields = []
def configure(self, oltIpAddress, oltPort, logDir):
self.__oltIpAddress = oltIpAddress
self.__oltPort = oltPort
self.__logDir = logDir
def preprovisionOlt(self):
print('Do PROVISIONING')
self.send_command_to_voltha_cli(
'preprovision_olt -t ponsim_olt -H %s:%s' %
(self.__oltIpAddress, self.__oltPort),
'voltha_preprovision_olt.log')
time.sleep(5)
def query_devices_before_enable(self):
self.send_command_to_voltha_cli('devices',
'voltha_devices_before_enable.log')
time.sleep(5)
grepCommand =\
"grep PREPROVISIONED %s/voltha_devices_before_enable.log " % self.__logDir
self.__statusLine = commands.getstatusoutput(grepCommand)[1]
self.__fields = self.parseFields(self.__statusLine)
self.__oltDeviceId = self.__fields[1].strip()
print ("OLT device id = %s" % self.__oltDeviceId)
def enable(self):
print('Enable %s OLT device' % self.__oltDeviceId)
self.send_command_to_voltha_cli('enable ' + self.__oltDeviceId,
'voltha_enable.log')
def query_devices_after_enable(self):
self.send_command_to_voltha_cli('devices',
'voltha_devices_after_enable.log')
def send_command_to_voltha_cli(self, cmd, logFile):
# os.system("docker exec -i -t compose_cli_1 sh -c 'echo \"" + cmd +
# "\" > /voltha_tmp_command.txt'")
os.system("docker exec compose_cli_1 sh -c 'echo \"" + cmd +
"\" > /voltha_tmp_command.txt'")
os.system("docker exec compose_cli_1 sh -c '/cli/cli/main.py -C "
"vconsul:8500 -L < /voltha_tmp_command.txt' > " +
self.__logDir + '/' + logFile)
def send_command_to_onos_cli(self, cmd, logFile):
os.system(
"sshpass -p karaf ssh -o StrictHostKeyChecking=no -p 8101 "
"karaf@localhost " + cmd + " 2>/dev/null > " +
self.__logDir + '/' + logFile)
def parseFields(self, statusLine):
statusList = statusLine.split("|")
return statusList
def runTest(oltIpAddress, oltPort, logDir):
preprovisioning = preprovisioningTest()
preprovisioning.configure(str(oltIpAddress), int(oltPort),
str(logDir))
preprovisioning.preprovisionOlt()
preprovisioning.query_devices_before_enable()
preprovisioning.enable()
preprovisioning.query_devices_after_enable()