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',
... invoice=order.id, memo=memo, txn_id='1234',
... mc_gross=order.total)
>>> response = client.post('/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 = fo.read()
if ret == "VERIFIED":
log.info("PayPal IPN data verification was successful.")
else:
log.info("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.end_headers()
self.wfile.write( "VERIFIED" )
class RunThread(Thread):
def __init__(self, port):
Thread.__init__(self)
self.server = HTTPServer(('127.0.0.1', port), PaypalHandler)
def run(self):
self.server.serve_forever()
server_thread = None
def start_server(port):
global server_thread
server_thread = RunThread(port)
server_thread.start()
def stop_server():
global server_thread
server_thread.server.socket.close()
Now in my test setup I can call
def setUp(suite):
testing.start_server(10080)
def tearDown(suite):
testing.stop_server()
