如何使用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?

- TheBeginner

---

显示完整的 Traceback - stovfl

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()

- Nick