[Python-checkins] cpython: asyncio: Refactor waitpid mocks. Patch by Anthony Baire.
guido.van.rossum
python-checkins at python.org
Fri Nov 15 01:16:36 CET 2013
http://hg.python.org/cpython/rev/a75b88048339
changeset: 87108:a75b88048339
user: Guido van Rossum <guido at dropbox.com>
date: Thu Nov 14 16:16:29 2013 -0800
summary:
asyncio: Refactor waitpid mocks. Patch by Anthony Baire.
files:
Lib/test/test_asyncio/test_unix_events.py | 425 ++++-----
1 files changed, 181 insertions(+), 244 deletions(-)
diff --git a/Lib/test/test_asyncio/test_unix_events.py b/Lib/test/test_asyncio/test_unix_events.py
--- a/Lib/test/test_asyncio/test_unix_events.py
+++ b/Lib/test/test_asyncio/test_unix_events.py
@@ -1,5 +1,6 @@
"""Tests for unix_events.py."""
+import collections
import gc
import errno
import io
@@ -705,8 +706,16 @@
NotImplementedError, watcher._do_waitpid, f)
+WaitPidMocks = collections.namedtuple("WaitPidMocks",
+ ("waitpid",
+ "WIFEXITED",
+ "WIFSIGNALED",
+ "WEXITSTATUS",
+ "WTERMSIG",
+ ))
+
+
class ChildWatcherTestsMixin:
- instance = None
ignore_warnings = unittest.mock.patch.object(unix_events.logger, "warning")
@@ -715,21 +724,12 @@
self.running = False
self.zombies = {}
- assert ChildWatcherTestsMixin.instance is None
- ChildWatcherTestsMixin.instance = self
-
with unittest.mock.patch.object(
self.loop, "add_signal_handler") as self.m_add_signal_handler:
self.watcher = self.create_watcher()
self.watcher.attach_loop(self.loop)
- def cleanup():
- ChildWatcherTestsMixin.instance = None
-
- self.addCleanup(cleanup)
-
- def waitpid(pid, flags):
- self = ChildWatcherTestsMixin.instance
+ def waitpid(self, pid, flags):
if isinstance(self.watcher, unix_events.SafeChildWatcher) or pid != -1:
self.assertGreater(pid, 0)
try:
@@ -747,33 +747,43 @@
def add_zombie(self, pid, returncode):
self.zombies[pid] = returncode + 32768
- def WIFEXITED(status):
+ def WIFEXITED(self, status):
return status >= 32768
- def WIFSIGNALED(status):
+ def WIFSIGNALED(self, status):
return 32700 < status < 32768
- def WEXITSTATUS(status):
- self = ChildWatcherTestsMixin.instance
- self.assertTrue(type(self).WIFEXITED(status))
+ def WEXITSTATUS(self, status):
+ self.assertTrue(self.WIFEXITED(status))
return status - 32768
- def WTERMSIG(status):
- self = ChildWatcherTestsMixin.instance
- self.assertTrue(type(self).WIFSIGNALED(status))
+ def WTERMSIG(self, status):
+ self.assertTrue(self.WIFSIGNALED(status))
return 32768 - status
def test_create_watcher(self):
self.m_add_signal_handler.assert_called_once_with(
signal.SIGCHLD, self.watcher._sig_chld)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_sigchld(self, m_waitpid, m_WIFEXITED, m_WIFSIGNALED,
- m_WEXITSTATUS, m_WTERMSIG):
+ def waitpid_mocks(func):
+ def wrapped_func(self):
+ def patch(target, wrapper):
+ return unittest.mock.patch(target, wraps=wrapper,
+ new_callable=unittest.mock.Mock)
+
+ with patch('os.WTERMSIG', self.WTERMSIG) as m_WTERMSIG, \
+ patch('os.WEXITSTATUS', self.WEXITSTATUS) as m_WEXITSTATUS, \
+ patch('os.WIFSIGNALED', self.WIFSIGNALED) as m_WIFSIGNALED, \
+ patch('os.WIFEXITED', self.WIFEXITED) as m_WIFEXITED, \
+ patch('os.waitpid', self.waitpid) as m_waitpid:
+ func(self, WaitPidMocks(m_waitpid,
+ m_WIFEXITED, m_WIFSIGNALED,
+ m_WEXITSTATUS, m_WTERMSIG,
+ ))
+ return wrapped_func
+
+ @waitpid_mocks
+ def test_sigchld(self, m):
# register a child
callback = unittest.mock.Mock()
@@ -782,33 +792,33 @@
self.watcher.add_child_handler(42, callback, 9, 10, 14)
self.assertFalse(callback.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# child is running
self.watcher._sig_chld()
self.assertFalse(callback.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# child terminates (returncode 12)
self.running = False
self.add_zombie(42, 12)
self.watcher._sig_chld()
- self.assertTrue(m_WIFEXITED.called)
- self.assertTrue(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertTrue(m.WIFEXITED.called)
+ self.assertTrue(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
callback.assert_called_once_with(42, 12, 9, 10, 14)
- m_WIFSIGNALED.reset_mock()
- m_WIFEXITED.reset_mock()
- m_WEXITSTATUS.reset_mock()
+ m.WIFSIGNALED.reset_mock()
+ m.WIFEXITED.reset_mock()
+ m.WEXITSTATUS.reset_mock()
callback.reset_mock()
# ensure that the child is effectively reaped
@@ -817,29 +827,24 @@
self.watcher._sig_chld()
self.assertFalse(callback.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WTERMSIG.called)
- m_WIFSIGNALED.reset_mock()
- m_WIFEXITED.reset_mock()
- m_WEXITSTATUS.reset_mock()
+ m.WIFSIGNALED.reset_mock()
+ m.WIFEXITED.reset_mock()
+ m.WEXITSTATUS.reset_mock()
# sigchld called again
self.zombies.clear()
self.watcher._sig_chld()
self.assertFalse(callback.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_sigchld_two_children(self, m_waitpid, m_WIFEXITED, m_WIFSIGNALED,
- m_WEXITSTATUS, m_WTERMSIG):
+ @waitpid_mocks
+ def test_sigchld_two_children(self, m):
callback1 = unittest.mock.Mock()
callback2 = unittest.mock.Mock()
@@ -850,10 +855,10 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# register child 2
with self.watcher:
@@ -861,20 +866,20 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# childen are running
self.watcher._sig_chld()
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# child 1 terminates (signal 3)
self.add_zombie(43, -3)
@@ -882,13 +887,13 @@
callback1.assert_called_once_with(43, -3, 7, 8)
self.assertFalse(callback2.called)
- self.assertTrue(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertTrue(m_WTERMSIG.called)
+ self.assertTrue(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertTrue(m.WTERMSIG.called)
- m_WIFSIGNALED.reset_mock()
- m_WIFEXITED.reset_mock()
- m_WTERMSIG.reset_mock()
+ m.WIFSIGNALED.reset_mock()
+ m.WIFEXITED.reset_mock()
+ m.WTERMSIG.reset_mock()
callback1.reset_mock()
# child 2 still running
@@ -896,10 +901,10 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# child 2 terminates (code 108)
self.add_zombie(44, 108)
@@ -908,13 +913,13 @@
callback2.assert_called_once_with(44, 108, 147, 18)
self.assertFalse(callback1.called)
- self.assertTrue(m_WIFEXITED.called)
- self.assertTrue(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertTrue(m.WIFEXITED.called)
+ self.assertTrue(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
- m_WIFSIGNALED.reset_mock()
- m_WIFEXITED.reset_mock()
- m_WEXITSTATUS.reset_mock()
+ m.WIFSIGNALED.reset_mock()
+ m.WIFEXITED.reset_mock()
+ m.WEXITSTATUS.reset_mock()
callback2.reset_mock()
# ensure that the children are effectively reaped
@@ -925,11 +930,11 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WTERMSIG.called)
- m_WIFSIGNALED.reset_mock()
- m_WIFEXITED.reset_mock()
- m_WEXITSTATUS.reset_mock()
+ m.WIFSIGNALED.reset_mock()
+ m.WIFEXITED.reset_mock()
+ m.WEXITSTATUS.reset_mock()
# sigchld called again
self.zombies.clear()
@@ -937,19 +942,13 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_sigchld_two_children_terminating_together(
- self, m_waitpid, m_WIFEXITED, m_WIFSIGNALED, m_WEXITSTATUS,
- m_WTERMSIG):
+ @waitpid_mocks
+ def test_sigchld_two_children_terminating_together(self, m):
callback1 = unittest.mock.Mock()
callback2 = unittest.mock.Mock()
@@ -960,10 +959,10 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# register child 2
with self.watcher:
@@ -971,20 +970,20 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# childen are running
self.watcher._sig_chld()
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# child 1 terminates (code 78)
# child 2 terminates (signal 5)
@@ -995,15 +994,15 @@
callback1.assert_called_once_with(45, 78, 17, 8)
callback2.assert_called_once_with(46, -5, 1147, 18)
- self.assertTrue(m_WIFSIGNALED.called)
- self.assertTrue(m_WIFEXITED.called)
- self.assertTrue(m_WEXITSTATUS.called)
- self.assertTrue(m_WTERMSIG.called)
+ self.assertTrue(m.WIFSIGNALED.called)
+ self.assertTrue(m.WIFEXITED.called)
+ self.assertTrue(m.WEXITSTATUS.called)
+ self.assertTrue(m.WTERMSIG.called)
- m_WIFSIGNALED.reset_mock()
- m_WIFEXITED.reset_mock()
- m_WTERMSIG.reset_mock()
- m_WEXITSTATUS.reset_mock()
+ m.WIFSIGNALED.reset_mock()
+ m.WIFEXITED.reset_mock()
+ m.WTERMSIG.reset_mock()
+ m.WEXITSTATUS.reset_mock()
callback1.reset_mock()
callback2.reset_mock()
@@ -1015,16 +1014,10 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WTERMSIG.called)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_sigchld_race_condition(
- self, m_waitpid, m_WIFEXITED, m_WIFSIGNALED, m_WEXITSTATUS,
- m_WTERMSIG):
+ @waitpid_mocks
+ def test_sigchld_race_condition(self, m):
# register a child
callback = unittest.mock.Mock()
@@ -1045,14 +1038,8 @@
self.assertFalse(callback.called)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_sigchld_replace_handler(
- self, m_waitpid, m_WIFEXITED, m_WIFSIGNALED, m_WEXITSTATUS,
- m_WTERMSIG):
+ @waitpid_mocks
+ def test_sigchld_replace_handler(self, m):
callback1 = unittest.mock.Mock()
callback2 = unittest.mock.Mock()
@@ -1063,10 +1050,10 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# register the same child again
with self.watcher:
@@ -1074,10 +1061,10 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# child terminates (signal 8)
self.running = False
@@ -1086,13 +1073,13 @@
callback2.assert_called_once_with(51, -8, 21)
self.assertFalse(callback1.called)
- self.assertTrue(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertTrue(m_WTERMSIG.called)
+ self.assertTrue(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertTrue(m.WTERMSIG.called)
- m_WIFSIGNALED.reset_mock()
- m_WIFEXITED.reset_mock()
- m_WTERMSIG.reset_mock()
+ m.WIFSIGNALED.reset_mock()
+ m.WIFEXITED.reset_mock()
+ m.WTERMSIG.reset_mock()
callback2.reset_mock()
# ensure that the child is effectively reaped
@@ -1102,15 +1089,10 @@
self.assertFalse(callback1.called)
self.assertFalse(callback2.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WTERMSIG.called)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_sigchld_remove_handler(self, m_waitpid, m_WIFEXITED,
- m_WIFSIGNALED, m_WEXITSTATUS, m_WTERMSIG):
+ @waitpid_mocks
+ def test_sigchld_remove_handler(self, m):
callback = unittest.mock.Mock()
# register a child
@@ -1119,19 +1101,19 @@
self.watcher.add_child_handler(52, callback, 1984)
self.assertFalse(callback.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# unregister the child
self.watcher.remove_child_handler(52)
self.assertFalse(callback.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# child terminates (code 99)
self.running = False
@@ -1141,13 +1123,8 @@
self.assertFalse(callback.called)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_sigchld_unknown_status(self, m_waitpid, m_WIFEXITED,
- m_WIFSIGNALED, m_WEXITSTATUS, m_WTERMSIG):
+ @waitpid_mocks
+ def test_sigchld_unknown_status(self, m):
callback = unittest.mock.Mock()
# register a child
@@ -1156,10 +1133,10 @@
self.watcher.add_child_handler(53, callback, -19)
self.assertFalse(callback.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# terminate with unknown status
self.zombies[53] = 1178
@@ -1167,14 +1144,14 @@
self.watcher._sig_chld()
callback.assert_called_once_with(53, 1178, -19)
- self.assertTrue(m_WIFEXITED.called)
- self.assertTrue(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertTrue(m.WIFEXITED.called)
+ self.assertTrue(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
callback.reset_mock()
- m_WIFEXITED.reset_mock()
- m_WIFSIGNALED.reset_mock()
+ m.WIFEXITED.reset_mock()
+ m.WIFSIGNALED.reset_mock()
# ensure that the child is effectively reaped
self.add_zombie(53, 101)
@@ -1183,13 +1160,8 @@
self.assertFalse(callback.called)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_remove_child_handler(self, m_waitpid, m_WIFEXITED,
- m_WIFSIGNALED, m_WEXITSTATUS, m_WTERMSIG):
+ @waitpid_mocks
+ def test_remove_child_handler(self, m):
callback1 = unittest.mock.Mock()
callback2 = unittest.mock.Mock()
callback3 = unittest.mock.Mock()
@@ -1221,8 +1193,8 @@
self.assertFalse(callback2.called)
callback3.assert_called_once_with(56, 2, 3)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_sigchld_unhandled_exception(self, m_waitpid):
+ @waitpid_mocks
+ def test_sigchld_unhandled_exception(self, m):
callback = unittest.mock.Mock()
# register a child
@@ -1231,7 +1203,7 @@
self.watcher.add_child_handler(57, callback)
# raise an exception
- m_waitpid.side_effect = ValueError
+ m.waitpid.side_effect = ValueError
with unittest.mock.patch.object(unix_events.logger,
"exception") as m_exception:
@@ -1239,15 +1211,8 @@
self.assertEqual(self.watcher._sig_chld(), None)
self.assertTrue(m_exception.called)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_sigchld_child_reaped_elsewhere(
- self, m_waitpid, m_WIFEXITED, m_WIFSIGNALED, m_WEXITSTATUS,
- m_WTERMSIG):
-
+ @waitpid_mocks
+ def test_sigchld_child_reaped_elsewhere(self, m):
# register a child
callback = unittest.mock.Mock()
@@ -1256,10 +1221,10 @@
self.watcher.add_child_handler(58, callback)
self.assertFalse(callback.called)
- self.assertFalse(m_WIFEXITED.called)
- self.assertFalse(m_WIFSIGNALED.called)
- self.assertFalse(m_WEXITSTATUS.called)
- self.assertFalse(m_WTERMSIG.called)
+ self.assertFalse(m.WIFEXITED.called)
+ self.assertFalse(m.WIFSIGNALED.called)
+ self.assertFalse(m.WEXITSTATUS.called)
+ self.assertFalse(m.WTERMSIG.called)
# child terminates
self.running = False
@@ -1268,13 +1233,13 @@
# waitpid is called elsewhere
os.waitpid(58, os.WNOHANG)
- m_waitpid.reset_mock()
+ m.waitpid.reset_mock()
# sigchld
with self.ignore_warnings:
self.watcher._sig_chld()
- callback.assert_called(m_waitpid)
+ callback.assert_called(m.waitpid)
if isinstance(self.watcher, unix_events.FastChildWatcher):
# here the FastChildWatche enters a deadlock
# (there is no way to prevent it)
@@ -1282,15 +1247,8 @@
else:
callback.assert_called_once_with(58, 255)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_sigchld_unknown_pid_during_registration(
- self, m_waitpid, m_WIFEXITED, m_WIFSIGNALED, m_WEXITSTATUS,
- m_WTERMSIG):
-
+ @waitpid_mocks
+ def test_sigchld_unknown_pid_during_registration(self, m):
# register two children
callback1 = unittest.mock.Mock()
callback2 = unittest.mock.Mock()
@@ -1310,15 +1268,8 @@
callback1.assert_called_once_with(591, 7)
self.assertFalse(callback2.called)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_set_loop(
- self, m_waitpid, m_WIFEXITED, m_WIFSIGNALED, m_WEXITSTATUS,
- m_WTERMSIG):
-
+ @waitpid_mocks
+ def test_set_loop(self, m):
# register a child
callback = unittest.mock.Mock()
@@ -1351,15 +1302,8 @@
callback.assert_called_once_with(60, 9)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_set_loop_race_condition(
- self, m_waitpid, m_WIFEXITED, m_WIFSIGNALED, m_WEXITSTATUS,
- m_WTERMSIG):
-
+ @waitpid_mocks
+ def test_set_loop_race_condition(self, m):
# register 3 children
callback1 = unittest.mock.Mock()
callback2 = unittest.mock.Mock()
@@ -1418,15 +1362,8 @@
self.assertFalse(callback2.called)
callback3.assert_called_once_with(622, 19)
- @unittest.mock.patch('os.WTERMSIG', wraps=WTERMSIG)
- @unittest.mock.patch('os.WEXITSTATUS', wraps=WEXITSTATUS)
- @unittest.mock.patch('os.WIFSIGNALED', wraps=WIFSIGNALED)
- @unittest.mock.patch('os.WIFEXITED', wraps=WIFEXITED)
- @unittest.mock.patch('os.waitpid', wraps=waitpid)
- def test_close(
- self, m_waitpid, m_WIFEXITED, m_WIFSIGNALED, m_WEXITSTATUS,
- m_WTERMSIG):
-
+ @waitpid_mocks
+ def test_close(self, m):
# register two children
callback1 = unittest.mock.Mock()
callback2 = unittest.mock.Mock()
--
Repository URL: http://hg.python.org/cpython
More information about the Python-checkins
mailing list