Merge branch 'master' into fix/1221-vpn-peer-cache-desync

Author: mn-ram

openwisp_controller/config/base/device.py

@@ -338,9 +338,9 @@ def _get_initial_values_for_checked_fields(self):
         if not present_values:
             return
         self.refresh_from_db(fields=present_values.keys())
-        for field in self._changed_checked_fields:
-            setattr(self, f"_initial_{field}", field)
-            setattr(self, field, present_values[field])
+        for field, value in present_values.items():
+            setattr(self, f"_initial_{field}", getattr(self, field))
+            setattr(self, field, value)
 
     def _check_name_changed(self):
         if self._initial_name == models.DEFERRED:

openwisp_controller/config/static/config/js/relevant_templates.js

@@ -138,7 +138,7 @@ django.jQuery(function ($) {
         resetTemplateOptions();
         var enabledTemplates = [],
           sortedm2mUl = $("ul.sortedm2m-items:first"),
-          sortedm2mPrefixUl = $("ul.sortedm2m-items:last");
+          sortedm2mPrefixUl = $("#config-empty ul.sortedm2m-items");
 
         // Adds "li" elements for templates
         Object.keys(data).forEach(function (templateId, index) {

openwisp_controller/config/tests/test_device.py

@@ -541,6 +541,20 @@ def test_changed_checked_fields_no_duplicates(self):
         device.__init__()
         self.assertEqual(device._changed_checked_fields.count("last_ip"), 1)
 
+    def test_deferred_fields_populated_correctly(self):
+        device = self._create_device(
+            name="deferred-test",
+            management_ip="10.0.0.1",
+        )
+        # Load the instance with deferred fields omitted
+        device = Device.objects.only("id").get(pk=device.pk)
+        device.management_ip = "10.0.0.55"
+        # Saving the device object will populate the deferred fields
+        device.save()
+        # Ensure `_initial_<field>` contains the actual value, not the field name
+        self.assertEqual(getattr(device, "_initial_management_ip"), "10.0.0.55")
+        self.assertNotEqual(getattr(device, "_initial_management_ip"), "management_ip")
+
     def test_exceed_organization_device_limit(self):
         org = self._get_org()
         org.config_limits.device_limit = 1

openwisp_controller/config/tests/test_selenium.py

@@ -1,6 +1,7 @@
 import os
 import time
 
+from django.contrib.auth.models import Group, Permission
 from django.contrib.staticfiles.testing import StaticLiveServerTestCase
 from django.test import tag
 from django.urls.base import reverse
@@ -45,9 +46,17 @@ def _verify_templates_visibility(self, hidden=None, visible=None):
         hidden = hidden or []
         visible = visible or []
         for template in hidden:
-            self.wait_for_invisibility(By.XPATH, f'//*[@value="{template.id}"]')
+            self.wait_for_invisibility(
+                By.XPATH,
+                f'//ul[contains(@class,"sortedm2m-items")]'
+                f'//input[@value="{template.id}"]',
+            )
         for template in visible:
-            self.wait_for_visibility(By.XPATH, f'//*[@value="{template.id}"]')
+            self.wait_for_visibility(
+                By.XPATH,
+                f'//ul[contains(@class,"sortedm2m-items")]'
+                f'//input[@value="{template.id}"]',
+            )
 
 
 @tag("selenium_tests")
@@ -389,6 +398,94 @@ def test_add_remove_templates(self):
             self.assertEqual(config.templates.count(), 0)
             self.assertEqual(config.status, "modified")
 
+    def test_relevant_templates_duplicates(self):
+        """
+        Test that a user with specific permissions can see shared templates
+        properly. Verifies that:
+        1. User with custom group permissions can access the admin
+        2. Multiple shared templates are displayed correctly
+        3. Each template appears only once in the sortedm2m list
+        """
+        # Define permission codenames for the custom group
+        permission_codenames = [
+            "view_group",
+            "change_config",
+            "view_config",
+            "add_device",
+            "change_device",
+            "delete_device",
+            "view_device",
+            "view_devicegroup",
+            "view_template",
+        ]
+        # Create a custom group with the specified permissions
+        permissions = Permission.objects.filter(codename__in=permission_codenames)
+        custom_group, _ = Group.objects.get_or_create(name="Custom Operator")
+        custom_group.permissions.set(permissions)
+        # Create a user and assign the custom group
+        user = self._create_user(
+            username="limited_user",
+            password="testpass123",
+            email="limited@test.com",
+            is_staff=True,
+        )
+        user.groups.add(custom_group)
+        org = self._get_org()
+        self._create_org_user(user=user, organization=org, is_admin=True)
+        # Create multiple shared templates (organization=None)
+        template1 = self._create_template(
+            name="Shared Template 1", organization=None, default=True
+        )
+        template2 = self._create_template(name="Shared Template 2", organization=None)
+        device = self._create_config(organization=org).device
+        # Login as the limited user
+        self.login(username="limited_user", password="testpass123")
+        # Navigate using Selenium
+        self.open(
+            reverse(f"admin:{self.config_app_label}_device_change", args=[device.id])
+            + "#config-group"
+        )
+        self.hide_loading_overlay()
+        with self.subTest(
+            "Regression precondition: empty Config inline is not rendered"
+        ):
+            self.assertFalse(self.web_driver.find_elements(By.ID, "config-empty"))
+
+        with self.subTest("All shared templates should be visible"):
+            self._verify_templates_visibility(visible=[template1, template2])
+
+        with self.subTest("Verify sortedm2m list has exactly 2 template items"):
+            # Check that ul.sortedm2m-items.sortedm2m.ui-sortable has exactly 2 children
+            # with .sortedm2m-item class
+            sortedm2m_items = self.find_elements(
+                by=By.CSS_SELECTOR,
+                value="ul.sortedm2m-items.sortedm2m.ui-sortable > li.sortedm2m-item",
+            )
+            self.assertEqual(
+                len(sortedm2m_items),
+                2,
+                (
+                    "Expected exactly 2 template items in sortedm2m list,"
+                    f" found {len(sortedm2m_items)}"
+                ),
+            )
+
+        with self.subTest(
+            "Verify checkbox inputs are rendered with expected attributes"
+        ):
+            for idx, template_id in enumerate([template1.id, template2.id]):
+                checkbox = self.find_element(
+                    by=By.ID, value=f"id_config-templates_{idx}"
+                )
+                self.assertEqual(checkbox.get_attribute("value"), str(template_id))
+                self.assertEqual(checkbox.get_attribute("data-required"), "false")
+
+        with self.subTest("Save operation completes successfully"):
+            # Scroll to the top of the page to ensure the save button is visible
+            self.web_driver.execute_script("window.scrollTo(0, 0);")
+            self.find_element(by=By.NAME, value="_save").click()
+            self.wait_for_presence(By.CSS_SELECTOR, ".messagelist .success", timeout=5)
+
 
 @tag("selenium_tests")
 class TestDeviceGroupAdmin(

openwisp_controller/geo/tests/test_admin.py

@@ -29,6 +29,12 @@ def setUp(self):
         """override TestAdminMixin.setUp"""
         pass
 
+    def _get_location_add_params(self, **kwargs):
+        params = super()._get_location_add_params(**kwargs)
+        if "organization" not in kwargs:
+            params["organization"] = self._get_org().id
+        return params
+
     def _create_multitenancy_test_env(self, vpn=False):
         org1 = self._create_organization(name="test1org")
         org2 = self._create_organization(name="test2org")

openwisp_controller/geo/tests/test_selenium.py

@@ -91,7 +91,11 @@ def setUp(self):
     def test_unsaved_changes_readonly(self):
         self.login()
         ol = self._create_object_location()
-        path = reverse("admin:config_device_change", args=[ol.device.id])
+        path = reverse(
+            f"admin:{self.object_model._meta.app_label}_"
+            f"{self.object_model._meta.model_name}_change",
+            args=[ol.device.id],
+        )
 
         with self.subTest("Alert should not be displayed without any change"):
             self.open(path)

openwisp_controller/subnet_division/rule_types/base.py

@@ -79,16 +79,36 @@ def _provision_receiver():
     def destroyer_receiver(cls, instance, **kwargs):
         cls.destroy_provisioned_subnets_ips(instance, **kwargs)
 
-    @staticmethod
-    def post_provision_handler(instance, provisioned, **kwargs):
+    @classmethod
+    def post_provision_handler(cls, instance, provisioned, **kwargs):
         """
-        This method should be overridden in inherited rule types to
-        perform any operation on provisioned subnets and IP addresses.
-        :param instance: object that triggered provisioning
-        :param provisioned: dictionary containing subnets and IP addresses
-            provisioned, None if nothing is provisioned
+        Hook for post-provisioning actions on subnets and IP addresses.
+
+        This method is intended to be extended by subclasses of rule types
+        to perform custom operations after subnets and IPs are provisioned.
+
+        Subnet provisioning is executed asynchronously in Celery workers.
+        If the device configuration references variables provided by the
+        subnet division rule, the current checksum may have been computed
+        using variable names instead of their provisioned values. In such cases,
+        `Config.checksum_db` (which tracks persisted configuration changes)
+        must be updated to reflect the actual provisioned values, and the
+        checksum cache invalidated to avoid stale data.
+
+        :param instance: The object that triggered the provisioning.
+        :param provisioned: Dictionary containing provisioned subnets and IPs,
+            or None if no provisioning occurred.
         """
-        pass
+        if not provisioned:
+            return
+        config = cls.get_config(instance)
+        config._invalidate_backend_instance_cache()
+        current_checksum = config.checksum
+        if current_checksum != config.checksum_db:
+            # Update checksum using the UPDATE query to avoid sending
+            # unnecessary signals that may be triggered by `save()` method.
+            config._update_checksum_db(current_checksum)
+            config.invalidate_checksum_cache()
 
     @staticmethod
     def subnet_provisioned_signal_emitter(instance, provisioned):

openwisp_controller/subnet_division/rule_types/vpn.py

@@ -41,8 +41,9 @@ def provision_for_existing_objects(cls, rule_obj):
         for vpn_client in qs:
             cls.provision_receiver(instance=vpn_client, created=True)
 
-    @staticmethod
-    def post_provision_handler(instance, provisioned, **kwargs):
+    @classmethod
+    def post_provision_handler(cls, instance, provisioned, **kwargs):
+        super().post_provision_handler(instance, provisioned, **kwargs)
         # Assign the first provisioned IP address to the VPNClient
         # only when subnets and IPs have been provisioned
         if provisioned and provisioned["ip_addresses"]: