Pytest`mutliprocessing.Process`实例不是以确定性方式创建的

2024-09-30 19:20:37 发布

您现在位置:Python中文网/ 问答频道 /正文

我的代码创建了SimpleFTPServer的两个实例,然后用multiprocessing.Process调用它们。测试用例同时使用了两台服务器,有时会相互替换。这会导致测试不稳定。 以下是测试和夹具代码:

# fixtures.py
import multiprocessing
import random
import time

import pytest

from pytest_localftpserver.servers import ProcessFTPServer, SimpleFTPServer


class ProcessFTPServer:

    def __init__(self, username, password, ftp_home, ftp_port, use_TLS=False):
        self._server = SimpleFTPServer(username, password, ftp_port=ftp_port,
                                       ftp_home=ftp_home, use_TLS=use_TLS)
        print(self._server)
        self.process = multiprocessing.Process(target=self._server.serve_forever)
        # This is a must in order to clear used sockets
        self.process.daemon = True
        # time.sleep(0.5) # and 200 out 200 runs pass ...?
        self.process.start()

    def stop(self):
        self.process.terminate()

   # adding this will cause the tests to fail less often
   #def __repr__(self):
   #    return f"{self._server.username}:{self._server.password}"

    
@pytest.fixture(scope="function")
def servers(request):
    port1, port2 = random.randint(1024, 2**16-1), random.randint(1024, 2**16-1)
    while port1 == port2:
        port2 = random.randint(1024, 2**16-1)

    server1 = ProcessFTPServer(username="benz", password="erni1",
                              ftp_home="/home/oznt/Music", ftp_port=port1) # uses explicit parameters
    request.addfinalizer(server1.stop)

    server2 = ProcessFTPServer(username="fakeusername", password="qweqwe",
                              ftp_home="/home/oznt/", ftp_port=port2) # uses explicit parameters
    request.addfinalizer(server2.stop)
    assert id(server1) != id(server2)
    return [server1, server2]

以及:

# tests.py
from ftplib import FTP
import ftplib

from fixtures import servers

def test_ftp(servers):
    ftpserver_from, ftpserver_to = servers
    try:
        ftp1 = FTP()
        ftp1.connect('localhost', ftpserver_from._server._ftp_port)
        ftp1.login(ftpserver_from._server.username, ftpserver_from._server.password)
        ftp2 = FTP()

        ftp2.connect('localhost', ftpserver_to._server._ftp_port)
        ftp2.login(ftpserver_to._server.username, ftpserver_to._server.password)
    except ftplib.error_perm:
        import pdb; pdb.set_trace()

以下是代码无法运行(触发异常)的示例:

$ for i in `seq 1 10`; do pytest -s . ; done
=============================================== test session starts ================================================
platform linux -- Python 3.9.4, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
rootdir: /home/oznt/Software/demo-pytest-ftpserver, configfile: pytest.ini
plugins: localftpserver-1.1.2, env-0.6.2
collected 1 item                                                                                                   

tests/test_two_servers.py <pytest_localftpserver.servers.SimpleFTPServer at 0x7f654a1714c0>
<pytest_localftpserver.servers.SimpleFTPServer at 0x7f654a171220>
{'servers': (<fixtures.ProcessFTPServer object at 0x7f654a171820>, <fixtures.ProcessFTPServer object at 0x7f654a171520>)}

>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> PDB set_trace >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
--Return--
> /home/oznt/Software/demo-pytest-ftpserver/tests/test_two_servers.py(19)test_ftp()->None
-> import pdb; pdb.set_trace()
(Pdb)

显然,ProcessFTPServer的实例以某种我无法控制的顺序“附加”到fixturesThe order of fixutres might be related。然而,我不知道为什么(我实际上只有一个夹具)。 奇怪的是,当我将__repr__方法添加到ProcessFTPServer中时,测试失败的次数就更少了。如果我添加一个小的time.sleep(0.5)来故意降低测试的速度,200次通过中会有200次运行。 有人能解释这种行为吗


Tags: tofromimportselfhomeserverpytestport