in tsqa/endpoint.py [0:0]
def __init__(self, port=0):
threading.Thread.__init__(self)
# dict to store request data in
self._tracked_requests = {}
# error in startup
self.error = None
self.daemon = True
self.port = port
self.ready = threading.Event()
# dict of pathname (no starting /) -> function
self._handlers = {}
self.app = flask.Flask(__name__)
self.app.debug = True
@self.app.before_request
def save_request():
'''
If the tracking header is set, save the request
'''
if flask.request.headers.get(self.TRACKING_HEADER):
self._tracked_requests[flask.request.headers[self.TRACKING_HEADER]] = {'request': flask.request}
@self.app.after_request
def save_response(response):
'''
If the tracking header is set, save the response
'''
if flask.request.headers.get(self.TRACKING_HEADER):
self._tracked_requests[flask.request.headers[self.TRACKING_HEADER]]['response'] = response
return response
@self.app.route('/', defaults={'path': ''})
@self.app.route('/<path:path>')
def catch_all(path=''):
# get path key
if path in self._handlers:
return self._handlers[path](flask.request)
# return a 404 since we didn't find it
return ('', 404)
# A little magic to make flask accept *all* methods on the catch_all path
for rule in self.app.url_map.iter_rules():
rule.methods = None
rule.refresh()