stop hardcoded FUNCTEST_REPO path everywhere
[functest.git] / testcases / Controllers / ONOS / Teston / onosfunctest.py
old mode 100644 (file)
new mode 100755 (executable)
index 38935c5..6b922fb
@@ -14,41 +14,49 @@ lanqinglong@huawei.com
 #
 """
 
-import argparse
 import datetime
 import os
 import re
 import time
-import yaml
 
+import argparse
 import functest.utils.functest_logger as ft_logger
 import functest.utils.functest_utils as functest_utils
+import functest.utils.openstack_utils as openstack_utils
+from functest.utils.functest_utils import FUNCTEST_REPO as REPO_PATH
+from neutronclient.v2_0 import client as neutronclient
 
 parser = argparse.ArgumentParser()
-parser.add_argument("-i", "--installer", help="Installer type")
+parser.add_argument("-t", "--testcase", help="Testcase name")
 args = parser.parse_args()
+
+
 """ logging configuration """
 logger = ft_logger.Logger("onos").getLogger()
 
-with open(os.environ["CONFIG_FUNCTEST_YAML"]) as f:
-    functest_yaml = yaml.safe_load(f)
-f.close()
-
 # onos parameters
-TEST_DB = functest_yaml.get("results").get("test_db_url")
-ONOS_REPO_PATH = functest_yaml.get("general").get("directories").get(
-    "dir_repos")
-ONOS_CONF_DIR = functest_yaml.get("general").get("directories").get(
-    "dir_functest_conf")
-REPO_PATH = ONOS_REPO_PATH + '/functest/'
-if not os.path.exists(REPO_PATH):
-    logger.error("Functest repository directory not found '%s'" % REPO_PATH)
-    exit(-1)
-
-ONOSCI_PATH = REPO_PATH + 'testcases/Controllers/ONOS/Teston/'
+TEST_DB = functest_utils.get_parameter_from_yaml(
+    "results.test_db_url")
+ONOS_REPO_PATH = functest_utils.get_parameter_from_yaml(
+    "general.directories.dir_repos")
+ONOS_CONF_DIR = functest_utils.get_parameter_from_yaml(
+    "general.directories.dir_functest_conf")
+
+ONOSCI_PATH = ONOS_REPO_PATH + "/"
 starttime = datetime.datetime.now()
 
 HOME = os.environ['HOME'] + "/"
+INSTALLER_TYPE = os.environ['INSTALLER_TYPE']
+DEPLOY_SCENARIO = os.environ['DEPLOY_SCENARIO']
+ONOSCI_PATH = ONOS_REPO_PATH + "/"
+GLANCE_IMAGE_NAME = functest_utils.get_parameter_from_yaml(
+    "onos_sfc.image_name")
+GLANCE_IMAGE_FILENAME = functest_utils.get_parameter_from_yaml(
+    "onos_sfc.image_file_name")
+GLANCE_IMAGE_PATH = functest_utils.get_parameter_from_yaml(
+    "general.directories.dir_functest_data") + "/" + GLANCE_IMAGE_FILENAME
+SFC_PATH = REPO_PATH + "/" + functest_utils.get_parameter_from_yaml(
+    "general.directories.dir_onos_sfc")
 
 
 def RunScript(testname):
@@ -57,12 +65,12 @@ def RunScript(testname):
     Parameters:
     testname: ONOS Testcase Name
     """
-    runtest = ONOSCI_PATH + "OnosSystemTest/TestON/bin/cli.py run " + testname
+    runtest = ONOSCI_PATH + "onos/TestON/bin/cli.py run " + testname
     logger.debug("Run script " + testname)
     os.system(runtest)
 
 
-def DownloadCodes(url="https://github.com/sunyulin/OnosSystemTest.git"):
+def DownloadCodes(url="https://github.com/wuwenbin2/OnosSystemTest.git"):
     """
     Download Onos Teston codes
     Parameters:
@@ -74,7 +82,7 @@ def DownloadCodes(url="https://github.com/sunyulin/OnosSystemTest.git"):
 
 
 def GetResult():
-    LOGPATH = ONOSCI_PATH + "OnosSystemTest/TestON/logs"
+    LOGPATH = ONOSCI_PATH + "onos/TestON/logs"
     cmd = "grep -rnh " + "Fail" + " " + LOGPATH
     Resultbuffer = os.popen(cmd).read()
     # duration = datetime.datetime.now() - starttime
@@ -156,25 +164,69 @@ def SetOnosIpForJoid():
 
 
 def CleanOnosTest():
-    TESTONPATH = ONOSCI_PATH + "OnosSystemTest/"
+    TESTONPATH = ONOSCI_PATH + "onos/"
     cmd = "rm -rf " + TESTONPATH
     os.system(cmd)
     time.sleep(2)
     logger.debug("Clean ONOS Teston")
 
 
-def main():
+def CreateImage():
+    glance_client = openstack_utils.get_glance_client()
+    image_id = openstack_utils.create_glance_image(glance_client,
+                                                   GLANCE_IMAGE_NAME,
+                                                   GLANCE_IMAGE_PATH)
+    EXIT_CODE = -1
+    if not image_id:
+        logger.error("Failed to create a Glance image...")
+        return(EXIT_CODE)
+    logger.debug("Image '%s' with ID=%s created successfully."
+                 % (GLANCE_IMAGE_NAME, image_id))
+
+
+def SfcTest():
+    cmd = "python " + SFC_PATH + "Sfc.py"
+    logger.debug("Run sfc tests")
+    os.system(cmd)
+
+
+def GetIp(type):
+    cmd = "openstack catalog show " + type + " | grep publicURL"
+    cmd_output = os.popen(cmd).read()
+    ip = re.search(r"\d+\.\d+\.\d+\.\d+", cmd_output).group()
+    return ip
+
+
+def Replace(before, after):
+    file = "Sfc_fun.py"
+    cmd = "sed -i 's/" + before + "/" + after + "/g' " + SFC_PATH + file
+    os.system(cmd)
+
+
+def SetSfcConf():
+    Replace("keystone_ip", GetIp("keystone"))
+    Replace("neutron_ip", GetIp("neutron"))
+    Replace("nova_ip", GetIp("nova"))
+    Replace("glance_ip", GetIp("glance"))
+    pwd = os.environ['OS_PASSWORD']
+    Replace("console", pwd)
+    creds_neutron = openstack_utils.get_credentials("neutron")
+    neutron_client = neutronclient.Client(**creds_neutron)
+    ext_net = openstack_utils.get_external_net(neutron_client)
+    Replace("admin_floating_net", ext_net)
+    logger.info("Modify configuration for SFC")
+
+
+def OnosTest():
     start_time = time.time()
     stop_time = start_time
-    DownloadCodes()
-    if args.installer == "joid":
+    if INSTALLER_TYPE == "joid":
         logger.debug("Installer is Joid")
         SetOnosIpForJoid()
     else:
         SetOnosIp()
     RunScript("FUNCvirNetNB")
     RunScript("FUNCvirNetNBL3")
-
     try:
         logger.debug("Push ONOS results into DB")
         # TODO check path result for the file
@@ -183,16 +235,16 @@ def main():
 
         # ONOS success criteria = all tests OK
         # i.e. FUNCvirNet & FUNCvirNetL3
-        status = "failed"
+        status = "FAIL"
         try:
             if (result['FUNCvirNet']['result'] == "Success" and
                     result['FUNCvirNetL3']['result'] == "Success"):
-                    status = "passed"
+                    status = "PASS"
         except:
             logger.error("Unable to set ONOS criteria")
 
         functest_utils.push_results_to_db("functest",
-                                          "ONOS",
+                                          "onos",
                                           logger,
                                           start_time,
                                           stop_time,
@@ -202,8 +254,19 @@ def main():
     except:
         logger.error("Error pushing results into Database")
 
-    CleanOnosTest()
+    if status == "FAIL":
+        EXIT_CODE = -1
+        exit(EXIT_CODE)
+
 
+def main():
+
+    if args.testcase == "sfc":
+        CreateImage()
+        SetSfcConf()
+        SfcTest()
+    else:
+        OnosTest()
 
 if __name__ == '__main__':
     main()