如何模拟线程中的内置模块
How to mock a builtin module in a Thread
问题如标题所示,我如何通过示例模拟 select.select 来测试我的线程 运行 功能。
测试功能失败
ready = select.select([self.sock], [], [], 5)
TypeError: fileno() returned a non-integer
打印类型给出
输入'builtin_function_or_method'
很明显 select.select 在线程范围内没有被模拟,而在测试中它是...(断言 isinstance)
import select
import threading
RECEIVE_BYTES = 256
class Bar(threading.Thread):
def __init__(self, sock):
threading.Thread.__init__(self)
self.sock = sock
def run(self):
print type(select.select)
ready = select.select([self.sock],[],[],5)
if ready[0]:
print self.sock.recv(RECEIVE_BYTES)
在另一个模块中测试如下
def test_run(self):
with patch("select.select"):
select.select.return_value = [True]
mock_sock = MagicMock()
foo = Bar(mock_sock)
assert isinstance(select.select, MagicMock)
foo.start()
测试是 运行 通过鼻子
简短的回答是在离开 with patch(...)
块之前调用 foo.join()
等待线程完成。该错误是由于在线程完成之前删除补丁造成的。
顺便说一下,如果你 post 可以是 运行 的例子,人们会更容易帮助你。您的示例不完整并且存在语法错误。
这是固定的测试。我添加了循环以便更容易重现错误。
import select
from mock import patch, MagicMock
from time import sleep
from scratch import Bar
IS_FIXED = True
def test_run():
for _ in range(20):
with patch("select.select"):
select.select.return_value = [True]
mock_sock = MagicMock()
foo = Bar(mock_sock)
assert isinstance(select.select, MagicMock)
foo.start()
if IS_FIXED:
foo.join()
sleep(0.1)
这里是 Bar
class 进行了一些语法修复。
import select
import threading
RECEIVE_BYTES = 256
class Bar(threading.Thread):
def __init__(self, sock):
threading.Thread.__init__(self)
self.sock = sock
def run(self):
print type(select.select)
ready = select.select([self.sock],[],[],5)
if ready[0]:
print self.sock.recv(RECEIVE_BYTES)
问题如标题所示,我如何通过示例模拟 select.select 来测试我的线程 运行 功能。 测试功能失败
ready = select.select([self.sock], [], [], 5)
TypeError: fileno() returned a non-integer
打印类型给出
输入'builtin_function_or_method'
很明显 select.select 在线程范围内没有被模拟,而在测试中它是...(断言 isinstance)
import select
import threading
RECEIVE_BYTES = 256
class Bar(threading.Thread):
def __init__(self, sock):
threading.Thread.__init__(self)
self.sock = sock
def run(self):
print type(select.select)
ready = select.select([self.sock],[],[],5)
if ready[0]:
print self.sock.recv(RECEIVE_BYTES)
在另一个模块中测试如下
def test_run(self):
with patch("select.select"):
select.select.return_value = [True]
mock_sock = MagicMock()
foo = Bar(mock_sock)
assert isinstance(select.select, MagicMock)
foo.start()
测试是 运行 通过鼻子
简短的回答是在离开 with patch(...)
块之前调用 foo.join()
等待线程完成。该错误是由于在线程完成之前删除补丁造成的。
顺便说一下,如果你 post 可以是 运行 的例子,人们会更容易帮助你。您的示例不完整并且存在语法错误。
这是固定的测试。我添加了循环以便更容易重现错误。
import select
from mock import patch, MagicMock
from time import sleep
from scratch import Bar
IS_FIXED = True
def test_run():
for _ in range(20):
with patch("select.select"):
select.select.return_value = [True]
mock_sock = MagicMock()
foo = Bar(mock_sock)
assert isinstance(select.select, MagicMock)
foo.start()
if IS_FIXED:
foo.join()
sleep(0.1)
这里是 Bar
class 进行了一些语法修复。
import select
import threading
RECEIVE_BYTES = 256
class Bar(threading.Thread):
def __init__(self, sock):
threading.Thread.__init__(self)
self.sock = sock
def run(self):
print type(select.select)
ready = select.select([self.sock],[],[],5)
if ready[0]:
print self.sock.recv(RECEIVE_BYTES)