--- /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