near-rt-ric-simulator/test/EXT_SRV/src/server.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 copy
  19 import datetime
  20 import json
  21 import logging
  22 import collections
  23 import time
  24
  25 from flask import Flask, request, Response
  26 from jsonschema import validate
  27 from var_declaration import a1_policy_instances, forced_settings
  28
  29
  30 #Constsants
  31 APPL_JSON='application/json'
  32 APPL_PROB_JSON='application/problem+json'
  33
  34 #Python implementation of EXT_SRV_api.yaml - Open API -
  35
  36 # API Function: Get all a1 policy ids
  37 def get_all_a1_policies():
  38
  39   if ((r := check_modified_response()) is not None):
  40     return r
  41
  42   res = list(a1_policy_instances)
  43   return (res, 200)
  44
  45 # API Function: Get A1 policy
  46 def get_a1_policy(a1policyId):
  47
  48   if ((r := check_modified_response()) is not None):
  49     return r
  50
  51   a1_policy_id=str(a1policyId)
  52
  53   policykeys=a1_policy_instances.keys()
  54   if (a1_policy_id not in policykeys):
  55     pjson=create_problem_json(None, "The A1 policy requested does not exist.", 404, None, a1_policy_id)
  56     return Response(json.dumps(pjson), 404, mimetype=APPL_PROB_JSON)
  57
  58   return Response(json.dumps(a1_policy_instances[a1_policy_id]), 200, mimetype=APPL_JSON)
  59
  60 # API Function: Create or update a a1policy
  61 def put_a1_policy(a1policyId):
  62
  63   if ((r := check_modified_response()) is not None):
  64     return r
  65
  66   a1_policy_id=str(a1policyId)
  67
  68   try:
  69     data = request.data
  70     data = json.loads(data)
  71   except Exception:
  72     pjson=create_problem_json(None, "The a1policy is corrupt or missing.", 400, None, a1_policy_id)
  73     return Response(json.dumps(pjson), 400, mimetype=APPL_PROB_JSON)
  74
  75   policykeys=a1_policy_instances.keys()
  76
  77   retcode=201
  78   if a1_policy_id in policykeys:
  79     retcode=200
  80
  81   a1_policy_instances[a1_policy_id]=data
  82
  83   if (retcode == 200):
  84     return Response(json.dumps(data), 200, mimetype=APPL_JSON)
  85   else:
  86     headers={}
  87     headers['Location']='/a1policy/' + a1_policy_id
  88     return Response(json.dumps(data), 201, headers=headers, mimetype=APPL_JSON)
  89
  90 # API Function: Delete a a1policy
  91 def delete_a1_policy(a1policyId):
  92
  93   if ((r := check_modified_response()) is not None):
  94     return r
  95
  96   a1_policy_id=str(a1policyId)
  97
  98   policykeys=a1_policy_instances.keys()
  99   if (a1_policy_id not in policykeys):
 100     pjson=create_problem_json(None, "The a1policy does not exist.", 404, None, a1_policy_id)
 101     return Response(json.dumps(pjson), 404, mimetype=APPL_PROB_JSON)
 102
 103   del a1_policy_instances[a1_policy_id]
 104   return Response('', 204, mimetype=APPL_JSON)
 105
 106
 107 # Helper: Create a response object if forced http response code is set
 108 def get_forced_response():
 109   if (forced_settings['code'] is not None):
 110     pjson=create_error_response(forced_settings['code'])
 111     return Response(json.dumps(pjson), pjson['status'], mimetype=APPL_PROB_JSON)
 112   return None
 113
 114
 115 # Helper: Delay if delayed response code is set
 116 def do_delay():
 117   if (forced_settings['delay'] is not None):
 118     try:
 119       val=int(forced_settings['delay'])
 120       time.sleep(val)
 121     except Exception:
 122       return
 123
 124
 125 # Helper: Check if response shall be delayed or a forced response shall be sent
 126 def check_modified_response():
 127   do_delay()
 128   return get_forced_response()
 129
 130
 131 # Helper: Create a problem json object
 132 def create_problem_json(type_of, title, status, detail, instance):
 133   error = {}
 134   if type_of is not None:
 135     error["type"] = type_of
 136   if title is not None:
 137     error["title"] = title
 138   if status is not None:
 139     error["status"] = status
 140   if detail is not None:
 141     error["detail"] = detail
 142   if instance is not None:
 143     error["instance"] = instance
 144   return error
 145
 146
 147 # Helper: Create a problem json based on a generic http response code
 148 def create_error_response(code):
 149     if code == 400:
 150       return(create_problem_json(None, "Bad request", 400, "Object in payload not properly formulated or not related to the method", None))
 151     elif code == 404:
 152       return(create_problem_json(None, "Not found", 404, "No resource found at the URI", None))
 153     elif code == 405:
 154       return(create_problem_json(None, "Method not allowed", 405, "Method not allowed for the URI", None))
 155     elif code == 409:
 156       return(create_problem_json(None, "Conflict", 409, "Request could not be processed in the current state of the resource", None))
 157     elif code == 429:
 158       return(create_problem_json(None, "Too many requests", 429, "Too many requests have been sent in a given amount of time", None))
 159     elif code == 507:
 160       return(create_problem_json(None, "Insufficient storage", 507, "The method could not be performed on the resource because the provider is unable to store the representation needed to successfully complete the request", None))
 161     elif code == 503:
 162       return(create_problem_json(None, "Service unavailable", 503, "The provider is currently unable to handle the request due to a temporary overload", None))
 163     else:
 164       return(create_problem_json(None, "Unknown", code, "Not implemented response code", None))