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()
No comments:
Post a Comment