ramon
/
vbv
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: datatrans cleanup & tests

This commit is contained in:
Livio Bieri 2023-11-26 22:28:45 +01:00 committed by Christian Cueni
parent 1281be2221
commit a9056a008b
7 changed files with 467 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
server/vbv_lernwelt/shop/migrations/0010_alter_checkoutinformation_state.py Normal file
View File

@ -0,0 +1,25 @@
# Generated by Django 3.2.20 on 2023-11-26 19:47
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("shop", "0009_alter_checkoutinformation_product_price"),
]
operations = [
migrations.AlterField(
model_name="checkoutinformation",
name="state",
field=models.CharField(
choices=[
("initialized", "Initialized"),
("paid", "Paid"),
("canceled", "Canceled"),
("failed", "Failed"),
],
max_length=50,
),
),
]

6
server/vbv_lernwelt/shop/models.py
View File

@ -58,6 +58,9 @@ class CheckoutState(models.TextChoices):
"""
The state of a checkout process transaction.
PAID: Datatrans transaction settled/transmitted.
Rest of the states are self-explanatory, same as in Datatrans docs.
1) We use the `autoSettle` feature of DataTrans! Therefore, there are less possible states:
-> https://docs.datatrans.ch/docs/after-the-payment
-> https://api-reference.datatrans.ch/#tag/v1transactions/operation/status
@ -69,8 +72,7 @@ class CheckoutState(models.TextChoices):
"""
INITIALIZED = "initialized"
SETTLED = "settled"
TRANSMITTED = "transmitted"
PAID = "paid"
CANCELED = "canceled"
FAILED = "failed"

1
server/vbv_lernwelt/shop/services.py
View File

@ -64,7 +64,6 @@ def init_transaction(
payload = {
# We use autoSettle=True, so that we don't have to settle the transaction:
# -> Be aware that autoSettle has implications of the possible transaction states
# -> see CheckoutState docstring.
"autoSettle": True,
"amount": amount_chf_centimes,
"currency": "CHF",

119
server/vbv_lernwelt/shop/tests/test_checkout_api.py
View File

@ -147,3 +147,122 @@ class CheckoutAPITestCase(APITestCase):
0,
CheckoutInformation.objects.count(),
)
def test_checkout_already_paid(self):
# GIVEN
CheckoutInformation.objects.create(
user=self.user,
product_sku=VV_PRODUCT_SKU,
product_price=0,
state=CheckoutState.PAID.value,
)
# WHEN
response = self.client.post(
path=reverse("checkout-vv"),
format="json",
data={
"redirect_url": REDIRECT_URL,
"address": TEST_ADDRESS_DATA,
},
)
# THEN
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
"/",
response.json()["next_step_url"],
)
def test_checkout_double_checkout(self):
# GIVEN
transaction_id = "1234567890"
CheckoutInformation.objects.create(
user=self.user,
product_sku=VV_PRODUCT_SKU,
product_price=0,
state=CheckoutState.INITIALIZED.value,
transaction_id=transaction_id,
)
# WHEN
response = self.client.post(
path=reverse("checkout-vv"),
format="json",
data={
"redirect_url": REDIRECT_URL,
"address": TEST_ADDRESS_DATA,
},
)
# THEN
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
f"https://pay.sandbox.datatrans.com/v1/start/{transaction_id}",
response.json()["next_step_url"],
)
@patch("vbv_lernwelt.shop.views.init_transaction")
def test_checkout_failed_creates_new(self, mock_init_transaction):
# GIVEN
state = CheckoutState.FAILED.value
transaction_id = "1234567890"
mock_init_transaction.return_value = transaction_id
CheckoutInformation.objects.create(
user=self.user,
product_sku=VV_PRODUCT_SKU,
product_price=0,
state=state,
transaction_id="0000000000",
)
# WHEN
response = self.client.post(
path=reverse("checkout-vv"),
format="json",
data={
"redirect_url": REDIRECT_URL,
"address": TEST_ADDRESS_DATA,
},
)
# THEN
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
f"https://pay.sandbox.datatrans.com/v1/start/{transaction_id}",
response.json()["next_step_url"],
)
@patch("vbv_lernwelt.shop.views.init_transaction")
def test_checkout_cancelled_creates_new(self, mock_init_transaction):
# GIVEN
state = CheckoutState.CANCELED.value
transaction_id = "1234567899"
mock_init_transaction.return_value = transaction_id
CheckoutInformation.objects.create(
user=self.user,
product_sku=VV_PRODUCT_SKU,
product_price=0,
state=state,
transaction_id="1111111111",
)
# WHEN
response = self.client.post(
path=reverse("checkout-vv"),
format="json",
data={
"redirect_url": REDIRECT_URL,
"address": TEST_ADDRESS_DATA,
},
)
# THEN
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
f"https://pay.sandbox.datatrans.com/v1/start/{transaction_id}",
response.json()["next_step_url"],
)

8
server/vbv_lernwelt/shop/tests/test_create_signature.py → server/vbv_lernwelt/shop/tests/test_datatrans_signature.py
View File

@ -3,8 +3,12 @@ from unittest import TestCase
from vbv_lernwelt.shop.services import is_signature_valid
class DatatransWebhookSigningTestCase(TestCase):
# Key is from their example in the docs, not ours! :D
class DatatransSigningTestCase(TestCase):
"""
Test based on the example from the docs.
Key is from their example, not ours!
"""
HMAC_KEY_FROM_THE_DOCS_NOT_HAZARDOUS = (
"861bbfc01e089259091927d6ad7f71c8"
"b46b7ee13499574e83c633b74cdc29e3"

252
server/vbv_lernwelt/shop/tests/test_datatrans_webhook.py Normal file
View File

@ -0,0 +1,252 @@
from unittest.mock import patch
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from vbv_lernwelt.core.admin import User
from vbv_lernwelt.course.creators.test_utils import create_course, create_course_session
from vbv_lernwelt.course.models import CourseSessionUser
from vbv_lernwelt.shop.models import (
CheckoutInformation,
CheckoutState,
Product,
VV_PRODUCT_SKU,
)
def create_checkout_information(
user: User,
transaction_id: str,
state: CheckoutState,
) -> CheckoutInformation:
return CheckoutInformation.objects.create(
user=user,
transaction_id=transaction_id,
product_sku=VV_PRODUCT_SKU,
product_price=300_00,
state=state.value,
)
class DatatransWebhookTestCase(APITestCase):
def setUp(self) -> None:
course, _ = create_course(title="VV")
create_course_session(course=course, title="Versicherungsvermittler/-in")
self.user = User.objects.create_user(
username="testuser",
email="test@user.com",
password="testpassword",
is_active=True,
)
self.product = Product.objects.create(
sku=VV_PRODUCT_SKU,
price=300_00,
description="VV",
name="VV",
)
def test_webhook_unsigned_payload(self):
# GIVEN
payload = {
"transactionId": "1234567890",
"status": "settled",
}
headers = {"Datatrans-Signature": "t=1605697463367,s0=guessed"}
# WHEN
response = self.client.post(
path=reverse("shop-transaction-webhook"),
format="json",
data=payload,
headers=headers,
)
# THEN
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(
response.json(),
{"status": "invalid signature"},
)
@patch("vbv_lernwelt.shop.views.is_signature_valid")
def test_webhook_creates_course_session_user(self, mock_is_signature_valid):
# GIVEN
transaction_id = "1234567890"
create_checkout_information(
user=self.user,
transaction_id=transaction_id,
state=CheckoutState.INITIALIZED,
)
mock_is_signature_valid.return_value = True
# WHEN
# ~immediately after successful payment
response_settled = self.client.post(
path=reverse("shop-transaction-webhook"),
format="json",
headers={"Datatrans-Signature": "<signature>"},
data={
"status": "settled",
"transactionId": transaction_id,
},
)
# ~24h later
response_transmitted = self.client.post(
path=reverse("shop-transaction-webhook"),
format="json",
headers={"Datatrans-Signature": "<signature>"},
data={
"status": "transmitted",
"transactionId": transaction_id,
},
)
# THEN
self.assertEqual(response_settled.status_code, status.HTTP_200_OK)
self.assertEqual(response_transmitted.status_code, status.HTTP_200_OK)
self.assertEqual(
CheckoutInformation.objects.get(transaction_id=transaction_id).state,
CheckoutState.PAID.value,
)
self.assertEqual(
1,
CourseSessionUser.objects.count(),
)
self.assertEqual(
CourseSessionUser.objects.first().user,
self.user,
)
@patch("vbv_lernwelt.shop.views.is_signature_valid")
def test_webhook_updates_webhook_history(self, mock_is_signature_valid):
# GIVEN
transaction_id = "1234567890"
create_checkout_information(
user=self.user,
transaction_id=transaction_id,
state=CheckoutState.INITIALIZED,
)
mock_is_signature_valid.return_value = True
# WHEN
response_1 = self.client.post(
path=reverse("shop-transaction-webhook"),
format="json",
headers={"Datatrans-Signature": "<signature>"},
data={
"transactionId": transaction_id,
"status": "failed",
"whatever": "1",
},
)
response_2 = self.client.post(
path=reverse("shop-transaction-webhook"),
format="json",
headers={"Datatrans-Signature": "<signature>"},
data={
"transactionId": transaction_id,
"status": "failed",
"whatever": "2",
},
)
self.assertEqual(response_1.status_code, status.HTTP_200_OK)
self.assertEqual(response_2.status_code, status.HTTP_200_OK)
# THEN
self.assertEqual(
CheckoutInformation.objects.get(
transaction_id=transaction_id
).webhook_history,
[
{
"transactionId": transaction_id,
"status": "failed",
"whatever": "1",
},
{
"transactionId": transaction_id,
"status": "failed",
"whatever": "2",
},
],
)
@patch("vbv_lernwelt.shop.views.is_signature_valid")
def test_webhook_failed(self, mock_is_signature_valid):
# GIVEN
transaction_id = "1234567890"
state_received = "failed"
create_checkout_information(
user=self.user,
transaction_id=transaction_id,
state=CheckoutState.INITIALIZED,
)
mock_is_signature_valid.return_value = True
# WHEN
response = self.client.post(
path=reverse("shop-transaction-webhook"),
format="json",
headers={"Datatrans-Signature": "<signature>"},
data={
"transactionId": transaction_id,
"status": state_received,
},
)
# THEN
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
CheckoutInformation.objects.get(transaction_id=transaction_id).state,
CheckoutState.FAILED.value,
)
@patch("vbv_lernwelt.shop.views.is_signature_valid")
def test_webhook_cancelled(self, mock_is_signature_valid):
# GIVEN
transaction_id = "1234567890"
state_received = "canceled"
create_checkout_information( # noqa
user=self.user,
transaction_id=transaction_id,
state=CheckoutState.INITIALIZED,
)
mock_is_signature_valid.return_value = True
# WHEN
response = self.client.post(
path=reverse("shop-transaction-webhook"),
format="json",
headers={"Datatrans-Signature": "<signature>"},
data={
"transactionId": transaction_id,
"status": state_received,
},
)
# THEN
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
CheckoutInformation.objects.get(transaction_id=transaction_id).state,
CheckoutState.CANCELED.value,
)

97
server/vbv_lernwelt/shop/views.py
View File

@ -5,7 +5,6 @@ from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from vbv_lernwelt.core.admin import User
from vbv_lernwelt.course.models import CourseSession, CourseSessionUser
from vbv_lernwelt.shop.models import (
BillingAddress,
@ -24,6 +23,8 @@ from vbv_lernwelt.shop.services import (
logger = structlog.get_logger(__name__)
COURSE_SESSION_TITLE_VV = "Versicherungsvermittler/-in"
@api_view(["GET"])
@permission_classes([IsAuthenticated])
@ -60,6 +61,12 @@ def update_billing_address(request):
@api_view(["POST"])
def transaction_webhook(request):
"""
VERY IMPORTANT: Datatrans had/has to set up stuff manually on their side:
Otherwise, this webhook was/will not be called for "initialized" -> "failed" state changes
For timed out transactions (cleaned up in 15 minute intervals, after 30 minutes by them),
"""
logger.info("Webhook: Datatrans called transaction webhook", body=request.body)
if not is_signature_valid(
@ -73,25 +80,23 @@ def transaction_webhook(request):
transaction_id = transaction["transactionId"]
transaction_status = transaction["status"]
# keep webhook history for debugging
checkout_info = CheckoutInformation.objects.get(transaction_id=transaction_id)
checkout_info.state = transaction_status
checkout_info.webhook_history.append(transaction)
checkout_info.save(update_fields=["state", "webhook_history"])
checkout_info.save(update_fields=["webhook_history"])
# be aware autoSettle has implications on possible transaction states we get!
# See https://api-reference.datatrans.ch/#tag/v1transactions/operation/status
if transaction_status in ["settled", "transmitted"]:
logger.info(
"Webhook: Create course session user", transaction_id=transaction_id
)
# FIXME make this "better"
cs_vv = CourseSession.objects.filter(
title="Versicherungsvermittler/-in"
).first()
CourseSessionUser.objects.get_or_create(
course_session=cs_vv,
user=User.objects.get(id=checkout_info.user_id),
role=CourseSessionUser.Role.MEMBER,
update_checkout_state(checkout_info=checkout_info, state=CheckoutState.PAID)
create_vv_course_session_user(checkout_info=checkout_info)
elif transaction_status == "failed":
update_checkout_state(checkout_info=checkout_info, state=CheckoutState.FAILED)
elif transaction_status == "canceled":
update_checkout_state(checkout_info=checkout_info, state=CheckoutState.CANCELED)
else:
logger.warning(
"Unhandled transaction status", transaction_status=transaction_status
)
return JsonResponse({"status": "ok"})
@ -113,13 +118,11 @@ def checkout_vv(request):
try:
product = Product.objects.get(sku=sku)
except Product.DoesNotExist:
return JsonResponse(
{
"next_step_url": checkout_error_url(
base_url=base_redirect_url,
message="vv_product_does_not_exist_needs_to_be_created",
)
},
return next_step_response(
url=checkout_error_url(
base_url=base_redirect_url,
message="vv_product_does_not_exist_needs_to_be_created",
),
)
checkouts = CheckoutInformation.objects.filter(
@ -127,15 +130,13 @@ def checkout_vv(request):
product_sku=sku,
)
# already purchased (settled or transmitted)
if checkouts.filter(
state__in=[CheckoutState.SETTLED, CheckoutState.TRANSMITTED]
).exists():
return JsonResponse({"next_step_url": "/"})
# already paid (successfully)-> redirect to home
if checkouts.filter(state__in=[CheckoutState.PAID]).exists():
return next_step_response(url="/")
# already initialized -> redirect to payment page again
if checkout := checkouts.filter(state=CheckoutState.INITIALIZED).first():
return JsonResponse({"next_step_url": get_payment_url(checkout.transaction_id)})
return next_step_response(url=get_payment_url(checkout.transaction_id))
# not yet initialized at all, or canceled/failed
# -> create new transaction and checkout
@ -149,12 +150,10 @@ def checkout_vv(request):
webhook_url=webhook_url(base_redirect_url),
)
except InitTransactionException as e:
return JsonResponse(
{
"next_step_url": checkout_error_url(
base_url=base_redirect_url,
)
},
return next_step_response(
url=checkout_error_url(
base_url=base_redirect_url,
),
)
CheckoutInformation.objects.create(
@ -170,7 +169,33 @@ def checkout_vv(request):
**request.data["address"],
)
return JsonResponse({"next_step_url": get_payment_url(transaction_id)})
return next_step_response(url=get_payment_url(transaction_id))
def update_checkout_state(checkout_info: CheckoutInformation, state: CheckoutState):
checkout_info.state = state.value
checkout_info.save(update_fields=["state"])
def create_vv_course_session_user(checkout_info: CheckoutInformation):
logger.info("Creating VV course session user", user_id=checkout_info.user_id)
CourseSessionUser.objects.get_or_create(
user=checkout_info.user,
role=CourseSessionUser.Role.MEMBER,
course_session=CourseSession.objects.filter(
title=COURSE_SESSION_TITLE_VV
).first(),
)
def next_step_response(
url: str,
) -> JsonResponse:
return JsonResponse(
{
"next_step_url": url,
},
)
def webhook_url(base_url: str) -> str: