Source Code for Module tlib.base.ApiMockServer

  1  from flask import jsonify, request, Flask, make_response 
  2  import logging 
  3  import re 
  4   
  5  mock_server = Flask(__name__) 
  6  _rules = [] 
  7   
  8  @mock_server.route("/mock/shutdown", methods=['GET']) 
  9 -def shutdown(): 
 10      func = request.environ.get('werkzeug.server.shutdown') 
 11      if func is None: 
 12          raise RuntimeError('Not running with the Werkzeug Server') 
 13      func() 
 14      return 'Server shutting down...' 
 15   
 16   
 17  @mock_server.route("/mock/responses", methods=['POST']) 
 18 -def add_response(): 
 19      """ 
 20      This method adds new responses to the mock. 
 21      To add a response send a POST request with a payload like this: 
 22   
 23      { 
 24          "url_filter": ".*", 
 25          "headers": { 
 26              "Accept": "text/xml" 
 27          }, 
 28          "body": "Sample body", 
 29          "status_code": 200 
 30      } 
 31   
 32      Server will validate each matching rule and apply the first match 
 33      If there is no match, it will return a 500 response 
 34      """ 
 35      try: 
 36          payload = request.get_json(force=True) 
 37      except: 
 38          logging.error("Payload is not a valid JSON string") 
 39          logging.error(request.data) 
 40          return "Payload is not valid JSON string", 400 
 41   
 42      #Parse data from request 
 43      if "url_filter" in payload.keys(): 
 44          try: 
 45              url_filter = re.compile(payload["url_filter"]) 
 46          except Exception as e: 
 47              logging.error("url_filter is not a valid regular expression") 
 48              logging.error(payload["url_filter"]) 
 49              return "url_filter is not a valid regular expression:\\n%s" % e.message, 400 
 50      else: 
 51          url_filter = re.compile('.*') 
 52   
 53      if "headers" in payload.keys(): 
 54          if type(payload["headers"]) is dict: 
 55              headers = payload["headers"] 
 56          else: 
 57              return "headers is not a dictionary:\\n%s" % payload["headers"], 400 
 58      else: 
 59          headers = {} 
 60   
 61      if "body" in payload.keys(): 
 62          body = payload["body"] 
 63      else: 
 64          body = "" 
 65   
 66      if "status_code" in payload.keys(): 
 67          status_code = payload["status_code"] 
 68      else: 
 69          status_code = 200 
 70   
 71      #Save parsed data 
 72      rule = {"url_filter": url_filter, "headers": headers, "body": body, "status_code": status_code} 
 73      _rules.append(rule) 
 74   
 75      return "OK" 
 76   
 77   
 78  @mock_server.route("/mock/responses", methods=['DELETE']) 
 79 -def clear_responses(): 
 80      """ 
 81      Delete existing responses 
 82      """ 
 83      del _rules[0:len(_rules)] 
 84      return "All rules were deleted" 
 85   
 86   
 87  @mock_server.route("/mock/responses", methods=['GET']) 
 88 -def get_responses(): 
 89      """ 
 90      Get all responses 
 91      """ 
 92      rules_as_text = [] 
 93      for rule in _rules: 
 94          #make a copy so we don't modify original 
 95          rule = rule.copy() 
 96   
 97          #Convert regex to str 
 98          rule["url_filter"] = rule["url_filter"].pattern 
 99   
100          #Add rule to list 
101          rules_as_text.append(rule) 
102   
103      return jsonify(rules=rules_as_text) 
104   
105   
106  @mock_server.route('/', defaults={'path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE']) 
107  @mock_server.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE']) 
108 -def catch_all(path): 
109      """ 
110      This method will catch all requests for which there are no explicit routes. 
111      Here is where we build responses based on the rules that have been configured 
112      It will go though the list of rules and apply one by one until a match is found. 
113      If there is no match, it will return a 500 response 
114      """ 
115      for rule in _rules: 
116          regex = rule["url_filter"] 
117          if regex.match(path): 
118              response = make_response() 
119              response .headers = rule["headers"] 
120              response .data = rule["body"] 
121              response .status_code = rule["status_code"] 
122              return response 
123              # return rule["body"], rule["status_code"] 
124   
125      # Default values returned when there wasn't a match in the rules 
126      return "Mock has not been configured", 500 
127   
128   
129  if __name__ == '__main__': 
130      mock_server.run(host='127.0.0.1', port=10000) 
131