test/csit/libraries/UtilLibrary.py

   1 __author__ = "Basheeruddin Ahmed"
   2 __copyright__ = "Copyright(c) 2014, Cisco Systems, Inc."
   3 __license__ = "New-style BSD"
   4 __email__ = "syedbahm@cisco.com"
   5
   6
   7 import requests
   8 from SSHLibrary import SSHLibrary
   9
  10 import robot
  11
  12 global _cache
  13
  14 #
  15 #Helps in making GET REST calls
  16 #
  17
  18 def get(url, userId=None, password=None):
  19
  20     headers = {}
  21     headers['Accept']= 'application/xml'
  22
  23     # Send the GET request
  24     session = _cache.switch("CLUSTERING_GET")
  25     resp = session.get(url,headers=headers)
  26
  27     # Read the response
  28     return resp
  29
  30 #
  31 #Helps in making POST REST calls without outputs
  32 #
  33 def nonprintpost(url, userId, password,data):
  34
  35     headers = {}
  36     headers['Content-Type'] = 'application/json'
  37     #headers['Accept']= 'application/xml'
  38
  39     session = _cache.switch("CLUSTERING_POST")
  40     resp = session.post(url,data.encode('utf-8'),headers=headers)
  41
  42
  43     return resp
  44
  45 #
  46 #Helps in making POST REST calls
  47 #
  48 def post(url, userId, password,data):
  49     print("post request with url "+url)
  50     print("post request with data "+data)
  51     headers = {}
  52     headers['Content-Type'] = 'application/json'
  53     #headers['Accept']= 'application/xml'
  54     session = _cache.switch("CLUSTERING_POST")
  55     resp = session.post(url,data.encode('utf-8'),headers=headers)
  56
  57     #print (resp.raise_for_status())
  58     print (resp.headers)
  59
  60     return resp
  61
  62 #
  63 #Helps in making DELET REST calls
  64 #
  65 def delete(url,userId,password):
  66     print("delete all resources belonging to url"+url)
  67     session = _cache.switch("CLUSTERING_DELETE")
  68     resp=session.delete(url)
  69
  70 def Should_Not_Be_Type_None(var):
  71     '''Keyword to check if the given variable is of type NoneType.  If the
  72         variable type does match  raise an assertion so the keyword will fail
  73     '''
  74     if var == None:
  75         raise AssertionError('the variable passed was type NoneType')
  76     return 'PASS'
  77
  78 # use username and password of controller server for ssh and need
  79 # karaf distribution location like /root/Documents/dist
  80 #
  81 def execute_ssh_command(ip, username, password, command):
  82     print "executing ssh command"
  83     lib = SSHLibrary()
  84     lib.open_connection(ip)
  85     lib.login(username=username,password=password)
  86     print "login done"
  87     lib.execute_command(command)
  88     print "command executed : " + command
  89     lib.close_connection()
  90
  91 def startcontroller(ip,username,password,karafhome):
  92     execute_ssh_command(ip, username, password, karafhome+"/bin/start")
  93
  94 def stopcontroller(ip,username,password,karafhome):
  95     execute_ssh_command(ip, username, password, karafhome+"/bin/stop")
  96
  97 def clean_journal(ip, username, password, karafHome):
  98     execute_ssh_command(ip, username, password, "rm -rf " + karafHome + "/journal")
  99
 100 def kill_controller(ip, username, password, karafHome):
 101     execute_ssh_command(ip, username, password, "ps axf | grep karaf | grep -v grep | awk '{print \"kill -9 \" $1}' | sh")
 102
 103 #
 104 # main invoked
 105 if __name__ != "__main__":
 106     _cache = robot.utils.ConnectionCache('No sessions created')
 107     # here create one session for each HTTP functions
 108     _cache.register(requests.session(), alias='CLUSTERING_GET')
 109     _cache.register(requests.session(),alias='CLUSTERING_POST')
 110     _cache.register(requests.session(),alias='CLUSTERING_DELETE')