Merge pull request #12040 from rtibbles/import_duplicate_username_sam…

…e_facility

Fix inaccurate duplicate username reporting in bulk import

kolibri/core/auth/management/commands/bulkimportusers.py

@@ -250,13 +250,14 @@ class Validator(object):
     Class to apply different validation checks on a CSV data reader.
     """
 
-    def __init__(self, header_translation):
+    def __init__(self, header_translation, facility):
         self._checks = []
         self.classrooms = {}
         self.coach_classrooms = {}
         self.users = {}
         self.header_translation = header_translation
         self.roles = {r: [] for r in roles_map.values() if r is not None}
+        self.facility = facility
 
     def add_check(self, header_name, check, message):
         """
@@ -275,7 +276,7 @@ def get_username(self, row):
 
         # Check if a user with the provided username exists (case-insensitive)
         existing_user = FacilityUser.objects.filter(
-            username__iexact=lowercase_username
+            username__iexact=lowercase_username, facility=self.facility
         ).first()
         # Convert existing keys in self.users to lowercase
         if existing_user and uuid == "":
@@ -408,9 +409,9 @@ def add_arguments(self, parser):
             help="File to store errors output (to be used in internal tests only)",
         )
 
-    def csv_values_validation(self, reader, header_translation):
+    def csv_values_validation(self, reader, header_translation, facility):
         per_line_errors = []
-        validator = Validator(header_translation)
+        validator = Validator(header_translation, facility)
         validator.add_check("UUID", valid_uuid(), MESSAGES[INVALID_UUID])
         validator.add_check(
             "FULL_NAME", value_length(125), MESSAGES[TOO_LONG].format("FULL_NAME")
@@ -890,7 +891,7 @@ def handle_async(self, *args, **options):
                 with csv_file as f:
                     reader = csv.DictReader(f, strict=True)
                     per_line_errors, classes, users, roles = self.csv_values_validation(
-                        reader, self.header_translation
+                        reader, self.header_translation, self.default_facility
                     )
             except (ValueError, FileNotFoundError, csv.Error) as e:
                 self.append_error(MESSAGES[FILE_READ_ERROR].format(e))

kolibri/core/auth/test/test_bulk_import.py

@@ -163,7 +163,7 @@ def test_dryrun_from_export_csv(self):
         with open_csv_for_reading(self.filepath) as source:
             reader = csv.DictReader(source, strict=True)
             per_line_errors, classes, users, roles = cmd.csv_values_validation(
-                reader, header_translation
+                reader, header_translation, self.facility
             )
 
         assert len(users) == 10  # admins have not been exported
@@ -233,7 +233,7 @@ def test_password_is_required(self):
         with open_csv_for_reading(new_filepath) as source:
             reader = csv.DictReader(source, strict=True)
             per_line_errors, classes, users, roles = cmd.csv_values_validation(
-                reader, header_translation
+                reader, header_translation, self.facility
             )
         assert len(per_line_errors) == 1
         assert (
@@ -345,6 +345,44 @@ def test_username_already_exists(self):
         # Check that the password of the existing user remains unchanged
         assert passwd2 == passwd1
 
+    def test_username_already_exists_on_different_facility(self):
+        _, first_filepath = tempfile.mkstemp(suffix=".csv")
+        rows = [
+            [
+                None,
+                "peter",  # Adding the first user with the username "peter"
+                "passwd1",
+                None,
+                "LEARNER",
+                None,
+                "2001",
+                "FEMALE",
+                "new_class",
+                None,
+            ],
+        ]
+        self.create_csv(first_filepath, rows)
+
+        data = create_dummy_facility_data(
+            classroom_count=CLASSROOMS, learnergroup_count=1
+        )
+
+        facility2 = data["facility"]
+
+        # First import this user into a different facility
+        call_command("bulkimportusers", first_filepath, facility=facility2.id)
+
+        # Then import into the main facility and confirm that it works!
+        call_command("bulkimportusers", first_filepath, facility=self.facility.id)
+
+        # Assert that we have created a user like this in both facilities.
+        assert FacilityUser.objects.filter(
+            username="peter", facility=facility2
+        ).exists()
+        assert FacilityUser.objects.filter(
+            username="peter", facility=self.facility
+        ).exists()
+
     def test_asterisk_in_password(self):
         _, first_filepath = tempfile.mkstemp(suffix=".csv")
         rows = [