python--线程enumerate

源码: tests/enumerate.py

# -.- coding:utf-8 -.-
import time
import unittest
import threading
from multiprocessing import current_process

py3 = hasattr(threading, "main_thread") is True
py2 = not py3


class TestEnumerate(unittest.TestCase):

    def test_one_thread(self):
        # 测试用例1: 查看线程总数
        enum = threading.enumerate()
        enum_array = list(enum)
        enum_len = len(enum_array)
        self.assertEqual(enum_len, 1)

    def test_pid_ne_tid(self):
        # 进程ID不等于线程ID, 线程ID隶属于进程ID,
        # 线程并不存在于windows的任务管理器或linux的ps进程管理器中.
        # 线程存在于进程的逻辑管理区, 也就是enumerate中.
        cp = current_process()
        tid = list(threading.enumerate())[0].ident
        self.assertNotEqual(cp.pid, tid)

    def test_multi_threads(self):
        # 启动5个线程, 并且验证enumerate()方法是否符合预期.
        def other_thread():
            time.sleep(10)

        number = 5
        ts = [threading.Thread(target=other_thread, args=())
              for i in range(number)]
        [t.start() for t in ts]

        threads_len = len(list(threading.enumerate()))
        self.assertEqual(threads_len, 5+1)
        [i.join() for i in ts]

    def test_main_thread_py2(self):
        # python2 和 python3 提取主线程.
        main_thread = [t.name == "MainThread" for t in threading.enumerate()]
        self.assertTrue(bool(main_thread))

    @unittest.skipIf(py2, u"python2不支持main_thread方法")
    def test_main_thread_py3(self):
        # python3 提取主线程.
        self.assertEqual(threading.main_thread(), "MainThread")

测试: tests/main.py

import unittest


TEST_MODULE = [
    "ln_threading.tests.enumerate",
]


if __name__ == '__main__':
    suite = unittest.defaultTestLoader.loadTestsFromNames(TEST_MODULE)
    runner = unittest.TextTestRunner(verbosity=2)
    runner.run(suite)