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 200 response with empty body 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 del _rules[0:len(_rules)] 81 return "All rules were deleted"
82
83 84 @mock_server.route("/mock/responses", methods=['GET']) 85 -def get_responses():
86 rules_as_text = [] 87 for rule in _rules: 88 #make a copy so we don't modify original 89 rule = rule.copy() 90 91 #Convert regex to str 92 rule["url_filter"] = rule["url_filter"].pattern 93 94 #Add rule to list 95 rules_as_text.append(rule) 96 97 return jsonify(rules=rules_as_text)
98
99 100 @mock_server.route('/', defaults={'path': ''}) 101 @mock_server.route('/<path:path>') 102 -def catch_all(path):
103 """ 104 This method handles all requests that are not defined below 105 It will go though the list of rules and apply one by one until a match is found. 106 If there is no match, it will return a 200 response with empty body 107 """ 108 for rule in _rules: 109 regex = rule["url_filter"] 110 if regex.match(path): 111 response = make_response() 112 response .headers = rule["headers"] 113 response .data = rule["body"] 114 response .status_code = rule["status_code"] 115 return response 116 # return rule["body"], rule["status_code"] 117 118 # Default values returned when there wasn't a match in the rules 119 return "", 200
120 121 122 if __name__ == '__main__': 123 mock_server.run(host='127.0.0.1', port=10000) 124