def test_watch_server_pids(self):
        class MockOs(object):
            WNOHANG = os.WNOHANG

            def __init__(self, pid_map=None):
                if pid_map is None:
                    pid_map = {}
                self.pid_map = {}
                for pid, v in pid_map.items():
                    self.pid_map[pid] = (x for x in v)

            def waitpid(self, pid, options):
                try:
                    rv = next(self.pid_map[pid])
                except StopIteration:
                    raise OSError(errno.ECHILD, os.strerror(errno.ECHILD))
                except KeyError:
                    raise OSError(errno.ESRCH, os.strerror(errno.ESRCH))
                if isinstance(rv, Exception):
                    raise rv
                else:
                    return rv

        class MockTime(object):
            def __init__(self, ticks=None):
                self.tock = time()
                if not ticks:
                    ticks = []

                self.ticks = (t for t in ticks)

            def time(self):
                try:
                    self.tock += next(self.ticks)
                except StopIteration:
                    self.tock += 1
                return self.tock

            def sleep(*args):
                return

        class MockServer(object):

            def __init__(self, pids, run_dir=manager.RUN_DIR, zombie=0):
                self.heartbeat = (pids for _ in range(zombie))

            def get_running_pids(self):
                try:
                    rv = next(self.heartbeat)
                    return rv
                except __HOLE__:
                    return {}

        _orig_os = manager.os
        _orig_time = manager.time
        _orig_server = manager.Server
        try:
            manager.time = MockTime()
            manager.os = MockOs()
            # this server always says it's dead when you ask for running pids
            server = MockServer([1])
            # list of pids keyed on servers to watch
            server_pids = {
                server: [1],
            }
            # basic test, server dies
            gen = manager.watch_server_pids(server_pids)
            expected = [(server, 1)]
            self.assertEqual([x for x in gen], expected)
            # start long running server and short interval
            server = MockServer([1], zombie=15)
            server_pids = {
                server: [1],
            }
            gen = manager.watch_server_pids(server_pids)
            self.assertEqual([x for x in gen], [])
            # wait a little longer
            gen = manager.watch_server_pids(server_pids, interval=15)
            self.assertEqual([x for x in gen], [(server, 1)])
            # zombie process
            server = MockServer([1], zombie=200)
            server_pids = {
                server: [1],
            }
            # test weird os error
            manager.os = MockOs({1: [OSError()]})
            gen = manager.watch_server_pids(server_pids)
            self.assertRaises(OSError, lambda: [x for x in gen])
            # test multi-server
            server1 = MockServer([1, 10], zombie=200)
            server2 = MockServer([2, 20], zombie=8)
            server_pids = {
                server1: [1, 10],
                server2: [2, 20],
            }
            pid_map = {
                1: [None for _ in range(10)],
                2: [None for _ in range(8)],
                20: [None for _ in range(4)],
            }
            manager.os = MockOs(pid_map)
            gen = manager.watch_server_pids(server_pids,
                                            interval=manager.KILL_WAIT)
            expected = [
                (server2, 2),
                (server2, 20),
            ]
            self.assertEqual([x for x in gen], expected)

        finally:
            manager.os = _orig_os
            manager.time = _orig_time
            manager.Server = _orig_server
dataset/ETHPy150Open openstack/swift/test/unit/common/test_manager.py/TestManagerModule.test_watch_server_pids