1

我有一个简单的代码库，如下所示：

import threading
import time

class Demo:
    def __init__(self):
        self._run_thread = False

    def start(self):
        if not self._run_thread:
            print("Thread has not run")
            threading.Thread(target=self.counter, args=(20, )).start()
            return 20
        else:
            print("Thread has run")
            return None


    def counter(self, value):
        self._run_thread = True
        for i in range(value):
            print(f"Inside counter with value {i}")
            time.sleep(3)



if __name__ == "__main__":
    d = Demo()
    d.start()
    d.start()

第一次start调用函数时，线程将运行，而下一次调用它时不会因为设置了 flag(_run_thread)。

代码本身运行得很好。

但是，当我尝试对此代码运行测试时，它无法按预期工作。下面是我的测试代码：

import app

def dummy_counter(self, value):
    self._run_thread = True

def test_app_start_function_runs_well(mocker):
    d = app.Demo()
    mocker.patch.object(
        d,
        "counter",
        dummy_counter
    )
    d.start()
    result = d.start()
    assert result is None

因此，由于我实际上不希望循环，因此我只想模拟该counter函数，然后在其中设置标志，以便实际线程不会运行。

我看到以下错误。

(roughwork) ➜  threads git:(master) ✗ pytest
============================================================ test session starts ============================================================
platform linux -- Python 3.7.6, pytest-5.4.3, py-1.8.1, pluggy-0.13.1
rootdir: /home/subhayan/Codes/Test/Python-remote/threads
plugins: mock-3.1.1
collected 1 item                                                                                                                            

test_app.py Exception in thread Thread-2:
Traceback (most recent call last):
  File "/home/subhayan/anaconda3/envs/roughwork/lib/python3.7/threading.py", line 926, in _bootstrap_inner
    self.run()
  File "/home/subhayan/anaconda3/envs/roughwork/lib/python3.7/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
TypeError: dummy_counter() missing 1 required positional argument: 'value'

F                                                                                                                         [100%]

================================================================= FAILURES ==================================================================
____________________________________________________________ test_app_runs_well _____________________________________________________________

mocker = <pytest_mock.plugin.MockFixture object at 0x7f922c05ee90>

    def test_app_runs_well(mocker):
        d = app.Demo()
        mocker.patch.object(
            d,
            "counter",
            dummy_counter
        )
        d.start()
        result = d.start()
>       assert result is None
E    assert 20 is None

test_app.py:15: AssertionError
----------------------------------------------------------- Captured stdout call ------------------------------------------------------------
Thread has not run
Thread has not run
----------------------------------------------------------- Captured stderr call ------------------------------------------------------------
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/home/subhayan/anaconda3/envs/roughwork/lib/python3.7/threading.py", line 926, in _bootstrap_inner
    self.run()
  File "/home/subhayan/anaconda3/envs/roughwork/lib/python3.7/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
TypeError: dummy_counter() missing 1 required positional argument: 'value'

========================================================== short test summary info ==========================================================
FAILED test_app.py::test_app_runs_well - assert 20 is None

我不知道我哪里出错了。任何线索，我将不胜感激。

4

1 回答 1

0

你几乎做对了所有事情。唯一的问题是它self.counter是实例的绑定方法，因此self填充了您的正确实例d。但是，dummy_counter未绑定 - 因此它不会自动获取第一个参数，因此您需要提供它。

一种方法是使用部分函数，​​如下所示：

from functools import partial

import app


def dummy_counter(self, value):
    self._run_thread = True


def test_app_start_function_runs_well(mocker):
    d = app.Demo()
    dummy_for_d = partial(dummy_counter, d)  # fills the first parameter with 'd'

    mocker.patch.object(
        d,
        "counter",
        dummy_for_d
    )
    d.start()
    result = d.start()
    assert result is None
于 2021-09-09T10:03:09.007 回答