如何在线程中模拟内置模块

stu*_*ioj 5 python unit-testing mocking python-multithreading python-2.7

问题与标题一样,如何通过示例模拟 select.select 以测试我的线程运行功能。测试功能失败

ready = select.select([self.sock], [], [], 5)
TypeError: fileno() returned a non-integer
Run Code Online (Sandbox Code Playgroud)

和类型打印给出

输入'builtin_function_or_method'

所以很明显 select.select 在线程的范围内没有被模拟,而在测试中它是......(assert 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)
Run Code Online (Sandbox Code Playgroud)

在另一个模块中测试如下

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()
Run Code Online (Sandbox Code Playgroud)

测试通过鼻子运行

Don*_*kby 2

简短的答案是foo.join()在离开块之前调用等待线程完成with patch(...)。该错误是由于在线程完成之前删除补丁而引起的。

顺便说一句,如果您发布一个可以运行的示例,人们会更容易帮助您。您的示例不完整并且存在语法错误。

这是固定的测试。我添加了循环以使其更容易重现错误。

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)
Run Code Online (Sandbox Code Playgroud)

这是Bar带有一些语法修复的类。

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)
Run Code Online (Sandbox Code Playgroud)