在django测试中并发执行django应用程序代码的帮助程序
django-concurrent-test-helper的Python项目详细描述
在django测试中并发执行django应用程序代码的帮助程序。
测试的python版本与Django supports：
| x | Py2.7 | Py3.4 | Py3.5 | Py3.6 |
|---|---|---|---|---|
| Django 1.4 | ||||
| Django 1.5 | ||||
| Django 1.6 | ||||
| Django 1.7 | ||||
| Django 1.8 | ||||
| Django 1.9 | ||||
| Django 1.10 | ||||
| Django 1.11 |
（除了Python3.2和3.3…它们不再受支持）
开始
pip install django-concurrent-test-helper
与https://github.com/box/flaky（pip install flaky）配合得很好，因为您可能希望在尝试触发罕见的竞争条件时多次运行测试。
您还需要将其添加到django项目设置中：
INSTALLED_APPS=(# ...'django_concurrent_tests',)
提供两个助手：
fromdjango_concurrent_tests.helpersimportcall_concurrentlydefis_success(result):returnresultisTrueandnotisinstance(result,Exception)deftest_concurrent_code():results=call_concurrently(5,racey_function,first_arg=1)# results contains the return value from each callsuccesses=list(filter(is_success,results))assertlen(successes)==1
以及：
fromdjango_concurrent_tests.helpersimportmake_concurrent_callsdefis_success(result):returnresultisTrueandnotisinstance(result,Exception)deftest_concurrent_code():calls=[(first_func,{'first_arg':1}),(second_func,{'other_arg':'wtf'}),]*3results=make_concurrent_calls(*calls)# results contains the return value from each callsuccesses=list(filter(is_success,results))assertlen(successes)==1
注意，如果调用的函数引发异常，则异常将包装在WrappedError异常中。这提供了一种访问原始回溯，甚至重新引发原始异常的方法。
importtypesfromdjango_concurrent_tests.errorsimportWrappedErrorfromdjango_concurrent_tests.helpersimportmake_concurrent_callsdeftest_concurrent_code():calls=[(first_func,{'first_arg':1}),(raises_error,{'other_arg':'wtf'}),]*3results=make_concurrent_calls(*calls)# results contains the return value from each callerrors=list(filter(lambdar:isinstance(r,Exception),results))assertlen(errors)==3assertisinstance(errors[0],WrappedError)assertisinstance(errors[0].error,ValueError)# the original errorassertisinstance(errors[0].traceback,types.TracebackType)# other things you can do with the WrappedError:# 1. print the tracebackerrors[0].print_tb()# 2. drop into a debugger (ipdb if installed, else pdb)errors[0].debug()ipdb># ...can explore the stack of original exception!# 3. re-raise the original exceptiontry:errors[0].reraise()exceptValueErrorase:# `e` will be the original error with original traceback
另一件要记住的事情是，如果您在测试中使用override_settingsdecorator。您还需要修饰调用的函数（因为子进程看不到主测试进程中重写的设置）：
fromdjango_concurrent_tests.helpersimportmake_concurrent_calls@override_settings(SPECIAL_SETTING=False)deftest_concurrent_code():calls=[(first_func,{'first_arg':1}),(raises_error,{'other_arg':'wtf'}),]*3results=make_concurrent_calls(*calls)@override_settings(SPECIAL_SETTING=False)deffirst_func(first_arg):returnfirst_arg*2defraises_error(other_arg):# can also be used as a context managerwithoverride_settings(SPECIAL_SETTING=False):raiseSomeError(other_arg)
另一方面，定制的环境变量将由子进程继承，并提供一个override_environment上下文管理器，用于您的测试：
fromdjango_concurrent_tests.helpersimportcall_concurrentlyfromdjango_concurrent_tests.utilsimportoverride_environmentdeffunc_to_test(first_arg):importosreturnos.getenv('SPECIAL_ENV')deftest_concurrent_code():withoverride_environment(SPECIAL_ENV='so special'):results=call_concurrently(1,func_to_test)assertresults[0]=='so special'
最后，可以将字符串导入路径传递给函数，而不是函数本身。格式为：'dotted module.path.to:function'（注意冒号分隔要导入的名称，位于点式模块路径之后）。
当您不想在测试中导入函数本身以通过测试时，这会很好。但更重要的是，它是essential在某些情况下，例如f是一个修饰函数，其修饰程序返回一个新对象（并且functools.wraps未使用）。在这种情况下，我们将无法从函数对象的__module__（它将指向decorator的模块）中内省导入路径，因此对于那些通过字符串调用的情况，是必需的。（用@app.task修饰的芹菜任务是一个需要通过字符串路径调用的示例）
fromdjango_concurrent_tests.helpersimportcall_concurrently@bad_decoratordefmyfunc():returnTruedeftest_concurrent_code():results=call_concurrently('mymodule.module:myfunc',3)# results contains the return value from each callresults=list(filter(None,results))assertlen(results)==3
注释
为什么是子流程？
我们最初只想使用multiprocessing.Pool调用要测试的函数来实现这个功能。如果成功的话，这个模块就不需要了。
不幸的是，这种方法遇到了一个问题：多处理通过分叉父进程来工作。分叉进程继承父进程的套接字，因此在django项目中，这将包括psycopg2打开的套接字到postgres数据库。但是，继承的套接字处于断开状态。有很多关于这个的问题，所以没有给出解决方案，看起来基本上你不能分叉一个django进程，然后对db做任何事情。
（注意在python 3中，您可能可以使用多处理的‘spawn’ start method来避免fork问题-没有尝试过这个方法）
因此，为了使这个工作，我们必须使用subprocess.Popen来运行未分叉的“virgin”进程。为了能够以这种方式测试任意函数，我们做了一个丑陋/聪明的黑客操作，并提供了一个manage.py concurrent_call_wrapper命令（这就是为什么您必须将此模块添加到INSTALLED_APPS中，该命令处理kwargs的序列化和返回值。
This does mean that your kwargs and return value must be pickleable.
另一个可能的问题是运行测试时是否使用sqlite db。在本例中，默认情况下django将对测试数据库使用:memory:。但这意味着并发进程都有自己的内存数据库，并且无法看到父测试运行创建的数据。
For these tests to work you need to be sure to set ^{tt15}$ for the SQLite db to a real filename in your ^{tt16}$ settings (in Django 1.9 this is a dict, i.e. ^{tt17}$).
最后，需要注意django的隐式事务，否则在父测试中创建的数据尚未提交，因此子进程看不到这些数据。
Ensure that you use Django’s ^{tt18}$ or a derivative (to prevent all the code in your test from being inside an uncommitted transaction).
欢迎加入QQ群-->： 979659372
