o1-vth/o1_vth.py

   1 #   Copyright (c) 2019 AT&T Intellectual Property.                             #\r
   2 #                                                                              #\r
   3 #   Licensed under the Apache License, Version 2.0 (the "License");            #\r
   4 #   you may not use this file except in compliance with the License.           #\r
   5 #   You may obtain a copy of the License at                                    #\r
   6 #                                                                              #\r
   7 #       http://www.apache.org/licenses/LICENSE-2.0                             #\r
   8 #                                                                              #\r
   9 #   Unless required by applicable law or agreed to in writing, software        #\r
  10 #   distributed under the License is distributed on an "AS IS" BASIS,          #\r
  11 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   #\r
  12 #   See the License for the specific language governing permissions and        #\r
  13 #   limitations under the License.                                             #\r
  14 ################################################################################\r
  15 # File name: o1-vth.py                                                        #\r
  16 # Description: Mainly used to get alarm list                                   #\r
  17 # Date created: 04/14/2020                                                     #\r
  18 # Python Version: 3.7                                                          #\r
  19 # Author: Jackie Chen (jv246a)                                                 #\r
  20 # Email: jv246a@att.com                                                        #\r
  21 ################################################################################\r
  22 \r
  23 import datetime\r
  24 from configparser import ConfigParser\r
  25 import os\r
  26 import logging\r
  27 from logging import FileHandler\r
  28 import requests\r
  29 from flask import Flask, request, jsonify\r
  30 \r
  31 # redirect http to https\r
  32 app = Flask(__name__)\r
  33 \r
  34 # Prevents print statement every time an endpoint is triggered.\r
  35 logging.getLogger("werkzeug").setLevel(logging.WARNING)\r
  36 \r
  37 \r
  38 def sendCallback(url, data):\r
  39     try:\r
  40         if type(data) is not dict:\r
  41             data = {"msg": data}\r
  42         app.logger.info("sending callback")\r
  43         requests.post(url, json= data)\r
  44     except Exception as e:\r
  45         app.logger.info(e)\r
  46     return\r
  47 \r
  48 def unix_time_millis(dt):\r
  49     epoch = datetime.datetime.utcfromtimestamp(0)\r
  50     return (dt - epoch).total_seconds() * 1000.0\r
  51 \r
  52 \r
  53 def _get_config(config_file_name):\r
  54     config = ConfigParser(os.environ)\r
  55     config.read(config_file_name)\r
  56     return config\r
  57 \r
  58 \r
  59 def _build_url(config):\r
  60     return config['resource']['base_address'] + config['resource']['get_config_alarms_list']\r
  61 \r
  62 \r
  63 def _send_request(url, config):\r
  64     # setup default values\r
  65     proxy_enabled = config.getboolean('resource', 'proxy_enabled')\r
  66     auth_enabled = config.getboolean('auth', 'auth_enabled')\r
  67     username = ''\r
  68     password = ''\r
  69 \r
  70     req_proxies = {\r
  71         'http': None,\r
  72         'https': None\r
  73     }\r
  74     # place proxy information\r
  75     if proxy_enabled:\r
  76         req_proxies['http'] = config['resource']['http_proxy']\r
  77         req_proxies['https'] = config['resource']['https_proxy']\r
  78     if auth_enabled:\r
  79         username = config['auth']['username']\r
  80         password = config['auth']['password']\r
  81     # get call for alarm list\r
  82     return requests.get(url,\r
  83                         auth=(username, password) if auth_enabled else None,\r
  84                         proxies=req_proxies if proxy_enabled else None)\r
  85 \r
  86 def _parse_resposne(response):\r
  87     try:\r
  88         return response.json()\r
  89     except ValueError:\r
  90         return response.text\r
  91 \r
  92 @app.route('/otf/vth/oran/smo/v1/alarm-list' , methods=['POST'])\r
  93 def get_alarm_list():\r
  94     response_data = {\r
  95         'vthResponse': {\r
  96             'testDuration': '',\r
  97             'dateTimeUTC': str(datetime.datetime.now()),\r
  98             'abstractMessage': '',\r
  99             'resultData': {}\r
 100         }\r
 101     }\r
 102     ret_url = request.args.get('retURL')\r
 103     startTime = unix_time_millis(datetime.datetime.now())\r
 104     try:\r
 105         # setup phase\r
 106         config = _get_config('config.ini')\r
 107         alarm_list_url = _build_url(config)\r
 108 \r
 109         # build initial response\r
 110         app.logger.info('Sending GET for alarm list')\r
 111         res = _send_request(alarm_list_url, config)\r
 112         app.logger.info('Status code from GET: {}'.format(res.status_code))\r
 113         app.logger.info('Response received from GET alarm-list: {}'.format(res.content))\r
 114         response_data['vthResponse']['abstractMessage'] = 'Result from GET alarm list request'\r
 115         response_data['vthResponse']['resultData']['status_code'] = res.status_code\r
 116         response_data['vthResponse']['resultData']['result_output'] = _parse_resposne(res)\r
 117     except Exception as ex:\r
 118         endTime = unix_time_millis(datetime.datetime.now())\r
 119         totalTime = endTime - startTime\r
 120         response_data['vthResponse']['testDuration'] = totalTime\r
 121         response_data['vthResponse']['abstractMessage'] = 'error: ' + str(ex)\r
 122         app.logger.error('ERROR:{}'.format(str(ex)))\r
 123         return jsonify(response_data)\r
 124 \r
 125     #finish up building response\r
 126     endTime = unix_time_millis(datetime.datetime.now())\r
 127     totalTime = endTime - startTime\r
 128     response_data['vthResponse']['testDuration'] = totalTime\r
 129     if ret_url is not None:\r
 130         sendCallback(ret_url, response_data)\r
 131         return '', 200\r
 132     return jsonify(response_data), 200\r
 133 \r
 134 @app.route("/otf/vth/oran/smo/v1/health", methods=['GET'])\r
 135 def getHealth():\r
 136     return 'UP'\r
 137 \r
 138 if __name__ == '__main__':\r
 139     logHandler = FileHandler('o1-vth.log', mode='a')\r
 140     logHandler.setLevel(logging.INFO)\r
 141     app.logger.setLevel(logging.INFO)\r
 142     app.logger.addHandler(logHandler)\r
 143     # context = ('opt/cert/otf.pem', 'opt/cert/privateKey.pem')\r
 144     # app.run(debug = False, host = '0.0.0.0', port = 5000, ssl_context = context)\r
 145     app.run(debug=False, host='0.0.0.0', port=5000)\r