YDA-6102: refactor csv import in group manager module

Co-authored-by: Lazlo Westerhof <[email protected]>

groups_import.py

@@ -1,8 +1,9 @@
 """Functions related to importing group data."""
 
-__copyright__ = 'Copyright (c) 2018-2024, Utrecht University'
+__copyright__ = 'Copyright (c) 2018-2025, Utrecht University'
 __license__   = 'GPLv3, see LICENSE'
 
+import csv
 from typing import Dict, List, Set, Tuple
 
 from iteration_utilities import duplicates, unique_everseen
@@ -170,15 +171,14 @@ def parse_data(ctx: 'rule.Context', csv_header_and_data: str) -> Tuple:
     extracted_data = []
 
     csv_lines = csv_header_and_data.splitlines()
-    header = csv_lines[0]
-    import_lines = csv_lines[1:]
+    csv_reader = csv.reader(csv_lines)
+    header_cols = next(csv_reader)
+    import_lines = list(csv_reader)
 
     # List of dicts each containing label / list of values pairs.
     lines = []
-    header_cols = header.split(',')
     for import_line in import_lines:
-        data = import_line.split(',')
-        if len(data) != len(header_cols):
+        if len(import_line) != len(header_cols):
             return [], 'Amount of header columns differs from data columns.'
         # A kind of MultiDict
         # each key is a header column
@@ -195,8 +195,8 @@ def parse_data(ctx: 'rule.Context', csv_header_and_data: str) -> Tuple:
             if header_cols[x] not in line_dict:
                 line_dict[header_cols[x]] = []
 
-            if len(data[x]):
-                line_dict[header_cols[x]].append(data[x])
+            if len(import_line[x]):
+                line_dict[header_cols[x]].append(import_line[x])
 
         lines.append(line_dict)
 

unit-tests/files/with-commas.csv

@@ -0,0 +1,5 @@
+category,subcategory,groupname,manager,member,viewer
+test-automation,csv-test,"csv,test,group1",[email protected],[email protected],[email protected]
+test-automation,csv-test,"csv,test,group2",[email protected],[email protected],[email protected]
+test-automation,csv-test,"csv,test,group3",[email protected],[email protected],[email protected]
+test-automation,csv-test,"csv,test,group4",[email protected],[email protected],[email protected]

unit-tests/files/without-commas.csv

@@ -0,0 +1,5 @@
+category,subcategory,groupname,manager,member,viewer
+test-automation,csv-test,csv-test-group1,[email protected],[email protected],[email protected]
+test-automation,csv-test,csv-test-group2,[email protected],[email protected],[email protected]
+test-automation,csv-test,csv-test-group3,[email protected],[email protected],[email protected]
+test-automation,csv-test,csv-test-group4,[email protected],[email protected],[email protected]

unit-tests/test_group_import.py

@@ -1,7 +1,6 @@
-"""Unit tests for the groups functionality
-"""
+"""Unit tests for the groups functionality."""
 
-__copyright__ = 'Copyright (c) 2019-2024, Utrecht University'
+__copyright__ = 'Copyright (c) 2019-2025, Utrecht University'
 __license__   = 'GPLv3, see LICENSE'
 
 import io
@@ -221,3 +220,14 @@ def test_parse_csv_file_duplicates(self):
         no_duplicate_data, no_duplicate_err = self.parse_csv_file("files/without-duplicates2.csv")
         self.assertNotEqual(no_duplicate_data, [])
         self.assertEqual(no_duplicate_err, '')
+
+    def test_parse_csv_file_commas(self):
+        # CSV file with commas
+        commas_data, commas_err = self.parse_csv_file("files/with-commas.csv")
+        self.assertEqual(commas_data, [])
+        self.assertIn("Data error", commas_err)
+
+        # CSV file without duplicates
+        no_commas_data, no_commas_err = self.parse_csv_file("files/without-commas.csv")
+        self.assertNotEqual(no_commas_data, [])
+        self.assertEqual(no_commas_err, '')