+import sys
import time
import unittest
+import threading
import concurrent.futures
-from test.support import threading_helper
+from test.support import setswitchinterval, threading_helper
from unittest.mock import patch, ThreadingMock
m.wait_until_any_call_with()
m.assert_called_once()
+ def test_call_count_thread_safe(self):
+ # See https://github.com/python/cpython/issues/142651.
+ m = ThreadingMock()
+ LOOPS = 100
+ THREADS = 10
+ def test_function():
+ for _ in range(LOOPS):
+ m()
+
+ oldswitchinterval = sys.getswitchinterval()
+ setswitchinterval(1e-6)
+ try:
+ threads = [threading.Thread(target=test_function) for _ in range(THREADS)]
+ with threading_helper.start_threads(threads):
+ pass
+ finally:
+ sys.setswitchinterval(oldswitchinterval)
+
+ self.assertEqual(m.call_count, LOOPS * THREADS)
+
if __name__ == "__main__":
unittest.main()
def _increment_mock_call(self, /, *args, **kwargs):
self.called = True
- self.call_count += 1
# handle call_args
# needs to be set here so assertions on call arguments pass before
_call = _Call((args, kwargs), two=True)
self.call_args = _call
self.call_args_list.append(_call)
+ self.call_count = len(self.call_args_list)
# initial stuff for method_calls:
do_method_calls = self._mock_parent is not None