near-rt-ric-simulator/test/EXT_SRV/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 a1_policy_instances, forced_settings, app
  25 from maincommon import check_apipath
  26
  27 #Constants
  28 TEXT_PLAIN='text/plain'
  29
  30 check_apipath()
  31
  32 # app is created in var_declarations
  33
  34 import payload_logging   # app var need to be initialized
  35
  36 #Check alive function
  37 @app.route('/', methods=['GET'])
  38 def test():
  39   return Response("OK", 200, mimetype=TEXT_PLAIN)
  40
  41 #Delete all created instances and status
  42 @app.route('/serveradmin/deleteinstances', methods=['POST'])
  43 def delete_instances():
  44     a1_policy_instances.clear()
  45     return Response("All a1 policy instances deleted", 200, mimetype=TEXT_PLAIN)
  46
  47 #Set|Reset force response to be returned from external server
  48 #/a1policy/forceresponse?code=<responsecode>
  49 @app.route('/serveradmin/forceresponse', methods=['POST'])
  50 def forceresponse():
  51
  52   query_param=request.args.get('code')
  53   forced_settings['code']=query_param
  54
  55   if (query_param is None):
  56     return Response("Force response code has been resetted for all external server responses", 200, mimetype=TEXT_PLAIN)
  57   else:
  58     return Response("Force response code: " + str(forced_settings['code']) + " set for all external server response until it is resetted", 200, mimetype=TEXT_PLAIN)
  59
  60 #Set|Reset force delay response, in seconds, for all external server responses
  61 #/a1policy/forcedelay?delay=<seconds>
  62 @app.route('/serveradmin/forcedelay', methods=['POST'])
  63 def forcedelay():
  64
  65   query_param=request.args.get('delay')
  66   forced_settings['delay']=query_param
  67
  68   if (query_param is None):
  69     return Response("Force delay has been resetted for all external server responses ", 200, mimetype=TEXT_PLAIN)
  70   else:
  71     return Response("Force delay: " + str(forced_settings['delay']) + " sec set for all external server responses until it is resetted ", 200, mimetype=TEXT_PLAIN)
  72
  73 port_number = 2222
  74 if len(sys.argv) >= 2:
  75   if isinstance(sys.argv[1], int):
  76     port_number = sys.argv[1]
  77
  78 #Import base RestFUL API functions from Open API
  79 app.add_api('EXT_SRV_api.yaml')
  80
  81 if __name__ == '__main__':
  82   app.run(port=port_number, host="127.0.0.1", threaded=False)