vbv/server/vbv_lernwelt/shop/datatrans_fake_server.py

import hashlib
import hmac
import json
import threading

from django.conf import settings
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
def fake_datatrans_api_view(request, api_url=""):
    # if api_url == "/redirect":
    #     fake_tamedia_token = request.GET.get("token")
    #     pai = fake_tamedia_token.split(":")[1]
    #     sub = SubhubCustomerSubscription.objects.filter(id=pai).first()
    #
    #     header = f"<h1>fake tamedia activation for {pai}"
    #
    #     if not sub:
    #         return HttpResponse(
    #             content=f"""
    #             {header}
    #             <p>no subscription found</p>
    #             """,
    #             status=404,
    #         )
    #
    #     if request.method == "GET":
    #         if (
    #             sub
    #             and sub.partner_status
    #             == SubhubCustomerSubscription.PARTNER_STATUS_ENROLLED
    #         ):
    #             return HttpResponse(
    #                 content=f"""
    #                 {header}
    #                 <div>
    #                 <input type="text" name="partnerSubscriptionId" value="{pai}">
    #                 <form action="{create_register_url(fake_tamedia_token)}" method="post">
    #                     <button type="submit" data-cy="register-button">Register</button>
    #                 </form>
    #                 </div>
    #                 """,
    #                 status=200,
    #             )
    #         else:
    #             return HttpResponse(
    #                 content=f"""
    #                 {header}
    #                 <p>already activated</p>
    #                 """,
    #                 status=200,
    #             )
    #     if request.method == "POST":
    #         if sub:
    #             response = requests.post(
    #                 f"{settings.APPLICATION_ABSOLUTE_URL}/subhub/ottwebhook",
    #                 json={
    #                     "PartnerIntegration": {
    #                         "effectiveDate": datetime.now().isoformat(),
    #                         "eventType": "activation",
    #                         "offerId": sub.subscription_choice.partner_product_id,
    #                         "optionalAttributes": None,
    #                         "pai": pai,
    #                         "partnerType": "Tamedia",
    #                         "transactionId": str(uuid.uuid4()),
    #                     },
    #                     "eventId": str(uuid.uuid4()),
    #                     "eventType": "OTT Partner Events",
    #                     "publisherId": "Partner Events",
    #                     "status": "new",
    #                     "timestamp": datetime.now().isoformat(),
    #                 },
    #                 auth=HTTPBasicAuth(
    #                     "swisscom_ott_webhook",
    #                     "swisscom-ott-webhook-rLaYG0btVJMPtfnzfLilZtm50",
    #                 ),
    #             )
    #             print(response)
    #         return redirect(f"{create_register_url(fake_tamedia_token)}")
    #
    # if api_url.startswith("/enroll") and request.method == "POST":
    #     return HttpResponse(status=204)
    #

    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"])
        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"
    ):
        import requests
        import time

        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/") and request.method == "GET":
        transaction_id = api_url.split("/")[-1]
        transaction_user = User.objects.filter(
            additional_json_data__datatrans_transaction_payload__refno=transaction_id
        ).first()

        redirect_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(redirect_url + f"?datatransTrxId={transaction_id}")

    return HttpResponse(
        content="unknown api url", content_type="application/json", status=400
    )