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
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
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
89 rule = rule.copy()
90
91
92 rule["url_filter"] = rule["url_filter"].pattern
93
94
95 rules_as_text.append(rule)
96
97 return jsonify(rules=rules_as_text)
98
99
100 @mock_server.route('/', defaults={'path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE'])
101 @mock_server.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE'])
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
117
118
119 return "", 200
120
121
122 if __name__ == '__main__':
123 mock_server.run(host='127.0.0.1', port=10000)
124