如何使用 py.test 测试并发性

Question

我想使用 py.test 测试函数的线程安全性我的努力：

def function_to_be_tested(arg1,arg2):
   some functionality

class Test:
   def setup()
   def teardown()
   def test_conc()
       p1=Process(taget=function_to_be_tested,args(arg1,arg2,))
       p2=Process (taget=function_to_be_tested,args(arg1,arg3,))
       p1.start()
       p2.start()
       p1.join()
       p2.join()

使用 py.test 命令执行上面的文件。这显示以下错误。

ExceptionPexpect: isalive() encountered condition where "terminated" is 0, but there was no child process. Did someone else call waitpid() on our process?

你能帮我解码这个错误，并就如何做到这一点提供指导吗？

谢谢这是我正在尝试的实际代码和堆栈跟踪：

import pytest
from multiprocessing import Process
from pexpect import pxssh

def func(cls,b):
    cls.s.sendline("bteq")
    cls.s.prompt()
    print b
    #some operations inside the bteq session

class Test:
    @classmethod
    def setup_class(cls):
        cls.s=pxssh.pxssh()
        cls.s.login("IP",'Username','Pwd')
    @classmethod
    def teardown_class(cls):
        cls.s.logout()
        print "teardown"

    def test_1(cls):
        p1=Process(target=func,args=(cls,13,))
        p2=Process(target=func,args=(cls,46,))
        p1.start()
        p2.start()
        p1.join()
        p2.join()

堆栈跟踪：

dipakw@dipakw-Inspiron-3558:~$ py.test -v -s test.py 
============================= test session starts ==============================
platform linux2 -- Python 2.7.6, pytest-3.0.6, py-1.4.32, pluggy-0.4.0 -- /usr/bin/python
cachedir: .cache
rootdir: /home/dipakw, inifile: 
plugins: xdist-1.15.0
collected 1 items 

test.py::Test::test_1 Process Process-1:
Traceback (most recent call last):
  File "/usr/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python2.7/multiprocessing/process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "/home/dipakw/test.py", line 7, in func
    cls.s.prompt()
  File "/usr/lib/python2.7/dist-packages/pexpect/pxssh.py", line 352, in prompt
Process Process-2:
Traceback (most recent call last):
  File "/usr/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python2.7/multiprocessing/process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "/home/dipakw/test.py", line 7, in func
    cls.s.prompt()
  File "/usr/lib/python2.7/dist-packages/pexpect/pxssh.py", line 352, in prompt
    i = self.expect([self.PROMPT, TIMEOUT], timeout=timeout)
    i = self.expect([self.PROMPT, TIMEOUT], timeout=timeout)
  File "/usr/lib/python2.7/dist-packages/pexpect/__init__.py", line 1418, in expect
  File "/usr/lib/python2.7/dist-packages/pexpect/__init__.py", line 1418, in expect
    timeout, searchwindowsize)
    timeout, searchwindowsize)
  File "/usr/lib/python2.7/dist-packages/pexpect/__init__.py", line 1433, in expect_list
  File "/usr/lib/python2.7/dist-packages/pexpect/__init__.py", line 1433, in expect_list
    timeout, searchwindowsize)
  File "/usr/lib/python2.7/dist-packages/pexpect/__init__.py", line 1502, in expect_loop
    timeout, searchwindowsize)
    c = self.read_nonblocking(self.maxread, timeout)
  File "/usr/lib/python2.7/dist-packages/pexpect/__init__.py", line 1502, in expect_loop
  File "/usr/lib/python2.7/dist-packages/pexpect/__init__.py", line 886, in read_nonblocking
    if not self.isalive():
    c = self.read_nonblocking(self.maxread, timeout)
  File "/usr/lib/python2.7/dist-packages/pexpect/__init__.py", line 1220, in isalive
    'on our process?')
  File "/usr/lib/python2.7/dist-packages/pexpect/__init__.py", line 886, in read_nonblocking
    if not self.isalive():
  File "/usr/lib/python2.7/dist-packages/pexpect/__init__.py", line 1220, in isalive
ExceptionPexpect: isalive() encountered condition where "terminated" is 0, but there was no child process. Did someone else call waitpid() on our process?
    'on our process?')
ExceptionPexpect: isalive() encountered condition where "terminated" is 0, but there was no child process. Did someone else call waitpid() on our process?

Answer 1

尝试按照以下方式测试并发/线程：

import threading
from functools import partial


@pytest.mark.django_db
def test_is_concurrency_safe():        
    # setup anything for the test...

    # recreate multiple threads calling the same function at the same time
   _call_concurrently(
        partial(my_function, args),
        partial(my_function, args),
    )

    # Test that the race condition didn't create duplicates, etc

def _call_concurrently(*callables):
    threads = [threading.Thread(target=callable_) for callable_ in callables]

    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()