源码: tests/local.py
# -.- coding:utf-8 -.-
import unittest
import threading
class TestLocal(unittest.TestCase):
def test_create_local_instance(self):
data = threading.local()
self.assertEqual(data.__dict__, {})
def test_add_kv_to_local_instance(self):
data = threading.local()
data.language = "python"
data.version = "2.7.13"
self.assertEqual(data.__dict__,
{"language": "python", "version": "2.7.13"})
def test_del_kv_to_local_instance(self):
data = threading.local()
data.language = "python"
data.version = "2.7.13"
del data.language
self.assertEqual(data.__dict__, {"version": "2.7.13"})
def test_cross_threads_access_local(self):
data = threading.local()
data.language = "python"
def other_thread():
# 在其他线程中, local 对象是一个空对象,
# 在该线程中创建的属性值, 其他线程(包括主线程)
# 都无法访问到, 只有当前线程能访问到.
#
# local 就好像是一个大字典, 每个线程一个命名空间,
# 表现形式: {"MainThread": {"language": "python"},
# "Thread-1": {"other_thread": 1}}
#
# 备注:
# local 只需在全局作用域中创建一次, 其他线程即可引用和访问.
# 无需在每个线程作用域中单独创建一次实例.
self.assertEqual(data.__dict__, {})
current_thread_id = threading.current_thread().ident
data.other_thread = current_thread_id
self.assertEqual(data.__dict__,
{"other_thread": current_thread_id})
t = threading.Thread(target=other_thread, args=())
t.start()
t.join()
self.assertEqual(data.__dict__, {"language": "python"})
测试: tests/main.py
import unittest
TEST_MODULE = [
"ln_threading.tests.local",
]
if __name__ == '__main__':
suite = unittest.defaultTestLoader.loadTestsFromNames(TEST_MODULE)
runner = unittest.TextTestRunner(verbosity=2)
runner.run(suite)
网友评论