[fix] Fixes by @coderabbitai

Author: pandafy

openwisp_users/admin.py

@@ -147,6 +147,12 @@ class OrganizationUserInline(admin.StackedInline):
     fields = ("organization", "is_admin")
     autocomplete_fields = ("organization",)
 
+    def get_queryset(self, request):
+        # OrganizationUserInlineFormSet.add_fields() reads
+        # instance.organization.is_active for every row; select_related
+        # folds that per-row query into this one.
+        return super().get_queryset(request).select_related("organization")
+
     def get_formset(self, request, obj=None, **kwargs):
         """
         In form dropdowns, display only active organizations;

openwisp_users/api/mixins.py

@@ -17,6 +17,10 @@
 
 Organization = swapper.load_model("openwisp_users", "Organization")
 
+DISABLED_ORGANIZATION_ERROR_MESSAGE = _(
+    'Organization with pk "{pk_value}" does not exist or is disabled.'
+)
+
 
 class OrgLookup:
     @property
@@ -165,8 +169,9 @@ def _user_attr(self):
     def filter_fields(self):
         user = self.context["request"].user
         # superuser can see everything, except disabled organizations
-        superuser = user.is_superuser or user.is_anonymous
-        if not superuser:
+        # The anonymouse use case exist so we don't run into errors with swagger
+        is_superuser_or_anonymous = user.is_superuser or user.is_anonymous
+        if not is_superuser_or_anonymous:
             # non superusers can see only items of organizations
             # they're related to
             organization_filter = getattr(user, self._user_attr)
@@ -176,15 +181,15 @@ def filter_fields(self):
                 # disabled organizations are excluded for everyone, superusers
                 # included, since they can only be re-enabled, not written to
                 queryset = self.fields[field].queryset.filter(is_active=True)
-                self.fields[field].error_messages["does_not_exist"] = _(
-                    'Organization with pk "{pk_value}" does not exist or is disabled.'
-                )
-                if not superuser:
+                self.fields[field].error_messages[
+                    "does_not_exist"
+                ] = DISABLED_ORGANIZATION_ERROR_MESSAGE
+                if not is_superuser_or_anonymous:
                     self.fields[field].allow_null = False
                     queryset = queryset.filter(pk__in=organization_filter)
                 self.fields[field].queryset = queryset
                 continue
-            if superuser:
+            if is_superuser_or_anonymous:
                 continue
             conditions = Q(**{self.organization_lookup: organization_filter})
             if self.include_shared:

openwisp_users/api/serializers.py

@@ -13,6 +13,8 @@
 
 from openwisp_utils.api.serializers import ValidatedModelSerializer
 
+from .mixins import DISABLED_ORGANIZATION_ERROR_MESSAGE
+
 Group = load_model("openwisp_users", "Group")
 Organization = load_model("openwisp_users", "Organization")
 User = get_user_model()
@@ -112,20 +114,30 @@ class Meta:
 
     def validate(self, data):
         if self.instance and not self.instance.is_active:
-            keys = set(data.keys())
             owner_data = data.get("owner") or {}
             owner_present = "owner" in data
             is_owner_unassignment = (
                 owner_present and owner_data.get("organization_user") is None
             )
             reenabling = data.get("is_active") is True
+            # A key whose submitted value matches the value already stored is
+            # not a change, so a read-modify-write PUT that resends every
+            # field unchanged except is_active must not be rejected just
+            # because e.g. "name" is present in the payload.
+            changed_keys = {
+                key
+                for key in data
+                if key != "owner" and getattr(self.instance, key) != data[key]
+            }
+            if owner_present:
+                changed_keys.add("owner")
             # While disabled, only re-enabling (Is active) and/or unassigning
             # the owner are allowed, and neither can be combined with any other
             # change (editing a field or assigning an owner). This matches the
             # admin interface, which locks every field except Is active, so the
             # admin, the API and the docs tell the same story.
             allowed = (
-                keys <= {"is_active", "owner"}
+                changed_keys <= {"is_active", "owner"}
                 and (not owner_present or is_owner_unassignment)
                 and (reenabling or is_owner_unassignment)
             )
@@ -218,9 +230,7 @@ def update(self, instance, validated_data):
 
 class OrgUserCustomPrimarykeyRelatedField(serializers.PrimaryKeyRelatedField):
     default_error_messages = {
-        "does_not_exist": _(
-            'Organization with pk "{pk_value}" does not exist or is disabled.'
-        ),
+        "does_not_exist": DISABLED_ORGANIZATION_ERROR_MESSAGE,
     }
 
     def get_queryset(self):

openwisp_users/base/models.py

@@ -501,12 +501,13 @@ def clean(self):
             # disabled organization must not fail.
             db_values = (
                 self._meta.model.objects.filter(pk=self.pk)
-                .values("organization_id", "is_admin")
+                .values("organization_id", "is_admin", "user_id")
                 .first()
             )
             if db_values is None or (
                 db_values["organization_id"] != self.organization_id
                 or db_values["is_admin"] != self.is_admin
+                or db_values["user_id"] != self.user_id
             ):
                 raise ValidationError(
                     _("Memberships of a disabled organization cannot be modified.")

openwisp_users/multitenancy.py

@@ -92,7 +92,7 @@ def has_change_permission(self, request, obj=None):
 
     def has_add_permission(self, request, *args, **kwargs):
         """
-        Hide the Add button from operators who manage no active organization:
+        Hide the Add button from admins who manage no active organization:
         the organization dropdown would be empty and the form could never be
         submitted. Does not apply to the user admin or to models without an
         organization (directly or through ``multitenant_parent``).
@@ -106,22 +106,15 @@ def has_add_permission(self, request, *args, **kwargs):
             and self.model != User
             and not request.user.organizations_managed
         ):
-            org_field = (
-                self.model._meta.get_field("organization")
-                if hasattr(self.model, "organization")
-                else None
-            )
-            # If the model has a required organization field (OrgMixin, not
-            # ShareableOrgMixin which allows null/blank), the dropdown would be
-            # empty — block. If it's optional or reached through a parent,
-            # the form can still be submitted without picking an org.
-            if org_field is not None:
-                return False
-            if org_field is None and self.multitenant_parent:
+            # Any model with an organization field (directly, or reached
+            # through multitenant_parent) is blocked: _edit_form() makes the
+            # field required for non-superusers, so the form could not be
+            # submitted without an active organization to pick anyway.
+            if hasattr(self.model, "organization") or self.multitenant_parent:
                 return False
         return super().has_add_permission(request, *args, **kwargs)
 
-    def _edit_form(self, request, form):
+    def _edit_form(self, request, form, obj=None):
         """
         Modifies the form querysets as follows;
         if current user is not superuser:
@@ -131,21 +124,35 @@ def _edit_form(self, request, form):
             * do not allow organization field to be empty (shared org)
         else show everything
         Organization choices always exclude disabled organizations,
-        superusers included.
+        superusers included, except an admin that opted out of write
+        protection (``disabled_organization_write_protection = False``)
+        keeps the edited object's own disabled organization selectable,
+        or the form could never be saved.
         """
         fields = form.base_fields
         user = request.user
         org_field = fields.get("organization")
+        keep_disabled_org_pk = None
+        if not self.disabled_organization_write_protection and obj is not None:
+            organization = self._get_object_organization(obj)
+            if organization is not None and not organization.is_active:
+                keep_disabled_org_pk = organization.pk
         if org_field:
-            org_field.queryset = org_field.queryset.filter(is_active=True)
+            allowed = Q(is_active=True)
+            if keep_disabled_org_pk is not None:
+                allowed |= Q(pk=keep_disabled_org_pk)
+            org_field.queryset = org_field.queryset.filter(allowed)
         if user.is_superuser and org_field and not org_field.required:
             org_field.empty_label = SHARED_SYSTEMWIDE_LABEL
         elif not user.is_superuser:
             orgs_pk = user.organizations_managed
             # organizations relation;
             # may be readonly and not present in field list
             if org_field:
-                org_field.queryset = org_field.queryset.filter(pk__in=orgs_pk)
+                managed = Q(pk__in=orgs_pk)
+                if keep_disabled_org_pk is not None:
+                    managed |= Q(pk=keep_disabled_org_pk)
+                org_field.queryset = org_field.queryset.filter(managed)
                 org_field.empty_label = None
                 org_field.required = True
             # other relations
@@ -160,7 +167,7 @@ def _edit_form(self, request, form):
 
     def get_form(self, request, obj=None, **kwargs):
         form = super().get_form(request, obj, **kwargs)
-        self._edit_form(request, form)
+        self._edit_form(request, form, obj)
         return form
 
     def get_formset(self, request, obj=None, **kwargs):

openwisp_users/tests/test_api/test_api.py

@@ -1,13 +1,10 @@
-from unittest import mock
-
 import django
 from allauth.account.models import EmailAddress
 from django.contrib import auth
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Permission
 from django.core import mail
-from django.core.exceptions import ValidationError as DjangoValidationError
-from django.test import TestCase
+from django.test import TestCase, TransactionTestCase
 from django.urls import reverse
 from django.utils.timezone import localdate, timedelta
 from swapper import load_model
@@ -140,6 +137,11 @@ def test_patch_disabled_organization_reenable_api(self):
         self.assertTrue(org1.is_active)
 
     def test_reenable_disabled_organization_with_field_edit_api(self):
+        """
+        Re-enabling (is_active) and editing another field in the same
+        request is rejected: re-enabling and editing must be two separate
+        requests, matching the admin and the docs.
+        """
         org1 = self._get_org()
         org1.is_active = False
         org1.save()
@@ -153,6 +155,27 @@ def test_reenable_disabled_organization_with_field_edit_api(self):
         self.assertEqual(org1.is_active, False)
         self.assertEqual(org1.name, "test org")
 
+    def test_reenable_disabled_organization_via_put_api(self):
+        org1 = self._get_org()
+        org1.is_active = False
+        org1.save()
+        path = reverse("users:organization_detail", args=(org1.pk,))
+        # a PUT always resends every required field, "name" included; since
+        # its value is unchanged it must not count as an edit and block the
+        # re-enable, the way a read-modify-write client would use PUT
+        data = {
+            "name": org1.name,
+            "is_active": True,
+            "slug": org1.slug,
+            "description": org1.description,
+            "email": org1.email,
+            "url": org1.url,
+        }
+        response = self.client.put(path, data, content_type="application/json")
+        self.assertEqual(response.status_code, 200)
+        org1.refresh_from_db()
+        self.assertTrue(org1.is_active)
+
     def test_create_organization_owner_api(self):
         user1 = self._create_user(username="user1", email="user1@email.com")
         org1 = self._create_org(name="org1")
@@ -658,41 +681,6 @@ def test_create_user_with_group_org_user_api(self):
         r = self.client.post(path, data, content_type="application/json")
         self.assertEqual(r.status_code, 201)
 
-    def test_create_user_organization_users_disabled_org_api(self):
-        path = reverse("users:user_list")
-        org1 = self._create_org(name="disabled-org", is_active=False)
-        data = {
-            "username": "tester",
-            "email": "tester@test.com",
-            "password": "password",
-            "organization_users": {"is_admin": False, "organization": org1.pk},
-        }
-        r = self.client.post(path, data, content_type="application/json")
-        self.assertEqual(r.status_code, 400)
-        self.assertEqual(User.objects.filter(username="tester").count(), 0)
-        self.assertEqual(OrganizationUser.objects.filter(organization=org1).count(), 0)
-
-    def test_create_user_membership_failure_rolls_back_user_api(self):
-        # A membership validation failure after the user row is written must
-        # roll the user back instead of leaving a half-created account behind.
-        path = reverse("users:user_list")
-        org1 = self._get_org()
-        data = {
-            "username": "rollbackuser",
-            "email": "rollbackuser@test.com",
-            "password": "password",
-            "organization_users": {"is_admin": False, "organization": org1.pk},
-        }
-        with mock.patch.object(
-            OrganizationUser,
-            "full_clean",
-            side_effect=DjangoValidationError("membership boom"),
-        ):
-            r = self.client.post(path, data, content_type="application/json")
-        self.assertEqual(r.status_code, 400)
-        self.assertEqual(User.objects.filter(username="rollbackuser").count(), 0)
-        self.assertEqual(OrganizationUser.objects.filter(organization=org1).count(), 0)
-
     def test_post_with_no_email(self):
         path = reverse("users:user_list")
         data = {"username": "", "email": "", "password": ""}
@@ -968,3 +956,25 @@ def test_expiration_date_none_api(self):
         self.assertIsNone(r.data["expiration_date"])
         user.refresh_from_db()
         self.assertIsNone(user.expiration_date)
+
+
+class TestUsersApiTransaction(TestOrganizationMixin, TransactionTestCase):
+    def setUp(self):
+        self.client.force_login(self._get_admin())
+
+    def test_create_user_organization_users_disabled_org_api(self):
+        # A membership validation failure (here, the disabled-organization
+        # guard) after the user row is written must roll the user back
+        # instead of leaving a half-created account behind.
+        path = reverse("users:user_list")
+        org1 = self._create_org(name="disabled-org", is_active=False)
+        data = {
+            "username": "tester",
+            "email": "tester@test.com",
+            "password": "password",
+            "organization_users": {"is_admin": False, "organization": org1.pk},
+        }
+        r = self.client.post(path, data, content_type="application/json")
+        self.assertEqual(r.status_code, 400)
+        self.assertEqual(User.objects.filter(username="tester").count(), 0)
+        self.assertEqual(OrganizationUser.objects.filter(organization=org1).count(), 0)

openwisp_users/tests/test_models.py

@@ -468,6 +468,21 @@ def test_organization_user_clean_disabled_organization(self):
             # Django admin cannot save a user who has a disabled-org membership
             org_user.full_clean()
 
+        with self.subTest("reassign the user of a disabled organization membership"):
+            org = self._create_org(name="test-org-reassign-user")
+            user = self._create_user(username="user6", email="user6@example.com")
+            other_user = self._create_user(username="user7", email="user7@example.com")
+            org_user = self._create_org_user(organization=org, user=user)
+            org.is_active = False
+            org.save()
+            org_user.refresh_from_db()
+            org_user.user = other_user
+            with self.assertRaisesMessage(
+                ValidationError,
+                "Memberships of a disabled organization cannot be modified.",
+            ):
+                org_user.full_clean()
+
     def test_organization_owner_clean_disabled_organization(self):
         with self.subTest("assign an owner to a disabled organization"):
             org = self._create_org(name="disabled-org-owner")

tests/testapp/admin.py

@@ -68,8 +68,13 @@ class TagAdmin(BaseAdmin):
     pass
 
 
+class LibraryParentAdmin(MultitenantAdminMixin, admin.ModelAdmin):
+    # Library has no organization field; it is reached through its Book parent
+    multitenant_parent = "book"
+
+
 admin.site.register(Shelf, ShelfAdmin)
 admin.site.register(Book, BookAdmin)
 admin.site.register(Template, TemplateAdmin)
-admin.site.register(Library)
+admin.site.register(Library, LibraryParentAdmin)
 admin.site.register(Tag, TagAdmin)