functest/opnfv_tests/vnf/ims/ixia/utils/IxChassisUtils.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2017 IXIA and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 import httplib2
  11 import json
  12 import logging
  13
  14
  15 okStates = [200, 201, 202]
  16 states = [
  17     'Queued',
  18     'In Progress',
  19     'Manual Step Required',
  20     'Error',
  21     'Finished',
  22     'Aborted',
  23     'Retried',
  24     'IRebooting',
  25     'Force Continue',
  26     'Pending',
  27     ]
  28 notStartedState = 'Not_Started'
  29 errorStates = ['Error', 'Aborted', 'Force Continue']
  30 finishedStates = ['Manual Step Required', 'Finished']
  31
  32 logger = logging.getLogger(__name__)
  33
  34
  35 class TestFailedError(Exception):
  36     pass
  37
  38
  39 class ChassisRestAPI:
  40     @staticmethod
  41     def postWithPayload(loginUrl, payload=None):
  42         urlHeadersJson = {'content-type': 'application/json'}
  43         try:
  44             h = httplib2.Http('.cache',
  45                               disable_ssl_certificate_validation=True)
  46             if payload is None:
  47                 logger.debug('POST: ' + loginUrl)
  48                 (response, content) = h.request(loginUrl, 'POST', '',
  49                                                 urlHeadersJson)
  50                 logger.debug(content)
  51             else:
  52                 logger.debug('POST: ' + loginUrl + ' <- Data: ' + str(payload))
  53                 (response, content) = h.request(loginUrl, 'POST',
  54                                                 body=payload,
  55                                                 headers=urlHeadersJson)
  56                 logger.debug(response)
  57                 logger.debug(content)
  58         except Exception, e:
  59             raise Exception('Got an error code: ', e)
  60         return content
  61
  62     @staticmethod
  63     def postWithPayloadAndHeaders(loginUrl, urlHeadersJson,
  64                                   payload=None):
  65         try:
  66             h = httplib2.Http('.cache',
  67                               disable_ssl_certificate_validation=True)
  68             if payload is None:
  69                 logger.debug('POST: ' + loginUrl)
  70                 (response, content) = h.request(loginUrl, 'POST', '',
  71                                                 urlHeadersJson)
  72             else:
  73                 logger.debug('POST: ' + loginUrl + ' <- Data: ' + str(payload))
  74                 (response, content) = h.request(loginUrl, 'POST',
  75                                                 body=payload,
  76                                                 headers=urlHeadersJson)
  77         except Exception, e:
  78             raise Exception('Got an error code: ', e)
  79         return content
  80
  81     @staticmethod
  82     def postOperation(url, apiKey, payload=''):
  83         urlHeadersJson = {'content-type': 'application/json',
  84                           'X-Api-Key': '%s' % str(apiKey)}
  85         try:
  86             h = httplib2.Http('.cache',
  87                               disable_ssl_certificate_validation=True)
  88             if payload is None:
  89                 logger.debug('POST: ' + url)
  90                 (response, content) = h.request(url, 'POST',
  91                                                 json.dumps(payload),
  92                                                 urlHeadersJson)
  93             else:
  94                 logger.debug('POST: ' + url + ' <- Data: ' + str(payload))
  95                 (response, content) = h.request(url, 'POST',
  96                                                 json.dumps(payload),
  97                                                 headers=urlHeadersJson)
  98         except Exception, e:
  99             raise Exception('Got an error code: ', e)
 100         return content
 101
 102     @staticmethod
 103     def patch(url, payload, apiKey):
 104         urlHeadersJson = {'content-type': 'application/json',
 105                           'X-Api-Key': '%s' % str(apiKey)}
 106         try:
 107             h = httplib2.Http('.cache',
 108                               disable_ssl_certificate_validation=True)
 109             logger.debug('PATCH: ' + url + ' <-- Attribute: ' +
 110                          str(payload))
 111             (response, content) = h.request(url, 'PATCH',
 112                                             json.dumps(payload),
 113                                             urlHeadersJson)
 114         except Exception, e:
 115
 116             # print (response, content)
 117
 118             raise Exception('Got an error code: ', e)
 119         return content
 120
 121     @staticmethod
 122     def delete(url, apiKey):
 123         urlHeadersJson = {'content-type': 'application/json',
 124                           'X-Api-Key': '%s' % str(apiKey)}
 125         try:
 126             h = httplib2.Http('.cache',
 127                               disable_ssl_certificate_validation=True)
 128             (response, content) = h.request(url, 'DELETE', '', urlHeadersJson)
 129             logger.debug('DELETE: ' + url)
 130         except Exception, e:
 131             raise Exception('Got an error code: ', e)
 132         if response.status not in okStates:
 133             raise TestFailedError(json.loads(content)['error'])
 134         return json.loads(content)
 135
 136     @staticmethod
 137     def getWithHeaders(url, apiKey):
 138         urlHeadersJson = {'content-type': 'application/json',
 139                           'X-Api-Key': '%s' % str(apiKey)}
 140         try:
 141             h = httplib2.Http('.cache',
 142                               disable_ssl_certificate_validation=True)
 143             logger.debug('GET: ' + url)
 144             (response, content) = h.request(url, 'GET', '', urlHeadersJson)
 145         except Exception, e:
 146             raise Exception('Got an error code: ', e)
 147         if response.status not in okStates:
 148             raise TestFailedError(json.loads(content)['error'])
 149         output = json.loads(content)
 150         return output