oran-ric-test-head/ric-test-head.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 \r
  16 import datetime\r
  17 import json\r
  18 import logging\r
  19 from logging import FileHandler\r
  20 \r
  21 import requests\r
  22 from flask import Flask, request, jsonify\r
  23 \r
  24 #redirect http to https\r
  25 app = Flask(__name__)\r
  26 \r
  27 \r
  28 # Prevents print statement every time an endpoint is triggered.\r
  29 logging.getLogger("werkzeug").setLevel(logging.WARNING)\r
  30 \r
  31 def unix_time_millis(dt):\r
  32         epoch = datetime.datetime.utcfromtimestamp(0)\r
  33         return (dt - epoch).total_seconds() * 1000.0\r
  34 \r
  35 \r
  36 @app.route("/otf/vth/oran/v1/health", methods=['GET'])\r
  37 def getHealth():\r
  38         return "UP"\r
  39 \r
  40 @app.route("/otf/vth/oran/ric/v1", methods =['POST'])\r
  41 def executeRicRequest():\r
  42 \r
  43         responseData = {\r
  44                 'vthResponse': {\r
  45                         'testDuration': '',\r
  46                         'dateTimeUTC': datetime.datetime.now(),\r
  47                         'abstractMessage': '',\r
  48                         'resultData': {}\r
  49                 }\r
  50         }\r
  51 \r
  52         startTime = unix_time_millis(datetime.datetime.now())\r
  53 \r
  54         try:\r
  55                 if not request.is_json:\r
  56                         raise ValueError("request must be json")\r
  57 \r
  58                 requestData = request.get_json()\r
  59 \r
  60                 app.logger.info("Ric requestData:"+str(requestData))\r
  61 \r
  62                 action = requestData['action'].lower()\r
  63                 possibleActions = ['alive','ready','list', 'deploy','delete']\r
  64                 responseData['vthResponse']['abstractMessage'] = 'Result from {}'.format(action)\r
  65 \r
  66                 if action not in possibleActions:\r
  67                         raise KeyError("invalid action")\r
  68                 if (action == 'deploy' or action == 'delete') and 'name' not in requestData:\r
  69                         raise KeyError("must include name")\r
  70 \r
  71                 with open('config.json') as configFile:\r
  72                         config = json.load(configFile)\r
  73 \r
  74                 baseAddress=  config['base_address']\r
  75 \r
  76                 if action == 'alive' or action == 'ready':\r
  77                         res = requests.get(baseAddress+config['actions_path'][action])\r
  78                         responseData['vthResponse']['resultData']['statusCode'] = res.status_code\r
  79                         responseData['vthResponse']['resultData']['resultOutput'] = res.text\r
  80                 elif action == 'list':\r
  81                         res = requests.get(baseAddress+config['actions_path'][action])\r
  82                         responseData['vthResponse']['resultData']['statusCode'] = res.status_code\r
  83                         responseData['vthResponse']['resultData']['resultOutput'] = res.json()\r
  84                 elif action == 'deploy':\r
  85                         payload = {'name': requestData['name']}\r
  86                         res = requests.post(baseAddress+config['actions_path'][action], data=payload)\r
  87                         responseData['vthResponse']['resultData']['statusCode'] = res.status_code\r
  88                         responseData['vthResponse']['resultData']['resultOutput'] = res.json()\r
  89                 elif action == 'delete':\r
  90                         path= baseAddress+config['actions_path'][action]+"{}".format(requestData['name'])\r
  91                         res = requests.delete(path)\r
  92                         responseData['vthResponse']['resultData']['resultOutput'] = res.text\r
  93                         responseData['vthResponse']['resultData']['statusCode'] = res.status_code\r
  94 \r
  95         except Exception as ex:\r
  96                 endTime = unix_time_millis(datetime.datetime.now())\r
  97                 totalTime = endTime - startTime\r
  98                 responseData['vthResponse']['testDuration'] = totalTime\r
  99                 responseData['vthResponse']['abstractMessage'] = str(ex)\r
 100                 return jsonify(responseData)\r
 101 \r
 102         endTime = unix_time_millis(datetime.datetime.now())\r
 103         totalTime= endTime-startTime\r
 104 \r
 105         responseData['vthResponse']['testDuration'] = totalTime\r
 106 \r
 107         return jsonify(responseData),200\r
 108 \r
 109 if __name__ == '__main__':\r
 110         # logHandler = FileHandler('otf/logs/pingVTH.log', mode='a')\r
 111         logHandler = FileHandler('ricVTH.log', mode='a')\r
 112         logHandler.setLevel(logging.INFO)\r
 113         app.logger.setLevel(logging.INFO)\r
 114         app.logger.addHandler(logHandler)\r
 115         # context = ('opt/cert/otf.pem', 'opt/cert/privateKey.pem')\r
 116         # app.run(debug = False, host = '0.0.0.0', port = 5000, ssl_context = context)\r
 117         app.run(debug = False, host = '0.0.0.0', port = 5000)\r