以官方文档中的第7章为起点.
# 7.2 The Testing Skeleton.
import os
import flaskr
import unittest
import tempfile
class FlaskTestCase(unittest.TestCase):
def setUp(self):
self.db_fd, flaskr.app.config['DATABASE'] = tempfile.mkstemp()
flaskr.app.config['TESTING'] = True
self.app = flaskr.app.test_client()
with flaskr.app.app_context():
flaskr.init_db()
def tearDown(self):
os.close(self.db_fd)
os.unlink(flaskr.app.config['DATABASE'])
if __name__ == '__main__':
unittest.main()
下面我们看一下self.app = flaskr.app.test_client()这行代码到底是干什么的.
# app.py
def test_client(self, use_cookies=True, **kwargs):
cls = self.test_client_class
if cls is None:
from flask.testing import FlaskClient as cls
return cls(self, self.response_class, use_cookies=use_cookies, **kwargs)
默认的test_client_class为None.所以看FlaskClient. FlaskClient继承自Werkzeug.test中的Client，但是额外附带了一些方法.
with flaskr.app.app_context(): flaskr.init_db() 中的app_context()方法已经剖析过了：一个AppContext中含有app, url_adapter, g. 既然创建的AppContext含有g变量，那么flaskr.init_db()就可以顺利的通过执行了。
在7.8节的Keeping the Context Around有小段代码：
app = flask.Flask(__name__)
with app.test_client() as c:
rv = c.get('/?tequila=42')
assert request.args['tequila'] == '42'
我们来看看flask是如何保持request可以正常工作而不是发出'Working outside of request context.'的错误.
在app.test_request_context():...中，很显然的RequestContext被push了。但是app.test_client()却绕了一下弯，不是那么明显.
# testing.py
def __enter__(self):
if self.preserve_context:
raise RuntimeError('Cannot nest client invocations')
self.preserve_context = True
return self
在enter函数中, 将self.preserve_context设置为True. 并返回FlaskClient实例.FlaskClient继承自Werkzeug中的Client.
# test.py in Werkzeug.
def get(self, *args, **kw):
"""Like open but method is enforced to GET."""
kw['method'] = 'GET'
return self.open(*args, **kw)
当我们使用c.get('/?tequila=42')调用get方法时，实际上是return self.open(*args, **kw).
def open(self, *args, **kwargs):
kwargs.setdefault('environ_overrides', {}) \
['flask._preserve_context'] = self.preserve_context
as_tuple = kwargs.pop('as_tuple', False)
buffered = kwargs.pop('buffered', False)
follow_redirects = kwargs.pop('follow_redirects', False)
builder = make_test_environ_builder(self.application, *args, **kwargs)
return Client.open(self, builder,
as_tuple=as_tuple,
buffered=buffered,
follow_redirects=follow_redirects)
FlaskClient重写了Client中的open方法.将kwargs中的flask._preserve_context参数设置为self.preserve_context，也就是True. 紧接着kwargs被传入make_test_environ_builder函数.
def make_test_environ_builder(app, path='/', base_url=None, *args, **kwargs):
http_host = app.config.get('SERVER_NAME')
app_root = app.config.get('APPLICATION_ROOT')
if base_url is None:
url = url_parse(path)
base_url = 'http://%s/' % (url.netloc or http_host or 'localhost')
if app_root:
base_url += app_root.lstrip('/')
if url.netloc:
path = url.path
if url.query:
path += '?' + url.query
return EnvironBuilder(path, base_url, *args, **kwargs)
下面我们看看传入make_test_environ_builder的参数都有哪些：
- app = Flask实例.
- path = '/?tequila=42'
- base_url = None
- args = ()
- kwargs = {'method': 'GET', 'environ_overrides': {'flask._preserve_context': True}}
make_test_environ_builder返回EnvironBuilder. 传入EnvironBuilder的参数除了base_url变为'http://localhost'， 其它都没有变.返回一个EnvironBuilder.示例.
接下来调用Client.open(...)，在Clien.open()函数中，我们主要看builder中的kwargs中的'flask._preserve_context'是怎么传入到wsgi_app中的。
# test.py in Werkzeug.
def open(self, *args, **kwargs):
as_tuple = kwargs.pop('as_tuple', False)
buffered = kwargs.pop('buffered', False)
follow_redirects = kwargs.pop('follow_redirects', False)
environ = None
if not kwargs and len(args) == 1:
if isinstance(args[0], EnvironBuilder):
# 在此处获得builder中的environ.
environ = args[0].get_environ() #<---
elif isinstance(args[0], dict):
environ = args[0]
if environ is None:
builder = EnvironBuilder(*args, **kwargs)
try:
environ = builder.get_environ()
finally:
builder.close()
# 在此处开始run_wsgi_app 并获得response.
response = self.run_wsgi_app(environ, buffered=buffered) # <---
# handle redirects
redirect_chain = []
while 1:
status_code = int(response[1].split(None, 1)[0])
if status_code not in (301, 302, 303, 305, 307) \
or not follow_redirects:
break
new_location = response[2]['location']
method = 'GET'
if status_code == 307:
method = environ['REQUEST_METHOD']
new_redirect_entry = (new_location, status_code)
if new_redirect_entry in redirect_chain:
raise ClientRedirectError('loop detected')
redirect_chain.append(new_redirect_entry)
environ, response = self.resolve_redirect(response, new_location,
environ,
buffered=buffered)
if self.response_wrapper is not None:
# 对response进行包装.
response = self.response_wrapper(*response) # <---
if as_tuple:
return environ, response
return response # <---
在self.run_wsgi_app中, ’flask._preserve_context‘就包含在environ中.
def run_wsgi_app(app, environ, buffered=False):
environ = _get_environ(environ)
response = []
buffer = []
...
app_rv = app(environ, start_response) # <---
...
现在回过头来，我们又进入了熟悉的Flask.wsgi_app(...)的处理流程中.
def wsgi_app(self, environ, start_response):
ctx = self.request_context(environ)
ctx.push()
...
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error) # <---
wsgi_app最后会自动pop ctx.下面我们仔细分析一下相关代码:
def auto_pop(self, exc):
if self.request.environ.get('flask._preserve_context') or \
(exc is not None and self.app.preserve_context_on_exception):
self.preserved = True
self._preserved_exc = exc
else:
self.pop(exc)
于是就真相大白了，原理auto_pop首行代码就是判断request.environ中是否有'flask._preserve_context'变量，另外两个判断条件可能与调试相关，暂时忽略.于是self.preserved = True. 当然， 也就不会执行self.pop(exc)了.不执行的结果就是，RequestContext仍然保存在_request_ctx_stack上面， 当然也就可以正常获取request.args['tequila']的值，而不会出现working outside of context 的报错了. 那什么时候pop ctx呢? 答案在此.
# testing.py
def __exit__(self, exc_type, exc_value, tb):
# 设置preserve-context变量为False.
self.preserve_context = False
# 获得ctx, 测试两个条件, 立即执行top.pop().
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
到此，我们也就搞清楚了官方文档7.8节的那段代码的背后原理.
网友评论