Permalink
Browse files

Merge pull request #15 from jasonculverhouse/master

Added signals for recurring_payment_skipped and recurring_payment_failed
  • Loading branch information...
dcramer committed Jan 30, 2012
2 parents 626254f + 8f58976 commit f32bef52f74599dffacd977718df58f01c55e57e
@@ -38,7 +38,11 @@ def send_signals(self):
recurring_payment.send(sender=self)
elif self.is_recurring_cancel():
recurring_cancel.send(sender=self)
- # Subscription signals:
+ elif self.is_recurring_skipped():
+ recurring_skipped.send(sender=self)
+ elif self.is_recurring_failed():
+ recurring_failed.send(sender=self)
+ # Subscription signals:
else:
if self.is_subscription_cancellation():
subscription_cancel.send(sender=self)
@@ -30,4 +30,8 @@
# recurring_payment
recurring_payment = Signal()
-recurring_cancel = Signal()
+recurring_cancel = Signal()
+
+recurring_skipped = Signal()
+
+recurring_failed = Signal()
@@ -6,7 +6,8 @@
from paypal.standard.models import ST_PP_CANCELLED
from paypal.standard.ipn.models import PayPalIPN
from paypal.standard.ipn.signals import (payment_was_successful,
- payment_was_flagged)
+ payment_was_flagged, recurring_skipped, recurring_failed,
+ recurring_create, recurring_payment, recurring_cancel)
IPN_POST_PARAMS = {
@@ -55,12 +56,41 @@ def setUp(self):
# Monkey patch over PayPalIPN to make it get a VERFIED response.
self.old_postback = PayPalIPN._postback
PayPalIPN._postback = lambda self: "VERIFIED"
+
+ self.payment_was_successful_receivers = payment_was_successful.receivers
+ self.payment_was_flagged_receivers = payment_was_flagged.receivers
+ self.recurring_skipped_receivers = recurring_skipped.receivers
+ self.recurring_failed_receivers = recurring_failed.receivers
+ self.recurring_create_receivers = recurring_create.receivers
+ self.recurring_failed_receivers = recurring_failed.receivers
+ self.recurring_payment_receivers = recurring_payment.receivers
+ self.recurring_cancel_receivers = recurring_cancel.receivers
+
+ payment_was_successful.receivers = []
+ payment_was_flagged.receivers = []
+ recurring_skipped.receivers = []
+ recurring_failed.receivers = []
+ recurring_create.receivers = []
+ recurring_failed.receivers = []
+ recurring_payment.receivers = []
+ recurring_cancel.receivers = []
+
def tearDown(self):
settings.DEBUG = self.old_debug
PayPalIPN._postback = self.old_postback
+
+ payment_was_successful.receivers =self.payment_was_successful_receivers
+ payment_was_flagged.receivers = self.payment_was_flagged_receivers
+ recurring_skipped.receivers = self.recurring_skipped_receivers
+ recurring_failed.receivers = self.recurring_failed_receivers
+ recurring_create.receivers = self.recurring_create_receivers
+ recurring_failed.receivers = self.recurring_failed_receivers
+ recurring_payment.receivers = self.recurring_payment_receivers
+ recurring_cancel.receivers = self.recurring_cancel_receivers
+
- def assertGotSignal(self, signal, flagged):
+ def assertGotSignal(self, signal, flagged, params=IPN_POST_PARAMS):
# Check the signal was sent. These get lost if they don't reference self.
self.got_signal = False
self.signal_obj = None
@@ -70,7 +100,7 @@ def handle_signal(sender, **kwargs):
self.signal_obj = sender
signal.connect(handle_signal)
- response = self.client.post("/ipn/", IPN_POST_PARAMS)
+ response = self.client.post("/ipn/", params)
self.assertEqual(response.status_code, 200)
ipns = PayPalIPN.objects.all()
self.assertEqual(len(ipns), 1)
@@ -123,3 +153,76 @@ def test_duplicate_txn_id(self):
ipn_obj = PayPalIPN.objects.order_by('-created_at', '-pk')[0]
self.assertEqual(ipn_obj.flag, True)
self.assertEqual(ipn_obj.flag_info, "Duplicate txn_id. (51403485VH153354B)")
+
+ def test_recurring_payment_skipped_ipn(self):
+ update = {
+ "recurring_payment_id": "BN5JZ2V7MLEV4",
+ "txn_type": "recurring_payment_skipped",
+ "txn_id": ""
+ }
+ params = IPN_POST_PARAMS.copy()
+ params.update(update)
+
+ self.assertGotSignal(recurring_skipped, False, params)
+
+ def test_recurring_payment_failed_ipn(self):
+ update = {
+ "recurring_payment_id": "BN5JZ2V7MLEV4",
+ "txn_type": "recurring_payment_failed",
+ "txn_id": ""
+ }
+ params = IPN_POST_PARAMS.copy()
+ params.update(update)
+
+ self.assertGotSignal(recurring_failed, False, params)
+
+ def test_recurring_payment_create_ipn(self):
+ update = {
+ "recurring_payment_id": "BN5JZ2V7MLEV4",
+ "txn_type": "recurring_payment_profile_created",
+ "txn_id": ""
+ }
+ params = IPN_POST_PARAMS.copy()
+ params.update(update)
+
+ self.assertGotSignal(recurring_create, False, params)
+
+ def test_recurring_payment_cancel_ipn(self):
+ update = {
+ "recurring_payment_id": "BN5JZ2V7MLEV4",
+ "txn_type": "recurring_payment_profile_cancel",
+ "txn_id": ""
+ }
+ params = IPN_POST_PARAMS.copy()
+ params.update(update)
+
+ self.assertGotSignal(recurring_cancel, False, params)
+
+ def test_recurring_payment_ipn(self):
+ """
+ The wat the code is written in
+ PayPalIPN.send_signals the recurring_payment
+ will never be sent because the paypal ipn
+ contains a txn_id, if this test failes you
+ might break some compatibility
+ """
+ update = {
+ "recurring_payment_id": "BN5JZ2V7MLEV4",
+ "txn_type": "recurring_payment",
+ }
+ params = IPN_POST_PARAMS.copy()
+ params.update(update)
+
+ self.got_signal = False
+ self.signal_obj = None
+
+ def handle_signal(sender, **kwargs):
+ self.got_signal = True
+ self.signal_obj = sender
+ recurring_payment.connect(handle_signal)
+
+ response = self.client.post("/ipn/", params)
+ self.assertEqual(response.status_code, 200)
+ ipns = PayPalIPN.objects.all()
+ self.assertEqual(len(ipns), 1)
+ self.assertFalse(self.got_signal)
@@ -220,6 +220,12 @@ def is_recurring_payment(self):
def is_recurring_cancel(self):
return self.txn_type == "recurring_payment_profile_cancel"
+ def is_recurring_skipped(self):
+ return self.txn_type == "recurring_payment_skipped"
+
+ def is_recurring_failed(self):
+ return self.txn_type == "recurring_payment_failed"
+
def set_flag(self, info, code=None):
"""Sets a flag on the transaction and also sets a reason."""
self.flag = True

0 comments on commit f32bef5

Please sign in to comment.