added oran ric test head

[it/otf.git] / oran-ric-test-head / ric-test-head.py

diff --git a/oran-ric-test-head/ric-test-head.py b/oran-ric-test-head/ric-test-head.py
new file mode 100644 (file)
index 0000000..f80d3d3
--- /dev/null
+++ b/oran-ric-test-head/ric-test-head.py
@@ -0,0 +1,117 @@
+#   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
+\r
+import datetime\r
+import json\r
+import logging\r
+from logging import FileHandler\r
+\r
+import requests\r
+from flask import Flask, request, jsonify\r
+\r
+#redirect http to https\r
+app = Flask(__name__)\r
+\r
+\r
+# Prevents print statement every time an endpoint is triggered.\r
+logging.getLogger("werkzeug").setLevel(logging.WARNING)\r
+\r
+def unix_time_millis(dt):\r
+       epoch = datetime.datetime.utcfromtimestamp(0)\r
+       return (dt - epoch).total_seconds() * 1000.0\r
+\r
+\r
+@app.route("/otf/vth/oran/v1/health", methods=['GET'])\r
+def getHealth():\r
+       return "UP"\r
+\r
+@app.route("/otf/vth/oran/ric/v1", methods =['POST'])\r
+def executeRicRequest():\r
+\r
+       responseData = {\r
+               'vthResponse': {\r
+                       'testDuration': '',\r
+                       'dateTimeUTC': datetime.datetime.now(),\r
+                       'abstractMessage': '',\r
+                       'resultData': {}\r
+               }\r
+       }\r
+\r
+       startTime = unix_time_millis(datetime.datetime.now())\r
+\r
+       try:\r
+               if not request.is_json:\r
+                       raise ValueError("request must be json")\r
+\r
+               requestData = request.get_json()\r
+\r
+               app.logger.info("Ric requestData:"+str(requestData))\r
+\r
+               action = requestData['action'].lower()\r
+               possibleActions = ['alive','ready','list', 'deploy','delete']\r
+               responseData['vthResponse']['abstractMessage'] = 'Result from {}'.format(action)\r
+\r
+               if action not in possibleActions:\r
+                       raise KeyError("invalid action")\r
+               if (action == 'deploy' or action == 'delete') and 'name' not in requestData:\r
+                       raise KeyError("must include name")\r
+\r
+               with open('config.json') as configFile:\r
+                       config = json.load(configFile)\r
+\r
+               baseAddress=  config['base_address']\r
+\r
+               if action == 'alive' or action == 'ready':\r
+                       res = requests.get(baseAddress+config['actions_path'][action])\r
+                       responseData['vthResponse']['resultData']['statusCode'] = res.status_code\r
+                       responseData['vthResponse']['resultData']['resultOutput'] = res.text\r
+               elif action == 'list':\r
+                       res = requests.get(baseAddress+config['actions_path'][action])\r
+                       responseData['vthResponse']['resultData']['statusCode'] = res.status_code\r
+                       responseData['vthResponse']['resultData']['resultOutput'] = res.json()\r
+               elif action == 'deploy':\r
+                       payload = {'name': requestData['name']}\r
+                       res = requests.post(baseAddress+config['actions_path'][action], data=payload)\r
+                       responseData['vthResponse']['resultData']['statusCode'] = res.status_code\r
+                       responseData['vthResponse']['resultData']['resultOutput'] = res.json()\r
+               elif action == 'delete':\r
+                       path= baseAddress+config['actions_path'][action]+"{}".format(requestData['name'])\r
+                       res = requests.delete(path)\r
+                       responseData['vthResponse']['resultData']['resultOutput'] = res.text\r
+                       responseData['vthResponse']['resultData']['statusCode'] = res.status_code\r
+\r
+       except Exception as ex:\r
+               endTime = unix_time_millis(datetime.datetime.now())\r
+               totalTime = endTime - startTime\r
+               responseData['vthResponse']['testDuration'] = totalTime\r
+               responseData['vthResponse']['abstractMessage'] = str(ex)\r
+               return jsonify(responseData)\r
+\r
+       endTime = unix_time_millis(datetime.datetime.now())\r
+       totalTime= endTime-startTime\r
+\r
+       responseData['vthResponse']['testDuration'] = totalTime\r
+\r
+       return jsonify(responseData),200\r
+\r
+if __name__ == '__main__':\r
+       # logHandler = FileHandler('otf/logs/pingVTH.log', mode='a')\r
+       logHandler = FileHandler('ricVTH.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