wip: Split up code, add attendance tests [skip ci]

This commit is contained in:
Christian Cueni 2024-06-03 15:52:09 +02:00
parent 0cad9666c5
commit 54d77264cb
9 changed files with 425 additions and 227 deletions

View File

@ -0,0 +1,227 @@
from dataclasses import dataclass
from io import BytesIO
import structlog
from openpyxl import Workbook
from vbv_lernwelt.assignment.models import (
Assignment,
AssignmentCompletion,
AssignmentType,
)
from vbv_lernwelt.course.models import CourseSession, CourseSessionUser
from vbv_lernwelt.course_session.models import (
CourseSessionAssignment,
CourseSessionEdoniqTest,
)
from vbv_lernwelt.course_session.services.export_attendance import (
add_user_export_data,
add_user_headers,
get_ordered_csus_by_course_session,
group_by_session_title,
make_export_filename,
sanitize_sheet_name,
)
from vbv_lernwelt.duedate.models import DueDate
from vbv_lernwelt.learnpath.models import LearningContent
logger = structlog.get_logger(__name__)
@dataclass
class CompetenceCertificateElement:
assignment: Assignment
date: DueDate
learning_content: LearningContent
course_session: CourseSession
def export_competence_certificates(
course_session_ids: list[str],
circle_ids: list[int] = None,
save_as_file: bool = False,
):
if len(course_session_ids) == 0:
return
COMPETENCE_ASSIGNMENT_TYPES = [
AssignmentType.CASEWORK.value,
AssignmentType.EDONIQ_TEST.value,
]
wb = Workbook()
# remove the first sheet is just easier than keeping track of the active sheet
wb.remove(wb.active)
competence_certificate_elements = _get_competence_certificate_elements(
course_session_ids
)
assignemnt_completions = AssignmentCompletion.objects.filter(
course_session_id__in=course_session_ids,
assignment__assignment_type__in=COMPETENCE_ASSIGNMENT_TYPES,
).order_by("course_session", "assignment")
# group all by the sessions title {session_id1: [...], session_id2: [...], ...}
grouped_cs_users = get_ordered_csus_by_course_session(course_session_ids)
grouped_cce = group_by_session_title(competence_certificate_elements)
grouped_ac = group_by_session_title(assignemnt_completions)
# create a sheet for each course session
for course_session_title, cs_users in grouped_cs_users.items():
logger.debug(
"export_assignment_completion",
data={
"course_session": course_session_title,
},
label="assignment_export",
)
_create_sheet(
wb,
course_session_title,
cs_users,
grouped_cce[course_session_title],
grouped_ac[course_session_title],
circle_ids,
)
if save_as_file:
wb.save(make_export_filename(name="competence_certificate_export"))
else:
output = BytesIO()
wb.save(output)
output.seek(0)
return output.getvalue()
def _create_sheet(
wb: Workbook,
title: str,
users: list[CourseSessionUser],
competence_certificate_element: list[CompetenceCertificateElement],
assignment_completions: list[AssignmentCompletion],
circle_ids: list[int],
):
sheet = wb.create_sheet(title=sanitize_sheet_name(title))
if len(users) == 0:
return sheet
# headers
# common user headers, Circle <title> <learningcontenttitle> bestanden, Circle <title> <learningcontenttitle> Resultat, ...
col_idx = add_user_headers(sheet)
ordered_assignement_ids = (
[]
) # keep track of the order of the columns when adding the rows
for cse in competence_certificate_element:
circle = cse.learning_content.get_circle()
if circle_ids and circle.id not in circle_ids:
continue
col_prefix = f'Circle "{circle.title}" {cse.learning_content.title} '
sheet.cell(
row=1,
column=col_idx,
value=f"{col_prefix} bestanden",
)
sheet.cell(
row=1,
column=col_idx + 1,
value=f"{col_prefix} Resultat %",
)
ordered_assignement_ids.append(cse.assignment.id)
col_idx += 2
# add rows with user results
_add_rows(sheet, users, ordered_assignement_ids, assignment_completions)
return sheet
def _add_rows(
sheet,
users: list[CourseSessionUser],
ordered_assignement_ids,
assignment_completions,
):
for row_idx, user in enumerate(users, start=2):
col_idx = add_user_export_data(sheet, user, row_idx)
for assignment_id in ordered_assignement_ids:
# get the completion for the user and the assignment
user_acs = [
ac
for ac in assignment_completions
if ac.assignment_id == assignment_id and ac.assignment_user == user.user
]
user_ac = user_acs[0] if user_acs else None
if user_ac:
status_text = (
"Bestanden" if user_ac.evaluation_passed else "Nicht bestanden"
)
sheet.cell(row=row_idx, column=col_idx, value=status_text)
try:
sheet.cell(
row=row_idx,
column=col_idx + 1,
value=round(
100
* user_ac.evaluation_points
/ user_ac.evaluation_max_points
),
)
except (ZeroDivisionError, TypeError):
sheet.cell(row=row_idx, column=col_idx + 1, value="Keine Daten")
else:
sheet.cell(row=row_idx, column=col_idx, value="Keine Daten")
sheet.cell(row=row_idx, column=col_idx + 1, value="Keine Daten")
col_idx += 2
def _get_competence_certificate_elements(
course_session_ids: list[str],
) -> list[CompetenceCertificateElement]:
course_session_assignments = CourseSessionAssignment.objects.filter(
course_session__id__in=course_session_ids,
learning_content__content_assignment__competence_certificate__isnull=False,
).order_by("course_session", "submission_deadline__start")
course_session_edoniqtests = CourseSessionEdoniqTest.objects.filter(
course_session__id__in=course_session_ids,
learning_content__content_assignment__competence_certificate__isnull=False,
).order_by("course_session", "deadline__start")
cse = [
CompetenceCertificateElement(
assignment=csa.learning_content.content_assignment,
date=csa.submission_deadline,
learning_content=csa.learning_content,
course_session=csa.course_session,
)
for csa in course_session_assignments
]
cse += [
CompetenceCertificateElement(
assignment=cset.learning_content.content_assignment,
date=cset.deadline,
learning_content=cset.learning_content,
course_session=cset.course_session,
)
for cset in course_session_edoniqtests
]
# order by course_session and submission_deadline
cse.sort(key=lambda x: (x.course_session.title, x.date.start))
return cse

View File

@ -1,7 +1,7 @@
import djclick as click
import structlog
from vbv_lernwelt.assignment.services import export_competence_certificates
from vbv_lernwelt.assignment.export import export_competence_certificates
logger = structlog.get_logger(__name__)

View File

@ -1,10 +1,7 @@
from copy import deepcopy
from dataclasses import dataclass
from io import BytesIO
import structlog
from django.utils import timezone
from openpyxl import Workbook
from rest_framework import serializers
from wagtail.models import Page
@ -19,39 +16,13 @@ from vbv_lernwelt.assignment.models import (
)
from vbv_lernwelt.core.models import User
from vbv_lernwelt.core.utils import find_first
from vbv_lernwelt.course.models import (
CourseCompletionStatus,
CourseSession,
CourseSessionUser,
)
from vbv_lernwelt.course.models import CourseCompletionStatus, CourseSession
from vbv_lernwelt.course.services import mark_course_completion
from vbv_lernwelt.course_session.models import (
CourseSessionAssignment,
CourseSessionEdoniqTest,
)
from vbv_lernwelt.course_session.services.export import (
add_user_export_data,
add_user_headers,
get_ordered_csus_by_course_session,
group_by_session_title,
make_export_filename,
sanitize_sheet_name,
)
from vbv_lernwelt.duedate.models import DueDate
from vbv_lernwelt.learnpath.models import LearningContent
from vbv_lernwelt.notify.services import NotificationService
logger = structlog.get_logger(__name__)
@dataclass
class CompetenceCertificateElement:
assignment: Assignment
date: DueDate
learning_content: LearningContent
course_session: CourseSession
def update_assignment_completion(
assignment_user: User,
assignment: Assignment,
@ -300,194 +271,3 @@ def _remove_unknown_entries(assignment, completion_data):
key: value for key, value in completion_data.items() if key in input_task_ids
}
return filtered_completion_data
def export_competence_certificates(
course_session_ids: list[str],
circle_ids: list[int] = None,
save_as_file: bool = False,
):
if len(course_session_ids) == 0:
return
COMPETENCE_ASSIGNMENT_TYPES = [
AssignmentType.CASEWORK.value,
AssignmentType.EDONIQ_TEST.value,
]
wb = Workbook()
# remove the first sheet is just easier than keeping track of the active sheet
wb.remove(wb.active)
competence_certificate_elements = _get_competence_certificate_elements(
course_session_ids
)
assignemnt_completions = AssignmentCompletion.objects.filter(
course_session_id__in=course_session_ids,
assignment__assignment_type__in=COMPETENCE_ASSIGNMENT_TYPES,
).order_by("course_session", "assignment")
# group all by the sessions title {session_id1: [...], session_id2: [...], ...}
grouped_cs_users = get_ordered_csus_by_course_session(course_session_ids)
grouped_cce = group_by_session_title(competence_certificate_elements)
grouped_ac = group_by_session_title(assignemnt_completions)
# create a sheet for each course session
for course_session, cs_users in grouped_cs_users.items():
logger.debug(
"export_assignment_completion",
data={
"course_session": course_session,
},
label="assignment_export",
)
_create_sheet(
wb,
course_session,
cs_users,
grouped_cce[course_session],
grouped_ac[course_session],
circle_ids,
)
if save_as_file:
wb.save(make_export_filename(name="competence_certificate_export"))
else:
output = BytesIO()
wb.save(output)
output.seek(0)
return output.getvalue()
def _create_sheet(
wb: Workbook,
title: str,
users: list[CourseSessionUser],
competence_certificate_element: list[CompetenceCertificateElement],
assignment_completions: list[AssignmentCompletion],
circle_ids: list[int],
):
sheet = wb.create_sheet(title=sanitize_sheet_name(title))
if len(users) == 0:
return sheet
# headers
# common user headers, Circle <title> <learningcontenttitle> bestanden, Circle <title> <learningcontenttitle> Resultat, ...
col_idx = add_user_headers(sheet)
ordered_assignement_ids = (
[]
) # keep track of the order of the columns when adding the rows
for cse in competence_certificate_element:
circle = cse.learning_content.get_circle()
if circle_ids and circle.id not in circle_ids:
continue
col_prefix = f'Circle "{circle.title}" {cse.learning_content.title} '
sheet.cell(
row=1,
column=col_idx,
value=f"{col_prefix} bestanden",
)
sheet.cell(
row=1,
column=col_idx + 1,
value=f"{col_prefix} Resultat %",
)
ordered_assignement_ids.append(cse.assignment.id)
col_idx += 2
# add rows with user results
_add_rows(sheet, users, ordered_assignement_ids, assignment_completions)
return sheet
def _add_rows(
sheet,
users: list[CourseSessionUser],
ordered_assignement_ids,
assignment_completions,
):
for row_idx, user in enumerate(users, start=2):
col_idx = add_user_export_data(sheet, user, row_idx)
for assignment_id in ordered_assignement_ids:
# get the completion for the user and the assignment
user_acs = [
ac
for ac in assignment_completions
if ac.assignment_id == assignment_id and ac.assignment_user == user.user
]
user_ac = user_acs[0] if user_acs else None
if user_ac:
status_text = (
"Bestanden" if user_ac.evaluation_passed else "Nicht bestanden"
)
sheet.cell(row=row_idx, column=col_idx, value=status_text)
try:
sheet.cell(
row=row_idx,
column=col_idx + 1,
value=round(
100
* user_ac.evaluation_points
/ user_ac.evaluation_max_points
),
)
except (ZeroDivisionError, TypeError):
sheet.cell(row=row_idx, column=col_idx + 1, value="Keine Daten")
else:
sheet.cell(row=row_idx, column=col_idx, value="Keine Daten")
sheet.cell(row=row_idx, column=col_idx + 1, value="Keine Daten")
col_idx += 2
def _get_competence_certificate_elements(
course_session_ids: list[str],
) -> list[CompetenceCertificateElement]:
course_session_assignments = CourseSessionAssignment.objects.filter(
course_session__id__in=course_session_ids,
learning_content__content_assignment__competence_certificate__isnull=False,
).order_by("course_session", "submission_deadline__start")
course_session_edoniqtests = CourseSessionEdoniqTest.objects.filter(
course_session__id__in=course_session_ids,
learning_content__content_assignment__competence_certificate__isnull=False,
).order_by("course_session", "deadline__start")
cse = [
CompetenceCertificateElement(
assignment=csa.learning_content.content_assignment,
date=csa.submission_deadline,
learning_content=csa.learning_content,
course_session=csa.course_session,
)
for csa in course_session_assignments
]
cse += [
CompetenceCertificateElement(
assignment=cset.learning_content.content_assignment,
date=cset.deadline,
learning_content=cset.learning_content,
course_session=cset.course_session,
)
for cset in course_session_edoniqtests
]
# order by course_session and submission_deadline
cse.sort(key=lambda x: (x.course_session.title, x.date.start))
return cse

View File

@ -0,0 +1,56 @@
from django.test import TestCase
from vbv_lernwelt.assignment.models import Assignment
from vbv_lernwelt.assignment.services import update_assignment_completion
from vbv_lernwelt.core.constants import TEST_STUDENT1_USER_ID, TEST_STUDENT2_USER_ID
from vbv_lernwelt.core.create_default_users import create_default_users
from vbv_lernwelt.core.models import User
from vbv_lernwelt.course.creators.test_course import create_test_course
from vbv_lernwelt.course.models import CourseSession
class AttendanceExportTestCase(TestCase):
def setUp(self):
create_default_users()
self.course = create_test_course(include_vv=False, with_sessions=True)
self.course_session_be = CourseSession.objects.get(title="Test Bern 2022 a")
self.course_session_zh = CourseSession.objects.get(title="Test Zürich 2022 a")
self.attendance_course_be = (
self.course_session_be.coursesessionattendancecourse_set.first()
)
some = (
self.course.coursepage.get_descendants()
.exact_type(Assignment)
.filter(assignment__assignment_type="CASEWORK")
)
self.assignment = (
self.course.coursepage.get_descendants()
.exact_type(Assignment)
.filter(assignment__assignment_type="CASEWORK")
.first()
.specific
)
self.trainer = User.objects.get(username="admin")
self.test_student1 = User.objects.get(id=TEST_STUDENT1_USER_ID)
self.test_student1.additional_json_data = {"Lehrvertragsnummer": 1234567890}
self.test_student1.save()
self.test_student2 = User.objects.get(id=TEST_STUDENT2_USER_ID)
self.test_student2.additional_json_data = {"Lehrvertragsnummer": 1987654321}
self.test_student2.save()
test_student3 = User.objects.get(email="test-student3@example.com")
update_assignment_completion(
assignment_user=self.test_student1,
assignment=self.assignment,
course_session=self.course_session_be,
completion_data={},
evaluation_points=20,
)
def test_attendance_export_single_cs(self):
self.assertTrue(True)

View File

@ -1,7 +1,7 @@
import djclick as click
import structlog
from vbv_lernwelt.course_session.services.export import export_attendance
from vbv_lernwelt.course_session.services.export_attendance import export_attendance
logger = structlog.get_logger(__name__)

View File

@ -28,7 +28,7 @@ def export_attendance(
grouped_cs_users = get_ordered_csus_by_course_session(course_session_ids)
# create dict with course_session as key and list of attendance_courses as value. Easier to access in the loop
# create dict with course_session_title as key and list of attendance_courses as value. Easier to access in the loop
grouped_attendance_course = {
key: list(group)
for key, group in groupby(

View File

@ -0,0 +1,135 @@
import io
from django.test import TestCase
from openpyxl import load_workbook
from vbv_lernwelt.core.constants import TEST_STUDENT1_USER_ID, TEST_STUDENT2_USER_ID
from vbv_lernwelt.core.create_default_users import create_default_users
from vbv_lernwelt.core.models import User
from vbv_lernwelt.course.creators.test_course import create_test_course
from vbv_lernwelt.course.models import CourseSession
from vbv_lernwelt.course_session.services.export_attendance import export_attendance
class AttendanceExportTestCase(TestCase):
def setUp(self):
create_default_users()
create_test_course(include_vv=False, with_sessions=True)
self.course_session_be = CourseSession.objects.get(title="Test Bern 2022 a")
self.course_session_zh = CourseSession.objects.get(title="Test Zürich 2022 a")
self.attendance_course_be = (
self.course_session_be.coursesessionattendancecourse_set.first()
)
self.attendance_course_zh = (
self.course_session_zh.coursesessionattendancecourse_set.first()
)
self.test_student1 = User.objects.get(id=TEST_STUDENT1_USER_ID)
self.test_student1.additional_json_data = {"Lehrvertragsnummer": 1234567890}
self.test_student1.save()
self.test_student2 = User.objects.get(id=TEST_STUDENT2_USER_ID)
self.test_student2.additional_json_data = {"Lehrvertragsnummer": 1987654321}
self.test_student2.save()
test_student3 = User.objects.get(email="test-student3@example.com")
self.attendance_course_be.attendance_user_list = [
{
"email": self.test_student1.email,
"status": "PRESENT",
"user_id": str(self.test_student1.id),
"last_name": self.test_student1.last_name,
"first_name": self.test_student1.first_name,
}
]
self.expected_data_be = [
self._make_header(self.attendance_course_be),
[
self.test_student1.first_name,
self.test_student1.last_name,
self.test_student1.email,
self.test_student1.additional_json_data["Lehrvertragsnummer"],
"Anwesend",
],
[
self.test_student2.first_name,
self.test_student2.last_name,
self.test_student2.email,
self.test_student2.additional_json_data["Lehrvertragsnummer"],
"Nicht anwesend",
],
[
test_student3.first_name,
test_student3.last_name,
test_student3.email,
None,
"Nicht anwesend",
],
]
self.attendance_course_be.save()
def _generate_workbook(self, course_session_ids):
export_data = io.BytesIO(
export_attendance(course_session_ids, save_as_file=False)
)
return load_workbook(export_data)
def _make_header(self, csac):
return (
[
"Vorname",
"Nachname",
"Email",
"Lehrvertragsnummer",
f"Anwesenheit {csac.get_circle().title} {csac.attendance_course_zh.due_date.start.strftime('%d.%m.%Y')}",
],
)
def _check_attendance_export(self, wb, expected_data, max_row, max_col):
for row in wb.active.iter_rows(max_col=max_col, max_row=max_row):
for cell in row:
self.assertEqual(
cell.value, expected_data[row[0].row - 1][row.index(cell)]
)
def test_attendance_export_single_cs(self):
wb = self._generate_workbook([self.course_session_be.id])
self.assertEqual(len(wb.sheetnames), 1)
self.assertEqual(wb.sheetnames[0], "Test Bern 2022 a")
self._check_attendance_export(wb, self.expected_data_be, 4, 5)
def test_attendance_export_multiple_cs(self):
self.attendance_course_zh.attendance_user_list = [
{
"email": self.test_student2.email,
"status": "PRESENT",
"user_id": str(self.test_student2.id),
"last_name": self.test_student2.last_name,
"first_name": self.test_student2.first_name,
}
]
expected_data_zh = [
self._make_header(self.attendance_course_zh),
[
self.test_student2.first_name,
self.test_student2.last_name,
self.test_student2.email,
self.test_student2.additional_json_data["Lehrvertragsnummer"],
"Anwesend",
],
]
self.attendance_course_zh.save()
wb = self._generate_workbook(
[self.course_session_be.id, self.course_session_zh.id]
)
self.assertEqual(len(wb.sheetnames), 2)
self.assertEqual(wb.sheetnames[0], "Test Bern 2022 a")
self.assertEqual(wb.sheetnames[1], "Test Zürich 2022 a")
wb.active = wb["Test Zürich 2022 a"]
self._check_attendance_export(wb, expected_data_zh, 2, 5)

View File

@ -9,11 +9,11 @@ from rest_framework.decorators import api_view
from rest_framework.exceptions import PermissionDenied
from rest_framework.response import Response
from vbv_lernwelt.assignment.export import export_competence_certificates
from vbv_lernwelt.assignment.models import (
AssignmentCompletion,
AssignmentCompletionStatus,
)
from vbv_lernwelt.assignment.services import export_competence_certificates
from vbv_lernwelt.competence.services import (
query_competence_course_session_assignments,
query_competence_course_session_edoniq_tests,
@ -25,7 +25,7 @@ from vbv_lernwelt.course.models import (
CourseSessionUser,
)
from vbv_lernwelt.course.views import logger
from vbv_lernwelt.course_session.services.export import (
from vbv_lernwelt.course_session.services.export_attendance import (
export_attendance,
make_export_filename,
)

View File

@ -10,7 +10,7 @@ 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 import (
from vbv_lernwelt.course_session.services.export_attendance import (
make_export_filename,
sanitize_sheet_name,
)