[Python-checkins] gh-61215: threadingmock: Improve test suite to avoid race conditions (#106822)
cjw296
webhook-mailer at python.org
Mon Jul 17 14:57:44 EDT 2023
https://github.com/python/cpython/commit/7e96370a946a2ca0f2f25af4ce5b3b59f020721b
commit: 7e96370a946a2ca0f2f25af4ce5b3b59f020721b
branch: main
author: Mario Corchero <mcorcherojim at bloomberg.net>
committer: cjw296 <chris at withers.org>
date: 2023-07-17T18:57:40Z
summary:
gh-61215: threadingmock: Improve test suite to avoid race conditions (#106822)
threadingmock: Improve test suite to avoid race conditions
Simplify tests and split them into multiple tests to prevent assertions
from triggering race conditions.
Additionally, we rely on calling the mocks without delay to validate the
functionality of matching calls.
files:
M Lib/test/test_unittest/testmock/testthreadingmock.py
diff --git a/Lib/test/test_unittest/testmock/testthreadingmock.py b/Lib/test/test_unittest/testmock/testthreadingmock.py
index b6e12bcb3cda9..94e71921d9bc0 100644
--- a/Lib/test/test_unittest/testmock/testthreadingmock.py
+++ b/Lib/test/test_unittest/testmock/testthreadingmock.py
@@ -8,6 +8,8 @@
threading_helper.requires_working_threading(module=True)
+VERY_SHORT_TIMEOUT = 0.1
+
class Something:
def method_1(self):
@@ -93,167 +95,86 @@ def test_no_name_clash(self):
waitable_mock.wait_until_called()
waitable_mock.wait_until_any_call_with("works")
- def test_wait_success(self):
+ def test_patch(self):
waitable_mock = self._make_mock(spec=Something)
- with patch(f"{__name__}.Something", waitable_mock):
- something = Something()
- self.run_async(something.method_1, delay=0.01)
- something.method_1.wait_until_called()
- something.method_1.wait_until_any_call_with()
- something.method_1.assert_called()
-
- def test_wait_success_with_instance_timeout(self):
- waitable_mock = self._make_mock(timeout=1)
-
- with patch(f"{__name__}.Something", waitable_mock):
- something = Something()
- self.run_async(something.method_1, delay=0.01)
- something.method_1.wait_until_called()
- something.method_1.wait_until_any_call_with()
- something.method_1.assert_called()
-
- def test_wait_failed_with_instance_timeout(self):
- waitable_mock = self._make_mock(timeout=0.01)
-
- with patch(f"{__name__}.Something", waitable_mock):
- something = Something()
- self.run_async(something.method_1, delay=0.5)
- self.assertRaises(AssertionError, waitable_mock.method_1.wait_until_called)
- self.assertRaises(
- AssertionError, waitable_mock.method_1.wait_until_any_call_with
- )
-
- def test_wait_success_with_timeout_override(self):
- waitable_mock = self._make_mock(timeout=0.01)
-
- with patch(f"{__name__}.Something", waitable_mock):
- something = Something()
- self.run_async(something.method_1, delay=0.05)
- something.method_1.wait_until_called(timeout=1)
-
- def test_wait_failed_with_timeout_override(self):
- waitable_mock = self._make_mock(timeout=1)
-
- with patch(f"{__name__}.Something", waitable_mock):
- something = Something()
- self.run_async(something.method_1, delay=0.5)
- with self.assertRaises(AssertionError):
- something.method_1.wait_until_called(timeout=0.05)
-
- def test_wait_success_called_before(self):
- waitable_mock = self._make_mock()
-
with patch(f"{__name__}.Something", waitable_mock):
something = Something()
something.method_1()
something.method_1.wait_until_called()
- something.method_1.wait_until_any_call_with()
- something.method_1.assert_called()
-
- def test_wait_magic_method(self):
- waitable_mock = self._make_mock()
- with patch(f"{__name__}.Something", waitable_mock):
- something = Something()
- self.run_async(something.method_1.__str__, delay=0.01)
- something.method_1.__str__.wait_until_called()
- something.method_1.__str__.assert_called()
-
- def test_wait_until_any_call_with_positional(self):
+ def test_wait_already_called_success(self):
waitable_mock = self._make_mock(spec=Something)
+ waitable_mock.method_1()
+ waitable_mock.method_1.wait_until_called()
+ waitable_mock.method_1.wait_until_any_call_with()
+ waitable_mock.method_1.assert_called()
- with patch(f"{__name__}.Something", waitable_mock):
- something = Something()
- self.run_async(something.method_1, 1, delay=0.2)
- self.assertNotIn(call(1), something.method_1.mock_calls)
- self.run_async(something.method_1, 2, delay=0.5)
- self.run_async(something.method_1, 3, delay=0.6)
-
- something.method_1.wait_until_any_call_with(1)
- something.method_1.assert_called_with(1)
- self.assertNotIn(call(2), something.method_1.mock_calls)
- self.assertNotIn(call(3), something.method_1.mock_calls)
-
- something.method_1.wait_until_any_call_with(3)
- self.assertIn(call(2), something.method_1.mock_calls)
- something.method_1.wait_until_any_call_with(2)
-
- def test_wait_until_any_call_with_keywords(self):
+ def test_wait_until_called_success(self):
waitable_mock = self._make_mock(spec=Something)
+ self.run_async(waitable_mock.method_1, delay=VERY_SHORT_TIMEOUT)
+ waitable_mock.method_1.wait_until_called()
- with patch(f"{__name__}.Something", waitable_mock):
- something = Something()
- self.run_async(something.method_1, a=1, delay=0.2)
- self.assertNotIn(call(a=1), something.method_1.mock_calls)
- self.run_async(something.method_1, b=2, delay=0.5)
- self.run_async(something.method_1, c=3, delay=0.6)
-
- something.method_1.wait_until_any_call_with(a=1)
- something.method_1.assert_called_with(a=1)
- self.assertNotIn(call(b=2), something.method_1.mock_calls)
- self.assertNotIn(call(c=3), something.method_1.mock_calls)
-
- something.method_1.wait_until_any_call_with(c=3)
- self.assertIn(call(b=2), something.method_1.mock_calls)
- something.method_1.wait_until_any_call_with(b=2)
-
- def test_wait_until_any_call_with_no_argument_fails_when_called_with_arg(self):
- waitable_mock = self._make_mock(timeout=0.01)
-
- with patch(f"{__name__}.Something", waitable_mock):
- something = Something()
- something.method_1(1)
-
- something.method_1.assert_called_with(1)
- with self.assertRaises(AssertionError):
- something.method_1.wait_until_any_call_with()
+ def test_wait_until_called_method_timeout(self):
+ waitable_mock = self._make_mock(spec=Something)
+ with self.assertRaises(AssertionError):
+ waitable_mock.method_1.wait_until_called(timeout=VERY_SHORT_TIMEOUT)
- something.method_1()
- something.method_1.wait_until_any_call_with()
+ def test_wait_until_called_instance_timeout(self):
+ waitable_mock = self._make_mock(spec=Something, timeout=VERY_SHORT_TIMEOUT)
+ with self.assertRaises(AssertionError):
+ waitable_mock.method_1.wait_until_called()
- def test_wait_until_any_call_with_global_default(self):
+ def test_wait_until_called_global_timeout(self):
with patch.object(ThreadingMock, "DEFAULT_TIMEOUT"):
- ThreadingMock.DEFAULT_TIMEOUT = 0.01
- m = self._make_mock()
+ ThreadingMock.DEFAULT_TIMEOUT = VERY_SHORT_TIMEOUT
+ waitable_mock = self._make_mock(spec=Something)
with self.assertRaises(AssertionError):
- m.wait_until_any_call_with()
- with self.assertRaises(AssertionError):
- m.wait_until_called()
+ waitable_mock.method_1.wait_until_called()
- m()
- m.wait_until_any_call_with()
- assert ThreadingMock.DEFAULT_TIMEOUT != 0.01
+ def test_wait_until_any_call_with_success(self):
+ waitable_mock = self._make_mock()
+ self.run_async(waitable_mock, delay=VERY_SHORT_TIMEOUT)
+ waitable_mock.wait_until_any_call_with()
- def test_wait_until_any_call_with_change_global_and_override(self):
- with patch.object(ThreadingMock, "DEFAULT_TIMEOUT"):
- ThreadingMock.DEFAULT_TIMEOUT = 0.01
+ def test_wait_until_any_call_with_instance_timeout(self):
+ waitable_mock = self._make_mock(timeout=VERY_SHORT_TIMEOUT)
+ with self.assertRaises(AssertionError):
+ waitable_mock.wait_until_any_call_with()
- m1 = self._make_mock()
- self.run_async(m1, delay=0.1)
+ def test_wait_until_any_call_global_timeout(self):
+ with patch.object(ThreadingMock, "DEFAULT_TIMEOUT"):
+ ThreadingMock.DEFAULT_TIMEOUT = VERY_SHORT_TIMEOUT
+ waitable_mock = self._make_mock()
with self.assertRaises(AssertionError):
- m1.wait_until_called()
+ waitable_mock.wait_until_any_call_with()
- m2 = self._make_mock(timeout=1)
- self.run_async(m2, delay=0.1)
- m2.wait_until_called()
-
- m3 = self._make_mock()
- self.run_async(m3, delay=0.1)
- m3.wait_until_called(timeout=1)
-
- m4 = self._make_mock()
- self.run_async(m4, delay=0.1)
- m4.wait_until_called(timeout=None)
+ def test_wait_until_any_call_positional(self):
+ waitable_mock = self._make_mock(timeout=VERY_SHORT_TIMEOUT)
+ waitable_mock.method_1(1, 2, 3)
+ waitable_mock.method_1.wait_until_any_call_with(1, 2, 3)
+ with self.assertRaises(AssertionError):
+ waitable_mock.method_1.wait_until_any_call_with(2, 3, 1)
+ with self.assertRaises(AssertionError):
+ waitable_mock.method_1.wait_until_any_call_with()
- m5 = self._make_mock(timeout=None)
- self.run_async(m5, delay=0.1)
- m5.wait_until_called()
+ def test_wait_until_any_call_kw(self):
+ waitable_mock = self._make_mock(timeout=VERY_SHORT_TIMEOUT)
+ waitable_mock.method_1(a=1, b=2)
+ waitable_mock.method_1.wait_until_any_call_with(a=1, b=2)
+ with self.assertRaises(AssertionError):
+ waitable_mock.method_1.wait_until_any_call_with(a=2, b=1)
+ with self.assertRaises(AssertionError):
+ waitable_mock.method_1.wait_until_any_call_with()
- assert ThreadingMock.DEFAULT_TIMEOUT != 0.01
+ def test_magic_methods_success(self):
+ waitable_mock = self._make_mock()
+ str(waitable_mock)
+ waitable_mock.__str__.wait_until_called()
+ waitable_mock.__str__.assert_called()
def test_reset_mock_resets_wait(self):
- m = self._make_mock(timeout=0.01)
+ m = self._make_mock(timeout=VERY_SHORT_TIMEOUT)
with self.assertRaises(AssertionError):
m.wait_until_called()
More information about the Python-checkins
mailing list