Thread-safe version of mock.call_count

Question

Похоже, что 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 самостоятельно ...

Answer 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, но результат ведет себя, как и ожидалось.