I ran into an issue with the socket module of the python standard library. This always comes as a surprise to me when I find a problem with something as mature as python. But it happens.

My own issue involved having a python based daemon running that does HTTP requests (using httplib) to another service. This daemon is restarted gracefully by sending SIGTERM, which it catches with a signal handler, finishes up what it was doing and exits. The problem arises if it receives a signal while in a system call, for example while receiving the response from an HTTP request. To correct behavior is to attempt the system call again, however the actual system call is abstracted away, so the caller, or even httplib can’t re-try.

The crux of the issue is the function readline() provided by a fileobject socket wrapper in socket.py

self._rbuf = StringIO()  # reset _rbuf.  we consume it via buf.
                data = None
                recv = self._sock.recv
                while data != "\n":
                    data = recv(1)
                    if not data:
                        break
                    buffers.append(data)
                return "".join(buffers)

I’m not the first to find this, as this issue even has a patch. But, due to the “test needed” status, it’s been siting there getting no attention for quite a while. Well I want it fixed, so let’s try to write a regression test!

The first step was to apply this patch to an appropriate development branch:

svn co http://svn.python.org/projects/python/branches/release26-maint python26
  cd python26/Lib
  patch -p0 < ~/socket.py.diff

Now it turned out, this didn’t apply cleanly, as the patch was from an earlier version. But it was easy enough to fix.

Secondly, I need to a test case to Lib/test/test_socket.py

There is already a test case for normal behavior of fileobject, however causing a real socket to generate a EINTR isn’t exactly easy. But I just need to test the error handling, this is unit test. Perfect case for using a mock object. Now there arn’t any handy mock object libraries in the standard python distribution, so i’ll just keep it simple:

class MockSocket(object):
            def __init__(self):
                # Build a generator that returns functions that we'll call and return for each
                # call to recv()
                def raise_error():
                    raise socket.error(errno.EINTR)
                self._step = iter([
                    lambda : "This is the first line\nAnd the sec",
                    raise_error,
                    lambda : "ond line is here\n",
                    lambda : None,
                ])

            def recv(self, size):
                return self._step.next()()

Now when I create my test case, I’ll just pass this mock socket in and call readline on it.

class FileObjectInterruptedTestCase(unittest.TestCase):
    """Test that the file object correctly handles being interrupted by a signal."""
    def setUp(self):
      ... create my mock socket ...

    def test(self):
        fo = socket._fileobject(self._mock_sock)
        self.assertEquals(fo.readline(), "This is the first line\n")
        self.assertEquals(fo.readline(), "And the second line is here\n")

Now to find out if this test case will allow this fix to be included……