To receive notifications about scheduled maintenance, please subscribe to the mailing-list gitlab-operations@sympa.ethz.ch. You can subscribe to the mailing-list at https://sympa.ethz.ch

Commit f533ff14 authored by Mathis Dedial's avatar Mathis Dedial Committed by Alexander Dietmüller
Browse files

Backend: Implement payment processing with Stripe including tests

parent d6dfbc6d
...@@ -31,7 +31,9 @@ from backend.signups import ( ...@@ -31,7 +31,9 @@ from backend.signups import (
patched_course, patched_course,
block_course_deletion, block_course_deletion,
mark_as_paid, mark_as_paid,
mark_as_unpaid,
) )
from backend.payments import create_payment
def create_app(config_file=None, **kwargs): def create_app(config_file=None, **kwargs):
...@@ -85,7 +87,9 @@ def create_app(config_file=None, **kwargs): ...@@ -85,7 +87,9 @@ def create_app(config_file=None, **kwargs):
application.on_updated_courses += patched_course application.on_updated_courses += patched_course
application.on_delete_item_courses += block_course_deletion application.on_delete_item_courses += block_course_deletion
application.on_insert_payments += create_payment
application.on_inserted_payments += mark_as_paid application.on_inserted_payments += mark_as_paid
application.on_deleted_item_payments += mark_as_unpaid
return application return application
......
"""Handle Stripe API interactions."""
from flask import abort
import stripe
from backend.settings import COURSE_PRICE, STRIPE_API_KEY
from backend.signups import mark_as_paid
stripe.api_key = STRIPE_API_KEY
def create_payment(payments):
"""Create a charge with the Stripe API.
on_insert_payments hook
items is guaranteed to only contain items that have passed validation.
"""
# We don't want payment to be a list
payment = payments[0]
# We don't accept free money
if not payment['signups']:
abort(400, 'Payment must be for at least one signup')
# For now we assume a constant price per course
amount = len(payment['signups']) * COURSE_PRICE
# If no token is set, we're dealing with an admin payment
# Thus, no call to the Stripe API is made
if not payment.get('token'):
return True
# Create a new charge
try:
charge = stripe.Charge.create(
amount=amount,
currency='CHF',
source=payment['token'],
)
except stripe.error.CardError:
# Something's wrong with the card
abort(422, 'Card declined')
except stripe.error.RateLimitError:
# Too many requests made to the API too quickly
abort(429, 'Too many requests')
except stripe.error.InvalidRequestError:
# Invalid parameters were supplied to Stripe's API
abort(500, 'Invalid call to Stripe API')
except stripe.error.AuthenticationError:
# Authentication with Stripe's API failed
abort(500, 'Authentication with Stripe API failed')
except stripe.error.APIConnectionError:
# Network communication with Stripe failed
abort(500, 'Failed to connect to Stripe API')
except stripe.error.StripeError:
# Some error to do with Stripe
abort(500, 'Stripe payment processing failed')
# Charge succeeded, save the charge id in the database
payment['charge_id'] = charge.id
# Mark corresponding signups as paid
mark_as_paid([payment])
return True
...@@ -35,6 +35,15 @@ RESOURCE_METHODS = ['GET', 'POST'] ...@@ -35,6 +35,15 @@ RESOURCE_METHODS = ['GET', 'POST']
ITEM_METHODS = ['GET', 'PATCH', 'DELETE'] ITEM_METHODS = ['GET', 'PATCH', 'DELETE']
# Stripe API Key
# TODO: Not a good idea to keep this in the repo
STRIPE_API_KEY = 'sk_test_KUiZO8E2VKGMmm94u4t5YPnL'
# Price per course in "rappen"
COURSE_PRICE = 1000
# ISO 8601 time format instead of rfc1123 # ISO 8601 time format instead of rfc1123
DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
...@@ -217,11 +226,17 @@ DOMAIN = { ...@@ -217,11 +226,17 @@ DOMAIN = {
}, },
'payments': { 'payments': {
# Dummy endpoint for payments. # Endpoint for payments via Stripe.
# TODO: Implement as soon as PSP is known.
# charge_id is a unique identifier which allows us to track the payment with Stripe
# Admins can however create payments without a charge_id
# Only admins can delete payments # Only admins can delete payments
'user_methods': ['GET', 'POST', 'PATCH'], # Also, there is no reason to ever change a payment.
'user_methods': ['GET', 'POST'],
# Bulk inserts don't make sense here, so we disallow them
'bulk_enabled': False,
'schema': { 'schema': {
'signups': { 'signups': {
...@@ -233,11 +248,32 @@ DOMAIN = { ...@@ -233,11 +248,32 @@ DOMAIN = {
'field': '_id', 'field': '_id',
'embeddable': True 'embeddable': True
}, },
# TODO: No duplicate entries 'no_waiting': True, # No signups on waiting list
# TODO: No courses on waiting list 'no_accepted': True, # No signups which have already been paid
}, },
'no_copies': True,
'required': True, 'required': True,
'nullable': False, 'nullable': False,
},
# Admins may leave this field empty
# However, it still has to be sent, even if it contains an empty string / None
'token': {
'type': 'string',
'unique': True,
'required': True,
'nullable': True,
'only_admin_empty': True,
},
'charge_id': { # Set by backend
'type': 'string',
'unique': True,
'required': False,
'nullable': True,
},
'amount': { # Set by backend
'type': 'integer',
'required': False,
'nullable': True,
} }
} }
} }
......
...@@ -134,6 +134,10 @@ def update_signups(course_id): ...@@ -134,6 +134,10 @@ def update_signups(course_id):
def mark_as_paid(payments): def mark_as_paid(payments):
"""After successful payment, set status to `accepted`.""" """After successful payment, set status to `accepted`."""
# Check if payments is not a list
if not isinstance(payments, list):
payments = [payments]
for payment in payments: for payment in payments:
for signup in payment['signups']: for signup in payment['signups']:
...@@ -143,3 +147,20 @@ def mark_as_paid(payments): ...@@ -143,3 +147,20 @@ def mark_as_paid(payments):
payload=data, payload=data,
concurrency_check=False, concurrency_check=False,
skip_validation=True) skip_validation=True)
def mark_as_unpaid(payments):
"""Before a payment is deleted, set status to `reserved`."""
# Check if payments is not a list
if not isinstance(payments, list):
payments = [payments]
for payment in payments:
for signup in payment['signups']:
data = {'status': 'reserved'}
patch_internal('signups',
_id=str(signup),
payload=data,
concurrency_check=False,
skip_validation=True)
...@@ -51,3 +51,27 @@ class APIValidator(Validator): ...@@ -51,3 +51,27 @@ class APIValidator(Validator):
"""Inhibit patching of the field, also copied from AMIVAPI.""" """Inhibit patching of the field, also copied from AMIVAPI."""
if enabled and (request.method == 'PATCH'): if enabled and (request.method == 'PATCH'):
self._error(field, "this field can not be changed with PATCH") self._error(field, "this field can not be changed with PATCH")
def _validate_only_admin_empty(self, only_admin_empty, field, value):
"""Allow the field to be empty only if the user is admin."""
if only_admin_empty and not value and not is_admin():
self._error(field, "only admins may leave this field empty")
def _validate_no_waiting(self, no_waiting, field, value):
"""Disallow signups which are on waiting list status."""
signup = current_app.data.driver.db['signups'].find_one({'_id': value})
if no_waiting and signup['status'] == 'waiting':
self._error(field, "this field may not contain signups " +
"which are still on the waiting list")
def _validate_no_accepted(self, no_accepted, field, value):
"""Disallow signups that have already been paid."""
signup = current_app.data.driver.db['signups'].find_one({'_id': value})
if no_accepted and signup['status'] == 'accepted':
self._error(field, "this field may not contain signups " +
"which have already been paid")
def _validate_no_copies(self, no_copies, field, value):
"""Ensure that each signup only appears once per payment."""
if no_copies and len(set(value)) != len(value):
self._error(field, "this field may not contain duplicate signups")
attrs==17.3.0
Cerberus==0.9.2
certifi==2017.11.5
chardet==3.0.4
click==6.7
Eve==0.7.8 Eve==0.7.8
Events==0.2.2 stripe==1.79
Flask==0.12
Flask-PyMongo==0.5.1
idna==2.6
itsdangerous==0.24
Jinja2==2.10
MarkupSafe==0.23
pluggy==0.6.0
py==1.5.2
pymongo==3.6.1
requests==2.18.4
simplejson==3.13.2
six==1.11.0
tox==2.9.1
urllib3==1.22
virtualenv==15.1.0
Werkzeug==0.11.15
"""Tests for the Stripe payment backend"""
import pytest
@pytest.fixture(autouse=True)
def base_data(app):
"""Build test data for the tests."""
with app.admin():
lecture_data = {
'title': 'Awesome Lecture',
'department': 'itet',
'year': 3,
'assistants': ['pablo', 'pablone'],
}
lecture = app.client.post('lectures',
data=lecture_data,
assert_status=201)
course = {
'lecture': lecture['_id'],
'assistant': 'anon',
'room': 'ETZ f 6',
'spots': 20,
'signup': {
'start': '2021-05-01T10:00:00Z',
'end': '2021-05-05T23:59:59Z',
},
'datetimes': [{
'start': '2021-06-05T10:00:00Z',
'end': '2021-06-05T12:00:00Z',
}, {
'start': '2021-06-06T10:00:00Z',
'end': '2021-06-06T12:00:00Z',
}],
}
response = app.client.post('courses',
data=course,
assert_status=201)
signup = {
'nethz': 'Something',
'course': response['_id']
}
app.client.post('signups',
data=signup,
assert_status=201)
def test_admin(app):
"""Test that an admin may create and delete payments unconditionally"""
with app.admin():
# Fetch the _ids of the signups we want to pay for
signups = [signup['_id'] for signup in app.client.get('signups')['_items']]
# Try to create a payment without a token
payment = {
'signups': signups,
'token': None,
}
payment_response = app.client.post('payments',
data=payment,
assert_status=201)
# Check that the signups have been marked as paid
signups = app.client.get('signups')['_items']
for signup in signups:
assert signup['status'] == 'accepted'
# Try to delete the payment
app.client.delete('payments/{}'.format(payment_response['_id']),
headers={'If-Match': payment_response['_etag']},
assert_status=204)
# Check that the signups have been marked as unpaid
signups = app.client.get('signups')['_items']
for signup in signups:
assert signup['status'] == 'reserved'
def test_no_multiple_payments_admin(app):
"""Make sure that no signup / course may be paid more than once by an admin"""
with app.admin():
# Fetch the signups we want to pay
signups = [signup['_id'] for signup in app.client.get('signups')['_items']]
# Pay the signups twice
payment = {
'signups': signups,
'token': None,
}
# First payment succeeds
app.client.post('payments', data=payment, assert_status=201)
# Second payment fails
app.client.post('payments', data=payment, assert_status=422)
# Check that the signups were marked as paid nonetheless
signups = app.client.get('signups')['_items']
for signup in signups:
assert signup['status'] == 'accepted'
def test_no_multiple_payments_user(app):
"""Make sure that no signup / course may be paid more than once by a user"""
with app.user():
# Fetch the signups we want to pay
signups = [signup['_id'] for signup in app.client.get('signups')['_items']]
# Pay the signups twice
payment = {
'signups': signups,
'token': 'tok_visa',
}
# First payment succeeds
app.client.post('payments', data=payment, assert_status=201)
# Second payment fails
app.client.post('payments', data=payment, assert_status=422)
# Check that the signups were marked as paid
signups = app.client.get('signups')['_items']
for signup in signups:
assert signup['status'] == 'accepted'
def test_users_require_token(app):
"""Test that regular users must provide a token to create a payment."""
with app.user():
# Fetch the signups we want to pay
signups = [signup['_id'] for signup in app.client.get('signups')['_items']]
# Try to create a payment without a token
# We expect this to fail with a validation error
payment = {
'signups': signups,
}
app.client.post('payments',
data=payment,
assert_status=422)
def test_valid_card(app):
"""Test that payment succeeds with a valid card.
'tok_visa' always corresponds to a valid Visa card.
"""
with app.user():
# Fetch the signups we want to pay
signups = [signup['_id'] for signup in app.client.get('signups')['_items']]
# Try to create a payment with the Visa test token
# We expect this to succeed
payment = {
'signups': signups,
'token': 'tok_visa',
}
app.client.post('payments',
data=payment,
assert_status=201)
# Check that the signups were marked as paid
signups = app.client.get('signups')['_items']
for signup in signups:
assert signup['status'] == 'accepted'
def test_invalid_card(app):
"""Test that invalid cards yield a status code of 422.
'tok_chargeDeclined' always causes the transaction to fail.
"""
with app.user():
# Fetch the signups we want to pay
signups = [signup['_id'] for signup in app.client.get('signups')['_items']]
# Try to create a payment with the charge declined token
# We expect this to fail
payment = {
'signups': signups,
'token': 'tok_chargeDeclined',
}
app.client.post('payments',
data=payment,
assert_status=422)
# Check that the signups were NOT marked as paid
signups = app.client.get('signups')['_items']
for signup in signups:
assert signup['status'] != 'accepted'
def test_users_cannot_delete(app):
"""Test that regular users cannot delete payments."""
# Let admin create a payment
with app.admin():
# Fetch the signups we want to pay
signups = [signup['_id'] for signup in app.client.get('signups')['_items']]
# Try to create a payment without a token
payment = {
'signups': signups,
'token': None,
}
payment_response = app.client.post('payments',
data=payment,
assert_status=201)
with app.user():
# Let the user try to delete the payment
app.client.delete('payments/{}'.format(payment_response['_id']),
assert_status=403)
# Make sure the signups are still marked as paid
signups = app.client.get('signups')['_items']
for signup in signups:
assert signup['status'] == 'accepted'
def test_signup_unique_per_payment(app):
"""Test that the same signup only appears once per payment"""
with app.admin():
# Fetch the signups we want to pay
signups = [signup['_id'] for signup in app.client.get('signups')['_items']]
# Duplicate the list so every signup appears twice
signups.extend(signups)
# Try to create a payment
# We expect a validation error
payment = {
'signups': signups,
'token': None,
}
app.client.post('payments',
data=payment,
assert_status=422)
def test_no_batch_payments(app):
"""Test that batch payments are disabled (Eve sends 400 in this case)."""
with app.admin():
batch = [{'signups': [], 'token': 'something'},
{'signups': [], 'token': 'something'}]
app.client.post('payments',
data=batch,
assert_status=400)
...@@ -45,12 +45,9 @@ def test_create(app): ...@@ -45,12 +45,9 @@ def test_create(app):
'nethz': "Pablito", 'nethz': "Pablito",
'course': course_response['_id'] 'course': course_response['_id']
} }
signup_response = app.client.post('signups', app.client.post('signups',
data=signup, data=signup,
assert_status=201) assert_status=201)
payment = {'signups': [signup_response['_id']]}
app.client.post('payments', data=payment, assert_status=201)
def test_no_double_signup(app): def test_no_double_signup(app):
......
...@@ -32,19 +32,11 @@ def test_success(app): ...@@ -32,19 +32,11 @@ def test_success(app):
assert signup_response['status'] == 'reserved' assert signup_response['status'] == 'reserved'
# Now pay
payment = {
'signups': [signup_response['_id']]
}
app.client.post('/payments',
data=payment,
assert_status=201)
# Check signup # Check signup
updated_signup = app.client.get('/signups/' + signup_response['_id'], updated_signup = app.client.get('/signups/' + signup_response['_id'],
assert_status=200) assert_status=200)
assert updated_signup['status'] == 'accepted' assert updated_signup['status'] == 'reserved'
def test_zero_spots(app): def test_zero_spots(app):
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment