conductor/conductor/data/plugins/inventory_provider/cps.py

   1 #
   2 # -------------------------------------------------------------------------
   3 #   Copyright (C) 2021 Wipro Limited.
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #       http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 #
  17 # -------------------------------------------------------------------------
  18 #
  19 from conductor.common import rest
  20 from conductor.i18n import _LE
  21 from oslo_config import cfg
  22 from oslo_log import log
  23 import time
  24 import uuid
  25
  26
  27 LOG = log.getLogger(__name__)
  28
  29 CONF = cfg.CONF
  30
  31 CPS_OPTS = [
  32     cfg.StrOpt('table_prefix',
  33                default='cps',
  34                help='Data Store table prefix.'),
  35     cfg.StrOpt('server_url',
  36                default='https://controller:8443/cps',
  37                help='Base URL for CPS, up to and not including '
  38                     'the version, and without a trailing slash.'),
  39     cfg.StrOpt('cps_rest_timeout',
  40                default='30',
  41                help='Timeout for CPS Rest Call'),
  42     cfg.StrOpt('cps_retries',
  43                default='3',
  44                help='Number of retry for CPS Rest Call'),
  45     # TODO(larry): follow-up with ONAP people on this (CPS basic auth username and password?)
  46     cfg.StrOpt('certificate_file',
  47                default='certificate.pem',
  48                help='SSL/TLS certificate file in pem format. '
  49                     'This certificate must be registered with the CPS '
  50                     'endpoint.'),
  51     cfg.StrOpt('certificate_key_file',
  52                default='certificate_key.pem',
  53                help='Private Certificate Key file in pem format.'),
  54     cfg.StrOpt('certificate_authority_bundle_file',
  55                default='certificate_authority_bundle.pem',
  56                help='Certificate Authority Bundle file in pem format. '
  57                     'Must contain the appropriate trust chain for the '
  58                     'Certificate file.'),
  59     cfg.StrOpt('username',
  60                default='',
  61                help='Username for CPS.'),
  62     cfg.StrOpt('password',
  63                default='',
  64                help='Password for CPS.'),
  65     cfg.StrOpt('get_ta_list_url',
  66                default='',
  67                help="url to get ta_list")
  68 ]
  69
  70 CONF.register_opts(CPS_OPTS, group='cps')
  71
  72
  73 class CPS(object):
  74
  75     def __init__(self):
  76         """Initializer"""
  77
  78         self.conf = CONF
  79
  80         self.base = self.conf.cps.server_url.rstrip('/')
  81         self.cert = self.conf.cps.certificate_file
  82         self.key = self.conf.cps.certificate_key_file
  83         self.verify = self.conf.cps.certificate_authority_bundle_file
  84         self.timeout = self.conf.cps.cps_rest_timeout
  85         self.retries = self.conf.cps.cps_retries
  86         self.username = self.conf.cps.username
  87         self.password = self.conf.cps.password
  88         self._init_python_request()
  89
  90     def _request(self, method='post', path='/', data=None,
  91                  context=None, value=None):
  92         """Performs HTTP request."""
  93         headers = {
  94             'X-FromAppId': 'CONDUCTOR',
  95             'X-TransactionId': str(uuid.uuid4()),
  96         }
  97         kwargs = {
  98             "method": method,
  99             "path": path,
 100             "headers": headers,
 101             "data": data,
 102         }
 103
 104         # TODO(jdandrea): Move timing/response logging into the rest helper?
 105         start_time = time.time()
 106         response = self.rest.request(**kwargs)
 107         elapsed = time.time() - start_time
 108         LOG.debug("Total time for CPS request "
 109                   "({0:}: {1:}): {2:.3f} sec".format(context, value, elapsed))
 110
 111         if response is None:
 112             LOG.error(_LE("No response from CPS ({}: {})").
 113                       format(context, value))
 114         elif response.status_code != 200:
 115             LOG.error(_LE("CPS request ({}: {}) returned HTTP "
 116                           "status {} {}, link: {}{}").
 117                       format(context, value,
 118                              response.status_code, response.reason,
 119                              self.base, path))
 120         return response
 121
 122     def _init_python_request(self):
 123
 124         kwargs = {
 125             "server_url": self.base,
 126             "retries": self.retries,
 127             "username": self.username,
 128             "password": self.password,
 129             "read_timeout": self.timeout,
 130         }
 131         self.rest = rest.REST(**kwargs)
 132
 133     def get_coveragearea_ta(self, args):
 134         response = self.get_cps_response(args)
 135         return response
 136
 137     def get_cps_response(self, args):
 138         path = self.conf.cps.get_ta_list_url
 139         data = {}
 140         data['input'] = {'zone_id': args}
 141         cps_response = self._request('post', path, data=data)
 142         if cps_response is None or cps_response.status_code != 200:
 143             return None
 144         if cps_response:
 145             return cps_response.json()