Refactor json data handling

server/vbv_lernwelt/core/models.py

@@ -1,4 +1,5 @@
 import uuid
+from typing import Any, Dict
 
 import structlog
 from django.contrib.auth.models import AbstractUser
@@ -6,6 +7,8 @@ from django.db import models
 from django.db.models import JSONField, Max
 from django.urls import reverse
 
+from vbv_lernwelt.core.utils import sanitize_json_data_input
+
 logger = structlog.get_logger(__name__)
 
 
@@ -140,6 +143,17 @@ class User(AbstractUser):
             logger.warn("could not create avatar url", label="security", exc_info=True)
         return "/static/avatars/myvbv-default-avatar.png"
 
+    def update_additional_json_data(self, data: Dict[str, Any]):
+        # Set E-Mail notification settings for new users
+        self.additional_json_data = (
+            self.additional_json_data
+            | sanitize_json_data_input(
+                {
+                    **data,
+                }
+            )
+        )
+
     @property
     def avatar_url(self):
         return self.create_avatar_url()

server/vbv_lernwelt/core/tests/test_utils.py

@@ -1,7 +1,10 @@
+from datetime import date, datetime, time
 from unittest import skip
 
 from django.test import TestCase
 
+from vbv_lernwelt.core.utils import sanitize_json_data_input
+
 
 class SimpleTestCase(TestCase):
     def test_simple(self):
@@ -10,3 +13,32 @@ class SimpleTestCase(TestCase):
     @skip("Do not fail in pipelines")
     def test_fail(self):
         self.assertEqual(1, 2)
+
+
+class SanitizerTestCase(TestCase):
+    def test_date(self):
+        a_date = date(2021, 1, 1)
+        user_dict = {"Name": "Rascher", "Datum": a_date}
+
+        expected_sanitized_data = {"Name": "Rascher", "Datum": a_date.isoformat()}
+
+        sanitized_data = sanitize_json_data_input(user_dict)
+        self.assertEqual(sanitized_data, expected_sanitized_data)
+
+    def test_datetime(self):
+        a_date = datetime(2021, 1, 1)
+        user_dict = {"Name": "Rascher", "Datum": a_date}
+
+        expected_sanitized_data = {"Name": "Rascher", "Datum": a_date.isoformat()}
+
+        sanitized_data = sanitize_json_data_input(user_dict)
+        self.assertEqual(sanitized_data, expected_sanitized_data)
+
+    def test_time(self):
+        a_date = time(23, 59, 59)
+        user_dict = {"Name": "Rascher", "Datum": a_date}
+
+        expected_sanitized_data = {"Name": "Rascher", "Datum": a_date.isoformat()}
+
+        sanitized_data = sanitize_json_data_input(user_dict)
+        self.assertEqual(sanitized_data, expected_sanitized_data)

server/vbv_lernwelt/core/utils.py

@@ -1,5 +1,7 @@
 import json
 import re
+from datetime import date, datetime, time
+from typing import Any, Dict
 
 from django.utils.safestring import mark_safe
 from rest_framework import serializers
@@ -65,3 +67,20 @@ def safe_deque_popleft(deq, default=None):
         return deq.popleft()
     except IndexError:
         return default
+
+
+def sanitize_json_data_input(data: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Saving additional_json_data fails if the data contains datetime objects.
+    This is a quick and dirty fix to convert datetime objects to iso strings.
+    """
+    for key, value in data.items():
+        if isinstance(value, datetime):
+            data[key] = value.isoformat()
+        elif isinstance(value, date):
+            data[key] = value.isoformat()
+        elif isinstance(value, time):
+            data[key] = value.isoformat()
+        else:
+            data[key] = value
+    return data

server/vbv_lernwelt/importer/services.py

@@ -1,4 +1,4 @@
-from datetime import date, datetime, time
+from datetime import date, datetime
 from typing import Any, Dict, List
 
 import structlog
@@ -25,7 +25,7 @@ from vbv_lernwelt.learnpath.models import (
     LearningContentEdoniqTest,
 )
 from vbv_lernwelt.notify.models import NotificationCategory
-from vbv_lernwelt.sso.role_sync.services import create_user
+from vbv_lernwelt.sso.role_sync.services import create_and_update_user, create_user
 
 logger = structlog.get_logger(__name__)
 
@@ -540,8 +540,8 @@ def create_or_update_user(
     user.last_name = last_name or user.last_name
     user.username = email
 
-    sso_data = {"intermediate_sso_id": intermediate_sso_id}
-    update_user_json_data(user, sso_data)
+    user.update_additional_json_data({"intermediate_sso_id": intermediate_sso_id})
+    init_notification_settings(user)
 
     user.set_unusable_password()
     user.save()
@@ -843,9 +843,8 @@ def create_or_update_trainer(course: Course, data: Dict[str, Any], language="de"
     user.language = data["Sprache"]
 
     # create user in intermediate sso i.e. Keycloak
-    sso_data = {"intermediate_sso_id": create_user(user)}
-    update_user_json_data(user, sso_data)
-
+    create_and_update_user(user)
+    init_notification_settings(user)
     user.save()
 
     group = data["Klasse"].strip()
@@ -1004,32 +1003,12 @@ def sync_students_from_t2l(data):
         except KeyError:
             pass
 
-    update_user_json_data(user, data)
+    user.update_additional_json_data(data)
     user.save()
 
 
-def update_user_json_data(user: User, data: Dict[str, Any]):
-    # Set E-Mail notification settings for new users
-    user.additional_json_data = user.additional_json_data | sanitize_json_data_input(
-        {
-            **data,
-            "email_notification_categories": [str(NotificationCategory.INFORMATION)],
-        }
-    )
-
-
-def sanitize_json_data_input(data: Dict[str, Any]) -> Dict[str, Any]:
-    """
-    Saving additional_json_data fails if the data contains datetime objects.
-    This is a quick and dirty fix to convert datetime objects to iso strings.
-    """
-    for key, value in data.items():
-        if isinstance(value, datetime):
-            data[key] = value.isoformat()
-        elif isinstance(value, date):
-            data[key] = value.isoformat()
-        elif isinstance(value, time):
-            data[key] = value.isoformat()
-        else:
-            data[key] = value
-    return data
+def init_notification_settings(user: User):
+    data = {
+        "email_notification_categories": [str(NotificationCategory.INFORMATION)],
+    }
+    user.update_additional_json_data(data)

server/vbv_lernwelt/importer/tests/test_t2l_sync.py

@@ -1,5 +1,4 @@
 import os
-from datetime import date, datetime, time
 
 from django.test import TestCase
 
@@ -7,7 +6,6 @@ from vbv_lernwelt.course.creators.test_course import create_test_course
 from vbv_lernwelt.course.models import CourseSession, CourseSessionUser
 from vbv_lernwelt.importer.services import (
     create_or_update_student,
-    sanitize_json_data_input,
     sync_students_from_t2l,
 )
 
@@ -151,32 +149,3 @@ class SyncT2lTestCase(TestCase):
             self.fail(
                 f"SyncT2lTestCase.test_ignors_wrong_contract_number: An exception was unexpectedly raised: {str(e)}"
             )
-
-
-class SanitizerTestCase(TestCase):
-    def test_date(self):
-        a_date = date(2021, 1, 1)
-        user_dict = {"Name": "Rascher", "Datum": a_date}
-
-        expected_sanitized_data = {"Name": "Rascher", "Datum": a_date.isoformat()}
-
-        sanitized_data = sanitize_json_data_input(user_dict)
-        self.assertEqual(sanitized_data, expected_sanitized_data)
-
-    def test_datetime(self):
-        a_date = datetime(2021, 1, 1)
-        user_dict = {"Name": "Rascher", "Datum": a_date}
-
-        expected_sanitized_data = {"Name": "Rascher", "Datum": a_date.isoformat()}
-
-        sanitized_data = sanitize_json_data_input(user_dict)
-        self.assertEqual(sanitized_data, expected_sanitized_data)
-
-    def test_time(self):
-        a_date = time(23, 59, 59)
-        user_dict = {"Name": "Rascher", "Datum": a_date}
-
-        expected_sanitized_data = {"Name": "Rascher", "Datum": a_date.isoformat()}
-
-        sanitized_data = sanitize_json_data_input(user_dict)
-        self.assertEqual(sanitized_data, expected_sanitized_data)

server/vbv_lernwelt/sso/admin.py

@@ -4,17 +4,18 @@ from django.utils.translation import gettext_lazy as _
 from keycloak.exceptions import KeycloakDeleteError, KeycloakPostError
 
 from vbv_lernwelt.course.models import CourseSessionUser
-from vbv_lernwelt.importer.services import update_user_json_data
 from vbv_lernwelt.sso.models import SsoSyncError, SsoUser
-from vbv_lernwelt.sso.role_sync.services import create_user, sync_roles_for_user
+from vbv_lernwelt.sso.role_sync.services import (
+    create_and_update_user,
+    sync_roles_for_user,
+)
 
 User = get_user_model()
 
 
 def create_sso_user_from_admin(user: User, request):
     try:
-        sso_data = {"intermediate_sso_id": create_user(user)}
-        update_user_json_data(user, sso_data)
+        create_and_update_user(user)  # noqa
         user.save()
         messages.add_message(
             request, messages.SUCCESS, f"Der Bentuzer wurde in Keycloak erstellt."

server/vbv_lernwelt/sso/role_sync/services.py

@@ -94,6 +94,13 @@ def create_user(user: User):
     return ""
 
 
+def create_and_update_user(user: User, save=False):
+    sso_data = {"intermediate_sso_id": create_user(user)}
+    user.update_additional_json_data(sso_data)
+    if save:
+        user.save()
+
+
 def get_roles_for_user(user_id: str):
     if settings.OAUTH_SYNC_ROLES:
         return keycloak_admin.get_realm_roles_of_user(

server/vbv_lernwelt/sso/signals.py

@@ -15,6 +15,7 @@ from vbv_lernwelt.sso.role_sync.services import (
 logger = structlog.get_logger(__name__)
 
 
+# CourseSessionUser
 @receiver(post_delete, sender=CourseSessionUser, dispatch_uid="delete_sso_roles_in_cs")
 def remove_sso_roles_in_cs(sender, instance, **kwargs):
     _remove_sso_role(instance.user, instance.course_session.course.slug, instance.role)
@@ -42,6 +43,7 @@ def update_sso_roles_in_cs(sender, instance: CourseSessionUser, **kwargs):
                 pass
 
 
+# CourseSessionGroup
 @receiver(
     post_delete, sender=CourseSessionGroup, dispatch_uid="delete_sso_roles_in_csg"
 )