python--线程event

源码: tests/event.py

# -.- coding:utf-8 -.-
import unittest
import threading


class TestEvent(unittest.TestCase):

    def test_create_event_instance(self):
        event = threading.Event()
        err_msg = u"Event对象默认应该是False."
        self.assertFalse(event.is_set(), err_msg)

    def test_set_true_to_event_instance(self):
        event = threading.Event()
        event.set()

        err_msg = u"Event被set()之后, 状态应该是True."
        self.assertTrue(event.is_set(), err_msg)

    def test_set_false_to_event_instance(self):
        event = threading.Event()
        event.set()
        self.assertTrue(event.is_set())

        event.clear()
        self.assertFalse(event.is_set())

    def test_sub_pub(self):
        # event是最原始和最简单的事件通知机制.
        # event通知机制的原理是, 其他线程调用event.wait()来等待通知.
        # event通过set()方法把flag改为True, 然后通知所有waiters去做其他事情.

        event = threading.Event()
        result = []

        def consumer(the_event):
            the_event.wait()
            result.append(threading.current_thread().ident)

        t = threading.Thread(target=consumer, args=(event, ))
        t.start()
        self.assertEqual(len(result), 0)

        event.set()
        t.join()
        self.assertEqual(len(result), 1)

 
 

测试: tests/main.py

import unittest


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


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