near-rt-ric-simulator/test/KAFKA_DISPATCHER/src/main.py

   1 #  ============LICENSE_START===============================================
   2 #  Copyright (C) 2022 Nordix Foundation. All rights reserved.
   3 #  ========================================================================
   4 #  Licensed under the Apache License, Version 2.0 (the "License");
   5 #  you may not use this file except in compliance with the License.
   6 #  You may obtain a copy of the License at
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #  Unless required by applicable law or agreed to in writing, software
  11 #  distributed under the License is distributed on an "AS IS" BASIS,
  12 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #  See the License for the specific language governing permissions and
  14 #  limitations under the License.
  15 #  ============LICENSE_END=================================================
  16 #
  17
  18 import json
  19 import sys
  20 import requests
  21
  22
  23 from flask import request, Response, Flask, json
  24 from var_declaration import forced_settings, app
  25 from maincommon import check_timeout, check_apipath
  26
  27 #Constants
  28 TEXT_PLAIN='text/plain'
  29
  30 check_apipath()
  31 check_timeout()
  32
  33 # app is created in var_declarations
  34
  35 import payload_logging   # app var need to be initialized
  36
  37 #Check alive function
  38 @app.route('/', methods=['GET'])
  39 def test():
  40   return Response("OK", 200, mimetype=TEXT_PLAIN)
  41
  42 #Set|Reset force response to be returned from dispatcher
  43 #/dispatcheradmin/forceresponse?code=<responsecode>
  44 @app.route('/dispatcheradmin/forceresponse', methods=['POST'])
  45 def forceresponse():
  46
  47   query_param=request.args.get('code')
  48   forced_settings['code']=query_param
  49
  50   if (query_param is None):
  51     return Response("Force response code has been resetted for dispatcher responses", 200, mimetype=TEXT_PLAIN)
  52   else:
  53     return Response("Force response code: " + str(forced_settings['code']) + " set for all dispatcher response until it is resetted", 200, mimetype=TEXT_PLAIN)
  54
  55 #Set|Reset force delay response, in seconds, for all external server responses
  56 #/a1policy/forcedelay?delay=<seconds>
  57 @app.route('/dispatcheradmin/forcedelay', methods=['POST'])
  58 def forcedelay():
  59
  60   query_param=request.args.get('delay')
  61   forced_settings['delay']=query_param
  62
  63   if (query_param is None):
  64     return Response("Force delay has been resetted for all dispatcher responses ", 200, mimetype=TEXT_PLAIN)
  65   else:
  66     return Response("Force delay: " + str(forced_settings['delay']) + " sec set for all dispatcher responses until it is resetted ", 200, mimetype=TEXT_PLAIN)
  67
  68 port_number = 7777
  69 if len(sys.argv) >= 2:
  70   if isinstance(sys.argv[1], int):
  71     port_number = sys.argv[1]
  72
  73 #Import base RestFUL API functions from Open API
  74 app.add_api('KAFKA_DISPATCHER_api.yaml')
  75
  76 if __name__ == '__main__':
  77   app.run(port=port_number, host="127.0.0.1", threaded=False)