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

This code produced the following output:

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 I can use call_count (or similar) within a multithreaded portion of code? I'd like to avoid having to rewrite MagicMock myself...

Answer 1

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

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

Consequently, I'm counting the number of calls of f instead of mock, but the result behaves as expected.