Add enrollment preprocessing CLI tool (#2011)

Co-authored-by: Richard Ebeling <[email protected]>

.github/workflows/tests.yml

@@ -74,7 +74,7 @@ jobs:
         uses: ./.github/setup_python
 
       - name: Run MyPy
-        run: mypy -p evap
+        run: mypy
 
   linter:
     runs-on: ubuntu-22.04
@@ -89,7 +89,7 @@ jobs:
         uses: ./.github/setup_python
 
       - name: Run linter
-        run: pylint evap -j 0
+        run: pylint evap tools
 
 
   formatter:

evap/evaluation/management/commands/lint.py

@@ -9,4 +9,4 @@ class Command(BaseCommand):
     requires_migrations_checks = False
 
     def handle(self, *args, **options):
-        subprocess.run(["pylint", "evap"], check=False)  # nosec
+        subprocess.run(["pylint", "evap", "tools"], check=False)  # nosec

evap/evaluation/management/commands/typecheck.py

@@ -9,4 +9,4 @@ class Command(BaseCommand):
     requires_migrations_checks = False
 
     def handle(self, *args, **options):
-        subprocess.run(["mypy", "-p", "evap"], check=True)  # nosec
+        subprocess.run(["mypy"], check=True)  # nosec

evap/evaluation/tests/test_commands.py

@@ -370,7 +370,7 @@ class TestLintCommand(TestCase):
     @patch("subprocess.run")
     def test_pylint_called(mock_subprocess_run):
         management.call_command("lint")
-        mock_subprocess_run.assert_called_once_with(["pylint", "evap"], check=False)
+        mock_subprocess_run.assert_called_once_with(["pylint", "evap", "tools"], check=False)
 
 
 class TestFormatCommand(TestCase):
@@ -392,7 +392,7 @@ class TestTypecheckCommand(TestCase):
     def test_mypy_called(self, mock_subprocess_run):
         management.call_command("typecheck")
         self.assertEqual(len(mock_subprocess_run.mock_calls), 1)
-        mock_subprocess_run.assert_has_calls([call(["mypy", "-p", "evap"], check=True)])
+        mock_subprocess_run.assert_has_calls([call(["mypy"], check=True)])
 
 
 class TestPrecommitCommand(TestCase):

evap/staff/fixtures/excel_files_test_data.py

@@ -1,4 +1,6 @@
+import csv
 import io
+from typing import TextIO
 
 import openpyxl
 
@@ -198,6 +200,14 @@
     ]
 }
 
+# user.title, user.last_name, user.first_name, user.email
+valid_user_courses_import_users = [
+    ['', 'Quid', 'Bastius', '[email protected]'],
+    ['', 'Sed', 'Diam', '[email protected]'],
+    ['', 'Manilium', 'Lucilia', '[email protected]'],
+    ['Prof. Dr.', 'Prorsus', 'Christoph', '[email protected]']
+]
+
 random_file_content = b"Hallo Welt\n"
 
 
@@ -221,3 +231,11 @@ def create_memory_excel_file(data):
                 sheet.cell(row=row_num, column=column_num).value = cell_data
     workbook.save(memory_excel_file)
     return memory_excel_file.getvalue()
+
+
+def create_memory_csv_file(data) -> TextIO:
+    memory_csv_file = io.StringIO()
+    writer = csv.writer(memory_csv_file, delimiter=";", lineterminator="\n")
+    writer.writerows(data)
+    memory_csv_file.seek(0)
+    return memory_csv_file

evap/staff/templates/staff_index.html

@@ -27,6 +27,7 @@ <h4 class="card-title">{% translate 'Users' %}</h4>
             <ul>
                 <li><a href="{% url 'staff:user_list' %}">{% translate 'All users' %}</a></li>
                 <li><a href="{% url 'staff:user_import' %}">{% translate 'Import users' %}</a></li>
+                <li><a href="{% url 'staff:user_export' %}">{% trans 'Export users' %}</a></li>
                 <li><a href="{% url 'staff:user_merge_selection' %}">{% translate 'Merge users' %}</a></li>
                 <li><a href="{% url 'staff:user_bulk_update' %}">{% translate 'Update users' %}</a></li>
             </ul>

evap/staff/templates/staff_user_index.html

@@ -17,6 +17,7 @@
                     <ul>
                         <li><a href="{% url 'staff:user_list' %}" class="">{% translate 'User list' %}</a></li>
                         <li><a href="{% url 'staff:user_import' %}" class="">{% translate 'Import users' %}</a></li>
+                        <li><a href="{% url 'staff:user_export' %}" class="">{% trans 'Export users' %}</a></li>
                         <li><a href="{% url 'staff:user_merge_selection' %}" class="">{% translate 'Merge users' %}</a></li>
                         <li><a href="{% url 'staff:user_bulk_update' %}" class="">{% translate 'Bulk update users' %}</a></li>
                     </ul>

evap/staff/tests/test_tools.py

@@ -1,17 +1,31 @@
+from io import BytesIO
+from itertools import cycle, repeat
+from unittest.mock import MagicMock, patch
+
 from django.contrib.auth.models import Group
 from django.test import TestCase
 from django.utils.html import escape
+from django_webtest import WebTest
 from model_bakery import baker
+from openpyxl import load_workbook
 
 from evap.evaluation.models import Contribution, Course, Evaluation, UserProfile
 from evap.evaluation.tests.tools import assert_no_database_modifications
+from evap.evaluation.tools import assert_not_none
 from evap.rewards.models import RewardPointGranting, RewardPointRedemption
+from evap.staff.fixtures.excel_files_test_data import (
+    create_memory_csv_file,
+    create_memory_excel_file,
+    valid_user_courses_import_filedata,
+    valid_user_courses_import_users,
+)
 from evap.staff.tools import (
     conditional_escape,
     merge_users,
     remove_user_from_represented_and_ccing_users,
     user_edit_link,
 )
+from tools.enrollment_preprocessor import run_preprocessor
 
 
 class MergeUsersTest(TestCase):
@@ -216,3 +230,76 @@ def test_conditional_escape(self):
         self.assertEqual(conditional_escape("<script>"), "&lt;script&gt;")
         self.assertEqual(conditional_escape(escape("<script>")), "&lt;script&gt;")
         self.assertEqual(conditional_escape("safe"), "safe")
+
+
+class EnrollmentPreprocessorTest(WebTest):
+    @classmethod
+    def setUpTestData(cls) -> None:
+        cls.imported_data = valid_user_courses_import_filedata
+        cls.csv = create_memory_csv_file(
+            [["Title", "Last name", "First name", "Email"]] + valid_user_courses_import_users
+        )
+
+    @patch("builtins.input", side_effect=cycle(("i", "e", "invalid")))
+    def test_xlsx_data_stripped(self, input_patch: MagicMock):
+        self.imported_data["MA Belegungen"][1][1] = " Accepted  "
+        self.imported_data["MA Belegungen"][1][8] = "   conflicts  "
+        self.imported_data["BA Belegungen"][1][2] = "   are    "
+        self.imported_data["BA Belegungen"][1][11] = "   stripped.   "
+        modified = run_preprocessor(BytesIO(create_memory_excel_file(self.imported_data)), self.csv)
+        self.assertIsNotNone(modified)
+        self.assertEqual(input_patch.call_count, 4)
+        workbook = load_workbook(assert_not_none(modified), read_only=True)
+        self.assertEqual(workbook["MA Belegungen"]["B2"].value, "Accepted")  # stripped conflict used
+        self.assertEqual(workbook["MA Belegungen"]["I2"].value, None)  # existing data kept
+        self.assertEqual(workbook["BA Belegungen"]["C2"].value, "are")  # stripped conflict used
+        self.assertEqual(workbook["BA Belegungen"]["L2"].value, "stripped.")  # different email is no conflict
+
+    @patch("builtins.input", side_effect=repeat("i"))
+    def test_empty_email_ignored(self, input_patch: MagicMock):
+        self.imported_data["MA Belegungen"][1][1] = " Add  "
+        self.imported_data["MA Belegungen"][1][8] = "   some  "
+        self.imported_data["BA Belegungen"][1][2] = "   conflicts    "
+        self.imported_data["MA Belegungen"][1][3] = " "
+        self.imported_data["MA Belegungen"][1][11] = "   "
+        self.imported_data["BA Belegungen"][1][3] = " \t    "
+        self.imported_data["BA Belegungen"][1][11] = "  \n   "
+        res = run_preprocessor(BytesIO(create_memory_excel_file(self.imported_data)), self.csv)
+        self.assertIsNone(res)
+        input_patch.assert_not_called()
+
+    @patch("builtins.input", side_effect=repeat("i"))
+    def test_deduplication(self, input_patch: MagicMock):
+        self.imported_data["MA Belegungen"][1][1] = "Some conflicts"
+        self.imported_data["MA Belegungen"][1][8] = "in all"
+        self.imported_data["BA Belegungen"][1][2] = "fields"
+        # copy data and pad with spaces
+        self.imported_data["MA Belegungen"].append([f" {data} " for data in self.imported_data["MA Belegungen"][1]])
+
+        res = run_preprocessor(BytesIO(create_memory_excel_file(self.imported_data)), self.csv)
+        self.assertIsNone(res)
+        self.assertEqual(input_patch.call_count, 3)  # conflicts are deduplicated.
+
+    @patch("builtins.input", side_effect=cycle(("i", "e", "e", "invalid")))
+    def test_changes_applied_globally(self, input_patch: MagicMock):
+        self.imported_data["MA Belegungen"][1][1] = "some conflicts"
+        self.imported_data["MA Belegungen"][1][8] = "in all"
+        self.imported_data["BA Belegungen"][1][2] = "fields"
+        # copy data and pad with spaces and add conflict
+        self.imported_data["MA Belegungen"].append([f" {data} " for data in self.imported_data["MA Belegungen"][1]])
+        self.imported_data["BA Belegungen"].append([f" {data} " for data in self.imported_data["BA Belegungen"][1]])
+        self.imported_data["MA Belegungen"][2][1] += "modified"
+        self.imported_data["MA Belegungen"][2][8] += "modified"
+        self.imported_data["BA Belegungen"][2][2] += "modified"
+        self.imported_data["MA Belegungen"].append([f" {data} " for data in self.imported_data["MA Belegungen"][2]])
+        self.imported_data["BA Belegungen"].append([f" {data} " for data in self.imported_data["BA Belegungen"][2]])
+        modified = run_preprocessor(BytesIO(create_memory_excel_file(self.imported_data)), self.csv)
+        self.assertIsNotNone(modified)
+        self.assertEqual(input_patch.call_count, 7)
+        workbook = load_workbook(assert_not_none(modified), read_only=True)
+        self.assertEqual(workbook["MA Belegungen"]["B2"].value, "some conflicts")
+        self.assertEqual(workbook["MA Belegungen"]["B3"].value, "some conflicts")
+        self.assertEqual(workbook["MA Belegungen"]["I2"].value, "in all modified")
+        self.assertEqual(workbook["MA Belegungen"]["I3"].value, "in all modified")
+        self.assertEqual(workbook["BA Belegungen"]["C2"].value, "Lucilia")
+        self.assertEqual(workbook["BA Belegungen"]["C3"].value, "Lucilia")

evap/staff/tests/test_views.py

@@ -1,3 +1,4 @@
+import csv
 import datetime
 import os
 from abc import ABC, abstractmethod
@@ -569,6 +570,33 @@ def test_wrong_files_dont_crash(self):
         self.assertIn("An error happened when processing the file", reply)
 
 
+class TestUserExportView(WebTestStaffMode):
+    url = reverse("staff:user_export")
+
+    @classmethod
+    def setUpTestData(cls) -> None:
+        cls.manager = make_manager()
+        # titles are not filled by baker because it has a default, see https://github.com/model-bakers/model_bakery/discussions/346
+        baker.make(
+            UserProfile,
+            _quantity=5,
+            _fill_optional=["first_name_given", "last_name", "email"],
+            title=iter(("", "Some", "Custom", "Titles", "")),
+        )
+
+    def test_export_all(self):
+        user_objects = {
+            (user.title or "", user.last_name or "", user.first_name or "", user.email or "")
+            for user in UserProfile.objects.iterator()
+        }
+        response = self.app.get(self.url, user=self.manager)
+
+        reader = csv.reader(response.text.strip().split("\n"), delimiter=";", lineterminator="\n")
+        # skip header
+        next(reader)
+        self.assertEqual({tuple(row) for row in reader}, user_objects)
+
+
 class TestUserImportView(WebTestStaffMode):
     url = "/staff/user/import"
 

evap/staff/urls.py

@@ -73,6 +73,7 @@
     path("user/", views.user_index, name="user_index"),
     path("user/create", views.UserCreateView.as_view(), name="user_create"),
     path("user/import", views.user_import, name="user_import"),
+    path("user/export", views.user_export, name="user_export"),
     path("user/<int:user_id>/edit", views.user_edit, name="user_edit"),
     path("user/list", views.user_list, name="user_list"),
     path("user/delete", views.user_delete, name="user_delete"),

evap/staff/views.py

@@ -2136,6 +2136,18 @@ def user_list(request):
     return render(request, "staff_user_list.html", {"users": users, "filter_users": filter_users})
 
 
+@manager_required
+def user_export(request):
+    response = AttachmentResponse("exported_users.csv")
+    writer = csv.writer(response, delimiter=";", lineterminator="\n")
+    header_row = (_("Title"), _("Last name"), _("First name"), _("Email"))
+    writer.writerow(header_row)
+    writer.writerows(
+        (user.title, user.last_name, user.first_name, user.email) for user in UserProfile.objects.iterator()
+    )
+    return response
+
+
 @manager_required
 class UserCreateView(SuccessMessageMixin, CreateView):
     model = UserProfile

pyproject.toml

@@ -5,7 +5,7 @@ requires-python = ">=3.10"
 
 [tool.black]
 line-length = 120
-include = 'evap.*\.pyi?$'
+include = '(evap|tools).*\.pyi?$'
 extend-exclude = """\
     .*/urls\\.py|\
     .*/migrations/.*\
@@ -15,7 +15,7 @@ extend-exclude = """\
 
 [tool.isort]
 profile = "black"
-src_paths = ["evap"]
+src_paths = ["evap", "tools"]
 line_length = 120
 skip_gitignore = true
 extend_skip_glob = ["**/migrations/*"]
@@ -74,14 +74,15 @@ disable = [
 [tool.coverage.run]
 branch = true
 omit = ["*migrations*", "*__init__.py"]
-source = ["evap"]
+source = ["evap", "tools"]
 
 [tool.coverage.report]
 omit = ["*/migrations/*", "*__init__.py"]
 
 ##############################################
 
 [tool.mypy]
+packages = ["evap", "tools"]
 plugins = ["mypy_django_plugin.main"]
 exclude = 'evap/.*/migrations/.*\.py$'
 no_implicit_optional = true
@@ -112,6 +113,6 @@ ignore_missing_imports = true
 # pytest-xdist worked for parallelizing tests.
 DJANGO_SETTINGS_MODULE = "evap.settings"
 python_files = ["tests.py", "test_*.py", "*_tests.py"]
-testpaths = ["evap"]
+testpaths = ["evap", "tools"]
 norecursedirs=["locale", "logs", "static", "static_collected", "upload"]
 addopts = "--reuse-db"