Add coingate
Alice
6 years ago
parent
98e8a508b6
commit
c90827878e
@ -0,0 +1,131 @@
|
||||
import string
|
||||
import random
|
||||
import requests
|
||||
import logging
|
||||
|
||||
from django.shortcuts import redirect
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
from django.urls import reverse
|
||||
from django.conf import settings as project_settings
|
||||
|
||||
from .base import BackendBase
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
prng = random.SystemRandom()
|
||||
|
||||
|
||||
def generate_token(length=16):
|
||||
charset = string.digits + string.ascii_letters
|
||||
return ''.join([prng.choice(charset) for _ in range(length)])
|
||||
|
||||
|
||||
class CoinGateBackend(BackendBase):
|
||||
backend_id = 'coingate'
|
||||
backend_verbose_name = _("CoinGate")
|
||||
backend_display_name = _("Cryptocurrencies")
|
||||
backend_has_recurring = False
|
||||
|
||||
def __init__(self, settings):
|
||||
self.api_token = settings.get('API_TOKEN')
|
||||
if not self.api_token:
|
||||
return
|
||||
|
||||
self.currency = settings.get('CURRENCY', 'EUR')
|
||||
self.title = settings.get('TITLE', 'VPN Payment')
|
||||
|
||||
if settings.get('SANDBOX'):
|
||||
self.api_base = "https://api-sandbox.coingate.com"
|
||||
else:
|
||||
default_base = "https://api.coingate.com"
|
||||
self.api_base = settings.get('API_BASE', default_base)
|
||||
|
||||
self.backend_enabled = True
|
||||
|
||||
def _post(self, endpoint, **kwargs):
|
||||
headers = {
|
||||
'Authorization': 'Token ' + self.api_token,
|
||||
}
|
||||
url = self.api_base + endpoint
|
||||
response = requests.post(url, headers=headers, **kwargs)
|
||||
response.raise_for_status()
|
||||
j = response.json()
|
||||
return j
|
||||
|
||||
def new_payment(self, payment):
|
||||
root_url = project_settings.ROOT_URL
|
||||
assert root_url
|
||||
|
||||
token = generate_token()
|
||||
|
||||
order = self._post('/v2/orders', data={
|
||||
'order_id': str(payment.id),
|
||||
'price_amount': payment.amount / 100,
|
||||
'price_currency': self.currency,
|
||||
'receive_currency': self.currency,
|
||||
'title': self.title,
|
||||
'callback_url': root_url + reverse('payments:cb_coingate', args=(payment.id,)),
|
||||
'cancel_url': root_url + reverse('payments:cancel', args=(payment.id,)),
|
||||
'success_url': root_url + reverse('payments:view', args=(payment.id,)),
|
||||
'token': token,
|
||||
})
|
||||
|
||||
url = order['payment_url']
|
||||
|
||||
payment.backend_data['coingate_id'] = order['id']
|
||||
payment.backend_data['coingate_url'] = url
|
||||
payment.backend_data['coingate_token'] = token
|
||||
payment.save()
|
||||
|
||||
return redirect(url)
|
||||
|
||||
def callback(self, payment, request):
|
||||
post_data = request.POST
|
||||
|
||||
# Verify callback authenticity
|
||||
|
||||
saved_token = payment.backend_data.get('coingate_token')
|
||||
if not saved_token:
|
||||
logger.debug("payment does not have a coingate_token")
|
||||
return False
|
||||
|
||||
token = post_data.get('token')
|
||||
if token != saved_token:
|
||||
logger.debug("unexpected token (%r != %r)", token, saved_token)
|
||||
return False
|
||||
|
||||
order_id = post_data.get('order_id')
|
||||
if order_id != str(payment.id):
|
||||
logger.debug("unexpected order_id (%r != %r)", order_id, str(payment.id))
|
||||
return False
|
||||
|
||||
# Handle event
|
||||
status = post_data.get('status')
|
||||
if status == 'new' or status == 'pending':
|
||||
payment.update_status('new')
|
||||
elif status == 'confirming':
|
||||
payment.update_status('new', _("Confirming transaction"))
|
||||
elif status == 'paid':
|
||||
if payment.status in {'new', 'cancelled', 'error'}:
|
||||
# we don't have the exact converted amount, but it's not
|
||||
# important. settings to the requested amount for consistency
|
||||
payment.paid_amount = payment.amount
|
||||
payment.confirm()
|
||||
elif status == 'invalid' or status == 'expired':
|
||||
if payment.status != 'confirmed':
|
||||
payment.update_status('cancelled')
|
||||
elif status == 'refunded':
|
||||
if payment.status == 'confirmed':
|
||||
payment.refund()
|
||||
else:
|
||||
logger.debug("unexpected payment status: %r", status)
|
||||
return False
|
||||
|
||||
payment.save()
|
||||
return True
|
||||
|
||||
def get_ext_url(self, payment):
|
||||
if not payment.backend_extid:
|
||||
return None
|
||||
return 'https://dashboard.stripe.com/payments/%s' % payment.backend_extid
|
||||
|
||||
|
||||
@ -0,0 +1,75 @@
|
||||
from datetime import timedelta
|
||||
|
||||
from django.test import TestCase, RequestFactory
|
||||
from django.http import HttpResponseRedirect
|
||||
from django.contrib.auth.models import User
|
||||
|
||||
from payments.models import Payment
|
||||
from payments.backends import CoinGateBackend
|
||||
|
||||
|
||||
class CoinGateBackendTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user('test', 'test_user@example.com', None)
|
||||
|
||||
self.backend_settings = dict(
|
||||
API_TOKEN='test',
|
||||
TITLE='Test Title',
|
||||
CURRENCY='EUR',
|
||||
)
|
||||
|
||||
def test_payment(self):
|
||||
payment = Payment.objects.create(
|
||||
user=self.user,
|
||||
time=timedelta(days=30),
|
||||
backend_id='coingate',
|
||||
amount=300
|
||||
)
|
||||
|
||||
def fake_post(_backend, *, data={}):
|
||||
self.assertEqual(data['order_id'], '1')
|
||||
self.assertEqual(data['price_amount'], 3.0)
|
||||
self.assertEqual(data['price_currency'], 'EUR')
|
||||
self.assertEqual(data['receive_currency'], 'EUR')
|
||||
return {'id': 42, 'payment_url': 'http://testtoken/'}
|
||||
|
||||
with self.settings(ROOT_URL='root'):
|
||||
backend = CoinGateBackend(self.backend_settings)
|
||||
backend._post = fake_post
|
||||
redirect = backend.new_payment(payment)
|
||||
|
||||
self.assertIsInstance(redirect, HttpResponseRedirect)
|
||||
self.assertEqual(redirect.url, 'http://testtoken/')
|
||||
self.assertEqual(payment.backend_data.get('coingate_id'), 42)
|
||||
|
||||
|
||||
# Test a standard successful payment callback flow
|
||||
|
||||
def post_callback(status):
|
||||
callback_data = {
|
||||
'token': payment.backend_data['coingate_token'],
|
||||
'order_id': str(payment.id),
|
||||
'status': status,
|
||||
}
|
||||
ipn_url = '/payments/callback/coingate/%d' % payment.id
|
||||
ipn_request = RequestFactory().post(
|
||||
ipn_url,
|
||||
data=callback_data)
|
||||
return backend.callback(payment, ipn_request)
|
||||
|
||||
r = post_callback('pending')
|
||||
self.assertTrue(r)
|
||||
self.assertEqual(payment.status, 'new')
|
||||
|
||||
r = post_callback('confirming')
|
||||
self.assertTrue(r)
|
||||
self.assertEqual(payment.status, 'new')
|
||||
|
||||
r = post_callback('paid')
|
||||
self.assertTrue(r)
|
||||
self.assertEqual(payment.status, 'confirmed')
|
||||
self.assertEqual(payment.paid_amount, 300)
|
||||
|
||||
time_left_s = self.user.vpnuser.time_left.total_seconds()
|
||||
self.assertAlmostEqual(time_left_s, payment.time.total_seconds(), delta=60)
|
||||
|
||||
Loading…
Reference in New Issue