Package tlib :: Package base :: Module ApiMockServer |
[hide private]
[frames] | no frames]

# 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