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

Add tests, refactor services

This commit is contained in:
Christian Cueni 2024-11-20 13:20:45 +01:00
parent 5e84490703
commit 6ca8469455
4 changed files with 237 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -16,7 +16,11 @@ from vbv_lernwelt.learning_mentor.services import (
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
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(
@ -65,9 +69,7 @@ class MentorInvitationAdmin(admin.ModelAdmin):
class OrganisationSupervisorAdmin(admin.ModelAdmin):
list_display = ["supervisor", "organisation", "role"]
search_fields = [
"supervisor"
]
search_fields = ["supervisor"]
raw_id_fields = [
"supervisor",

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

@ -83,27 +83,41 @@ class OrganisationSupervisor(models.Model):
)
def save(self, *args, **kwargs):
if self.role == OrganisationSupervisortRoleType.AUSBILDUNGSVERANTWORTLICHER.value:
from vbv_lernwelt.core.admin import (
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)
create_or_sync_ausbildungsverantwortlicher(
self.supervisor, self.organisation
)
else:
from vbv_lernwelt.learning_mentor.services import (
create_or_sync_berufsbildner,
)
create_or_sync_berufsbildner(self.supervisor, self.organisation)
super().save(*args, **kwargs)
def delete(self, *args, **kwargs):
if self.role == OrganisationSupervisortRoleType.AUSBILDUNGSVERANTWORTLICHER.value:
from vbv_lernwelt.core.admin import (
if (
self.role
== OrganisationSupervisortRoleType.AUSBILDUNGSVERANTWORTLICHER.value
):
from vbv_lernwelt.learning_mentor.services import (
delete_ausbildungsverantwortlicher_relation,
)
delete_ausbildungsverantwortlicher_relation(self.supervisor, self.organisation)
delete_ausbildungsverantwortlicher_relation(
self.supervisor, self.organisation
)
else:
from vbv_lernwelt.learning_mentor.services import (
delete_berufsbildner_relation,
)
delete_berufsbildner_relation(self.supervisor, self.organisation)
super().delete(*args, **kwargs)

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

@ -1,3 +1,5 @@
from typing import List
import structlog
from vbv_lernwelt.core.models import Organisation, User
@ -7,7 +9,7 @@ from vbv_lernwelt.course.consts import (
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,60 @@ logger = structlog.get_logger(__name__)
UK_COURSES = [COURSE_UK, COURSE_UK_FR, COURSE_UK_IT]
def uk_cs_users_by_org(org: Organisation) -> set[CourseSessionUser]:
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=True)
.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)
)
def uk_cs_users_by_org(org: Organisation) -> set[CourseSessionUser]:
return users_by_org(org, True, UK_COURSES, excluded_course_sessions=[4, 5, 6])
def vv_cs_users_by_org(org: Organisation) -> set[CourseSessionUser]:
return set(
CourseSessionUser.objects.filter(user__organisation=org)
.filter(course_session__course__configuration__is_uk=False)
.filter(role=CourseSessionUser.Role.MEMBER.value)
.filter(course_session__course_id__in=VV_COURSE_IDS)
)
return users_by_org(org, False, VV_COURSE_IDS)
def create_or_sync_berufsbildner(berufsbildner: User, organisation: Organisation) -> bool:
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)
return create_or_sync_learning_mentor(berufsbildner, org_members, organisation)
def create_or_sync_ausbildungsverantwortlicher(
ausbildungsverantwortlicher: User,
organisation: Organisation
ausbildungsverantwortlicher: User, organisation: Organisation
) -> bool:
org_members = vv_cs_users_by_org(organisation)
return create_or_sync_learning_mentor(ausbildungsverantwortlicher, org_members)
return create_or_sync_learning_mentor(
ausbildungsverantwortlicher, org_members, organisation
)
def create_or_sync_learning_mentor(
agent: User, org_members: set[CourseSessionUser]
agent: User, org_members: set[CourseSessionUser], organisation: Organisation
) -> bool:
logger.info(
"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)
if organisation and organisation.organisation_id < 4:
logger.error("Invalid organisation", org=organisation)
return False
# Get existing connections (full relation objects)
@ -85,12 +97,14 @@ def create_or_sync_learning_mentor(
return True
def delete_berufsbildner_relation(berufsbildner: User, organisation: Organisation) -> bool:
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) -> bool:
def delete_ausbildungsverantwortlicher_relation(
ausbildungsverantwortlicher: User, organisation: Organisation
):
org_members = vv_cs_users_by_org(organisation)
delete_org_supervisor_relation(ausbildungsverantwortlicher, org_members)
@ -101,7 +115,9 @@ def delete_org_supervisor_relation(
):
# 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)
relations_to_delete = agent.agentparticipantrelation_set.filter(
participant__in=org_members
)
# Bulk delete the identified relations
deleted_count, _ = relations_to_delete.delete()

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

@ -0,0 +1,172 @@
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(name_de="Die Mobiliar")
self.baloise = Organisation.objects.get(name_de="Baloise")
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)