[Python-Dev] unittest and sockets. Ugh!?

[Michael Gilfix]

  Could someone please offer some advice/comments on this?  While
restructuring test_socket.py, I've come across the following
(limitation?) problem with the unittest module. To test socket stuff,
I really need to use two separate processes or threads.  Since
fork() is much better supported, here's my attempt at porting the
_fileobject tests into unittest. I really like the structure of the
test (hopefully you guys agree) and this is how I'd like to lay it out
ideally for testing things like accept/connect, etc, using the various
layers of my inhertiance hierarchy.

  However, this test won't work because I get an error binding to the
socket in the setUp function because the socket is already taken. Is
this because unittest dispatches the tests at roughly the same
time? I'm not quite sure why this is failing in sequence (perhaps I'm
missing something). In addition, I thought that the setUp/tearDown
functions were shared between all tests within a class, not called
for each test but this does not seem true. If I want the setup to be
shared between all tests, do I have to override __init__? Another
issue is that unittest doesn't seem to like that I've forked. It
considers it to be the equivalent of two tests. Perhaps I shouldn't
care, provided that they all pass anyway.

  Any other stuff that I've seen that uses forking/threading doesn't
seem to use the unittest style framework. Perhaps I shouldn't be using
this and should just write outside of it? That would be a shame since
I like many of the features of the framework but some seem limiting.

                    -- Mike

=======================================================================

#!/usr/bin/env python

import unittest
import test_support

import socket
import os
import time

PORT = 50007
HOST = 'localhost'

class SocketTest(unittest.TestCase):

    def setUp(self):
        canfork = hasattr (os, 'fork')
        if not canfork:
            raise test_support.TestSkipped, \
                  "Platform does not support forking."

        # Use this to figure out who we are in the tests
        self.parent = os.fork()

        if self.parent:
            self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.s.bind((HOST, PORT))
            self.s.listen(1)
        else:
            self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        time.sleep(1) # So we can catch up

    def tearDown(self):
        self.s.close ()
        self.s = None

class SocketConnectedTest(SocketTest):

    SYNCH_MSG = 'Michael Gilfix was here'

    def setUp(self):
        SocketTest.setUp(self)
        if self.parent:
            conn, addr = self.s.accept()
            self.conn = conn
        else:
            self.s.connect((HOST, PORT))
            self.conn = self.s

    def tearDown(self):
        if self.parent:
            self.conn.close()
        self.conn = None
        SocketTest.tearDown(self)

    def synchronize(self):
        time.sleep(1)
        if self.parent:
            msg = self.conn.recv(len(self.SYNCH_MSG))
            self.assertEqual(msg, self.SYNCH_MSG, "Parent synchronization error")
            self.conn.send(msg)
        else:
            self.conn.send(msg)
            msg = self.conn.recv(len(self.SYNCH_MSG))
            self.assertEqual(msg, self.SYNCH_MSG, "Child synchronization error")
        time.sleep(1)

class FileObjectClassTestCase(SocketConnectedTest):

    def setUp(self):
        SocketConnectedTest.setUp(self)
        # Create a file object for both the parent/client processes
        self.f = socket._fileobject(self.conn, 'rb', 8192)

    def tearDown(self):
        self.f.close()
        SocketConnectedTest.tearDown(self)

    def testSmallRead(self):
        """Performing small read test."""
        if self.parent:
            first_seg = self.f.read(7)
            second_seg = self.f.read(25)
            msg = ''.join((first_seg, second_seg))
            self.assertEqual(msg, self.SYNCH_MSG, "Error performing small read.")
        else:
            self.f.write(self.SYNCH_MSG)
            self.f.flush()

    def testUnbufferedRead(self):
        """Performing unbuffered read test."""
        if self.parent:
            buf = ''
            while 1:
                char = self.f.read(1)
                self.failIf(not char, "Error performing unbuffered read.")
                buf += char
                if buf == self.SYNCH_MSG:
                    break
        else:
            self.f.write(self.SYNCH_MSG)
            self.f.flush()

    def testReadline(self):
        """Performing readline test."""
        if self.parent:
            line = self.f.readline()
            self.assertEqual(line, self.SYNCH_MSG, "Error performing readline.")
        else:
            self.f.write(self.SYNCH_MSG)
            self.f.flush()

def suite():
    suite = unitest.TestSuite()
    suite.addTest(unittest.makeSuite(FileObjectClassTestCase))
    return suite

if __name__ == '__main__':
    unittest.main()

--
Michael Gilfix
mgilfix@eecs.tufts.edu

For my gpg public key:
http://www.eecs.tufts.edu/~mgilfix/contact.html