Streaming version of mock.call_count

Question

It appears that Mock.call_count does not work correctly with threads. For instance:

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

The following output appeared in this code:

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

Is there a way to use call_count(or similar) in the multi-threaded part of the code? I would like to avoid having to rewrite MagicMock myself ...

Answer 1

I finally did the job using a counter associated with the side effect method and blocking.

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

Therefore, I am counting the number of calls finstead mock, but the result behaves as expected.