[fix] Fixes by @coderabbitai

Author: pandafy

openwisp_users/management/commands/export_users.py

@@ -16,20 +16,6 @@ def normalize_field(field):
     return {"name": field}
 
 
-def serialize_related(manager, subfields):
-    """Serialize a RelatedManager queryset using the given subfields.
-
-    Single subfield → comma-separated values: val1,val2,...
-    Multiple subfields → tuple-per-row format: ((v1,v2),(v3,v4))
-    """
-    rows = [[str(getattr(obj, f, "")) for f in subfields] for obj in manager.all()]
-    if not rows:
-        return ""
-    if len(subfields) == 1:
-        return ",".join(row[0] for row in rows)
-    return "(" + ",".join("(" + ",".join(row) + ")" for row in rows) + ")"
-
-
 class Command(BaseCommand):
     help = "Exports user data to a CSV file"
 
@@ -51,15 +37,15 @@ def add_arguments(self, parser):
         )
 
     def handle(self, *args, **options):
-        fields = app_settings.EXPORT_USERS_COMMAND_CONFIG.get("fields", []).copy()
+        raw_fields = app_settings.EXPORT_USERS_COMMAND_CONFIG.get("fields", []).copy()
         # Get the fields to be excluded from the command-line argument
         exclude_fields = [
             t.strip() for t in options.get("exclude_fields").split(",") if t.strip()
         ]
         # Remove excluded fields from the export fields (match on the field name)
         fields = [
             field
-            for field in fields
+            for field in raw_fields
             if normalize_field(field)["name"] not in exclude_fields
         ]
         # Fetch all user data using select_related and prefetch_related
@@ -89,6 +75,40 @@ def handle(self, *args, **options):
             self.style.SUCCESS(f"User data exported successfully to {filename}!")
         )
 
+    def serialize_related(self, manager, subfields):
+        """Serialize a RelatedManager queryset using the given subfields.
+
+        Single subfield → comma-separated values: val1,val2,...
+        Multiple subfields → tuple-per-row format: ((v1,v2),(v3,v4))
+        """
+        rows = [
+            [str(self._get_nested_attr(obj, f)) for f in subfields]
+            for obj in manager.all()
+        ]
+        if not rows:
+            return ""
+        if len(subfields) == 1:
+            return ",".join(row[0] for row in rows)
+        return "(" + ",".join("(" + ",".join(row) + ")" for row in rows) + ")"
+
+    def _get_nested_attr(self, obj, attr_path):
+        if not attr_path:
+            return obj
+        parts = attr_path.split(".")
+        current = obj
+        for i, part in enumerate(parts):
+            try:
+                current = getattr(current, part)
+            except (ObjectDoesNotExist, AttributeError):
+                return ""
+            if hasattr(current, "iterator") and i < len(parts) - 1:
+                remaining_path = ".".join(parts[i + 1 :])
+                return ",".join(
+                    str(self._get_nested_attr(item, remaining_path))
+                    for item in current.iterator()
+                )
+        return current
+
     def _get_field_value(self, user, field):
         normalized = normalize_field(field)
         name = normalized["name"]
@@ -101,30 +121,14 @@ def _get_field_value(self, user, field):
             except Exception as e:
                 raise CommandError(f"Error calling function for field '{name}': {e}")
         if subfields is not None:
-            try:
-                attr = getattr(user, name)
-            except ObjectDoesNotExist:
-                return ""
+            attr = self._get_nested_attr(user, name)
             if attr is None:
                 return ""
             if hasattr(attr, "iterator"):
-                return serialize_related(attr, subfields)
-            return ",".join(str(getattr(attr, f, "")) for f in subfields)
+                return self.serialize_related(attr, subfields)
+            return ",".join(str(self._get_nested_attr(attr, f)) for f in subfields)
 
         # Dot-notation: e.g. "auth_token.key" or "profile.phone_number"
         if "." in name:
-            model_attr, sub_attr = name.split(".", 1)
-            try:
-                intermediate = getattr(user, model_attr)
-            except ObjectDoesNotExist:
-                return ""
-            if hasattr(intermediate, "iterator"):
-                # Related manager accessed via dot notation → comma-separated values
-                return ",".join(
-                    str(getattr(obj, sub_attr, "")) for obj in intermediate.iterator()
-                )
-            try:
-                return getattr(intermediate, sub_attr)
-            except ObjectDoesNotExist:
-                return ""
-        return getattr(user, name)
+            return self._get_nested_attr(user, name)
+        return self._get_nested_attr(user, name)

openwisp_users/tests/test_commands.py

@@ -10,7 +10,7 @@
 from openwisp_utils.tests import capture_stdout
 
 from .. import settings as app_settings
-from ..management.commands.export_users import Command, normalize_field
+from ..management.commands.export_users import Command
 from .utils import TestOrganizationMixin
 
 
@@ -47,9 +47,25 @@ def test_export_users(self):
 
         # 3 user and 1 header
         self.assertEqual(len(csv_data), 4)
+        # Expected headers are the keys produced by normalize_field for the
+        # default EXPORT_USERS_COMMAND_CONFIG. These are stable values and
+        # asserted explicitly to avoid mirroring production code in the test.
         expected_headers = [
-            normalize_field(f)["name"]
-            for f in app_settings.EXPORT_USERS_COMMAND_CONFIG["fields"]
+            "id",
+            "username",
+            "email",
+            "password",
+            "first_name",
+            "last_name",
+            "is_staff",
+            "is_active",
+            "date_joined",
+            "phone_number",
+            "birth_date",
+            "location",
+            "notes",
+            "language",
+            "organizations",
         ]
         self.assertEqual(csv_data[0], expected_headers)
         # Ensuring ordering
@@ -67,9 +83,24 @@ def test_exclude_fields(self):
         call_command(
             "export_users",
             filename=self.temp_file.name,
+            # Exclude all fields except "id"
             exclude_fields=",".join(
-                normalize_field(f)["name"]
-                for f in app_settings.EXPORT_USERS_COMMAND_CONFIG["fields"][1:]
+                [
+                    "username",
+                    "email",
+                    "password",
+                    "first_name",
+                    "last_name",
+                    "is_staff",
+                    "is_active",
+                    "date_joined",
+                    "phone_number",
+                    "birth_date",
+                    "location",
+                    "notes",
+                    "language",
+                    "organizations",
+                ]
             ),
         )
         with open(self.temp_file.name, "r") as temp_file:
@@ -107,10 +138,9 @@ def test_related_fields(self):
 
         # 3 user and 1 header
         self.assertEqual(len(csv_data), 2)
-        expected_headers = [
-            normalize_field(f)["name"]
-            for f in app_settings.EXPORT_USERS_COMMAND_CONFIG["fields"]
-        ]
+        # When fields are ["id", "auth_token.key"] the expected headers
+        # are the literal keys used to identify columns in the CSV.
+        expected_headers = ["id", "auth_token.key"]
         self.assertEqual(csv_data[0], expected_headers)
         self.assertEqual(csv_data[1][0], str(user.id))
         self.assertEqual(csv_data[1][1], str(token.key))
@@ -152,9 +182,10 @@ def _broken_callable(user):
         }
         self._create_user()
         stderr = StringIO()
-        with patch.object(
-            app_settings, "EXPORT_USERS_COMMAND_CONFIG", config
-        ), self.assertRaises(Exception) as context:
+        with (
+            patch.object(app_settings, "EXPORT_USERS_COMMAND_CONFIG", config),
+            self.assertRaises(Exception) as context,
+        ):
             call_command("export_users", filename=self.temp_file.name, stderr=stderr)
         self.assertIn(
             "Error calling function for field 'broken'", str(context.exception)