test_process.py 文件源码

def test_num_threads(self):
        # on certain platforms such as Linux we might test for exact
        # thread number, since we always have with 1 thread per process,
        # but this does not apply across all platforms (OSX, Windows)
        p = psutil.Process()
        if OPENBSD:
            try:
                step1 = p.num_threads()
            except psutil.AccessDenied:
                raise unittest.SkipTest("on OpenBSD this requires root access")
        else:
            step1 = p.num_threads()

        thread = ThreadTask()
        thread.start()
        try:
            step2 = p.num_threads()
            self.assertEqual(step2, step1 + 1)
            thread.stop()
        finally:
            if thread._running:
                thread.stop()