smo-o1-vth/o1_vth.py

#   Copyright (c) 2019 AT&T Intellectual Property.                             #
#                                                                              #
#   Licensed under the Apache License, Version 2.0 (the "License");            #
#   you may not use this file except in compliance with the License.           #
#   You may obtain a copy of the License at                                    #
#                                                                              #
#       http://www.apache.org/licenses/LICENSE-2.0                             #
#                                                                              #
#   Unless required by applicable law or agreed to in writing, software        #
#   distributed under the License is distributed on an "AS IS" BASIS,          #
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   #
#   See the License for the specific language governing permissions and        #
#   limitations under the License.                                             #
################################################################################
# File name: smo-o1-vth.py                                                        #
# Description: Mainly used to get alarm list                                   #
# Date created: 04/14/2020                                                     #
# Python Version: 3.7                                                          #
# Author: Jackie Chen (jv246a)                                                 #
# Email: jv246a@att.com                                                        #
################################################################################

import datetime
from configparser import ConfigParser
import os
import logging
from logging import FileHandler
import requests
from flask import Flask, request, jsonify

# redirect http to https
app = Flask(__name__)

# Prevents print statement every time an endpoint is triggered.
logging.getLogger("werkzeug").setLevel(logging.WARNING)


def sendCallback(url, data):
    try:
        if type(data) is not dict:
            data = {"msg": data}
        app.logger.info("sending callback")
        requests.post(url, json= data)
    except Exception as e:
        app.logger.info(e)
    return

def unix_time_millis(dt):
    epoch = datetime.datetime.utcfromtimestamp(0)
    return (dt - epoch).total_seconds() * 1000.0


def _get_config(config_file_name):
    config = ConfigParser(os.environ)
    config.read(config_file_name)
    return config


def _build_url(config):
    return config['resource']['base_address'] + config['resource']['get_config_alarms_list']


def _send_request(url, config):
    # setup default values
    proxy_enabled = config.getboolean('resource', 'proxy_enabled')
    auth_enabled = config.getboolean('auth', 'auth_enabled')
    username = ''
    password = ''

    req_proxies = {
        'http': None,
        'https': None
    }
    # place proxy information
    if proxy_enabled:
        req_proxies['http'] = config['resource']['http_proxy']
        req_proxies['https'] = config['resource']['https_proxy']
    if auth_enabled:
        username = config['auth']['username']
        password = config['auth']['password']
    # get call for alarm list
    return requests.get(url,
                        auth=(username, password) if auth_enabled else None,
                        proxies=req_proxies if proxy_enabled else None)

def _parse_resposne(response):
    try:
        return response.json()
    except ValueError:
        return response.text

@app.route('/otf/vth/oran/smo/v1/alarm-list' , methods=['POST'])
def get_alarm_list():
    response_data = {
        'vthResponse': {
            'testDuration': '',
            'dateTimeUTC': str(datetime.datetime.now()),
            'abstractMessage': '',
            'resultData': {}
        }
    }
    ret_url = request.args.get('retURL')
    startTime = unix_time_millis(datetime.datetime.now())
    try:
        # setup phase
        config = _get_config('config.ini')
        alarm_list_url = _build_url(config)

        # build initial response
        app.logger.info('Sending GET for alarm list')
        res = _send_request(alarm_list_url, config)
        app.logger.info('Status code from GET: {}'.format(res.status_code))
        app.logger.info('Response received from GET alarm-list: {}'.format(res.content))
        response_data['vthResponse']['abstractMessage'] = 'Result from GET alarm list request'
        response_data['vthResponse']['resultData']['status_code'] = res.status_code
        response_data['vthResponse']['resultData']['result_output'] = _parse_resposne(res)
    except Exception as ex:
        endTime = unix_time_millis(datetime.datetime.now())
        totalTime = endTime - startTime
        response_data['vthResponse']['testDuration'] = totalTime
        response_data['vthResponse']['abstractMessage'] = 'error: ' + str(ex)
        app.logger.error('ERROR:{}'.format(str(ex)))
        return jsonify(response_data)

    #finish up building response
    endTime = unix_time_millis(datetime.datetime.now())
    totalTime = endTime - startTime
    response_data['vthResponse']['testDuration'] = totalTime
    if ret_url is not None:
        sendCallback(ret_url, response_data)
        return '', 200
    return jsonify(response_data), 200

@app.route("/otf/vth/oran/smo/v1/health", methods=['GET'])
def getHealth():
    return 'UP'

if __name__ == '__main__':
    logHandler = FileHandler('smo-o1-vth.log', mode='a')
    logHandler.setLevel(logging.INFO)
    app.logger.setLevel(logging.INFO)
    app.logger.addHandler(logHandler)
    # context = ('opt/cert/otf.pem', 'opt/cert/privateKey.pem')
    # app.run(debug = False, host = '0.0.0.0', port = 5000, ssl_context = context)
    app.run(debug=False, host='0.0.0.0', port=5000)