Thread-safe version of mock.call_count
<p>It appears that <a href="https://docs.python.org/3/library/unittest.mock.html#unittest.mock.Mock.call_count" rel="nofollow" title="Mock.call_count">Mock.call_count</a> does not work correctly with threads. For instance:</p>

<pre><code>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()
</code></pre>

<p>This code produced the following output:</p>

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

<p>Is there a way I can use <code>call_count</code> (or similar) within a multithreaded portion of code? I'd like to avoid having to rewrite MagicMock myself...</p>

<p>I finally made it work by using a counter linked to the side effect method and a lock.</p>

<pre><code>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()
</code></pre>

<p>Consequently, I'm counting the number of calls of <code>f</code> instead of <code>mock</code>, but the result behaves as expected.</p>

python multithreading unit-testing mocking