def test_supervisor_shutdown(self):
        """ Test shutdown joining/forcing with timeouts. """
        sup = PyonThreadManager()
        sup.start()

        import gevent
        self.assertIs(time.sleep, gevent.hub.sleep)

        # Test that it takes at least the given timeout to join_children, but not much more
        proc_sleep_secs, proc_count = 0.01, 5
        [sup.spawn(time.sleep, seconds=proc_sleep_secs) for i in xrange(5)]
        elapsed = sup.shutdown(2*proc_sleep_secs)
        # MM, 1/12: Ok, I loosened the timing boundaries. Do the tests still work?
        # Enabled 0.2s of slack for all tests

        self.assertLess(elapsed - proc_sleep_secs, 0.2)

        # this could be trouble
        self.assertLess(elapsed, 0.2 + proc_sleep_secs*3)

        # Test that a small timeout forcibly shuts down without waiting
        wait_secs = 0.0001
        [sup.spawn(time.sleep, seconds=proc_sleep_secs) for i in xrange(5)]
        elapsed = sup.shutdown(wait_secs)
        self.assertLess(elapsed - wait_secs, 0.2)

        # this could be trouble too
        self.assertLess(elapsed, 0.2 + proc_sleep_secs)

        # Test that no timeout waits until all finished
        [sup.spawn(time.sleep, seconds=proc_sleep_secs) for i in xrange(5)]
        elapsed = sup.shutdown()
        self.assertLess(elapsed - proc_sleep_secs, 0.2)
dataset/ETHPy150Open ooici/pyon/pyon/core/test/test_thread.py/ProcessTest.test_supervisor_shutdown/original