functest/opnfv_tests/Controllers/ONOS/Teston/adapters/client.py

   1 """
   2 Description:
   3     This file is used to run testcase
   4     lanqinglong@huawei.com
   5
   6 #
   7 # All rights reserved. This program and the accompanying materials
   8 # are made available under the terms of the Apache License, Version 2.0
   9 # which accompanies this distribution, and is available at
  10 # http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 """
  13 import json
  14 import pexpect
  15 import requests
  16 import time
  17
  18 from environment import environment
  19 import functest.utils.functest_logger as ft_logger
  20
  21
  22 class client(environment):
  23
  24     logger = ft_logger.Logger("client").getLogger()
  25
  26     def __init__(self):
  27         environment.__init__(self)
  28         self.loginfo = environment()
  29         self.testcase = ''
  30
  31     def RunScript(self, handle, testname, timeout=300):
  32         """
  33         Run ONOS Test Script
  34         Parameters:
  35         testname: ONOS Testcase Name
  36         masterusername: The server username of running ONOS
  37         masterpassword: The server password of running ONOS
  38         """
  39         self.testcase = testname
  40         self.ChangeTestCasePara(testname, self.masterusername,
  41                                 self.masterpassword)
  42         runhandle = handle
  43         runtest = (self.home + "/OnosSystemTest/TestON/bin/cli.py run " +
  44                    testname)
  45         runhandle.sendline(runtest)
  46         circletime = 0
  47         lastshowscreeninfo = ''
  48         while True:
  49             Result = runhandle.expect(["PEXPECT]#", pexpect.EOF,
  50                                        pexpect.TIMEOUT])
  51             curshowscreeninfo = runhandle.before
  52             if(len(lastshowscreeninfo) != len(curshowscreeninfo)):
  53                 self.loginfo.log(str(curshowscreeninfo)
  54                                  [len(lastshowscreeninfo)::])
  55                 lastshowscreeninfo = curshowscreeninfo
  56             if Result == 0:
  57                 self.logger.info("Done!")
  58                 return
  59             time.sleep(1)
  60             circletime += 1
  61             if circletime > timeout:
  62                 break
  63         self.loginfo.log("Timeout when running the test, please check!")
  64
  65     def onosstart(self):
  66         # This is the compass run machine user&pass,you need to modify
  67
  68         self.logger.info("Test Begin.....")
  69         self.OnosConnectionSet()
  70         masterhandle = self.SSHlogin(self.localhost, self.masterusername,
  71                                      self.masterpassword)
  72         self.OnosEnvSetup(masterhandle)
  73         return masterhandle
  74
  75     def onosclean(self, handle):
  76         self.SSHRelease(handle)
  77         self.loginfo.log('Release onos handle Successful')
  78
  79     def push_results_to_db(self, payload, pushornot=1):
  80         if pushornot != 1:
  81             return 1
  82         url = self.Result_DB + "/results"
  83         params = {"project_name": "functest", "case_name": "ONOS-" +
  84                   self.testcase, "pod_name": 'huawei-build-2',
  85                   "details": payload}
  86
  87         headers = {'Content-Type': 'application/json'}
  88         try:
  89             r = requests.post(url, data=json.dumps(params), headers=headers)
  90             self.loginfo.log(r)
  91         except:
  92             self.loginfo.log('Error pushing results into Database')