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

Merged in feature/VBV-772-automatic-ab-bb-assignment-model (pull request #426) 

Noch nicht mergen: Feature/VBV-772 automatic ab bb assignment model

Approved-by: Daniel Egger
This commit is contained in:
Christian Cueni 2024-11-25 08:14:00 +00:00
parent d361dabd16 1a909431cf
commit 9692c71fbc
8 changed files with 476 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
scripts/reset_password_to_test.py Normal file
View File

@ -0,0 +1,24 @@
import json
import os
import sys
import django
from django.db import transaction
sys.path.append("../server")
os.environ.setdefault("IT_APP_ENVIRONMENT", "local")
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.base")
django.setup()
from vbv_lernwelt.core.models import User
# Get the user whose password you want to use as the reference
reference_user = User.objects.get(email='axel.manderbach@lernetz.ch')
reference_user.set_password('test')
reference_user.save()
# Update the password for all users
with transaction.atomic():
User.objects.update(password=reference_user.password)

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

@ -1,4 +1,4 @@
from django.contrib import admin, messages
from django.contrib import admin
from django.contrib.auth import admin as auth_admin
from django.contrib.auth import get_user_model
from django.utils.translation import gettext_lazy as _
@ -11,12 +11,6 @@ from vbv_lernwelt.core.models import (
SecurityRequestResponseLog,
)
from vbv_lernwelt.core.utils import pretty_print_json
from vbv_lernwelt.learning_mentor.services import (
create_or_sync_ausbildungsverantwortlicher as create_or_sync_av,
)
from vbv_lernwelt.learning_mentor.services import (
create_or_sync_berufsbildner as create_or_sync_bb,
)
User = get_user_model()
@ -38,45 +32,6 @@ class LogAdmin(admin.ModelAdmin):
return pretty_print_json(json_string)
@admin.action(description="Berufsbildner: Create or Sync")
def create_or_sync_berufsbildner(modeladmin, request, queryset):
# keep it easy
success = []
for user in queryset:
success.append(create_or_sync_bb(user))
if all(success):
messages.add_message(
request,
messages.SUCCESS,
"Berufsbildner erfolgreich erstellt oder synchronisiert",
)
else:
messages.add_message(
request,
messages.ERROR,
"Einige Berufsbildner konnten nicht erstellt oder synchronisiert werden",
)
@admin.action(description="Ausbildungsverantwortlicher: Create or Sync")
def create_or_sync_ausbildungsverantwortlicher(modeladmin, request, queryset):
success = []
for user in queryset:
success.append(create_or_sync_av(user))
if all(success):
messages.add_message(
request,
messages.SUCCESS,
"Ausbildungsverantwortlicher erfolgreich erstellt oder synchronisiert",
)
else:
messages.add_message(
request,
messages.ERROR,
"Einige Ausbildungsverantwortliche konnten nicht erstellt oder synchronisiert werden",
)
@admin.register(User)
class UserAdmin(auth_admin.UserAdmin):
fieldsets = (
@ -139,7 +94,6 @@ class UserAdmin(auth_admin.UserAdmin):
]
list_filter = ("is_staff", "is_superuser", "is_active", "groups", "organisation")
search_fields = ["first_name", "last_name", "email", "username", "sso_id"]
actions = [create_or_sync_berufsbildner, create_or_sync_ausbildungsverantwortlicher]
@admin.register(JobLog)

46
server/vbv_lernwelt/learning_mentor/admin.py
View File

@ -1,9 +1,39 @@
from django.contrib import admin
from django.contrib import admin, messages
from vbv_lernwelt.learning_mentor.models import (
AgentParticipantRelation,
MentorInvitation,
OrganisationSupervisor,
OrganisationSupervisortRoleType,
)
from vbv_lernwelt.learning_mentor.services import (
create_or_sync_ausbildungsverantwortlicher,
create_or_sync_berufsbildner,
)
@admin.action(description="Organisation Supervisor: Sync")
def create_or_sync_org_supervisor(_modeladmin, request, queryset):
success = []
for supervisor in queryset:
sync_fn = (
create_or_sync_berufsbildner
if supervisor.role == OrganisationSupervisortRoleType.BERUFSBILDNER.value
else create_or_sync_ausbildungsverantwortlicher
)
success.append(sync_fn(supervisor.supervisor, supervisor.organisation))
if all(success):
messages.add_message(
request,
messages.SUCCESS,
"Organisation Supervisor synchronisiert",
)
else:
messages.add_message(
request,
messages.ERROR,
"Einige Organisation Supervisors konnten nicht synchronisiert werden",
)
@admin.register(AgentParticipantRelation)
@ -33,3 +63,17 @@ class MentorInvitationAdmin(admin.ModelAdmin):
list_display = ["id", "email", "participant", "created"]
readonly_fields = ["id", "created", "email", "participant"]
search_fields = ["email"]
@admin.register(OrganisationSupervisor)
class OrganisationSupervisorAdmin(admin.ModelAdmin):
list_display = ["supervisor", "organisation", "role"]
search_fields = ["supervisor"]
raw_id_fields = [
"supervisor",
]
list_filter = ["role", "organisation"]
actions = [create_or_sync_org_supervisor]

58
server/vbv_lernwelt/learning_mentor/migrations/0011_organisationsupervisor.py Normal file
View File

@ -0,0 +1,58 @@
# Generated by Django 4.2.13 on 2024-11-20 06:22
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("core", "0012_auto_20240621_1626"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
("learning_mentor", "0010_alter_agentparticipantrelation_role"),
]
operations = [
migrations.CreateModel(
name="OrganisationSupervisor",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
(
"role",
models.CharField(
choices=[
(
"AUSBILDUNGSVERANTWORTLICHER",
"AUSBILDUNGSVERANTWORTLICHER",
),
("BERUFSBILDNER", "BERUFSBILDNER"),
],
default="AUSBILDUNGSVERANTWORTLICHER",
max_length=255,
),
),
(
"organisation",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
to="core.organisation",
),
),
(
"supervisor",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
to=settings.AUTH_USER_MODEL,
),
),
],
),
]

38
server/vbv_lernwelt/learning_mentor/models.py
View File

@ -4,7 +4,7 @@ from enum import Enum
from django.db import models
from django_extensions.db.models import TimeStampedModel
from vbv_lernwelt.core.models import User
from vbv_lernwelt.core.models import Organisation, User
from vbv_lernwelt.course.models import CourseSessionUser
@ -65,3 +65,39 @@ class MentorInvitation(TimeStampedModel):
verbose_name = "Lernbegleiter Einladung"
verbose_name_plural = "Lernbegleiter Einladungen"
unique_together = [["email", "participant"]]
class OrganisationSupervisortRoleType(Enum):
AUSBILDUNGSVERANTWORTLICHER = "AUSBILDUNGSVERANTWORTLICHER"
BERUFSBILDNER = "BERUFSBILDNER"
class OrganisationSupervisor(models.Model):
supervisor = models.ForeignKey(User, on_delete=models.CASCADE)
organisation = models.ForeignKey(Organisation, on_delete=models.CASCADE)
role = models.CharField(
max_length=255,
choices=[(t.value, t.value) for t in OrganisationSupervisortRoleType],
default=OrganisationSupervisortRoleType.AUSBILDUNGSVERANTWORTLICHER.value,
)
def save(self, *args, **kwargs):
if (
self.role
== OrganisationSupervisortRoleType.AUSBILDUNGSVERANTWORTLICHER.value
):
from vbv_lernwelt.learning_mentor.services import (
create_or_sync_ausbildungsverantwortlicher,
)
create_or_sync_ausbildungsverantwortlicher(
self.supervisor, self.organisation
)
elif self.role == OrganisationSupervisortRoleType.BERUFSBILDNER.value:
from vbv_lernwelt.learning_mentor.services import (
create_or_sync_berufsbildner,
)
create_or_sync_berufsbildner(self.supervisor, self.organisation)
super().save(*args, **kwargs)

122
server/vbv_lernwelt/learning_mentor/services.py
View File

@ -1,13 +1,15 @@
from typing import List
import structlog
from vbv_lernwelt.core.models import User
from vbv_lernwelt.core.models import Organisation, User
from vbv_lernwelt.course.consts import (
COURSE_UK,
COURSE_UK_FR,
COURSE_UK_IT,
VV_COURSE_IDS,
)
from vbv_lernwelt.course.models import CourseSessionUser
from vbv_lernwelt.course.models import Course, CourseSessionUser
from vbv_lernwelt.learning_mentor.models import (
AgentParticipantRelation,
AgentParticipantRoleType,
@ -18,50 +20,75 @@ logger = structlog.get_logger(__name__)
UK_COURSES = [COURSE_UK, COURSE_UK_FR, COURSE_UK_IT]
def create_or_sync_berufsbildner(berufsbildner: User) -> bool:
new_members = set(
CourseSessionUser.objects.filter(user__organisation=berufsbildner.organisation)
.filter(course_session__course__configuration__is_uk=True)
def users_by_org(
org: Organisation,
is_uk: bool,
courses: List[Course],
excluded_course_sessions: List[int] = None,
) -> set[CourseSessionUser]:
if not excluded_course_sessions:
excluded_course_sessions = []
return set(
CourseSessionUser.objects.filter(user__organisation=org)
.filter(course_session__course__configuration__is_uk=is_uk)
.filter(role=CourseSessionUser.Role.MEMBER.value)
.filter(course_session__course_id__in=UK_COURSES)
.exclude(course_session_id__in=[4, 5, 6])
.filter(course_session__course_id__in=courses)
.exclude(course_session_id__in=excluded_course_sessions)
)
return create_or_sync_learning_mentor(berufsbildner, new_members)
def uk_cs_users_by_org(org: Organisation) -> set[CourseSessionUser]:
return users_by_org(
org,
is_uk=True,
courses=UK_COURSES,
# ignore "Demo" course sessions
excluded_course_sessions=[4, 5, 6],
)
def vv_cs_users_by_org(org: Organisation) -> set[CourseSessionUser]:
return users_by_org(org, False, VV_COURSE_IDS)
def create_or_sync_berufsbildner(
berufsbildner: User, organisation: Organisation
) -> bool:
org_members = uk_cs_users_by_org(organisation)
return create_or_sync_learning_mentor(berufsbildner, org_members, organisation)
def create_or_sync_ausbildungsverantwortlicher(
ausbildungsverantwortlicher: User,
ausbildungsverantwortlicher: User, organisation: Organisation
) -> bool:
new_members = set(
CourseSessionUser.objects.filter(
user__organisation=ausbildungsverantwortlicher.organisation
)
.filter(course_session__course__configuration__is_uk=False)
.filter(role=CourseSessionUser.Role.MEMBER.value)
.filter(course_session__course_id__in=VV_COURSE_IDS)
org_members = vv_cs_users_by_org(organisation)
return create_or_sync_learning_mentor(
ausbildungsverantwortlicher, org_members, organisation
)
return create_or_sync_learning_mentor(ausbildungsverantwortlicher, new_members)
def create_or_sync_learning_mentor(
agent: User, new_members: set[CourseSessionUser]
agent: User, org_members: set[CourseSessionUser], organisation: Organisation
) -> bool:
logger.info(
"Creating or syncing berufsbildner",
"Creating or syncing berufsbildner/ausbildungsverantwortlicher",
berufsbildner=agent,
org=agent.organisation.name_de,
org=organisation.name_de,
)
# check if it is a valid organisation
if agent.organisation and agent.organisation.organisation_id < 4:
logger.error("Invalid organisation", org=agent.organisation)
# Check if it is a valid organisation
# ids < 4 are "andere Broker/Krankenversicherer"
if organisation and organisation.organisation_id < 4:
logger.error("Invalid organisation", org=organisation)
return False
# get existing connections
existing_members = set(agent.agentparticipantrelation_set.all())
# Get existing connections (full relation objects)
existing_relations = set(agent.agentparticipantrelation_set.all())
# add new relations that are not in existing relations
for csu in new_members:
# Add new relations that are not in existing relations
existing_members = {relation.participant for relation in existing_relations}
for csu in org_members:
if csu not in existing_members:
AgentParticipantRelation.objects.get_or_create(
agent=agent,
@ -69,9 +96,42 @@ def create_or_sync_learning_mentor(
role=AgentParticipantRoleType.BERUFSBILDNER.value,
)
# remove old relations that are not in the new relations
for relation in existing_members:
if relation.participant not in new_members:
# Remove old relations that are not in the new relations
for relation in existing_relations:
if relation.participant not in org_members:
relation.delete()
return True
def delete_berufsbildner_relation(berufsbildner: User, organisation: Organisation):
org_members = uk_cs_users_by_org(organisation)
delete_org_supervisor_relation(berufsbildner, org_members)
def delete_ausbildungsverantwortlicher_relation(
ausbildungsverantwortlicher: User, organisation: Organisation
):
org_members = vv_cs_users_by_org(organisation)
delete_org_supervisor_relation(ausbildungsverantwortlicher, org_members)
def delete_org_supervisor_relation(
agent: User,
org_members: set[CourseSessionUser],
):
# As the key berufsbildner is used in several courses, we use org_members to select the ones from the correct
# course sessions
relations_to_delete = agent.agentparticipantrelation_set.filter(
participant__in=org_members
)
# Bulk delete the identified relations
deleted_count, _ = relations_to_delete.delete()
# Log the result
logger.info(
"Deleted ausbildungsverantwortlicher relations",
agent=agent,
deleted_count=deleted_count,
)

36
server/vbv_lernwelt/learning_mentor/signals.py Normal file
View File

@ -0,0 +1,36 @@
import structlog
from django.db.models.signals import pre_delete
from django.dispatch import receiver
from vbv_lernwelt.learning_mentor.models import (
OrganisationSupervisor,
OrganisationSupervisortRoleType,
)
logger = structlog.get_logger(__name__)
# CourseSessionGroup
@receiver(
pre_delete,
sender=OrganisationSupervisor,
dispatch_uid="remove_org_supervisor_relations",
)
def remove_org_supervisor_relations(sender, instance: OrganisationSupervisor, **kwargs):
if (
instance.role
== OrganisationSupervisortRoleType.AUSBILDUNGSVERANTWORTLICHER.value
):
from vbv_lernwelt.learning_mentor.services import (
delete_ausbildungsverantwortlicher_relation,
)
delete_ausbildungsverantwortlicher_relation(
instance.supervisor, instance.organisation
)
elif instance.role == OrganisationSupervisortRoleType.BERUFSBILDNER.value:
from vbv_lernwelt.learning_mentor.services import (
delete_berufsbildner_relation,
)
delete_berufsbildner_relation(instance.supervisor, instance.organisation)

184
server/vbv_lernwelt/learning_mentor/tests/test_organisation_supervisor.py Normal file
View File

@ -0,0 +1,184 @@
from typing import Dict, List, Optional
from rest_framework.test import APITestCase
from vbv_lernwelt.core.admin import User
from vbv_lernwelt.core.models import Organisation
from vbv_lernwelt.course.creators.test_utils import (
add_course_session_user,
create_course,
create_course_session,
create_user,
)
from vbv_lernwelt.course.models import CourseConfiguration, CourseSessionUser
from vbv_lernwelt.learning_mentor.services import (
create_or_sync_learning_mentor,
users_by_org,
)
def get_completion_for_user(
completions: List[Dict[str, str]], user: User
) -> Optional[Dict[str, str]]:
for completion in completions:
if completion["user_id"] == str(user.id):
return completion
return None
class OrganisationSupervisorTestCase(APITestCase):
def setUp(self) -> None:
self.course, self.course_page = create_course("Test Course")
self.course_session = create_course_session(course=self.course, title="Test VV")
self.course_config = CourseConfiguration.objects.get(course=self.course)
self.mobi = Organisation.objects.get_or_create(
name_de="Die Mobiliar",
defaults={
"organisation_id": 100,
"name_de": "Die Mobiliar",
},
)[0]
self.baloise = Organisation.objects.get_or_create(
name_de="Baloise",
defaults={
"organisation_id": 101,
"name_de": "Baloise",
},
)[0]
self.supervisor = create_user("supervisor")
self.participant_1 = add_course_session_user(
self.course_session,
create_user("participant_1"),
role=CourseSessionUser.Role.MEMBER,
)
self.participant_2 = add_course_session_user(
self.course_session,
create_user("participant_2"),
role=CourseSessionUser.Role.MEMBER,
)
self.participant_3 = add_course_session_user(
self.course_session,
create_user("participant_3"),
role=CourseSessionUser.Role.MEMBER,
)
self.participant_4 = add_course_session_user(
self.course_session,
create_user("participant_4"),
role=CourseSessionUser.Role.MEMBER,
)
self.participant_1.user.organisation = self.mobi
self.participant_1.user.save()
self.participant_2.user.organisation = self.mobi
self.participant_2.user.save()
self.participant_3.user.organisation = self.mobi
self.participant_3.user.save()
self.participant_4.user.organisation = self.baloise
self.participant_4.user.save()
def test_add_can_berufsbildner(self) -> None:
# GIVEN
self.course_config.is_uk = True
self.course_config.save()
# WHEN
org_members = users_by_org(self.mobi, True, [self.course])
success = create_or_sync_learning_mentor(
self.supervisor, org_members, self.mobi
)
agent_participant_relations = self.supervisor.agentparticipantrelation_set.all()
# THEN
self.assertTrue(success)
self.assertEqual(len(agent_participant_relations), 3)
def test_add_cannot_berufsbildner_if_excluded(self) -> None:
# GIVEN
self.course_config.is_uk = True
self.course_config.save()
# WHEN
org_members = users_by_org(
self.mobi,
True,
[self.course],
excluded_course_sessions=[self.course_session.id],
)
success = create_or_sync_learning_mentor(
self.supervisor, org_members, self.mobi
)
agent_participant_relations = self.supervisor.agentparticipantrelation_set.all()
# THEN
self.assertTrue(success)
self.assertEqual(len(agent_participant_relations), 0)
def test_add_cannot_berufsbildner_if_not_uk(self) -> None:
# GIVEN
self.course_config.is_uk = False
self.course_config.save()
# WHEN
org_members = users_by_org(self.mobi, True, [self.course])
success = create_or_sync_learning_mentor(
self.supervisor, org_members, self.mobi
)
agent_participant_relations = self.supervisor.agentparticipantrelation_set.all()
# THEN
self.assertTrue(success)
self.assertEqual(len(agent_participant_relations), 0)
def test_add_can_ausbildungsverantwortlicher(self) -> None:
# GIVEN
self.course_config.is_uk = False
self.course_config.save()
# WHEN
org_members = users_by_org(self.mobi, False, [self.course])
success = create_or_sync_learning_mentor(
self.supervisor, org_members, self.mobi
)
agent_participant_relations = self.supervisor.agentparticipantrelation_set.all()
# THEN
self.assertTrue(success)
self.assertEqual(len(agent_participant_relations), 3)
def test_add_cannot_ausbildungsverantwortlicher_if_excluded(self) -> None:
# GIVEN
self.course_config.is_uk = False
self.course_config.save()
# WHEN
org_members = users_by_org(
self.mobi,
False,
[self.course],
excluded_course_sessions=[self.course_session.id],
)
success = create_or_sync_learning_mentor(
self.supervisor, org_members, self.mobi
)
agent_participant_relations = self.supervisor.agentparticipantrelation_set.all()
# THEN
self.assertTrue(success)
self.assertEqual(len(agent_participant_relations), 0)
def test_add_cannot_ausbildungsverantwortlicher_if_not_vv(self) -> None:
# GIVEN
self.course_config.is_uk = True
self.course_config.save()
# WHEN
org_members = users_by_org(self.mobi, False, [self.course])
success = create_or_sync_learning_mentor(
self.supervisor, org_members, self.mobi
)
agent_participant_relations = self.supervisor.agentparticipantrelation_set.all()
# THEN
self.assertTrue(success)
self.assertEqual(len(agent_participant_relations), 0)