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  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 # @mock_server.route("/test", methods=['GET']) 18 # def test(): 19 # return "OK" 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 #Parse data from request 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 #Save parsed data 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 #make a copy so we don't modify original 111 rule = rule.copy() 112 113 #Convert regex to str 114 rule["url_filter"] = rule["url_filter"].pattern 115 116 #Add rule to list 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 # Default values returned when there wasn't a match in the rules 158 return "Mock has not been configured", 500
159
160 161 -class ApiMockServer(Process):
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
169 - def __init__(self, port):
170 Process.__init__(self) 171 self.port = port 172 self.host = "0.0.0.0" # host IP to be server by the Flask server 173 self.localhost = self.host_ip() # IP to communicate with server 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
179 - def run(self):
180 """ 181 start the mock server 182 """ 183 self.server.run(host=self.host, port=self.port)
184
185 - def stop(self):
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 # if payload is provided, all other other parameters woll be ignored 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
221 - def get_responses(self):
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
232 - def clear_responses(self):
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
240 - def get_requests(self):
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
251 - def clear_requests(self):
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
259 - def host_ip(self):
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
267 - def _is_server_running(self):
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
275 - def wait_for_server_to_start(self):
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
282 - def get_base_url(self):
283 return self.base_url
284 285 if __name__ == '__main__': 286 #to us as standalone application 287 #run 'python ApiMockServer.py' 288 api_server = ApiMockServer(port=10000) 289 api_server.start() 290