testcases/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 from environment import environment
  14 import time
  15 import pexpect
  16 import requests
  17 import json
  18
  19
  20 class client(environment):
  21
  22     def __init__(self):
  23         environment.__init__(self)
  24         self.loginfo = environment()
  25         self.testcase = ''
  26
  27     def RunScript(self, handle, testname, timeout=300):
  28         """
  29         Run ONOS Test Script
  30         Parameters:
  31         testname: ONOS Testcase Name
  32         masterusername: The server username of running ONOS
  33         masterpassword: The server password of running ONOS
  34         """
  35         self.testcase = testname
  36         self.ChangeTestCasePara(testname, self.masterusername,
  37                                 self.masterpassword)
  38         runhandle = handle
  39         runtest = (self.home + "/OnosSystemTest/TestON/bin/cli.py run " +
  40                    testname)
  41         runhandle.sendline(runtest)
  42         circletime = 0
  43         lastshowscreeninfo = ''
  44         while True:
  45             Result = runhandle.expect(["PEXPECT]#", pexpect.EOF,
  46                                        pexpect.TIMEOUT])
  47             curshowscreeninfo = runhandle.before
  48             if(len(lastshowscreeninfo) != len(curshowscreeninfo)):
  49                 self.loginfo.log(str(curshowscreeninfo)
  50                                  [len(lastshowscreeninfo)::])
  51                 lastshowscreeninfo = curshowscreeninfo
  52             if Result == 0:
  53                 print "Done!"
  54                 return
  55             time.sleep(1)
  56             circletime += 1
  57             if circletime > timeout:
  58                 break
  59         self.loginfo.log("Timeout when running the test, please check!")
  60
  61     def onosstart(self):
  62         # This is the compass run machine user&pass,you need to modify
  63
  64         print "Test Begin....."
  65         self.OnosConnectionSet()
  66         masterhandle = self.SSHlogin(self.localhost, self.masterusername,
  67                                      self.masterpassword)
  68         self.OnosEnvSetup(masterhandle)
  69         return masterhandle
  70
  71     def onosclean(self, handle):
  72         self.SSHRelease(handle)
  73         self.loginfo.log('Release onos handle Successful')
  74
  75     def push_results_to_db(self, payload, pushornot=1):
  76         if pushornot != 1:
  77             return 1
  78         url = self.Result_DB + "/results"
  79         params = {"project_name": "functest", "case_name": "ONOS-" +
  80                   self.testcase, "pod_name": 'huawei-build-2',
  81                   "details": payload}
  82
  83         headers = {'Content-Type': 'application/json'}
  84         try:
  85             r = requests.post(url, data=json.dumps(params), headers=headers)
  86             self.loginfo.log(r)
  87         except:
  88             self.loginfo.log('Error pushing results into Database')