[VOL-1129] Refactor Voltha Test Automation Robot
Add missing License Header to files
Change-Id: I04311b17ad2f4c7af77c8bfd9539970915fbfea5
diff --git a/tests/atests/common/preprovisioningTest.py b/tests/atests/common/preprovisioningTest.py
new file mode 100755
index 0000000..a8bbe3e
--- /dev/null
+++ b/tests/atests/common/preprovisioningTest.py
@@ -0,0 +1,105 @@
+#
+# 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()
+
+
+
+