On Fri, Apr 18, 2008 at 3:36 PM, Don Dwiggins ddwiggins@advpubtech.com wrote:

I'm offering the following as an experience report and a draft of a howto article:

---

I've been developing an XML-RPC server using Twisted, and unit testing it with the distribution unittest, with each test connecting as a client and exercising a particular method. I ran into a problem, however:

My server is essentially a special-purpose front end to a database, providing limited access to it. Some of the server methods modify the database (I'm using adbapi for DB access). For unit testing purposes, I want each test to leave the test database unchanged when it's done. In unit testing stored procedures, for example, I create a connection in SetUp, use it in the test function to send a SQL "exec", and do a rollback in TearDown. Nice and easy, since I'm using the same connection throughout. In testing my server, however, I don't have access to the connection it used to access the database (and in fact shouldn't have, since my unit test functions are just clients to the server, and know nothing of the database itself).

This led me to the following approach, using Twisted's Trial extension of unittest. I converted my unittest module to be run under Trial, as follows: rather than running the server, the test module imports the server module, giving it access to the XMLRPC class itself, and the ability to directly call its methods. Since the server's methods return Deferreds, it's easy enough to call a method, then attach a callback to it that does the checking of its results. Here's an example:

---

import MyServer

class MyServerTests(unittest.TestCase) def setUp(self): # Instantiate object to be tested here self.srvr = MyServer.XMLRPCServer()

def testFrobulate(self):
    d = self.srvr.xmlrpc_frobulate(theFrobulatee)
    def checkResult(info):
        # Test that the frobulation occurred correctly
        pass
    d.addCallback(checkFrobulation)

def checkFrobulation(self, resultOfFrobulation):
    # Test whether it turned out OK

---

This works beautifully when frobulate doesn't modify the database; when it does, however, I have the same problem as before: the actual connection used is hidden in the guts of adbapi.

For this case, I changed the coding a bit to allow testing. Rather than ConnectionPool.runQuery(), I use .runInteraction(), passing a function that expects a DBAPI cursor as its first argument (when running in the server, the function will be called in the context of a subthread). The code in the server module then looks like this (I'm running on Py 2.4):

---

@defer.deferredGenerator
def xmlrpc_frobulate(theVictim):
    frobInProgress = defer.waitForDeferred(
        self._dbpool.runInteraction(
            self.frobulateInteraction, theVictim) )
    yield frobInProgress
    didItWork = frobInProgress.getResult()
    yield didItWork
    return

def frobulateInteraction(self, cursor, theVictim):
    cursor.execute("exec FrobulateOn " + theVictim)
    # Check the results, return True or False

---

Now, in the unit test, I can call self.srvr.frobulateInteraction like this:

def testFrobulate(self):
    self.cursor = self.connection.cursor()
    d = self.srvr.frobulateInteraction(self.cursor,
                                       self.theFrobulatee)
    def checkResult(info):
        # Test that the frobulation occurred correctly
        pass
    d.addCallback(checkFrobulation)

---

Now, since frobulateInteraction is using the connection from SetUp, the rollback in TearDown will restore the state of the DB.

In effect, by going "under the covers" of the server, I'm bypassing the parts of the server that are supplied by Twisted, and focusing on the code that I've written, which is exactly what I wanted to test.

Feedback solicited...

-- Don Dwiggins Advanced Publishing Technology

---

Twisted-Python mailing list Twisted-Python@twistedmatrix.com http://twistedmatrix.com/cgi-bin/mailman/listinfo/twisted-python

One question: are you really not returning each Deferred from your test methods, or was this a typo in the above?