mock.call_count 的线程安全版本答案

【问题标题】：Thread-safe version of mock.call_countmock.call_count 的线程安全版本
【发布时间】：2016-09-05 13:53:25
【问题描述】：

Mock.call_count 似乎无法与线程一起正常工作。例如：

import threading
import time
from mock import MagicMock


def f():
    time.sleep(0.1)

def test_1():
    mock = MagicMock(side_effect=f)
    nb_threads = 100000
    threads = []
    for _ in range(nb_threads):
        thread = threading.Thread(target=mock)
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

    assert mock.call_count == nb_threads, mock.call_count

test_1()

此代码产生以下输出：

Traceback (most recent call last):
  File "test1.py", line 24, in <module>
    test_1()
  File "test1.py", line 21, in test_1
    assert mock.call_count == nb_threads, mock.call_count
AssertionError: 99994

有没有办法可以在代码的多线程部分中使用call_count（或类似的）？我想避免自己重写 MagicMock...

【问题讨论】：

标签： python multithreading unit-testing mocking

【解决方案1】：

我终于通过使用链接到副作用方法和锁的计数器使其工作。

import threading
import time
from mock import MagicMock

lock_side_effect = threading.Lock()

def f():
    with lock_side_effect:
        f.call_count += 1
    time.sleep(0.1)

f.call_count = 0

def test_1():
    mock = MagicMock(side_effect=f)
    nb_threads = 100000
    threads = []
    for _ in range(nb_threads):
        thread = threading.Thread(target=mock)
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

    assert f.call_count == nb_threads, f.call_count

test_1()

因此，我计算的是 f 而不是 mock 的调用次数，但结果表现符合预期。

【讨论】：

【解决方案2】：

我写了一个小库来处理这个。

重构大量测试套件可能会出现问题，因此该库允许您修补基类调用以使用 RLock 模拟 __getattr__。

https://github.com/AtakamaLLC/tsmock

你可以这样做：

from tsmock import MagicMock

或者这个：

from tsmock import thread_safe_mocks
thread_safe_mocks()

任何一个都能为您提供正确的通话次数。

【讨论】：