1 from flask import jsonify, request, Flask, make_response
2 import logging
3 import re
4 from multiprocessing import Process
5 import requests
6 from requests.exceptions import RequestException
7 import json
8 import socket
9 import time
10 from collections import OrderedDict
11
12 mock_server = Flask(__name__)
13 _rules = OrderedDict()
14 _requests = []
15
16
17
18
19
20
21 @mock_server.route("/mock/shutdown", methods=['GET'])
22 -def shutdown():
23 func = request.environ.get('werkzeug.server.shutdown')
24 if func is None:
25 raise RuntimeError('Not running with the Werkzeug Server')
26 func()
27 return 'Server shutting down...'
28
29
30 @mock_server.route("/mock/responses", methods=['POST'])
31 -def add_response():
32 """
33 This method adds new responses to the mock.
34 To add a response send a POST request with a payload like this:
35
36 {
37 "url_filter": ".*",
38 "headers": {
39 "Accept": "text/xml"
40 },
41 "body": "Sample body",
42 "status_code": 200
43 }
44
45 Server will validate each matching rule and apply the first match
46 If there is no match, it will return a 500 response
47 """
48 try:
49 payload = request.get_json(force=True)
50 assert isinstance(payload, dict)
51 except:
52 logging.error("Payload is not a valid JSON string")
53 logging.error(request.data)
54 return "Payload is not valid JSON string", 400
55
56
57 if "url_filter" in payload.keys():
58 key = payload["url_filter"]
59 try:
60 url_filter = re.compile(payload["url_filter"])
61 except Exception as e:
62 logging.error("url_filter is not a valid regular expression")
63 logging.error(payload["url_filter"])
64 return "url_filter is not a valid regular expression:\\n%s" % e.message, 400
65 else:
66 url_filter = re.compile('.*')
67 key = '.*'
68
69 if "headers" in payload.keys():
70 if type(payload["headers"]) is dict:
71 headers = payload["headers"]
72 else:
73 return "headers is not a dictionary:\\n%s" % payload["headers"], 400
74 else:
75 headers = {}
76
77 if "body" in payload.keys():
78 body = payload["body"]
79 else:
80 body = ""
81
82 if "status_code" in payload.keys():
83 status_code = payload["status_code"]
84 else:
85 status_code = 200
86
87
88 new_rule = {"url_filter": url_filter, "headers": headers, "body": body, "status_code": status_code}
89 _rules[key] = new_rule
90
91 return "OK"
92
93
94 @mock_server.route("/mock/responses", methods=['DELETE'])
95 -def clear_responses():
96 """
97 Delete existing responses
98 """
99 _rules.clear()
100 return "All rules were deleted"
101
102
103 @mock_server.route("/mock/responses", methods=['GET'])
104 -def get_responses():
105 """
106 Get all responses
107 """
108 rules_as_text = []
109 for rule in _rules.values():
110
111 rule = rule.copy()
112
113
114 rule["url_filter"] = rule["url_filter"].pattern
115
116
117 rules_as_text.append(rule)
118
119 return jsonify(rules=rules_as_text)
120
121
122 @mock_server.route("/mock/requests", methods=['DELETE'])
123 -def clear_requests():
124 """
125 Delete existing requests
126 """
127 del _requests[0:len(_requests)]
128 return "All requests were deleted"
129
130
131 @mock_server.route("/mock/requests", methods=['GET'])
132 -def get_requests():
133 """
134 Get all requests
135 """
136 return jsonify(requests=_requests)
137
138 @mock_server.route('/', defaults={'path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE'])
139 @mock_server.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE'])
140 -def catch_all(path):
141 """
142 This method will catch all requests for which there are no explicit routes.
143 Here is where we build responses based on the rules that have been configured
144 It will go though the list of rules and apply one by one until a match is found.
145 If there is no match, it will return a 500 response
146 """
147 _requests.append(request.url)
148 for rule in _rules.values():
149 regex = rule["url_filter"]
150 if regex.search("/" + path):
151 response = make_response()
152 response .headers = rule["headers"]
153 response .data = rule["body"]
154 response .status_code = rule["status_code"]
155 return response
156
157
158 return "Mock has not been configured", 500
159
162 """
163 Helper Class to interact with the ApiMockServer
164 Provides methods to add, remove mock objects as well as incoming requests
165 """
166
167 server = mock_server
168
170 Process.__init__(self)
171 self.port = port
172 self.host = "0.0.0.0"
173 self.localhost = self.host_ip()
174 self.base_url = "http://{host}:{port}".format(host=self.localhost, port=self.port)
175 self.responses_url = self.base_url + "/mock/responses"
176 self.requests_url = self.base_url + "/mock/requests"
177 self.shutdown_url = self.base_url + "/mock/shutdown"
178
180 """
181 start the mock server
182 """
183 self.server.run(host=self.host, port=self.port)
184
186 """
187 Shutdown the server and terminate process
188 :return:
189 """
190 self.wait_for_server_to_start()
191 requests.get(self.shutdown_url)
192 self.terminate()
193
194 - def add_response(self, url_filter=".*", status_code=200, headers=None, body="", data=None):
195 """
196 Add the response to the mock server
197 if the payload is provided, other parameters are ignored
198
199 :param url_filter: Regular expression to match the url path
200 :param status_code: Expected status code
201 :param headers: Headers as a dictionary
202 :param body: response body as plain text
203 :param payload: a python dictionary
204 :return: HTTP status code or error message if any error occurred
205 """
206
207 self.wait_for_server_to_start()
208 if data is None:
209 if not headers:
210 headers = {}
211 data = {"url_filter": url_filter,
212 "status_code": status_code,
213 "headers": headers,
214 "body": body}
215 try:
216 response = requests.post(self.responses_url, data=json.dumps(data))
217 return response
218 except RequestException as ex:
219 return ex.message
220
222 """
223 Returns all the responses stored on the MockServer
224 :return:
225 """
226 self.wait_for_server_to_start()
227 response = requests.get(self.responses_url)
228 if response:
229 return response
230 return None
231
233 """
234 Delete all the responses stored on the MockServer
235 :return:
236 """
237 self.wait_for_server_to_start()
238 return requests.delete(self.responses_url)
239
241 """
242 Returns all the requests stored on the MockServer
243 :return:
244 """
245 self.wait_for_server_to_start()
246 response = requests.get(self.requests_url)
247 if response:
248 return response
249 return None
250
252 """
253 Delete all the requests stored on the MockServer
254 :return:
255 """
256 self.wait_for_server_to_start()
257 return requests.delete(self.requests_url)
258
260 """
261 Dirty hack to return the localhost IP,
262 works only when you have an internet connection
263 :return:
264 """
265 return [(s.connect(('8.8.8.8', 80)), s.getsockname()[0], s.close()) for s in [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]][0][1]
266
268 """
269 check if the server is running and port is open
270 """
271 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
272 result = sock.connect_ex((self.host_ip(),self.port))
273 return result == 0
274
276 """
277 Latency to avoid the access when the server is not started yet
278 """
279 while not self._is_server_running():
280 time.sleep(0.1)
281
284
285 if __name__ == '__main__':
286
287
288 api_server = ApiMockServer(port=10000)
289 api_server.start()
290