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」。