[fix] Fixed hanlding of multiple sub-fields for OneToOne/ForeignKey

Author: pandafy

openwisp_users/management/commands/export_users.py

@@ -89,12 +89,20 @@ def handle(self, *args, **options):
             )
         )
 
-    def serialize_related(self, manager, subfields):
-        """Serialize a RelatedManager queryset using the given subfields.
+    def _format_rows(self, rows):
+        """Format rows into a cell string.
 
         Single subfield → comma-separated values: val1,val2,...
         Multiple subfields → tuple-per-row format: ((v1,v2),(v3,v4))
         """
+        if not rows:
+            return ""
+        if len(rows[0]) == 1:
+            return ",".join(row[0] for row in rows)
+        return "\n".join("(" + ",".join(row) + ")" for row in rows)
+
+    def serialize_related(self, manager, subfields):
+        """Serialize a RelatedManager queryset using the given subfields."""
         rows = []
         # We use manager.all() instead of manager.iterator() to utilize the
         # prefetch_related queryset cache. The iterator() method would bypass the cache
@@ -106,11 +114,7 @@ def serialize_related(self, manager, subfields):
                 val = self._get_nested_attr(obj, f)
                 row.append(self._normalize_value(val))
             rows.append(row)
-        if not rows:
-            return ""
-        if len(subfields) == 1:
-            return ",".join(row[0] for row in rows)
-        return "(" + ",".join("(" + ",".join(row) + ")" for row in rows) + ")"
+        return self._format_rows(rows)
 
     def _get_nested_attr(self, obj, attr_path):
         """Resolve a dotted attribute path on an object.
@@ -170,9 +174,10 @@ def _get_field_value(self, user, field):
                 return ""
             if isinstance(attr, (QuerySet, BaseManager)):
                 return self.serialize_related(attr, subfields)
-            return ",".join(
+            row = [
                 self._normalize_value(self._get_nested_attr(attr, f)) for f in subfields
-            )
+            ]
+            return self._format_rows([row])
         val = self._get_nested_attr(user, name)
         if isinstance(val, (QuerySet, BaseManager)):
             return ""

openwisp_users/tests/test_commands.py

@@ -1,4 +1,5 @@
 import csv
+import re
 from io import StringIO
 from unittest.mock import patch
 
@@ -226,8 +227,10 @@ def _broken_command_error_callable(user):
         {
             "fields": [
                 "id",
-                # single related object with subfields
+                # single related object with single subfield
                 {"name": "auth_token", "fields": ["key"]},
+                # single related object with multiple subfields
+                {"name": "auth_token", "fields": ["key", "created"]},
                 # nullable field
                 {"name": "birth_date", "fields": ["year"]},
                 # manager with single subfield
@@ -264,18 +267,34 @@ def test_subfields_dict_field(self):
         self.assertEqual(len(csv_data), 3)
         # user1: token present, birth_date None, one org membership
         self.assertEqual(csv_data[1][0], str(user1.id))
-        self.assertEqual(csv_data[1][1], str(token.key))  # subfields, single obj
-        self.assertEqual(csv_data[1][2], "")  # birth_date is None
-        self.assertEqual(csv_data[1][3], str(org.id))  # single-subfield manager
-        self.assertEqual(csv_data[1][4], f"(({org.id},True))")  # multi-subfield manager
-        self.assertEqual(csv_data[1][5], str(org.id))  # dot-notation manager
+        # subfields, single obj
+        self.assertEqual(csv_data[1][1], str(token.key))
+        # single-obj multi-subfield: wrapped shape ((key,created))
+        self.assertRegex(
+            csv_data[1][2],
+            rf"^\({re.escape(token.key)},",
+        )
+        # birth_date is None
+        self.assertEqual(csv_data[1][3], "")
+        # single-subfield manager
+        self.assertEqual(csv_data[1][4], str(org.id))
+        # multi-subfield manager
+        self.assertEqual(csv_data[1][5], f"({org.id},True)")
+        # dot-notation manager
+        self.assertEqual(csv_data[1][6], str(org.id))
         # user2: no token, no org membership
         self.assertEqual(csv_data[2][0], str(user2.id))
-        self.assertEqual(csv_data[2][1], "")  # ObjectDoesNotExist auth_token
-        self.assertEqual(csv_data[2][2], "")  # birth_date is None
-        self.assertEqual(csv_data[2][3], "")  # empty manager
-        self.assertEqual(csv_data[2][4], "")  # empty manager
-        self.assertEqual(csv_data[2][5], "")  # empty dot-notation manager
+        # ObjectDoesNotExist auth_token
+        self.assertEqual(csv_data[2][1], "")
+        # ObjectDoesNotExist auth_token multi-subfield
+        self.assertEqual(csv_data[2][2], "")
+        # birth_date is None
+        self.assertEqual(csv_data[2][3], "")
+        # empty manager
+        self.assertEqual(csv_data[2][4], "")
+        self.assertEqual(csv_data[2][5], "")
+        # empty dot-notation manager
+        self.assertEqual(csv_data[2][6], "")
 
     def test_dot_notation_objectdoesnotexist_on_sub_attr(self):
         """Returns empty string when sub-attribute access raises ObjectDoesNotExist."""