5

マルチプロセッシングを多用するモジュールのコンテキストで、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 を使用して子の例外検出を自動化するきちんとした方法があるかどうか疑問に思っています。

親で子の終了コードを確認する必要がありますか? これが唯一の方法ですか？

4

2 に答える 2

6

さらに検討した結果、子で予想される例外をキャッチし、子の終了状態をチェックするだけで、テストに追加の 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 ===========================
于 2012-11-15T15:39:23.153 に答える
3 
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" を受け入れることができます。

于 2012-11-20T07:41:43.480 に答える