+++ /dev/null
-# Copyright (c) 2019 AT&T Intellectual Property. #\r
-# #\r
-# Licensed under the Apache License, Version 2.0 (the "License"); #\r
-# you may not use this file except in compliance with the License. #\r
-# You may obtain a copy of the License at #\r
-# #\r
-# http://www.apache.org/licenses/LICENSE-2.0 #\r
-# #\r
-# Unless required by applicable law or agreed to in writing, software #\r
-# distributed under the License is distributed on an "AS IS" BASIS, #\r
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #\r
-# See the License for the specific language governing permissions and #\r
-# limitations under the License. #\r
-################################################################################\r
-# File name: o1-vth.py #\r
-# Description: Mainly used to get alarm list #\r
-# Date created: 04/14/2020 #\r
-# Python Version: 3.7 #\r
-# Author: Jackie Chen (jv246a) #\r
-# Email: jv246a@att.com #\r
-################################################################################\r
-\r
-import datetime\r
-from configparser import ConfigParser\r
-import os\r
-import logging\r
-from logging import FileHandler\r
-import requests\r
-from flask import Flask, request, jsonify\r
-\r
-# redirect http to https\r
-app = Flask(__name__)\r
-\r
-# Prevents print statement every time an endpoint is triggered.\r
-logging.getLogger("werkzeug").setLevel(logging.WARNING)\r
-\r
-\r
-def sendCallback(url, data):\r
- try:\r
- if type(data) is not dict:\r
- data = {"msg": data}\r
- app.logger.info("sending callback")\r
- requests.post(url, json= data)\r
- except Exception as e:\r
- app.logger.info(e)\r
- return\r
-\r
-def unix_time_millis(dt):\r
- epoch = datetime.datetime.utcfromtimestamp(0)\r
- return (dt - epoch).total_seconds() * 1000.0\r
-\r
-\r
-def _get_config(config_file_name):\r
- config = ConfigParser(os.environ)\r
- config.read(config_file_name)\r
- return config\r
-\r
-\r
-def _build_url(config):\r
- return config['resource']['base_address'] + config['resource']['get_config_alarms_list']\r
-\r
-\r
-def _send_request(url, config):\r
- # setup default values\r
- proxy_enabled = config.getboolean('resource', 'proxy_enabled')\r
- auth_enabled = config.getboolean('auth', 'auth_enabled')\r
- username = ''\r
- password = ''\r
-\r
- req_proxies = {\r
- 'http': None,\r
- 'https': None\r
- }\r
- # place proxy information\r
- if proxy_enabled:\r
- req_proxies['http'] = config['resource']['http_proxy']\r
- req_proxies['https'] = config['resource']['https_proxy']\r
- if auth_enabled:\r
- username = config['auth']['username']\r
- password = config['auth']['password']\r
- # get call for alarm list\r
- return requests.get(url,\r
- auth=(username, password) if auth_enabled else None,\r
- proxies=req_proxies if proxy_enabled else None)\r
-\r
-def _parse_resposne(response):\r
- try:\r
- return response.json()\r
- except ValueError:\r
- return response.text\r
-\r
-@app.route('/otf/vth/oran/smo/v1/alarm-list' , methods=['POST'])\r
-def get_alarm_list():\r
- response_data = {\r
- 'vthResponse': {\r
- 'testDuration': '',\r
- 'dateTimeUTC': str(datetime.datetime.now()),\r
- 'abstractMessage': '',\r
- 'resultData': {}\r
- }\r
- }\r
- ret_url = request.args.get('retURL')\r
- startTime = unix_time_millis(datetime.datetime.now())\r
- try:\r
- # setup phase\r
- config = _get_config('config.ini')\r
- alarm_list_url = _build_url(config)\r
-\r
- # build initial response\r
- app.logger.info('Sending GET for alarm list')\r
- res = _send_request(alarm_list_url, config)\r
- app.logger.info('Status code from GET: {}'.format(res.status_code))\r
- app.logger.info('Response received from GET alarm-list: {}'.format(res.content))\r
- response_data['vthResponse']['abstractMessage'] = 'Result from GET alarm list request'\r
- response_data['vthResponse']['resultData']['status_code'] = res.status_code\r
- response_data['vthResponse']['resultData']['result_output'] = _parse_resposne(res)\r
- except Exception as ex:\r
- endTime = unix_time_millis(datetime.datetime.now())\r
- totalTime = endTime - startTime\r
- response_data['vthResponse']['testDuration'] = totalTime\r
- response_data['vthResponse']['abstractMessage'] = 'error: ' + str(ex)\r
- app.logger.error('ERROR:{}'.format(str(ex)))\r
- return jsonify(response_data)\r
-\r
- #finish up building response\r
- endTime = unix_time_millis(datetime.datetime.now())\r
- totalTime = endTime - startTime\r
- response_data['vthResponse']['testDuration'] = totalTime\r
- if ret_url is not None:\r
- sendCallback(ret_url, response_data)\r
- return '', 200\r
- return jsonify(response_data), 200\r
-\r
-@app.route("/otf/vth/oran/smo/v1/health", methods=['GET'])\r
-def getHealth():\r
- return 'UP'\r
-\r
-if __name__ == '__main__':\r
- logHandler = FileHandler('o1-vth.log', mode='a')\r
- logHandler.setLevel(logging.INFO)\r
- app.logger.setLevel(logging.INFO)\r
- app.logger.addHandler(logHandler)\r
- # context = ('opt/cert/otf.pem', 'opt/cert/privateKey.pem')\r
- # app.run(debug = False, host = '0.0.0.0', port = 5000, ssl_context = context)\r
- app.run(debug=False, host='0.0.0.0', port=5000)\r
Code Review / it / otf.git / blobdiff