151 lines
5.1 KiB
Python
151 lines
5.1 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
import threading
|
|
import time
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
from django.db import transaction
|
|
from django.http import HttpResponse, JsonResponse
|
|
from django.shortcuts import redirect
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
|
|
from vbv_lernwelt.core.middleware.auth import django_view_authentication_exempt
|
|
from vbv_lernwelt.core.models import User
|
|
|
|
|
|
@csrf_exempt
|
|
@django_view_authentication_exempt
|
|
@transaction.non_atomic_requests
|
|
def fake_datatrans_api_view(request, api_url=""):
|
|
if api_url == "/v1/transactions" and request.method == "POST":
|
|
data = json.loads(request.body.decode("utf-8"))
|
|
user = User.objects.get(id=data["user_id"])
|
|
|
|
if "customer" in data and not user.abacus_debitor_number:
|
|
user.set_increment_abacus_debitor_number()
|
|
data["customer"]["id"] = user.abacus_debitor_number
|
|
|
|
user.additional_json_data["datatrans_transaction_payload"] = data
|
|
user.save()
|
|
|
|
return JsonResponse({"transactionId": data["refno"]}, status=201)
|
|
|
|
return HttpResponse(
|
|
content="unknown api url", content_type="application/json", status=400
|
|
)
|
|
|
|
|
|
@csrf_exempt
|
|
@django_view_authentication_exempt
|
|
def fake_datatrans_pay_view(request, api_url=""):
|
|
def call_transaction_complete_webhook(
|
|
webhook_url, transaction_id, datatrans_status="settled"
|
|
):
|
|
time.sleep(1)
|
|
|
|
payload = {
|
|
"transactionId": transaction_id,
|
|
"status": datatrans_status,
|
|
}
|
|
key_hex_bytes = bytes.fromhex(settings.DATATRANS_HMAC_KEY)
|
|
|
|
# Create sign with timestamp and payload
|
|
sign = hmac.new(
|
|
key_hex_bytes, bytes(str(1) + json.dumps(payload), "utf-8"), hashlib.sha256
|
|
)
|
|
|
|
response = requests.post(
|
|
url=webhook_url,
|
|
json=payload,
|
|
headers={"Datatrans-Signature": f"t=1,s0={sign.hexdigest()}"},
|
|
)
|
|
print(response)
|
|
|
|
if api_url.startswith("/v1/start/"):
|
|
transaction_id = api_url.split("/")[-1]
|
|
transaction_user = User.objects.filter(
|
|
additional_json_data__datatrans_transaction_payload__refno=transaction_id
|
|
).first()
|
|
|
|
if transaction_user is None:
|
|
return HttpResponse(
|
|
content=f"""
|
|
<h1>Fake Datatrans Payment</h1>
|
|
<p>No active transaction found for {transaction_id}</p>
|
|
""",
|
|
status=404,
|
|
)
|
|
|
|
if request.method == "GET":
|
|
return HttpResponse(
|
|
content=f"""
|
|
<h1>Fake Datatrans Payment</h1>
|
|
<form action="{request.build_absolute_uri()}" method="post">
|
|
<fieldset>
|
|
<legend>Datatrans payment result status</legend>
|
|
<div>
|
|
<input type="radio" name="payment" value="settled" checked/>
|
|
<label>settled</label>
|
|
|
|
<input type="radio" name="payment" value="cancelled" />
|
|
<label>cancelled</label>
|
|
|
|
<input type="radio" name="payment" value="failed" />
|
|
<label>failed</label>
|
|
</div>
|
|
<div>
|
|
<button type="submit" data-cy="pay-button">
|
|
Pay with selected status
|
|
</button>
|
|
</div>
|
|
</fieldset>
|
|
</form>
|
|
"""
|
|
)
|
|
|
|
elif request.method == "POST":
|
|
payment_status = request.POST.get("payment", "settled")
|
|
|
|
if payment_status == "settled":
|
|
success_url = transaction_user.additional_json_data[
|
|
"datatrans_transaction_payload"
|
|
]["redirect"]["successUrl"]
|
|
|
|
# start new thread which will call webhook after 2 seconds
|
|
webhook_url = transaction_user.additional_json_data[
|
|
"datatrans_transaction_payload"
|
|
]["webhook"]["url"]
|
|
thread = threading.Thread(
|
|
target=call_transaction_complete_webhook,
|
|
args=(
|
|
webhook_url,
|
|
transaction_id,
|
|
),
|
|
)
|
|
thread.start()
|
|
|
|
# redirect to url
|
|
return redirect(success_url + f"?datatransTrxId={transaction_id}")
|
|
|
|
if payment_status == "cancelled":
|
|
cancel_url = transaction_user.additional_json_data[
|
|
"datatrans_transaction_payload"
|
|
]["redirect"]["cancelUrl"]
|
|
|
|
# redirect to url
|
|
return redirect(cancel_url + f"?datatransTrxId={transaction_id}")
|
|
|
|
if payment_status == "failed":
|
|
error_url = transaction_user.additional_json_data[
|
|
"datatrans_transaction_payload"
|
|
]["redirect"]["errorUrl"]
|
|
|
|
# redirect to url
|
|
return redirect(error_url + f"?datatransTrxId={transaction_id}")
|
|
|
|
return HttpResponse(
|
|
content="unknown api url", content_type="application/json", status=400
|
|
)
|