Add circle permission check, refactor

This commit is contained in:
Christian Cueni 2024-06-11 21:04:23 +02:00
parent 90393e76d0
commit 672464b8c9
5 changed files with 291 additions and 166 deletions

View File

@ -164,7 +164,7 @@ class AssignmentCompletionExportTestCase(ExportBaseTestCase):
self._check_export(wb, expected_data, 4, 6)
def test_export_multiple_cs(self):
csa = CourseSessionAssignment.objects.create(
_csa = CourseSessionAssignment.objects.create(
course_session=self.course_session_zh,
learning_content=LearningContentAssignment.objects.get(
slug=f"{self.course.slug}-lp-circle-fahrzeug-lc-überprüfen-einer-motorfahrzeug-versicherungspolice"

View File

@ -24,14 +24,15 @@ from vbv_lernwelt.course.creators.test_utils import (
)
from vbv_lernwelt.course.models import Course, CourseSession, CourseSessionUser
from vbv_lernwelt.dashboard.views import (
_get_allowed_course_session_ids_for_user,
_get_course_sessions_with_roles_for_user,
_get_mentee_count,
_get_mentor_open_tasks_count,
_get_permitted_circles_ids_for_user_and_course_session,
get_course_config,
get_course_sessions_with_roles_for_user,
)
from vbv_lernwelt.learning_mentor.models import LearningMentor
from vbv_lernwelt.learnpath.models import LearningUnit
from vbv_lernwelt.learnpath.models import Circle, LearningUnit
from vbv_lernwelt.self_evaluation_feedback.models import SelfEvaluationFeedback
@ -445,34 +446,69 @@ class GetMentorOpenTasksTestCase(BaseMentorAssignmentTestCase):
class ExportXlsTestCase(TestCase):
def setUp(self):
create_default_users()
create_test_course(include_vv=False, with_sessions=True)
create_test_course(include_vv=True, with_sessions=True)
self.ALLOWED_ROLES = ["EXPERT", "SUPERVISOR"]
def test_can_export_cs_dats(self):
# supervisor sees all cs in region
supervisor = User.objects.get(id=TEST_SUPERVISOR1_USER_ID)
requested_cs_ids = [TEST_COURSE_SESSION_ZURICH_ID, TEST_COURSE_SESSION_BERN_ID]
allowed_cs_id = _get_allowed_course_session_ids_for_user(
supervisor, requested_cs_ids
allowed_csrs_ids = _get_course_sessions_with_roles_for_user(
supervisor, self.ALLOWED_ROLES, requested_cs_ids
)
self.assertCountEqual(requested_cs_ids, allowed_cs_id)
self.assertCountEqual(requested_cs_ids, [csr.id for csr in allowed_csrs_ids])
def test_student_cannot_export_data(self):
# student cannot export any data
student = User.objects.get(id=TEST_STUDENT1_USER_ID)
requested_cs_ids = [TEST_COURSE_SESSION_ZURICH_ID]
allowed_cs_id = _get_allowed_course_session_ids_for_user(
student, requested_cs_ids
allowed_csrs_ids = _get_course_sessions_with_roles_for_user(
student, self.ALLOWED_ROLES, requested_cs_ids
)
self.assertCountEqual([], allowed_cs_id)
self.assertCountEqual([], allowed_csrs_ids)
def test_trainer_cannot_export_other_cs(self):
# trainer can only export cs where she is assigned
student = User.objects.get(email="test-trainer2@example.com")
trainer = User.objects.get(email="test-trainer2@example.com")
requested_cs_ids = [TEST_COURSE_SESSION_BERN_ID, TEST_COURSE_SESSION_ZURICH_ID]
allowed_cs_id = _get_allowed_course_session_ids_for_user(
student, requested_cs_ids
allowed_csrs_ids = _get_course_sessions_with_roles_for_user(
trainer, self.ALLOWED_ROLES, requested_cs_ids
)
self.assertCountEqual([TEST_COURSE_SESSION_ZURICH_ID], allowed_cs_id)
self.assertCountEqual(
[TEST_COURSE_SESSION_ZURICH_ID], [csr.id for csr in allowed_csrs_ids]
)
def test_trainer_can_get_circles_where_expert(self):
trainer = User.objects.get(email="test-trainer2@example.com")
circle = Circle.objects.get(slug="test-lehrgang-lp-circle-fahrzeug")
requested_cs_ids = [TEST_COURSE_SESSION_ZURICH_ID]
allowed_csrs_ids = _get_course_sessions_with_roles_for_user(
trainer, self.ALLOWED_ROLES, requested_cs_ids
)
allowed_circles = _get_permitted_circles_ids_for_user_and_course_session(
trainer, allowed_csrs_ids, [circle.id]
)
self.assertEqual(
[(TEST_COURSE_SESSION_ZURICH_ID, [circle.id])], allowed_circles
)
def test_trainer_cannot_get_circles_where_not_expert(self):
trainer = User.objects.get(email="test-trainer2@example.com")
circle = Circle.objects.get(slug="test-lehrgang-lp-circle-reisen")
requested_cs_ids = [TEST_COURSE_SESSION_ZURICH_ID]
allowed_csrs_ids = _get_course_sessions_with_roles_for_user(
trainer, self.ALLOWED_ROLES, requested_cs_ids
)
allowed_circles = _get_permitted_circles_ids_for_user_and_course_session(
trainer, allowed_csrs_ids, [circle.id]
)
self.assertEqual([(TEST_COURSE_SESSION_ZURICH_ID, [])], allowed_circles)

View File

@ -1,8 +1,9 @@
from dataclasses import asdict, dataclass
from datetime import date
from enum import Enum
from typing import List, Set
from typing import List, Set, Tuple
from django.db.models import Q
from django.http import HttpResponse
from rest_framework import status
from rest_framework.decorators import api_view
@ -32,7 +33,9 @@ from vbv_lernwelt.course_session.services.export_attendance import (
from vbv_lernwelt.course_session_group.models import CourseSessionGroup
from vbv_lernwelt.duedate.models import DueDate
from vbv_lernwelt.duedate.serializers import DueDateSerializer
from vbv_lernwelt.feedback.export import export_feedback_with_circle_restriction
from vbv_lernwelt.learning_mentor.models import LearningMentor
from vbv_lernwelt.learnpath.models import Circle
from vbv_lernwelt.self_evaluation_feedback.models import SelfEvaluationFeedback
@ -537,26 +540,54 @@ def _get_mentor_open_tasks_count(course_id: str, mentor: User) -> int:
@api_view(["POST"])
def export_attendance_as_xsl(request):
return _generate_xls_export(request, export_attendance)
circle_ids = request.data.get("circleIds", None)
requested_course_session_ids = request.data.get("courseSessionIds", [])
course_sessions_with_roles = _get_permitted_courses_sessions_for_user(
request.user, requested_course_session_ids
)
data = export_attendance(
[cs.id for cs in course_sessions_with_roles], circle_ids=circle_ids
)
return _make_excel_response(data)
@api_view(["POST"])
def export_competence_certificate_as_xsl(request):
return _generate_xls_export(request, export_competence_certificates)
def _generate_xls_export(request, export_fn) -> HttpResponse:
requested_course_session_ids = request.data.get("courseSessionIds", [])
circle_ids = request.data.get("circleIds", None)
if not requested_course_session_ids:
return Response({"error": "no_cs_ids"}, status=status.HTTP_400_BAD_REQUEST)
course_session_ids = _get_allowed_course_session_ids_for_user(
requested_course_session_ids = request.data.get("courseSessionIds", [])
course_sessions_with_roles = _get_permitted_courses_sessions_for_user(
request.user, requested_course_session_ids
)
data = export_competence_certificates(
course_sessions_with_roles, circle_ids=circle_ids
)
return _make_excel_response(data)
@api_view(["POST"])
def export_feedback_as_xsl(request):
circle_ids = request.data.get("circleIds", None)
requested_course_session_ids = request.data.get("courseSessionIds", [])
course_sessions_with_roless = _get_permitted_courses_sessions_for_user(
request.user, requested_course_session_ids
)
data = export_feedback_with_circle_restriction()
return _make_excel_response(data)
def _get_permitted_courses_sessions_for_user(
user: User, requested_coursesession_ids: List[str]
) -> List[CourseSessionWithRoles]:
ALLOWED_ROLES = ["EXPERT", "SUPERVISOR"]
user_course_sessions_with_roles = _get_course_sessions_with_roles_for_user(
user, ALLOWED_ROLES, requested_coursesession_ids
) # noqa
data = export_fn(course_session_ids, circle_ids=circle_ids)
return user_course_sessions_with_roles
def _make_excel_response(data: bytes) -> HttpResponse:
response = HttpResponse(
data,
content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
@ -565,16 +596,40 @@ def _generate_xls_export(request, export_fn) -> HttpResponse:
return response
def _get_allowed_course_session_ids_for_user(
user: User, requested_cs_ids: List[str]
) -> List[str]:
ALLOWED_ROLES = ["EXPERT", "SUPERVISOR"]
# 1. get course sessions for user with allowed roles
# 2. get overlapping course sessions with given course_session_ids
# Note: We don't care about the circle_ids as it's ok-ish that trainers could export other data
all_cs_ids_for_user = [
csr._original.id
def _get_course_sessions_with_roles_for_user(
user: User, allowed_roles: List[str], requested_cs_ids: List[str]
) -> List[CourseSessionWithRoles]:
all_cs_roles_for_user = [
csr
for csr in get_course_sessions_with_roles_for_user(user)
if any(role in ALLOWED_ROLES for role in csr.roles)
if any(role in allowed_roles for role in csr.roles)
and csr.id in requested_cs_ids
] # noqa
return list(set(requested_cs_ids) & set(all_cs_ids_for_user))
return all_cs_roles_for_user
def _get_permitted_circles_ids_for_user_and_course_session(
user: User,
user_course_sessions_with_roles: List[CourseSessionWithRoles],
requested_circle_ids: Tuple[int, int],
):
allowed_circles_for_sessions = []
for cswr in user_course_sessions_with_roles:
if "SUPERVISOR" in cswr.roles:
allowed_circles_for_sessions.append(requested_circle_ids)
else:
course_session_users = CourseSessionUser.objects.filter(
course_session=cswr.id,
user=user,
)
allowed_circles = (
Circle.objects.filter(
Q(expert__in=course_session_users) & Q(id__in=requested_circle_ids)
)
.distinct()
.values_list("id", flat=True)
)
allowed_circles_for_sessions.append((cswr.id, list(allowed_circles)))
return allowed_circles_for_sessions

View File

@ -0,0 +1,160 @@
from io import BytesIO
from itertools import groupby
from operator import attrgetter
from typing import List, Tuple
import structlog
from django.db.models import QuerySet
from openpyxl import Workbook
from vbv_lernwelt.course_session.services.export_attendance import (
make_export_filename,
sanitize_sheet_name,
)
from vbv_lernwelt.feedback.models import FeedbackResponse
logger = structlog.get_logger(__name__)
VV_FEEDBACK_QUESTIONS = [
("satisfaction", "Zufriedenheit insgesamt"),
("goal_attainment", "Zielerreichung insgesamt"),
(
"proficiency",
"Wie beurteilst du deine Sicherheit bezüglichen den Themen nach dem Circle?",
),
("preparation_task_clarity", "Waren die Praxisaufträge klar und verständlich?"),
("would_recommend", "Würdest du den Circle weiterempfehlen?"),
("course_positive_feedback", "Was hat dir besonders gut gefallen?"),
("course_negative_feedback", "Wo siehst du Verbesserungspotential?"),
]
UK_FEEDBACK_QUESTIONS = [
("satisfaction", "Zufriedenheit insgesamt"),
("goal_attainment", "Zielerreichung insgesamt"),
(
"proficiency",
"Wie beurteilst du deine Sicherheit bezüglichen den Themen nach dem Kurs?",
),
(
"preparation_task_clarity",
"Waren die Vorbereitungsaufträge klar und verständlich?",
),
(
"instructor_competence",
"Wie beurteilst du die Themensicherheit und Fachkompetenz des Kursleiters/der Kursleiterin?",
),
(
"instructor_respect",
"Wurden Fragen und Anregungen der Kursteilnehmenden ernst genommen und aufgegriffen?",
),
(
"instructor_open_feedback",
"Was möchtest du dem Kursleiter/der Kursleiterin sonst noch sagen?",
),
("would_recommend", "Würdest du den Kurs weiterempfehlen?"),
("course_positive_feedback", "Was hat dir besonders gut gefallen?"),
("course_negative_feedback", "Wo siehst du Verbesserungspotential?"),
]
def export_feedback(course_session_ids: list[str], save_as_file: bool, circles=None):
"""
Export for django view, all circles are allowed
"""
if circles:
feedback_unordered = FeedbackResponse.objects.filter(
course_session_id__in=course_session_ids,
circle__in=circles,
submitted=True,
)
else:
feedback_unordered = FeedbackResponse.objects.filter(
course_session_id__in=course_session_ids,
submitted=True,
)
return _generate_feedback_export(feedback_unordered, save_as_file)
def export_feedback_with_circle_restriction(
course_sessions_with_circles: List[Tuple[int, List[int]]], save_as_file: bool
):
"""
Export for user export, only circles in specified course sessions are allowed
"""
feedback_unordered = FeedbackResponse.objects.none()
for course_session_with_circles in course_sessions_with_circles:
feedback_unordered = feedback_unordered | FeedbackResponse.objects.filter(
course_session_id__in=course_session_with_circles[0],
circle__in=course_session_with_circles[1],
submitted=True,
)
return _generate_feedback_export(feedback_unordered, save_as_file)
def _generate_feedback_export(feedback_unordered: QuerySet, save_as_file: bool):
wb = Workbook()
# remove the first sheet is just easier than keeping track of the active sheet
wb.remove(wb.active)
feedbacks = feedback_unordered.order_by("circle", "course_session", "updated_at")
grouped_feedbacks = groupby(feedbacks, key=attrgetter("circle"))
for circle, group_feedbacks in grouped_feedbacks:
group_feedbacks = list(group_feedbacks)
logger.debug(
"export_feedback_for_circle",
data={
"circle": circle.id,
"count": len(group_feedbacks),
},
label="feedback_export",
)
_create_sheet(wb, circle.title, group_feedbacks)
if save_as_file:
wb.save(make_export_filename(name="feedback_export"))
else:
output = BytesIO()
wb.save(output)
output.seek(0)
return output.getvalue()
def _create_sheet(wb: Workbook, title: str, data: list[FeedbackResponse]):
sheet = wb.create_sheet(title=sanitize_sheet_name(title))
if len(data) == 0:
return sheet
# we instruct the users not to mix exports of different courses, so we can assume the questions are the same and of the first type
question_data = (
UK_FEEDBACK_QUESTIONS
if data[0].data["feedback_type"] == "uk"
else VV_FEEDBACK_QUESTIONS
)
# add header
sheet.cell(row=1, column=1, value="Durchführung")
sheet.cell(row=1, column=2, value="Datum")
questions = [q[1] for q in question_data]
for col_idx, title in enumerate(questions, start=3):
sheet.cell(row=1, column=col_idx, value=title)
_add_rows(sheet, data, question_data)
return sheet
def _add_rows(sheet, data, question_data):
for row_idx, feedback in enumerate(data, start=2):
sheet.cell(row=row_idx, column=1, value=feedback.course_session.title)
sheet.cell(
row=row_idx, column=2, value=feedback.updated_at.date().strftime("%d.%m.%Y")
)
for col_idx, question in enumerate(question_data, start=3):
response = feedback.data.get(question[0], "")
sheet.cell(row=row_idx, column=col_idx, value=response)

View File

@ -1,19 +1,13 @@
from io import BytesIO
from itertools import groupby
from operator import attrgetter
from typing import Union
import structlog
from django.http import HttpResponse
from openpyxl import Workbook
from vbv_lernwelt.core.models import User
from vbv_lernwelt.course.models import CourseCompletionStatus, CourseSession
from vbv_lernwelt.course.services import mark_course_completion
from vbv_lernwelt.course_session.services.export_attendance import (
make_export_filename,
sanitize_sheet_name,
)
from vbv_lernwelt.course_session.services.export_attendance import make_export_filename
from vbv_lernwelt.feedback.export import export_feedback
from vbv_lernwelt.feedback.models import FeedbackResponse
from vbv_lernwelt.learnpath.models import (
LearningContentFeedbackUK,
@ -22,47 +16,6 @@ from vbv_lernwelt.learnpath.models import (
logger = structlog.get_logger(__name__)
VV_FEEDBACK_QUESTIONS = [
("satisfaction", "Zufriedenheit insgesamt"),
("goal_attainment", "Zielerreichung insgesamt"),
(
"proficiency",
"Wie beurteilst du deine Sicherheit bezüglichen den Themen nach dem Circle?",
),
("preparation_task_clarity", "Waren die Praxisaufträge klar und verständlich?"),
("would_recommend", "Würdest du den Circle weiterempfehlen?"),
("course_positive_feedback", "Was hat dir besonders gut gefallen?"),
("course_negative_feedback", "Wo siehst du Verbesserungspotential?"),
]
UK_FEEDBACK_QUESTIONS = [
("satisfaction", "Zufriedenheit insgesamt"),
("goal_attainment", "Zielerreichung insgesamt"),
(
"proficiency",
"Wie beurteilst du deine Sicherheit bezüglichen den Themen nach dem Kurs?",
),
(
"preparation_task_clarity",
"Waren die Vorbereitungsaufträge klar und verständlich?",
),
(
"instructor_competence",
"Wie beurteilst du die Themensicherheit und Fachkompetenz des Kursleiters/der Kursleiterin?",
),
(
"instructor_respect",
"Wurden Fragen und Anregungen der Kursteilnehmenden ernst genommen und aufgegriffen?",
),
(
"instructor_open_feedback",
"Was möchtest du dem Kursleiter/der Kursleiterin sonst noch sagen?",
),
("would_recommend", "Würdest du den Kurs weiterempfehlen?"),
("course_positive_feedback", "Was hat dir besonders gut gefallen?"),
("course_negative_feedback", "Wo siehst du Verbesserungspotential?"),
]
def update_feedback_response(
feedback_user: User,
@ -154,85 +107,6 @@ def initial_data_for_feedback_page(
return {}
def export_feedback(course_session_ids: list[str], save_as_file: bool, circles=None):
wb = Workbook()
# remove the first sheet is just easier than keeping track of the active sheet
wb.remove(wb.active)
if circles:
feedback_unordered = FeedbackResponse.objects.filter(
course_session_id__in=course_session_ids,
circle__in=circles,
submitted=True,
)
else:
feedback_unordered = FeedbackResponse.objects.filter(
course_session_id__in=course_session_ids,
submitted=True,
)
feedbacks = feedback_unordered.order_by("circle", "course_session", "updated_at")
grouped_feedbacks = groupby(feedbacks, key=attrgetter("circle"))
for circle, group_feedbacks in grouped_feedbacks:
group_feedbacks = list(group_feedbacks)
logger.debug(
"export_feedback_for_circle",
data={
"circle": circle.id,
"course_session_ids": course_session_ids,
"count": len(group_feedbacks),
},
label="feedback_export",
)
_create_sheet(wb, circle.title, group_feedbacks)
if save_as_file:
wb.save(make_export_filename(name="feedback_export"))
else:
output = BytesIO()
wb.save(output)
output.seek(0)
return output.getvalue()
def _create_sheet(wb: Workbook, title: str, data: list[FeedbackResponse]):
sheet = wb.create_sheet(title=sanitize_sheet_name(title))
if len(data) == 0:
return sheet
# we instruct the users not to mix exports of different courses, so we can assume the questions are the same and of the first type
question_data = (
UK_FEEDBACK_QUESTIONS
if data[0].data["feedback_type"] == "uk"
else VV_FEEDBACK_QUESTIONS
)
# add header
sheet.cell(row=1, column=1, value="Durchführung")
sheet.cell(row=1, column=2, value="Datum")
questions = [q[1] for q in question_data]
for col_idx, title in enumerate(questions, start=3):
sheet.cell(row=1, column=col_idx, value=title)
_add_rows(sheet, data, question_data)
return sheet
def _add_rows(sheet, data, question_data):
for row_idx, feedback in enumerate(data, start=2):
sheet.cell(row=row_idx, column=1, value=feedback.course_session.title)
sheet.cell(
row=row_idx, column=2, value=feedback.updated_at.date().strftime("%d.%m.%Y")
)
for col_idx, question in enumerate(question_data, start=3):
response = feedback.data.get(question[0], "")
sheet.cell(row=row_idx, column=col_idx, value=response)
# used as admin action, that's why it's not in the views.py
def get_feedbacks_for_course_sessions(_modeladmin, _request, queryset):
file_name = "feedback_export_durchfuehrungen"