1 # Copyright (c) 2019 AT&T Intellectual Property. #
\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
7 # http://www.apache.org/licenses/LICENSE-2.0 #
\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: smo-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
24 from configparser import ConfigParser
\r
27 from logging import FileHandler
\r
29 from flask import Flask, request, jsonify
\r
31 # redirect http to https
\r
32 app = Flask(__name__)
\r
34 # Prevents print statement every time an endpoint is triggered.
\r
35 logging.getLogger("werkzeug").setLevel(logging.WARNING)
\r
38 def sendCallback(url, data):
\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
48 def unix_time_millis(dt):
\r
49 epoch = datetime.datetime.utcfromtimestamp(0)
\r
50 return (dt - epoch).total_seconds() * 1000.0
\r
53 def _get_config(config_file_name):
\r
54 config = ConfigParser(os.environ)
\r
55 config.read(config_file_name)
\r
59 def _build_url(config):
\r
60 return config['resource']['base_address'] + config['resource']['get_config_alarms_list']
\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
74 # place proxy information
\r
76 req_proxies['http'] = config['resource']['http_proxy']
\r
77 req_proxies['https'] = config['resource']['https_proxy']
\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
86 def _parse_resposne(response):
\r
88 return response.json()
\r
90 return response.text
\r
92 @app.route('/otf/vth/oran/smo/v1/alarm-list' , methods=['POST'])
\r
93 def get_alarm_list():
\r
97 'dateTimeUTC': str(datetime.datetime.now()),
\r
98 'abstractMessage': '',
\r
102 ret_url = request.args.get('retURL')
\r
103 startTime = unix_time_millis(datetime.datetime.now())
\r
106 config = _get_config('config.ini')
\r
107 alarm_list_url = _build_url(config)
\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
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
132 return jsonify(response_data), 200
\r
134 @app.route("/otf/vth/oran/smo/v1/health", methods=['GET'])
\r
138 if __name__ == '__main__':
\r
139 logHandler = FileHandler('smo-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