def test_sigchld_child_reaped_elsewhere(self, m):
        # register a child
        callback = mock.Mock()

        with self.watcher:
            self.running = True
            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)

        # child terminates
        self.running = False
        self.add_zombie(58, 4)

        # waitpid is called elsewhere
        os.waitpid(58, os.WNOHANG)

        m.waitpid.reset_mock()

        # sigchld
        with self.ignore_warnings:
            self.watcher._sig_chld()

        if isinstance(self.watcher, asyncio.FastChildWatcher):
            # here the FastChildWatche enters a deadlock
            # (there is no way to prevent it)
            self.assertFalse(callback.called)
        else:
            callback.assert_called_once_with(58, 255) 

def test_close(self, m):
        # register two children
        callback1 = mock.Mock()

        with self.watcher:
            self.running = True
            # child 1 terminates
            self.add_zombie(63, 9)
            # other child terminates
            self.add_zombie(65, 18)
            self.watcher._sig_chld()

            self.watcher.add_child_handler(63, callback1)
            self.watcher.add_child_handler(64, callback1)

            self.assertEqual(len(self.watcher._callbacks), 1)
            if isinstance(self.watcher, asyncio.FastChildWatcher):
                self.assertEqual(len(self.watcher._zombies), 1)

            with mock.patch.object(
                    self.loop,
                    "remove_signal_handler") as m_remove_signal_handler:

                self.watcher.close()

                m_remove_signal_handler.assert_called_once_with(
                    signal.SIGCHLD)
                self.assertFalse(self.watcher._callbacks)
                if isinstance(self.watcher, asyncio.FastChildWatcher):
                    self.assertFalse(self.watcher._zombies) 

def create_watcher(self):
        return asyncio.FastChildWatcher() 

def test_get_child_watcher_after_set(self):
        policy = self.create_policy()
        watcher = asyncio.FastChildWatcher()

        policy.set_child_watcher(watcher)
        self.assertIs(policy._watcher, watcher)
        self.assertIs(watcher, policy.get_child_watcher()) 

def main(cls, loop=None, argv=sys.argv):
        """
        Runs cli commands in asyncio loop and outputs in appropriate format
        """
        if loop is None:
            loop = asyncio.get_event_loop()
            # In Python 3.8 ThreadedChildWatcher becomes the default which
            # should work fine for us. However, in Python 3.7 SafeChildWatcher
            # is the default and may cause BlockingIOErrors when many
            # subprocesses are created
            # https://docs.python.org/3/library/asyncio-policy.html#asyncio.FastChildWatcher
            if (
                sys.version_info.major == 3
                and sys.version_info.minor == 7
                and sys.platform != "win32"
            ):
                watcher = asyncio.FastChildWatcher()
                asyncio.set_child_watcher(watcher)
                watcher.attach_loop(loop)
        result = None
        try:
            result = loop.run_until_complete(cls._main(*argv[1:]))
        except KeyboardInterrupt:  # pragma: no cover
            pass  # pragma: no cover
        loop.run_until_complete(loop.shutdown_asyncgens())
        loop.close() 

def test_sigchld_child_reaped_elsewhere(self, m):
        # register a child
        callback = mock.Mock()

        with self.watcher:
            self.running = True
            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)

        # child terminates
        self.running = False
        self.add_zombie(58, 4)

        # waitpid is called elsewhere
        os.waitpid(58, os.WNOHANG)

        m.waitpid.reset_mock()

        # sigchld
        with self.ignore_warnings:
            self.watcher._sig_chld()

        if isinstance(self.watcher, asyncio.FastChildWatcher):
            # here the FastChildWatche enters a deadlock
            # (there is no way to prevent it)
            self.assertFalse(callback.called)
        else:
            callback.assert_called_once_with(58, 255) 

def test_close(self, m):
        # register two children
        callback1 = mock.Mock()

        with self.watcher:
            self.running = True
            # child 1 terminates
            self.add_zombie(63, 9)
            # other child terminates
            self.add_zombie(65, 18)
            self.watcher._sig_chld()

            self.watcher.add_child_handler(63, callback1)
            self.watcher.add_child_handler(64, callback1)

            self.assertEqual(len(self.watcher._callbacks), 1)
            if isinstance(self.watcher, asyncio.FastChildWatcher):
                self.assertEqual(len(self.watcher._zombies), 1)

            with mock.patch.object(
                    self.loop,
                    "remove_signal_handler") as m_remove_signal_handler:

                self.watcher.close()

                m_remove_signal_handler.assert_called_once_with(
                    signal.SIGCHLD)
                self.assertFalse(self.watcher._callbacks)
                if isinstance(self.watcher, asyncio.FastChildWatcher):
                    self.assertFalse(self.watcher._zombies) 

def create_watcher(self):
        return asyncio.FastChildWatcher() 

def test_get_child_watcher_after_set(self):
        policy = self.create_policy()
        watcher = asyncio.FastChildWatcher()

        policy.set_child_watcher(watcher)
        self.assertIs(policy._watcher, watcher)
        self.assertIs(watcher, policy.get_child_watcher()) 

def test_sigchld_child_reaped_elsewhere(self, m):
        # register a child
        callback = mock.Mock()

        with self.watcher:
            self.running = True
            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)

        # child terminates
        self.running = False
        self.add_zombie(58, 4)

        # waitpid is called elsewhere
        os.waitpid(58, os.WNOHANG)

        m.waitpid.reset_mock()

        # sigchld
        with self.ignore_warnings:
            self.watcher._sig_chld()

        if isinstance(self.watcher, asyncio.FastChildWatcher):
            # here the FastChildWatche enters a deadlock
            # (there is no way to prevent it)
            self.assertFalse(callback.called)
        else:
            callback.assert_called_once_with(58, 255) 

def test_close(self, m):
        # register two children
        callback1 = mock.Mock()

        with self.watcher:
            self.running = True
            # child 1 terminates
            self.add_zombie(63, 9)
            # other child terminates
            self.add_zombie(65, 18)
            self.watcher._sig_chld()

            self.watcher.add_child_handler(63, callback1)
            self.watcher.add_child_handler(64, callback1)

            self.assertEqual(len(self.watcher._callbacks), 1)
            if isinstance(self.watcher, asyncio.FastChildWatcher):
                self.assertEqual(len(self.watcher._zombies), 1)

            with mock.patch.object(
                    self.loop,
                    "remove_signal_handler") as m_remove_signal_handler:

                self.watcher.close()

                m_remove_signal_handler.assert_called_once_with(
                    signal.SIGCHLD)
                self.assertFalse(self.watcher._callbacks)
                if isinstance(self.watcher, asyncio.FastChildWatcher):
                    self.assertFalse(self.watcher._zombies) 

def create_watcher(self):
        return asyncio.FastChildWatcher() 

def test_get_child_watcher_after_set(self):
        policy = self.create_policy()
        watcher = asyncio.FastChildWatcher()

        policy.set_child_watcher(watcher)
        self.assertIs(policy._watcher, watcher)
        self.assertIs(watcher, policy.get_child_watcher()) 

def test_sigchld_child_reaped_elsewhere(self, m):
        # register a child
        callback = mock.Mock()

        with self.watcher:
            self.running = True
            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)

        # child terminates
        self.running = False
        self.add_zombie(58, 4)

        # waitpid is called elsewhere
        os.waitpid(58, os.WNOHANG)

        m.waitpid.reset_mock()

        # sigchld
        with self.ignore_warnings:
            self.watcher._sig_chld()

        if isinstance(self.watcher, asyncio.FastChildWatcher):
            # here the FastChildWatche enters a deadlock
            # (there is no way to prevent it)
            self.assertFalse(callback.called)
        else:
            callback.assert_called_once_with(58, 255) 

def test_close(self, m):
        # register two children
        callback1 = mock.Mock()

        with self.watcher:
            self.running = True
            # child 1 terminates
            self.add_zombie(63, 9)
            # other child terminates
            self.add_zombie(65, 18)
            self.watcher._sig_chld()

            self.watcher.add_child_handler(63, callback1)
            self.watcher.add_child_handler(64, callback1)

            self.assertEqual(len(self.watcher._callbacks), 1)
            if isinstance(self.watcher, asyncio.FastChildWatcher):
                self.assertEqual(len(self.watcher._zombies), 1)

            with mock.patch.object(
                    self.loop,
                    "remove_signal_handler") as m_remove_signal_handler:

                self.watcher.close()

                m_remove_signal_handler.assert_called_once_with(
                    signal.SIGCHLD)
                self.assertFalse(self.watcher._callbacks)
                if isinstance(self.watcher, asyncio.FastChildWatcher):
                    self.assertFalse(self.watcher._zombies) 

def create_watcher(self):
        return asyncio.FastChildWatcher() 

def test_get_child_watcher_after_set(self):
        policy = self.create_policy()
        watcher = asyncio.FastChildWatcher()

        policy.set_child_watcher(watcher)
        self.assertIs(policy._watcher, watcher)
        self.assertIs(watcher, policy.get_child_watcher())