# https://github.com/stripe/stripe-python # https://stripe.com/docs/api?lang=python#top import logging import json import re import unittest from datetime import timedelta from unittest import TestCase import stripe #django imports from django.conf import settings from django.core.mail import send_mail from django.http import HttpResponse from django.utils.timezone import now #regluit imports from regluit.payment import baseprocessor from regluit.payment.models import Account, Transaction, PaymentResponse from regluit.payment.parameters import ( PAYMENT_HOST_STRIPE, TRANSACTION_STATUS_ACTIVE, TRANSACTION_STATUS_COMPLETE, TRANSACTION_STATUS_ERROR, PAYMENT_TYPE_AUTHORIZATION, PAYMENT_TYPE_INSTANT, TRANSACTION_STATUS_CANCELED ) from regluit.payment.signals import transaction_charged, transaction_failed STRIPE_EVENT_TYPES = [ 'account.updated', 'account.application.deauthorized', 'balance.available', 'charge.succeeded', 'charge.failed', 'charge.refunded', 'charge.captured', 'charge.dispute.created', 'charge.dispute.updated', 'charge.dispute.closed', 'customer.created', 'customer.updated', 'customer.deleted', 'customer.card.created', 'customer.card.updated', 'customer.card.deleted', 'customer.source.created', 'customer.source.deleted', 'customer.source.expiring', 'customer.source.updated', 'customer.subscription.created', 'customer.subscription.updated', 'customer.subscription.deleted', 'customer.subscription.trial_will_end', 'customer.discount.created', 'customer.discount.updated', 'customer.discount.deleted', 'invoice.created', 'invoice.updated', 'invoice.payment_succeeded', 'invoice.payment_failed', 'invoiceitem.created', 'invoiceitem.updated', 'invoiceitem.deleted', 'plan.created', 'plan.updated', 'plan.deleted', 'coupon.created', 'coupon.deleted', 'transfer.created', 'transfer.updated', 'transfer.paid', 'transfer.failed', 'payment_method.attached', 'payment_method.card_automatically_updated', 'ping', ] logger = logging.getLogger(__name__) # https://stackoverflow.com/questions/2348317/how-to-write-a-pager-for-python-iterators/2350904#2350904 def grouper(iterable, page_size): page = [] for item in iterable: page.append(item) if len(page) == page_size: yield page page = [] if len(page) > 0: yield page class StripelibError(baseprocessor.ProcessorError): pass STRIPE_PK = settings.STRIPE_PK STRIPE_SK = settings.STRIPE_SK # set default stripe api_key to that of unglue.it stripe.api_key = STRIPE_SK # maybe we should be able to set this in django.settings... # to start with, let's try hard-coding the api_version # https://stripe.com/docs/upgrades?since=2012-07-09#api-changelog #API_VERSION = '2012-07-09' API_VERSION = '2013-02-13' stripe.api_version = API_VERSION # https://stripe.com/docs/testing TEST_CARDS = ( ('4242424242424242', 'Visa'), ('4012888888881881', 'Visa'), ('5555555555554444', 'MasterCard'), ('5105105105105100', 'MasterCard'), ('378282246310005', 'American Express'), ('371449635398431', 'American Express'), ('6011111111111117', 'Discover'), ('6011000990139424', 'Discover'), ('30569309025904', "Diner's Club"), ('38520000023237', "Diner's Club"), ('3530111333300000', 'JCB'), ('3566002020360505', 'JCB') ) ERROR_TESTING = dict(( ('ADDRESS1_ZIP_FAIL', ('4000000000000010', 'address_line1_check and address_zip_check will both fail')), ('ADDRESS1_FAIL', ('4000000000000028', 'address_line1_check will fail.')), ('ADDRESS_ZIP_FAIL', ('4000000000000036', 'address_zip_check will fail.')), ('CVC_CHECK_FAIL', ('4000000000000101', 'cvc_check will fail.')), ('BAD_ATTACHED_CARD', ('4000000000000341', 'Attaching this card to a Customer object will succeed, but attempts to charge the customer will fail.')), ('CHARGE_DECLINE', ('4000000000000002', 'Charges with this card will always be declined.')) )) CARD_FIELDS_TO_COMPARE = ('exp_month', 'exp_year', 'name', 'address_line1', 'address_line2', 'address_zip', 'address_state') # types of errors / when they can be handled #card_declined: Use this special card number - 4000000000000002. #incorrect_number: Use a number that fails the Luhn check, e.g. 4242424242424241. #invalid_expiry_month: Use an invalid month e.g. 13. #invalid_expiry_year: Use a year in the past e.g. 1970. #invalid_cvc: Use a two digit number e.g. 99. def filter_none(d): return dict([(k, v) for (k, v) in d.items() if v is not None]) # if you create a Customer object, then you'll be able to charge multiple times. # You can create a customer with a token. # https://en.wikipedia.org/wiki/Luhn_algorithm#Implementation_of_standard_Mod_10 def luhn_checksum(card_number): def digits_of(n): return [int(d) for d in str(n)] digits = digits_of(card_number) odd_digits = digits[-1::-2] even_digits = digits[-2::-2] checksum = 0 checksum += sum(odd_digits) for d in even_digits: checksum += sum(digits_of(d*2)) return checksum % 10 def is_luhn_valid(card_number): return luhn_checksum(card_number) == 0 # https://stripe.com/docs/tutorials/charges def card(number=TEST_CARDS[0][0], exp_month=1, exp_year=2030, cvc=None, name=None, address_line1=None, address_line2=None, address_zip=None, address_state=None, address_country=None): """Note: there is no place to enter address_city in the API""" card = { "number": number, "exp_month": int(exp_month), "exp_year": int(exp_year), "cvc": int(cvc) if cvc is not None else None, "name": name, "address_line1": address_line1, "address_line2": address_line2, "address_zip": address_zip, "address_state": address_state, "address_country": address_country } return filter_none(card) def _isListableAPIResource(x): """test whether x is an instance of the stripe.ListableAPIResource class""" try: return issubclass(x, stripe.abstract.ListableAPIResource) except: return False class StripeClient(object): def __init__(self, api_key=STRIPE_SK): self.api_key = api_key # key entities: Charge, Customer, Token, Event @property def charge(self): return stripe.Charge(api_key=self.api_key) @property def customer(self): return stripe.Customer(api_key=self.api_key) @property def token(self): return stripe.Token(api_key=self.api_key) @property def transfer(self): return stripe.Transfer(api_key=self.api_key) @property def event(self): return stripe.Event(api_key=self.api_key) def create_token(self, card): return stripe.Token(api_key=self.api_key).create(card=card) def create_customer(self, card=None, description=None, email=None, account_balance=None, plan=None, trial_end=None): """card is a dictionary or a token""" # https://stripe.com/docs/api?lang=python#create_customer customer = stripe.Customer(api_key=self.api_key).create( card=card, description=description, email=email, account_balance=account_balance, plan=plan, trial_end=trial_end ) # customer.id is useful to save in db return customer def create_charge(self, amount, currency="usd", customer=None, card=None, description=None): # https://stripe.com/docs/api?lang=python#create_charge # customer.id or card required but not both # charge the Customer instead of the card # amount in cents charge = stripe.Charge(api_key=self.api_key).create( amount=int(100*amount), # in cents currency=currency, customer=customer, card=card, description=description ) return charge def refund_charge(self, charge_id): # https://stripe.com/docs/api?lang=python#refund_charge ch = stripe.Charge(api_key=self.api_key).retrieve(charge_id) ch.refund() return ch def _all_objs(self, class_type, **kwargs): """a generic iterator for all classes of type stripe.ListableAPIResource""" # type=None, created=None, count=None, offset=0 # obj_type: one of 'Charge','Coupon','Customer', 'Event','Invoice', # 'InvoiceItem', 'Plan', 'Transfer' try: stripe_class = getattr(stripe, class_type) except: yield StopIteration else: if _isListableAPIResource(stripe_class): kwargs2 = kwargs.copy() kwargs2.setdefault('offset', 0) kwargs2.setdefault('count', 100) more_items = True while more_items: items = stripe_class(api_key=self.api_key).list(**kwargs2)['data'] for item in items: yield item if len(items) > 0: kwargs2['offset'] += len(items) else: more_items = False else: yield StopIteration def __getattribute__(self, name): """ handle list_* calls""" mapping = {'list_charges':"Charge", 'list_coupons': "Coupon", 'list_customers':"Customer", 'list_events':"Event", 'list_invoices':"Invoice", 'list_invoiceitems':"InvoiceItem", 'list_plans':"Plan", 'list_transfers':"Transfer"} if name in mapping.keys(): class_value = mapping[name] def list_events(**kwargs): for e in self._all_objs(class_value, **kwargs): yield e return list_events return object.__getattribute__(self, name) # can't test Transfer in test mode: "There are no transfers in test mode." # pledge scenario # bad card -- what types of erros to handle? # https://stripe.com/docs/api#errors # https://stripe.com/docs/api#event_types # events of interest -- especially ones that do not directly arise immediately (synchronously) # from something we do -- I think especially: charge.disputed # I think following (charge.succeeded, charge.failed, charge.refunded) pretty much sychronous # to our actions customer.created, customer.updated, customer.deleted # transfer # I expect the ones related to transfers all happen asynchronously: # transfer.created, transfer.updated, transfer.failed # When will the money I charge with Stripe end up in my bank account? # Every day, we transfer the money that you charged seven days previously? that is, # you receive the money for your March 1st charges on March 8th. # pending payments? # how to tell whether money transferred to bank account yet # best practices for calling Events -- not too often. # Errors we still need to catch: # # * invalid_number -- can't get stripe to generate for us. What it means: # # * that the card has been cancelled (or never existed to begin with # # * the card is technically correct (Luhn valid?) # # * the first 6 digits point to a valid bank # # * but the account number (the rest of the digits) doesn't correspond to a credit account # with that bank # * Brian of stripe.com suggests we could treat it the same way as we'd treat card_declined # # * processing_error: # # * means: something went wrong when stripe tried to make the charge (it could be that the # card's issuing bank is down, or our connection to the bank isn't working properly) # * we can retry -- e.g., a minute later, then 30 minutes, then an hour, 3 hours, a day. # * we shouldn't see processing_error very often # # * expired_card -- also not easily simulatable in test mode class StripeErrorTest(TestCase): """Make sure the exceptions returned by stripe act as expected""" def test_cc_test_numbers_luhn_valid(self): """Show that the test CC numbers supplied for testing as valid numbers are indeed Luhn valid""" self.assertTrue(all([is_luhn_valid(c[0]) for c in ERROR_TESTING.values()])) def test_good_token(self): """ verify normal operation """ sc = StripeClient() card1 = card(number=TEST_CARDS[0][0], exp_month=1, exp_year='2020', cvc='123', name='Don Giovanni', address_line1="100 Jackson St.", address_line2="", address_zip="94706", address_state="CA", address_country=None) # good card token1 = sc.create_token(card=card1) # use the token id -- which is what we get from JavaScript api -- and retrieve the token token2 = sc.token.retrieve(id=token1.id) self.assertEqual(token2.id, token1.id) # make sure token id has a form tok_ self.assertEqual(token2.id[:4], "tok_") # should be only test mode self.assertEqual(token2.livemode, False) # token hasn't been used yet self.assertEqual(token2.used, False) # test that card info matches up with what was fed in. for k in CARD_FIELDS_TO_COMPARE: self.assertEqual(token2.card[k], card1[k]) # last4 self.assertEqual(token2.card.last4, TEST_CARDS[0][0][-4:]) # fingerprint self.assertGreaterEqual(len(token2.card.fingerprint), 16) # now charge the token charge1 = sc.create_charge(10, 'usd', card=token2.id) self.assertEqual(charge1.amount, 1000) self.assertEqual(charge1.id[:3], "ch_") # dispute, failure_message, fee, fee_details self.assertEqual(charge1.dispute, None) self.assertEqual(charge1.failure_message, None) self.assertEqual(charge1.fee, 59) self.assertEqual(charge1.refunded, False) def test_error_creating_customer_with_declined_card(self): """Test whether we can get a charge decline error""" sc = StripeClient() card1 = card(number=ERROR_TESTING['CHARGE_DECLINE'][0]) try: cust1 = sc.create_customer(card=card1, description="This card should fail") self.fail("Attempt to create customer did not throw expected exception.") except stripe.error.CardError as e: self.assertEqual(e.code, "card_declined") self.assertEqual(e.args[0], "Your card was declined") def test_charge_bad_cust(self): # expect the card to be declined -- and for us to get CardError sc = StripeClient() # bad card card1 = card(number=ERROR_TESTING['BAD_ATTACHED_CARD'][0]) # attaching card should be ok cust1 = sc.create_customer(card=card1, description="test bad customer", email="rdhyee@gluejar.com") # trying to charge the card should fail self.assertRaises(stripe.error.CardError, sc.create_charge, 10, customer=cust1.id, description="$10 for bad cust") def test_bad_cc_number(self): """send a bad cc and should get an error when trying to create a token""" BAD_CC_NUM = '4242424242424241' # reason for decline is number is not Luhn valid self.assertFalse(is_luhn_valid(BAD_CC_NUM)) sc = StripeClient() card1 = card(number=BAD_CC_NUM, exp_month=1, exp_year=2020, cvc='123', name='Don Giovanni', address_line1="100 Jackson St.", address_line2="", address_zip="94706", address_state="CA", address_country=None) # good card try: token1 = sc.create_token(card=card1) self.fail("Attempt to create token with bad cc number did not throw expected exception.") except stripe.error.CardError as e: self.assertEqual(e.code, "incorrect_number") self.assertEqual(e.args[0], "Your card number is incorrect") def test_invalid_expiry_month(self): """Use an invalid month e.g. 13.""" sc = StripeClient() card1 = card(number=TEST_CARDS[0][0], exp_month=13, exp_year=2020, cvc='123', name='Don Giovanni', address_line1="100 Jackson St.", address_line2="", address_zip="94706", address_state="CA", address_country=None) try: token1 = sc.create_token(card=card1) self.fail("Attempt to create token with invalid expiry month did not throw expected exception.") except stripe.error.CardError as e: self.assertEqual(e.code, "invalid_expiry_month") self.assertEqual(e.args[0], "Your card's expiration month is invalid") def test_invalid_expiry_year(self): """Use a year in the past e.g. 1970.""" sc = StripeClient() card1 = card(number=TEST_CARDS[0][0], exp_month=12, exp_year=1970, cvc='123', name='Don Giovanni', address_line1="100 Jackson St.", address_line2="", address_zip="94706", address_state="CA", address_country=None) try: token1 = sc.create_token(card=card1) self.fail("Attempt to create token with invalid expiry year did not throw expected exception.") except stripe.error.CardError as e: self.assertEqual(e.code, "invalid_expiry_year") self.assertEqual(e.args[0], "Your card's expiration year is invalid") def test_invalid_cvc(self): """Use a two digit number e.g. 99.""" sc = StripeClient() card1 = card(number=TEST_CARDS[0][0], exp_month=12, exp_year=2020, cvc='99', name='Don Giovanni', address_line1="100 Jackson St.", address_line2="", address_zip="94706", address_state="CA", address_country=None) try: token1 = sc.create_token(card=card1) self.fail("Attempt to create token with invalid cvc did not throw expected exception.") except stripe.error.CardError as e: self.assertEqual(e.code, "invalid_cvc") self.assertEqual(e.args[0], "Your card's security code is invalid") def test_missing_card(self): """There is no card on a customer that is being charged""" sc = StripeClient() # create a Customer with no attached card cust1 = sc.create_customer(description="test cust w/ no card") try: sc.create_charge(10, customer=cust1.id, description="$10 for cust w/ no card") except stripe.error.CardError as e: self.assertEqual(e.code, "missing") self.assertEqual(e.args[0], "Cannot charge a customer that has no active card") class PledgeScenarioTest(TestCase): @classmethod def setUpClass(cls): cls._sc = StripeClient(api_key=STRIPE_SK) # valid card card0 = card() cls._good_cust = cls._sc.create_customer(card=card0, description="test good customer", email="raymond.yee@gmail.com") # bad card test_card_num_to_get_BAD_ATTACHED_CARD = ERROR_TESTING['BAD_ATTACHED_CARD'][0] card1 = card(number=test_card_num_to_get_BAD_ATTACHED_CARD) cls._cust_bad_card = cls._sc.create_customer(card=card1, description="test bad customer", email="rdhyee@gluejar.com") def test_charge_good_cust(self): charge = self._sc.create_charge(10, customer=self._good_cust.id, description="$10 for good cust") self.assertEqual(type(charge.id), str) # print out all the pieces of Customer and Charge objects print(dir(charge)) print(dir(self._good_cust)) def test_error_creating_customer_with_declined_card(self): # should get a CardError upon attempt to create Customer with this card _card = card(number=card(ERROR_TESTING['CHARGE_DECLINE'][0])) self.assertRaises(stripe.error.CardError, self._sc.create_customer, card=_card) def test_charge_bad_cust(self): # expect the card to be declined -- and for us to get CardError self.assertRaises(stripe.error.CardError, self._sc.create_charge, 10, customer=self._cust_bad_card.id, description="$10 for bad cust") @classmethod def tearDownClass(cls): # clean up stuff we create in test -- right now list current objects pass #cls._good_cust.delete() class StripePaymentRequest(baseprocessor.BasePaymentRequest): """so far there is no need to have a separate class here""" pass class Processor(baseprocessor.Processor): def make_account(self, user=None, token=None, email=None): """returns a payment.models.Account based on stripe token and user""" if token is None or len(token) == 0: raise StripelibError("input token is None", None) sc = StripeClient() # create customer and charge id and then charge the customer try: if user: customer = sc.create_customer(card=token, description=user.username, email=user.email) else: customer = sc.create_customer(card=token, description='anonymous user', email=email) except stripe.error.StripeError as e: raise StripelibError(e.args, e) account = Account(host=PAYMENT_HOST_STRIPE, account_id=customer.id, card_last4=customer.active_card.last4, card_type=customer.active_card.type, card_exp_month=customer.active_card.exp_month, card_exp_year=customer.active_card.exp_year, card_fingerprint=customer.active_card.fingerprint, card_country=customer.active_card.country, user=user ) if user and user.profile.account: user.profile.account.deactivate() account.save() account.recharge_failed_transactions() else: account.save() return account class Preapproval(StripePaymentRequest, baseprocessor.Processor.Preapproval): def __init__(self, transaction, amount, expiry=None, return_url=None, paymentReason=""): # set the expiration date for the preapproval if not passed in. # This is what the paypal library does self.transaction = transaction now_val = now() if expiry is None: expiry = now_val + timedelta(days=settings.PREAPPROVAL_PERIOD) transaction.date_authorized = now_val transaction.date_expired = expiry # let's figure out what part of transaction can be used to store info # try placing charge id in transaction.pay_key # need to set amount # how does transaction.max_amount get set? # -- coming from /pledge/xxx/ -> manager.process_transaction # max_amount is set -- but I don't think we need it for stripe # ASSUMPTION: a user has any given moment one and only one active payment Account account = transaction.user.profile.account if not account: logger.warning("user %s has no active payment account", transaction.user) raise StripelibError("user %s has no active payment account" % transaction.user) logger.info("user: %s customer.id is %s", transaction.user, account.account_id) # settings to apply to transaction for TRANSACTION_STATUS_ACTIVE # should approved be set to False and wait for a webhook? transaction.approved = True transaction.type = PAYMENT_TYPE_AUTHORIZATION transaction.host = PAYMENT_HOST_STRIPE transaction.status = TRANSACTION_STATUS_ACTIVE transaction.preapproval_key = account.account_id transaction.currency = 'USD' transaction.amount = amount transaction.save() def key(self): return self.transaction.preapproval_key def next_url(self): """return None because no redirection to stripe is required""" return None class Pay(StripePaymentRequest, baseprocessor.Processor.Pay): ''' The pay function generates a redirect URL to approve the transaction If the transaction has a null user (is_anonymous), then a token musr be supplied ''' def __init__(self, transaction, return_url=None, amount=None, paymentReason="", token=None): self.transaction = transaction self.url = return_url now_val = now() transaction.date_authorized = now_val # ASSUMPTION: a user has any given moment one and only one active payment Account if token: # user is anonymous try: account = transaction.get_payment_class().make_account( token=token, email=transaction.receipt) except StripelibError as e: self.errorMessage = str(e) return else: account = transaction.user.profile.account if not account: logger.warning("user %s has no active payment account", transaction.user) raise StripelibError("user %s has no active payment account" % transaction.user) logger.info("user: %s customer.id is %s", transaction.user, account.account_id) # settings to apply to transaction for TRANSACTION_STATUS_ACTIVE # should approved be set to False and wait for a webhook? transaction.approved = True transaction.type = PAYMENT_TYPE_INSTANT transaction.host = PAYMENT_HOST_STRIPE transaction.preapproval_key = account.account_id transaction.currency = 'USD' transaction.amount = amount transaction.save() # execute the transaction p = transaction.get_payment_class().Execute(transaction) if p.success() and not p.error(): transaction.pay_key = p.key() transaction.save() else: self.errorMessage = p.errorMessage #pass error message up logger.info("execute_transaction Error: %s", p.error_string()) def amount(self): return self.transaction.amount def key(self): return self.transaction.pay_key def next_url(self): return self.url class Execute(StripePaymentRequest): ''' The Execute function attempts to charge the credit card of stripe Customer associated with user connected to transaction. ''' def __init__(self, transaction=None): self.transaction = transaction # make sure transaction hasn't already been executed if transaction.status == TRANSACTION_STATUS_COMPLETE: return # make sure we are dealing with a stripe transaction if transaction.host != PAYMENT_HOST_STRIPE: raise StripelibError( "transaction.host %s is not the expected %s".format( transaction.host, PAYMENT_HOST_STRIPE) ) sc = StripeClient() # look first for transaction.user.profile.account.account_id try: customer_id = transaction.user.profile.account.account_id except: customer_id = transaction.preapproval_key if customer_id is not None: try: # useful things to put in description: transaction.id, transaction.user.id, # customer_id, transaction.amount charge = sc.create_charge( transaction.amount, customer=customer_id, description=json.dumps( {"t.id": transaction.id, "email": transaction.user.email if transaction.user else transaction.receipt, "cus.id":customer_id, "tc.id": transaction.campaign.id if transaction.campaign else '0', "amount": float(transaction.amount)} ) ) except stripe.error.StripeError as e: # what to record in terms of errors? (error log?) # use PaymentResponse to store error r = PaymentResponse.objects.create(api="stripelib.Execute", correlation_id=None, timestamp=now(), info=e.args[0], status=TRANSACTION_STATUS_ERROR, transaction=transaction) transaction.status = TRANSACTION_STATUS_ERROR self.errorMessage = e.args # manager puts this on transaction transaction.save() # fire off the fact that transaction failed -- should actually do so # only if not a transient error # if card_declined or expired card, ask user to update account if isinstance(e, stripe.error.CardError) and e.code in ( 'card_declined', 'expired_card', 'incorrect_number', 'processing_error'): transaction_failed.send(sender=self, transaction=transaction) # otherwise, report exception to us else: logger.exception("transaction id %s, exception: %s", transaction.id, e.args) else: self.charge = charge transaction.status = TRANSACTION_STATUS_COMPLETE transaction.pay_key = charge.id transaction.date_payment = now() transaction.save() # fire signal for sucessful transaction transaction_charged.send(sender=self, transaction=transaction) else: # nothing to charge raise StripelibError( "No customer id available to charge for transaction {0}".format(transaction.id), None) def api(self): return "Base Pay" def key(self): # IN paypal land, our key is updated from a preapproval to a pay key here, # just return the existing key return self.transaction.pay_key class PreapprovalDetails(StripePaymentRequest): ''' Get details about an authorized token This api must set 4 different class variables to work with the code in manager.py status - one of the global transaction status codes approved - boolean value currency - not used in this API, but we can get some more info via other APIs - TODO amount - not used in this API, but we can get some more info via other APIs - TODO ''' def __init__(self, transaction): self.transaction = transaction self.status = self.transaction.status if self.status == TRANSACTION_STATUS_CANCELED: self.approved = False else: self.approved = True # Set the other fields that are expected. We don't have values for these now, # so just copy the transaction self.currency = transaction.currency self.amount = transaction.amount def ProcessIPN(self, request): # retrieve the request's body and parse it as JSON in, e.g. Django try: event_json = json.loads(request.body) except ValueError as e: # not able to parse request.body -- throw a "Bad Request" error logger.warning("Non-json being sent to Stripe IPN: %s", e) return HttpResponse(status=400) else: # now parse out pieces of the webhook event_id = event_json.get("id") # use Stripe to ask for details -- ignore what we're sent for security sc = StripeClient() try: event = sc.event.retrieve(event_id) except stripe.error.InvalidRequestError: logger.warning("Invalid Event ID: %s", event_id) return HttpResponse(status=400) else: event_type = event.get("type") if event_type not in STRIPE_EVENT_TYPES: logger.warning("Unrecognized Stripe event type %s for event %s", event_type, event_id) # is this the right code to respond with? return HttpResponse(status=400) # https://stripe.com/docs/api?lang=python#event_types -- type to delegate things # parse out type as resource.action try: (resource, action) = re.match(r"^(.+)\.([^\.]*)$", event_type).groups() except Exception as e: logger.warning("Parsing of event_type into resource, action failed: %s", e) return HttpResponse(status=400) try: ev_object = event.data.object except Exception as e: logger.warning("attempt to retrieve event object failed: %s", e) return HttpResponse(status=400) if event_type == 'account.updated': # should we alert ourselves? # how about account.application.deauthorized ? pass elif resource == 'charge': # we need to handle: succeeded, failed, refunded, disputed if action == 'succeeded': # TO DO: delete this logic since we don't do anything # but look up transaction. logger.info("charge.succeeded webhook for %s", ev_object.get("id")) # try to parse description of object to pull related transaction if any # wrapping this in a try statement because it possible that we have a # charge.succeeded outside context of unglue.it try: charge_meta = json.loads(ev_object["description"]) transaction = Transaction.objects.get(id=charge_meta["t.id"]) # now check that account associated with the transaction matches # ev.data.object.id, t.pay_key if ev_object.id == transaction.pay_key: logger.info("ev_object.id == transaction.pay_key: %s", ev_object.id) else: logger.warning("ev_object.id %s != transaction.pay_key %s", ev_object.id, transaction.pay_key) except Exception as e: logger.warning(e) elif action == 'failed': # TO DO: delete this logic since we don't do anything but # look up transaction. logger.info("charge.failed webhook for %s", ev_object.get("id")) try: charge_meta = json.loads(ev_object["description"]) transaction = Transaction.objects.get(id=charge_meta["t.id"]) # now check that account associated with the transaction matches # ev.data.object.id, t.pay_key if ev_object.id == transaction.pay_key: logger.info("ev_object.id == transaction.pay_key: %s", ev_object.id) else: logger.warning("ev_object.id %s != transaction.pay_key %s", ev_object.id, transaction.pay_key) except Exception as e: logger.warning(e) elif action == 'refunded': pass elif action == 'disputed': pass else: # unexpected pass elif resource == 'customer': if action == 'created': # test application: email support # do we have a flag to indicate production vs non-production? # -- or does it matter? # email support whenever a new Customer created # -- we probably want to replace this with some other # more useful long tem action. send_mail( "Stripe Customer (id {0}; description: {1}) created".format( ev_object.get("id"), ev_object.get("description")), "Stripe Customer email: {0}".format(ev_object.get("email")), "notices@gluejar.com", ["unglueit@ebookfoundation.org"]) logger.info("email sent for customer.created for %s", ev_object.get("id")) # handle updated, deleted else: pass else: # other events pass return HttpResponse("event_id: {0} event_type: {1}".format(event_id, event_type)) def suite(): testcases = [PledgeScenarioTest, StripeErrorTest] #testcases = [StripeErrorTest] suites = unittest.TestSuite([ unittest.TestLoader().loadTestsFromTestCase(testcase) for testcase in testcases ]) return suites # IPNs/webhooks: https://stripe.com/docs/webhooks # how to use pending_webhooks ? # all events # https://stripe.com/docs/api?lang=python#list_events if __name__ == '__main__': #unittest.main() suites = suite() #suites = unittest.defaultTestLoader.loadTestsFromModule(__import__('__main__')) unittest.TextTestRunner().run(suites)