How to mock a buffered file handle for unit testing

Question

I want this to work correctly without buffering:

$ tail -f logfile | python parser.py

Previously I used:

for line in sys.stdin:
    parse(line)

But now I have implemented https://stackoverflow.com/a/18235323/604515 :

while True:
    line = sys.stdin.readline()
    if not line: break # EOF
    parse(line)

Question: I cannot think of a way to unit test my change without sleep(). How do I simulate a buffered filehandle? I can easily mock stdin, but then it will never buffer, so that won't prove anything.

Edit: I've looked at StringIO but that doesn't seem to have the same buffering behaviour as a normal filehandle.

I want a unit test that verifies that it never buffers, in case someone changes the code back to ```for line in stdin``` — Willem, Mar 26 '15 at 12:01
You can make your own custom class and set sys.stdout = class — Padraic Cunningham, Mar 26 '15 at 12:15
On a side note `for line in iter(sys.stdin.readline,"")` will do the same as your while loop — Padraic Cunningham, Mar 26 '15 at 12:59
If someone changes the line to `for line in sys.stdin` without doing some magic with subprocess or pseudo terminals then it is always going to buffer output. — Padraic Cunningham, Mar 26 '15 at 13:01

score 1 · Accepted Answer · answered Mar 27 '15 at 08:45

First of all consider to parameterize your function by pass the stream. You can use default to take sys.stdout. Your method become something like that:

def my_reader(feed=sys.stdin):
    while True:
        line = feed.readline()
        if not line:
            break
        parse(line)

Now you can mock your feed by mock framework and patch your parse method. Unfortunately you cannot patch directly sys.stdin.readline() method because is a read only attribute.

Now the steps in your test are: create a mock of stdin in setUp() and in your tests configure the it and do your test. Follow an example.

#If you trust parse and you have no other way to test parse instead check its calls you need to patch it
@patch(__name__ + ".parse")
class MyTestCase(unittest.TestCase):
    def setUp(self):
        #Our 'feed'
        self.mock_feed = MagicMock()
        #readline mock
        self.mock_readline = self.mock_feed.readline

    def test_my_reader_empty(self, mock_parse):
        #setting up an empty stream
        self.mock_readline.return_value = ""
        my_reader(self.mock_feed)
        #Our test is that parse was never called
        self.assertFalse(mock_parse.called)

    def test_my_reader_some_lines(self, mock_parse):
        #setting up lines sequence and EOF
        self.mock_readline.side_effect = ["hi,\n", "I'm your standard input\n", "cheers", ""]
        my_reader(self.mock_feed)
        #Call parse with every lines
        mock_parse.assert_has_calls([call("hi,\n"), call("I'm your standard input\n"), call("cheers")])

Just a note: I patched the class instead patching all method as documented here.

How to mock a buffered file handle for unit testing

1 Answers1