py.test:如何自动检测子进程中的异常

Question

我正在Linux上运行py.test，在一个大量使用多处理的模块的上下文中。子进程中的异常不会被检测为错误。示例测试文件pytest_mp_test.py：

import multiprocessing
def test_mp():
    p = multiprocessing.Process(target=child)
    p.start()
    p.join()
def child():
    assert False

执行：

$ py.test pytest_mp_test.py 
================================== test session starts ===================================
platform linux2 -- Python 2.7.3 -- pytest-2.3.3
plugins: cov
collected 1 items 
pytest_mp_test.py .
================================ 1 passed in 0.04 seconds ================================

未检测到错误。使用 -s 调用时打印异常：

$ py.test -s pytest_mp_test.py 
================================== test session starts ===================================
platform linux2 -- Python 2.7.3 -- pytest-2.3.3
plugins: cov
collected 1 items 
pytest_mp_test.py Process Process-1:
Traceback (most recent call last):
  File "/apps11/bioinfp/Python-2.7.3/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/apps11/bioinfp/Python-2.7.3/lib/python2.7/multiprocessing/process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "/home/bioinfp/jang/hobbyproggorn/gevent-messagepipe/gevent-messagepipe/pytest_mp_test.py", line 9, in child
    assert False
AssertionError: assert False
.
================================ 1 passed in 0.04 seconds ================================

当我手动查看测试日志时，我意识到何时出现问题。但是，我想知道是否有一种巧妙的方法可以在使用 py.test 的孩子中自动进行异常检测。

我必须在家长中验证孩子的退出代码吗？这是唯一的办法吗？

Answer 1

经过进一步的考虑，我认为只在孩子中捕获预期的异常并检查孩子的退出状态是一个非常干净可靠的解决方案，它不会在测试中添加额外的 IPC 组件。示例代码：

import multiprocessing
from py.test import raises
class CustomError(Exception):
    pass
def test_mp_expected_fail():
    p = multiprocessing.Process(target=child_expected_fail)
    p.start()
    p.join()
    assert not p.exitcode
def test_mp_success():
    p = multiprocessing.Process(target=child)
    p.start()
    p.join()
    assert not p.exitcode
def test_mp_unexpected_fail():
    p = multiprocessing.Process(target=child_unexpected_fail)
    p.start()
    p.join()
    assert not p.exitcode

def child_expected_fail():
    with raises(CustomError):
        raise CustomError
def child_unexpected_fail():
    raise TypeError
def child():
    pass

执行：

$ py.test pytest_mp_test.py 
================================== test session starts ===================================
platform linux2 -- Python 2.7.3 -- pytest-2.3.3
plugins: cov
collected 3 items 
pytest_mp_test.py ..F
======================================== FAILURES ========================================
________________________________ test_mp_unexpected_fail _________________________________
    def test_mp_unexpected_fail():
        p = multiprocessing.Process(target=child_unexpected_fail)
        p.start()
        p.join()
>       assert not p.exitcode
E       assert not 1
E        +  where 1 = <Process(Process-3, stopped[1])>.exitcode
pytest_mp_test.py:23: AssertionError
------------------------------------ Captured stderr -------------------------------------
Process Process-3:
Traceback (most recent call last):
  File "/apps11/bioinfp/Python-2.7.3/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/apps11/bioinfp/Python-2.7.3/lib/python2.7/multiprocessing/process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "/home/bioinfp/jang/hobbyproggorn/gevent-messagepipe/gevent-messagepipe/pytest_mp_test.py", line 31, in child_unexpected_fail
    raise TypeError
TypeError
=========================== 1 failed, 2 passed in 0.07 seconds ===========================

Answer 2

import pytest
@pytest.fixture
def runproc(request):
    import multiprocessing
    def Process(target):
        p = multiprocessing.Process(target=target)
        p.start()
        p.join()
        return p
    return Process
class CustomError(Exception):
    pass

def test_mp_expected_fail(runproc):
    p = runproc(child_expected_fail)
    assert not p.exitcode
def test_mp_success(runproc):
    p = runproc(target=child)
    assert not p.exitcode
def test_mp_unexpected_fail(runproc):
    p = runproc(child_unexpected_fail)
    assert not p.exitcode
def child_expected_fail():
    with pytest.raises(CustomError):
        raise CustomError
def child_unexpected_fail():
    raise TypeError
def child():
    pass

如果将此"runproc"装置放入 conftest.py 文件中，则可以在项目的任何函数中接受"runproc"。