CODEWITHSUNDEEP
pythonpython-2.7unit-testingmockingpython-multithreading
studioj
studioj

Reputation: 1390

How to mock a builtin module in a Thread

Question is as in the heading, how can I mock select.select by example to test my thread run function. the test function fails with 

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

and the type print gives

type 'builtin_function_or_method'

so clearly select.select is NOT mocked in the thread's scope while in the test it is...(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)

the test is as follows in another module

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()

tests are run via nose

Upvotes: 5

Views: 4385

Answers (1)

Don Kirkby
Don Kirkby

Reputation: 56590

The short answer is to call foo.join() to wait for the thread to finish before leaving the with patch(...) block. The error was caused by removing the patch before the thread had finished.

By the way, it's much easier for people to help you if you post an example that can be run. Your example was incomplete and had syntax errors.

Here's the fixed up test. I added the loop to make it easier to reproduce the error.

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)

And here's the Bar class with some syntax fixes.

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)

Upvotes: 2

Related Questions