Wednesday, April 7, 2010

Running a HTTP Server in a Thread

I'm building a shop using Django and Satchmo and I ran into an interesting problem while building tests for the Paypal payment module.

The user is sent off to Paypal to process the payment and are returned to the success page. Paypal notifies the site of payment through an 'ipn' url on the shop. Faking the call to the ipn url is easy enough in the test:

>>> postdata = dict(payment_status='Completed',
..., memo=memo, txn_id='1234',
>>> response ='/checkout/paypal/ipn/', postdata)

But the problem then arose because the ipn django view then requests verification from paypal:

def confirm_ipn_data(data, PP_URL):
    # data is the form data that was submitted to the IPN URL.

    newparams = {}
    for key in data.keys():
        newparams[key] = data[key]

    newparams['cmd'] = "_notify-validate"
    params = urlencode(newparams)

    req = urllib2.Request(PP_URL)
    req.add_header("Content-type", "application/x-www-form-urlencoded")
    fo = urllib2.urlopen(req, params)

    ret =
    if ret == "VERIFIED":"PayPal IPN data verification was successful.")
    else:"PayPal IPN data verification failed.")
        log.debug("HTTP code %s, response text: '%s'" % (fo.code, ret))
        return False

    return True

The problem here is opening the url using urllib2. I couldn't use the http://testserver/ that is 'running' in the Django test code. So I tried to start up a simple HTTPServer instance but this would lock up the testrunner. Finally then I came up with the following code which does the job nicely:

# simple server to use as fake verification server
from BaseHTTPServer import BaseHTTPRequestHandler,HTTPServer
from threading import Thread

class PaypalHandler(BaseHTTPRequestHandler):
    Simple server to return verified value for our paypal tests
    def do_POST(self):
        self.send_response(200, 'OK')
        self.send_header('Content-type', 'text/html')
        self.wfile.write( "VERIFIED" )

class RunThread(Thread):
    def __init__(self, port):
        self.server = HTTPServer(('', port), PaypalHandler)
    def run(self):

server_thread = None

def start_server(port):
    global server_thread
    server_thread = RunThread(port)

def stop_server():
    global server_thread

Now in my test setup I can call

def setUp(suite):

def tearDown(suite):

And it works as expected.