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

Add more logging in app/database

This commit is contained in:
Daniel Egger 2024-06-21 14:56:55 +02:00
parent b9f8e5d771
commit 9d91a9102a
22 changed files with 575 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
client/src/components/onboarding/DatatransCembraDeviceFingerprint.vue
View File

@ -43,5 +43,4 @@ export default {
};
</script>
<style scoped>
</style>
<style scoped></style>

2
client/src/main.ts
View File

@ -1,4 +1,5 @@
import { i18nextInit } from "@/i18nextWrapper";
import { generateLocalSessionKey } from "@/statistics";
import * as Sentry from "@sentry/vue";
import i18next from "i18next";
import I18NextVue from "i18next-vue";
@ -9,7 +10,6 @@ import type { Router } from "vue-router";
import "../tailwind.css";
import App from "./App.vue";
import router from "./router";
import { generateLocalSessionKey } from "@/statistics";
declare module "pinia" {
export interface PiniaCustomProperties {

2
server/config/urls.py
View File

@ -69,7 +69,7 @@ from vbv_lernwelt.importer.views import (
from vbv_lernwelt.media_files.views import user_image
from vbv_lernwelt.notify.views import email_notification_settings
from vbv_lernwelt.shop.datatrans_fake_server import (
from vbv_lernwelt.shop.datatrans.datatrans_fake_server import (
fake_datatrans_api_view,
fake_datatrans_pay_view,
)

2
server/requirements/requirements-dev.txt
View File

@ -576,6 +576,8 @@ typing-extensions==4.7.1
# wagtail-localize
typing-inspect==0.9.0
# via libcst
ua-parser==0.18.0
# via -r requirements.in
ufmt==2.2.0
# via -r requirements-dev.in
uritemplate==4.1.1

1
server/requirements/requirements.in
View File

@ -8,6 +8,7 @@ redis # https://github.com/redis/redis-py
uvicorn[standard] # https://github.com/encode/uvicorn
environs
click
ua-parser
# Django
# ------------------------------------------------------------------------------

2
server/requirements/requirements.txt
View File

@ -324,6 +324,8 @@ typing-extensions==4.7.1
# dj-database-url
# uvicorn
# wagtail-localize
ua-parser==0.18.0
# via -r requirements.in
uritemplate==4.1.1
# via drf-spectacular
urllib3==1.26.16

5
server/vbv_lernwelt/api/tests/test_profile.py
View File

@ -43,7 +43,8 @@ class ProfileViewTest(APITestCase):
# THEN
self.assertEqual(response.status_code, status.HTTP_200_OK)
profile = response.data
self.assertEqual(
self.maxDiff = None
self.assertDictEqual(
profile,
{
"id": str(self.user.id),
@ -62,6 +63,8 @@ class ProfileViewTest(APITestCase):
"postal_code": "",
"city": "",
"country": None,
"phone_number": "",
"birth_date": None,
"organisation_detail_name": "",
"organisation_street": "",
"organisation_street_number": "",

97
server/vbv_lernwelt/core/admin.py
View File

@ -2,7 +2,13 @@ from django.contrib import admin
from django.contrib.auth import admin as auth_admin, get_user_model
from django.utils.translation import gettext_lazy as _
from vbv_lernwelt.core.models import Country, JobLog, Organisation
from vbv_lernwelt.core.models import (
Country,
ExternalApiRequestLog,
JobLog,
Organisation,
SecurityRequestResponseLog,
)
from vbv_lernwelt.core.utils import pretty_print_json
User = get_user_model()
@ -109,6 +115,95 @@ class JobLogAdmin(LogAdmin):
return None
@admin.register(SecurityRequestResponseLog)
class SecurityRequestResponseLogAdmin(LogAdmin):
date_hierarchy = "created"
list_display = (
"created",
"label",
"type",
"action",
"ref",
"request_username",
"request_full_path",
"request_method",
"response_status_code",
"request_client_ip",
"request_elapse_time",
"request_trace_id",
"session_key",
"user_agent_os",
"user_agent_browser",
)
list_filter = [
"label",
"type",
"action",
"request_full_path",
"request_method",
"response_status_code",
"user_agent_os",
"user_agent_browser",
]
search_fields = [
"request_username",
"request_full_path",
"request_client_ip",
"request_trace_id",
"additional_json_data",
]
def additional_json_data_pretty(self, instance):
return self.pretty_print_json(instance.additional_json_data)
def get_readonly_fields(self, request, obj=None):
return super().get_readonly_fields(request, obj) + [
"additional_json_data_pretty"
]
@admin.register(ExternalApiRequestLog)
class ExternalApiRequestLogAdmin(LogAdmin):
date_hierarchy = "created"
list_display = (
"created",
"api_request_verb",
"api_url",
"api_response_status_code",
"elapsed_time",
"request_trace_id",
"request_username",
)
search_fields = [
"request_username",
"request_trace_id",
"api_request_data",
"api_response_data",
"additional_json_data",
]
list_filter = [
"api_response_status_code",
"api_request_verb",
"api_url",
]
def api_request_data_pretty(self, instance):
return self.pretty_print_json(instance.api_request_data)
def api_response_data_pretty(self, instance):
return self.pretty_print_json(instance.api_response_data)
def additional_data_pretty(self, instance):
return self.pretty_print_json(instance.additional_data)
def get_readonly_fields(self, request, obj=None):
return super().get_readonly_fields(request, obj) + [
"api_request_data_pretty",
"api_response_data_pretty",
"additional_data_pretty",
]
@admin.register(Organisation)
class OrganisationAdmin(admin.ModelAdmin):
list_display = (

85
server/vbv_lernwelt/core/external_request_logger.py Normal file
View File

@ -0,0 +1,85 @@
import time
from functools import wraps
import structlog
from vbv_lernwelt.core.models import ExternalApiRequestLog
logger = structlog.get_logger(__name__)
class ExternalApiRequestLogDecorator:
def __init__(self, base_url, store_request_data=True, store_response_data=True):
self.base_url = base_url
self.store_request_data = store_request_data
self.store_response_data = store_response_data
def __call__(self, fn):
@wraps(fn)
def wrapper(*args, **kwargs):
store_request_data = self.store_request_data
store_response_data = self.store_response_data
request_data = kwargs.get("json", None)
if not request_data:
request_data = kwargs.get("data", None)
if not request_data:
request_data = kwargs.get("params", None)
if not request_data:
request_data = args[2] if len(args) > 2 else {}
url = args[1]
context_data = kwargs.get("context_data", {}) or {}
if not store_request_data:
request_data = {"hidden": True}
logger.debug(
"try to call subhub external api",
request_url=args[1],
request_data=request_data,
base_url=self.base_url,
label="subhub_external_communication",
)
log = ExternalApiRequestLog()
log.api_url = f"{self.base_url}{url}"
log.api_request_data = request_data
log.request_trace_id = context_data.get("request_trace_id", "")
log.request_username = context_data.get("request_username", "")
log.additional_json_data = context_data
# user = get_request_user()
# if user:
# log.user = user.username
log.save()
start = time.time()
api_response = fn(*args, **kwargs)
log.elapsed_time = round(time.time() - start, 3)
log.api_request_verb = str(api_response.request.method)
log.api_response_status_code = api_response.status_code
if store_response_data:
log.api_response_data = str(api_response.text)
else:
log.api_response_data = "hidden"
log.save()
logger.info(
"call to subhub external api successful",
api_request_verb=log.api_request_verb,
api_url=log.api_url,
api_response_status_code=log.api_response_status_code,
api_time=log.elapsed_time,
api_request_data=log.api_request_data,
api_response_data=log.api_response_data,
label="subhub_external_communication",
)
return api_response
return wrapper

93
server/vbv_lernwelt/core/middleware/security.py
View File

@ -1,12 +1,13 @@
import time
import uuid
import structlog
from django.core.exceptions import PermissionDenied
from django.http import Http404
from ipware import get_client_ip
from structlog.threadlocal import bind_threadlocal, clear_threadlocal
from ua_parser import user_agent_parser
from vbv_lernwelt.core.models import SecurityRequestResponseLog
from vbv_lernwelt.importer.utils import try_parse_int
logger = structlog.get_logger(__name__)
@ -30,31 +31,69 @@ class SecurityRequestResponseLoggingMiddleware:
def create_logging_threadlocalbind(self, request):
request_username = request.user.username if hasattr(request, "user") else ""
request_trace_id = uuid.uuid4().hex
bind_threadlocal(
request_method=request.method,
request_full_path=request.get_full_path(),
request_username=request_username,
request_client_ip=request.META.get("REMOTE_ADDR"),
request_trace_id=uuid.uuid4().hex,
request_trace_id=request_trace_id,
)
def create_database_security_request_response_log(self, request, response):
return request_trace_id
def create_database_security_request_response_log(
self, request, response, elapsed_time, request_trace_id=""
):
try:
entry = SecurityRequestResponseLog()
entry.label = getattr(request, "security_request_logging", "")
entry.type = getattr(request, "type", "")
entry.request_method = request.method
entry.request_full_path = request.get_full_path()[:255]
entry.request_full_path = request.get_full_path()[0:250]
entry.request_username = (
request.user.username if hasattr(request, "user") else ""
)
entry.request_client_ip = request.META.get("REMOTE_ADDR")
entry.request_scn = getattr(request, "scn", "")
entry.response_status_code = response.status_code
entry.additional_json_data = getattr(
request, "log_additional_json_data", {}
entry.request_trace_id = request_trace_id
entry.request_elapse_time = elapsed_time
entry.save()
# save predefined fields
additional_json_data = getattr(request, "log_additional_json_data", {})
entry.category = additional_json_data.pop("category", "")[0:255].strip()
entry.action = additional_json_data.pop("action", "")[0:255].strip()
entry.name = additional_json_data.pop("name", "")[0:255].strip()
entry.ref = additional_json_data.pop("ref", "")[0:255].strip()
entry.local_url = additional_json_data.pop("local_url", "")[0:255].strip()
entry.session_key = additional_json_data.pop("session_key", "")[
0:255
].strip()
_, value = try_parse_int(additional_json_data.pop("value", 0), 0)
entry.value = value
entry.save()
user_agent = request.headers.get("User-Agent", "")
entry.user_agent = user_agent[0:4096].strip()
try:
ua_parsed = user_agent_parser.Parse(user_agent)
entry.user_agent_parsed = user_agent_parser.Parse(user_agent)
entry.user_agent_os = ua_parsed.get("os").get("family")
entry.user_agent_browser = ua_parsed.get("user_agent").get("family")
# pylint: disable=broad-except
except Exception as e:
logger.warning(
"error while parsing user agent",
label="analytics",
user_agent=user_agent,
error=str(e),
)
entry.additional_json_data = additional_json_data
entry.save()
# pylint: disable=broad-except
@ -63,18 +102,32 @@ class SecurityRequestResponseLoggingMiddleware:
def log_request_response(self, request):
clear_threadlocal()
self.create_logging_threadlocalbind(request)
request_trace_id = self.create_logging_threadlocalbind(request)
request.request_trace_id = request_trace_id
logger.info(
"url access initialized",
label="security",
)
start_time = time.time()
response = self.get_response(request)
elapsed_time = round(time.time() - start_time, 3)
security_request_logging = getattr(request, "security_request_logging", None)
try:
security_request_logging = getattr(
request, "security_request_logging", None
)
if security_request_logging:
self.create_database_security_request_response_log(request, response)
self.create_database_security_request_response_log(
request,
response,
elapsed_time=elapsed_time,
request_trace_id=request_trace_id,
)
# pylint: disable=broad-except
except Exception:
logger.warn("could not create db entry", label="security", exc_info=True)
logger.info(
"url access finished",
@ -82,6 +135,7 @@ class SecurityRequestResponseLoggingMiddleware:
response_status_code=response.status_code,
request_ratelimited=getattr(request, "limited", False),
request_finished=True,
elapsed_time=elapsed_time,
)
clear_threadlocal()
@ -90,18 +144,3 @@ class SecurityRequestResponseLoggingMiddleware:
def __call__(self, request):
return self.log_request_response(request)
def process_exception(self, request, exception):
if isinstance(exception, (Http404, PermissionDenied)):
# We don't log an exception here, and we don't set that we handled
# an error as we want the standard `request_finished` log message
# to be emitted.
return
self._raised_exception = True
logger.exception(
"request_failed",
label="security",
response_status_code=500,
)

13
server/vbv_lernwelt/core/migrations/0010_auto_20240620_1625.py
View File

@ -4,20 +4,19 @@ from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0009_country_refactor'),
("core", "0009_country_refactor"),
]
operations = [
migrations.AddField(
model_name='user',
name='birth_date',
model_name="user",
name="birth_date",
field=models.DateField(blank=True, null=True),
),
migrations.AddField(
model_name='user',
name='phone_number',
field=models.CharField(blank=True, default='', max_length=255),
model_name="user",
name="phone_number",
field=models.CharField(blank=True, default="", max_length=255),
),
]

15
server/vbv_lernwelt/core/migrations/0011_delete_securityrequestresponselog.py Normal file
View File

@ -0,0 +1,15 @@
# Generated by Django 3.2.20 on 2024-06-21 13:10
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("core", "0010_auto_20240620_1625"),
]
operations = [
migrations.DeleteModel(
name="SecurityRequestResponseLog",
),
]

126
server/vbv_lernwelt/core/migrations/0012_auto_20240621_1626.py Normal file
View File

@ -0,0 +1,126 @@
# Generated by Django 3.2.20 on 2024-06-21 14:26
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("core", "0011_delete_securityrequestresponselog"),
]
operations = [
migrations.CreateModel(
name="ExternalApiRequestLog",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("created", models.DateTimeField(auto_now_add=True)),
("api_url", models.TextField(blank=True, default="")),
("api_request_data", models.JSONField(blank=True, default=dict)),
(
"api_request_verb",
models.CharField(blank=True, default="", max_length=255),
),
("api_response_status_code", models.IntegerField(default=0)),
("api_response_data", models.TextField(blank=True, default="")),
(
"request_username",
models.CharField(blank=True, default="", max_length=255),
),
(
"request_trace_id",
models.CharField(blank=True, default="", max_length=255),
),
("elapsed_time", models.FloatField(default=0)),
("additional_json_data", models.JSONField(blank=True, default=dict)),
],
),
migrations.CreateModel(
name="SecurityRequestResponseLog",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("created", models.DateTimeField(auto_now_add=True)),
("label", models.CharField(blank=True, default="", max_length=255)),
("type", models.CharField(blank=True, default="", max_length=255)),
(
"request_trace_id",
models.CharField(blank=True, default="", max_length=255),
),
(
"request_method",
models.CharField(blank=True, default="", max_length=255),
),
(
"request_full_path",
models.CharField(blank=True, default="", max_length=255),
),
(
"request_username",
models.CharField(blank=True, default="", max_length=255),
),
(
"request_client_ip",
models.CharField(blank=True, default="", max_length=255),
),
("request_elapse_time", models.FloatField(default=0)),
(
"response_status_code",
models.CharField(blank=True, default="", max_length=255),
),
("category", models.CharField(blank=True, default="", max_length=255)),
("action", models.CharField(blank=True, default="", max_length=255)),
("name", models.CharField(blank=True, default="", max_length=255)),
("ref", models.CharField(blank=True, default="", max_length=255)),
("value", models.IntegerField(default=0)),
("local_url", models.CharField(blank=True, default="", max_length=255)),
(
"session_key",
models.CharField(blank=True, default="", max_length=255),
),
("user_agent", models.TextField(blank=True, default="")),
("user_agent_parsed", models.JSONField(blank=True, default=dict)),
(
"user_agent_os",
models.CharField(blank=True, default="", max_length=255),
),
(
"user_agent_browser",
models.CharField(blank=True, default="", max_length=255),
),
("additional_json_data", models.JSONField(blank=True, default=dict)),
],
),
migrations.AddIndex(
model_name="securityrequestresponselog",
index=models.Index(fields=["type"], name="core_securi_type_009727_idx"),
),
migrations.AddIndex(
model_name="securityrequestresponselog",
index=models.Index(fields=["label"], name="core_securi_label_dd1821_idx"),
),
migrations.AddIndex(
model_name="securityrequestresponselog",
index=models.Index(
fields=["category"], name="core_securi_categor_1b776c_idx"
),
),
migrations.AddIndex(
model_name="securityrequestresponselog",
index=models.Index(fields=["action"], name="core_securi_action_5df089_idx"),
),
]

51
server/vbv_lernwelt/core/models.py
View File

@ -154,16 +154,63 @@ class User(AbstractUser):
class SecurityRequestResponseLog(models.Model):
label = models.CharField(max_length=255, blank=True, default="")
created = models.DateTimeField(auto_now_add=True)
label = models.CharField(max_length=255, blank=True, default="")
type = models.CharField(max_length=255, blank=True, default="")
request_trace_id = models.CharField(max_length=255, blank=True, default="")
request_method = models.CharField(max_length=255, blank=True, default="")
request_full_path = models.CharField(max_length=255, blank=True, default="")
request_username = models.CharField(max_length=255, blank=True, default="")
request_client_ip = models.CharField(max_length=255, blank=True, default="")
request_elapse_time = models.FloatField(default=0)
response_status_code = models.CharField(max_length=255, blank=True, default="")
additional_json_data = JSONField(default=dict, blank=True)
category = models.CharField(max_length=255, blank=True, default="")
action = models.CharField(max_length=255, blank=True, default="")
name = models.CharField(max_length=255, blank=True, default="")
ref = models.CharField(max_length=255, blank=True, default="")
value = models.IntegerField(default=0)
local_url = models.CharField(max_length=255, blank=True, default="")
session_key = models.CharField(max_length=255, blank=True, default="")
user_agent = models.TextField(blank=True, default="")
user_agent_parsed = models.JSONField(blank=True, default=dict)
user_agent_os = models.CharField(max_length=255, blank=True, default="")
user_agent_browser = models.CharField(max_length=255, blank=True, default="")
additional_json_data = models.JSONField(default=dict, blank=True)
class Meta:
indexes = [
models.Index(fields=["type"]),
models.Index(fields=["label"]),
models.Index(fields=["category"]),
models.Index(fields=["action"]),
]
class ExternalApiRequestLog(models.Model):
created = models.DateTimeField(auto_now_add=True)
api_url = models.TextField(blank=True, default="")
api_request_data = models.JSONField(default=dict, blank=True)
api_request_verb = models.CharField(max_length=255, blank=True, default="")
api_response_status_code = models.IntegerField(default=0)
api_response_data = models.TextField(blank=True, default="")
request_username = models.CharField(max_length=255, blank=True, default="")
request_trace_id = models.CharField(max_length=255, blank=True, default="")
elapsed_time = models.FloatField(default=0)
additional_json_data = models.JSONField(default=dict, blank=True)
def __str__(self):
return f"{self.api_request_verb} {self.api_response_status_code} {self.api_url}"
class JobLog(models.Model):

0
server/vbv_lernwelt/shop/datatrans/__init__.py Normal file
View File

42
server/vbv_lernwelt/shop/datatrans/datatrans_api_client.py Normal file
View File

@ -0,0 +1,42 @@
# -*- coding: utf-8 -*-
import requests
import structlog
from django.conf import settings
from vbv_lernwelt.core.external_request_logger import ExternalApiRequestLogDecorator
logger = structlog.getLogger(__name__)
class DatatransApiClient:
def __init__(self):
self.base_url = settings.DATATRANS_API_ENDPOINT
@ExternalApiRequestLogDecorator(
base_url=settings.DATATRANS_API_ENDPOINT,
)
def _get_request(self, url, params, headers=None, context_data=None):
complete_url = self.base_url + url
return requests.get(complete_url, params=params, headers=headers)
@ExternalApiRequestLogDecorator(
base_url=settings.DATATRANS_API_ENDPOINT,
)
def _post_request(self, url, json, headers=None, context_data=None):
complete_url = self.base_url + url
return requests.post(complete_url, json=json, headers=headers)
def post_initialize_transactions(
self, json_payload: dict, context_data: dict = None
):
url = "/v1/transactions"
return self._post_request(
url,
json=json_payload,
headers={
"Authorization": f"Basic {settings.DATATRANS_BASIC_AUTH_KEY}",
"Content-Type": "application/json",
},
context_data=context_data,
)

0
server/vbv_lernwelt/shop/datatrans_fake_server.py → server/vbv_lernwelt/shop/datatrans/datatrans_fake_server.py
View File

35
server/vbv_lernwelt/shop/migrations/0015_cembra_fields.py
View File

@ -4,40 +4,39 @@ from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('shop', '0014_checkoutinformation_abacus_ssh_upload_done'),
("shop", "0014_checkoutinformation_abacus_ssh_upload_done"),
]
operations = [
migrations.AddField(
model_name='checkoutinformation',
name='birth_date',
model_name="checkoutinformation",
name="birth_date",
field=models.DateField(blank=True, null=True),
),
migrations.AddField(
model_name='checkoutinformation',
name='cembra_invoice',
model_name="checkoutinformation",
name="cembra_invoice",
field=models.BooleanField(default=False),
),
migrations.AddField(
model_name='checkoutinformation',
name='device_fingerprint_session_key',
field=models.CharField(blank=True, default='', max_length=255),
model_name="checkoutinformation",
name="device_fingerprint_session_key",
field=models.CharField(blank=True, default="", max_length=255),
),
migrations.AddField(
model_name='checkoutinformation',
name='email',
field=models.CharField(blank=True, default='', max_length=255),
model_name="checkoutinformation",
name="email",
field=models.CharField(blank=True, default="", max_length=255),
),
migrations.AddField(
model_name='checkoutinformation',
name='ip_address',
field=models.CharField(blank=True, default='', max_length=255),
model_name="checkoutinformation",
name="ip_address",
field=models.CharField(blank=True, default="", max_length=255),
),
migrations.AddField(
model_name='checkoutinformation',
name='phone_number',
field=models.CharField(blank=True, default='', max_length=255),
model_name="checkoutinformation",
name="phone_number",
field=models.CharField(blank=True, default="", max_length=255),
),
]

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

@ -7,11 +7,36 @@ import structlog
from django.conf import settings
from vbv_lernwelt.core.admin import User
from vbv_lernwelt.shop.datatrans.datatrans_api_client import DatatransApiClient
from vbv_lernwelt.shop.models import CheckoutState
logger = structlog.get_logger(__name__)
def create_context_data_log(
request,
action: str,
):
request._request.security_request_logging = "shop_view"
request._request.type = "datatrans"
request_trace_id = getattr(request._request, "request_trace_id", "")
context_data = {
"action": action,
"session_key": (request.data.get("session_key", "") or "")[0:255].strip(),
"ref": (request.data.get("ref", "") or "")[0:255].strip(),
"request_trace_id": (request_trace_id or "")[0:255].strip(),
"request_username": (
request._request.user.username if hasattr(request._request, "user") else ""
),
}
request._request.log_additional_json_data = context_data
log = logger.bind(label="shop", **context_data)
return context_data, log
class InitTransactionException(Exception):
pass
@ -54,6 +79,7 @@ def init_datatrans_transaction(
webhook_url: str,
datatrans_customer_data: dict = None,
datatrans_int_data: dict = None,
context_data: dict = None,
):
payload = {
# We use autoSettle=True, so that we don't have to settle the transaction:
@ -82,13 +108,9 @@ def init_datatrans_transaction(
logger.info("Initiating transaction", payload=payload)
response = requests.post(
url=f"{settings.DATATRANS_API_ENDPOINT}/v1/transactions",
json=payload,
headers={
"Authorization": f"Basic {settings.DATATRANS_BASIC_AUTH_KEY}",
"Content-Type": "application/json",
},
api_client = DatatransApiClient()
response = api_client.post_initialize_transactions(
json_payload=payload, context_data=context_data
)
if response.status_code == 201:

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

@ -83,13 +83,22 @@ class CheckoutAPITestCase(APITestCase):
self.assertEqual(ci.state, "ongoing")
self.assertEqual(ci.transaction_id, "1234567890")
mock_init_transaction.assert_called_once_with(
user=self.user,
amount_chf_centimes=324_30,
redirect_url_success=f"{REDIRECT_URL}/onboarding/{VV_DE_PRODUCT_SKU}/checkout/complete",
redirect_url_error=f"{REDIRECT_URL}/onboarding/{VV_DE_PRODUCT_SKU}/checkout/address?error",
redirect_url_cancel=f"{REDIRECT_URL}/",
webhook_url=f"{REDIRECT_URL}/api/shop/transaction/webhook/",
mock_init_transaction.assert_called_once()
call_kwargs = mock_init_transaction.call_args[1]
print(call_kwargs)
self.assertEqual(call_kwargs["user"], self.user)
self.assertEqual(call_kwargs["amount_chf_centimes"], 324_30)
self.assertEqual(
call_kwargs["redirect_url_success"],
f"{REDIRECT_URL}/onboarding/{VV_DE_PRODUCT_SKU}/checkout/complete",
)
self.assertEqual(
call_kwargs["redirect_url_error"],
f"{REDIRECT_URL}/onboarding/{VV_DE_PRODUCT_SKU}/checkout/address?error",
)
self.assertEqual(call_kwargs["redirect_url_cancel"], f"{REDIRECT_URL}/")
self.assertEqual(
call_kwargs["webhook_url"], f"{REDIRECT_URL}/api/shop/transaction/webhook/"
)
@patch("vbv_lernwelt.shop.views.init_datatrans_transaction")

25
server/vbv_lernwelt/shop/tests/test_datatrans_service.py
View File

@ -48,26 +48,11 @@ class DatatransServiceTest(TestCase):
self.assertEqual(1234567890, transaction_id)
# THEN
mock_post.assert_called_once_with(
url="https://api.sandbox.datatrans.com/v1/transactions",
json={
"autoSettle": True,
"amount": 324_30,
"currency": "CHF",
"language": self.user.language,
"refno": str(mock_uuid()),
"webhook": {"url": f"{REDIRECT_URL}/webhook"},
"redirect": {
"successUrl": f"{REDIRECT_URL}/success",
"errorUrl": f"{REDIRECT_URL}/error",
"cancelUrl": f"{REDIRECT_URL}/cancel",
},
},
headers={
"Authorization": "Basic BASIC_AUTH_KEY",
"Content-Type": "application/json",
},
)
mock_post.assert_called_once()
call_kwargs = mock_post.call_args[1]
print(call_kwargs)
self.assertEqual(call_kwargs["json"]["autoSettle"], True)
self.assertEqual(call_kwargs["json"]["amount"], 324_30)
@patch("vbv_lernwelt.shop.services.requests.post")
def test_init_transaction_500(self, mock_post):

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

@ -16,6 +16,7 @@ from vbv_lernwelt.shop.const import (
)
from vbv_lernwelt.shop.models import CheckoutInformation, CheckoutState, Product
from vbv_lernwelt.shop.services import (
create_context_data_log,
datatrans_state_to_checkout_state,
get_payment_url,
init_datatrans_transaction,
@ -82,10 +83,12 @@ def checkout_vv(request):
bei Browser Back redirections zu vermeiden."
"""
context_data, log = create_context_data_log(request, "checkout_vv")
sku = request.data["product"]
base_redirect_url = request.data["redirect_url"]
logger.info(f"Checkout requested: sku", user_id=request.user.id, sku=sku)
log.info(f"Checkout requested: sku", user_id=request.user.id, sku=sku)
try:
product = Product.objects.get(sku=sku)
@ -116,7 +119,6 @@ def checkout_vv(request):
datatrans_customer_data = None
datatrans_int_data = None
if with_cembra_invoice:
if "fakeapi" not in settings.DATATRANS_API_ENDPOINT:
request.user.set_increment_abacus_debitor_number()
@ -155,9 +157,11 @@ def checkout_vv(request):
webhook_url=webhook_url(base_redirect_url),
datatrans_customer_data=datatrans_customer_data,
datatrans_int_data=datatrans_int_data,
context_data=context_data,
)
except InitTransactionException as e:
if not settings.DEBUG:
log.error("Transaction initiation failed", exc_info=True, error=str(e))
capture_exception(e)
return next_step_response(
url=checkout_error_url(
@ -190,7 +194,9 @@ def checkout_vv(request):
email=email,
ip_address=ip_address,
cembra_invoice=with_cembra_invoice,
device_fingerprint_session_key=request.data["device_fingerprint_session_key"],
device_fingerprint_session_key=request.data.get(
"device_fingerprint_session_key", ""
),
# address
**request.data["address"],
)