[changes] Perform lookup and location for last_ip updation only

Signed-off-by: DragnEmperor <[email protected]>

Author: DragnEmperor

docs/user/whois.rst

@@ -40,25 +40,23 @@ A WHOIS lookup is triggered automatically when:
 However, the lookup will only run if **all** the following conditions are
 met:
 
+- The device is either **newly created** or has a **changed last IP**.
 - The device's last IP address is **public**.
 - There is **no existing WHOIS record** for that IP.
 - WHOIS lookup is **enabled** for the device's organization.
 
-Behavior with Shared IP Addresses
----------------------------------
+Managing WHOIS Records
+----------------------
 
-If multiple devices share the same public IP address and one of them
-switches to a different IP, the following occurs:
-
-- A lookup is triggered for the **new IP**.
-- The WHOIS record for the **old IP** is deleted.
-- The next time a device still using the old IP fetches its checksum, a
-  new lookup is triggered, ensuring up-to-date data.
+If a device updates its last IP address, lookup is triggered for the **new
+IP** and the **WHOIS record for the old IP** is deleted if no active
+devices are associated with that IP address.
 
 .. note::
 
     When a device with an associated WHOIS record is deleted, its WHOIS
-    record is automatically removed.
+    record is automatically removed only if no active devices are
+    associated with that IP address.
 
 .. _controller_setup_whois_lookup:
 

openwisp_controller/config/base/device.py

@@ -287,7 +287,7 @@ def save(self, *args, **kwargs):
         state_adding = self._state.adding
         super().save(*args, **kwargs)
         if app_settings.WHOIS_CONFIGURED:
-            self._check_last_ip()
+            self._check_last_ip(creating=state_adding)
         if state_adding and self.group and self.group.templates.exists():
             self.create_default_config()
         # The value of "self._state.adding" will always be "False"
@@ -510,9 +510,10 @@ def whois_service(self):
         """
         return WHOISService(self)
 
-    def _check_last_ip(self):
+    def _check_last_ip(self, creating=False):
         """Trigger WHOIS lookup if last_ip is not deferred."""
         if self._initial_last_ip == models.DEFERRED:
             return
-        self.whois_service.trigger_whois_lookup()
+        if creating or self.last_ip != self._initial_last_ip:
+            self.whois_service.process_ip_data_and_location()
         self._initial_last_ip = self.last_ip

openwisp_controller/config/base/whois.py

@@ -6,6 +6,7 @@
 from django.db import models, transaction
 from django.utils.translation import gettext_lazy as _
 from jsonfield import JSONField
+from swapper import load_model
 
 from openwisp_utils.base import TimeStampedEditableModel
 
@@ -103,7 +104,17 @@ def device_whois_info_delete_handler(instance, **kwargs):
         Delete WHOIS information for a device when the last IP address is removed or
         when device is deleted.
         """
-        if instance._get_organization__config_settings().whois_enabled:
+        Device = load_model("config", "Device")
+
+        last_ip = instance.last_ip
+        existing_devices = Device.objects.filter(_is_deactivated=False).filter(
+            last_ip=last_ip
+        )
+        if (
+            last_ip
+            and instance._get_organization__config_settings().whois_enabled
+            and not existing_devices.exists()
+        ):
             transaction.on_commit(lambda: delete_whois_record.delay(instance.last_ip))
 
     # this method is kept here instead of in OrganizationConfigSettings because

openwisp_controller/config/controller/views.py

@@ -158,7 +158,7 @@ def get(self, request, pk):
         # trigger the WHOIS lookup which checks if WHOIS lookup is required
         # or not
         elif app_settings.WHOIS_CONFIGURED:
-            device.whois_service.trigger_whois_lookup()
+            device.whois_service.process_ip_data_and_location()
         checksum_requested.send(
             sender=device.__class__, instance=device, request=request
         )

openwisp_controller/config/whois/service.py

@@ -129,10 +129,12 @@ def get_device_whois_info(self):
 
         return self._get_whois_info_from_db(ip_address=ip_address).first()
 
-    def trigger_whois_lookup(self):
+    def process_ip_data_and_location(self):
         """
-        Trigger WHOIS lookup based on the conditions of `_need_whois_lookup`.
-        Task is triggered on commit to ensure redundant data is not created.
+        Trigger WHOIS lookup based on the conditions of `_need_whois_lookup`
+        and also manage estimated locations based on the conditions of
+        `_need_estimated_location_management`.
+        Tasks are triggered on commit to ensure redundant data is not created.
         """
         new_ip = self.device.last_ip
         if self._need_whois_lookup(new_ip):

openwisp_controller/config/whois/tasks.py

@@ -128,14 +128,12 @@ def fetch_whois_details(self, device_pk, initial_ip_address, new_ip_address):
                 device_pk=device_pk, ip_address=new_ip_address
             )
 
-        # the following check ensures that for a case when device last_ip
-        # is not changed and there is no related WHOIS record, we do not
-        # delete the newly created record as both `initial_ip_address` and
-        # `new_ip_address` would be same for such case.
-        if initial_ip_address != new_ip_address:
-            # If any active devices are linked to the following record,
-            # then they will trigger this task and new record gets created
-            # with latest data.
+        # delete WHOIS record for initial IP if no devices are linked to it
+        if (
+            not Device.objects.filter(_is_deactivated=False)
+            .filter(last_ip=initial_ip_address)
+            .exists()
+        ):
             delete_whois_record(ip_address=initial_ip_address)
 
 

openwisp_controller/config/whois/test_whois.py

@@ -491,6 +491,7 @@ def _verify_whois_details(instance, ip_address):
 
         with self.subTest(
             "Test WHOIS create & deletion of old record when last ip is updated"
+            " when no other devices are linked to the old ip address"
         ):
             old_ip_address = device.last_ip
             device.last_ip = "172.217.22.10"
@@ -508,12 +509,61 @@ def _verify_whois_details(instance, ip_address):
                 WHOISInfo.objects.filter(ip_address=old_ip_address).count(), 0
             )
 
-        with self.subTest("Test WHOIS delete when device is deleted"):
+        with self.subTest(
+            "Test WHOIS create & deletion of old record when last ip is updated"
+            " when other devices are linked to the old ip address"
+        ):
+            old_ip_address = device.last_ip
+            self._create_device(
+                name="11:22:33:44:55:66",
+                mac_address="11:22:33:44:55:66",
+                last_ip="172.217.22.11",
+            )
+            device.last_ip = "172.217.22.11"
+            device.save()
+            self.assertEqual(mock_info.call_count, 1)
+            mock_info.reset_mock()
+            device.refresh_from_db()
+
+            _verify_whois_details(
+                device.whois_service.get_device_whois_info(), device.last_ip
+            )
+
+            # details related to old ip address should be not be deleted
+            self.assertEqual(
+                WHOISInfo.objects.filter(ip_address=old_ip_address).count(), 1
+            )
+
+        with self.subTest(
+            "Test WHOIS not deleted when device is deleted and"
+            " other active devices are linked to the last_ip"
+        ):
             ip_address = device.last_ip
             device.delete(check_deactivated=False)
             self.assertEqual(mock_info.call_count, 0)
             mock_info.reset_mock()
 
+            # WHOIS related to the device's last_ip should be deleted
+            self.assertEqual(WHOISInfo.objects.filter(ip_address=ip_address).count(), 1)
+
+        Device.objects.all().delete()
+        with self.subTest(
+            "Test WHOIS deleted when device is deleted and"
+            " no other active devices are linked to the last_ip"
+        ):
+            device1 = self._create_device(last_ip="172.217.22.11")
+            device2 = self._create_device(
+                name="11:22:33:44:55:66",
+                mac_address="11:22:33:44:55:66",
+                last_ip="172.217.22.11",
+            )
+            device2.deactivate()
+            mock_info.reset_mock()
+            ip_address = device1.last_ip
+            device1.delete(check_deactivated=False)
+            self.assertEqual(mock_info.call_count, 0)
+            mock_info.reset_mock()
+
             # WHOIS related to the device's last_ip should be deleted
             self.assertEqual(WHOISInfo.objects.filter(ip_address=ip_address).count(), 0)
 

openwisp_controller/config/whois/tests_utils.py

@@ -80,6 +80,12 @@ def _task_called(self, mocked_task, task_name="WHOIS lookup"):
                 mocked_get.assert_called_once()
         mocked_task.reset_mock()
 
+        with self.subTest(f"{task_name} task not called when last_ip not updated"):
+            device.name = "default.test.Device2"
+            device.save()
+            mocked_task.assert_not_called()
+        mocked_task.reset_mock()
+
         with self.subTest(f"{task_name} task not called when last_ip is private"):
             device.last_ip = "10.0.0.1"
             device.save()

openwisp_controller/geo/estimated_location/tasks.py

@@ -21,8 +21,8 @@ def manage_estimated_locations(device_pk, ip_address):
     - If the current device has no location or location is estimate, either update
     to an existing location; if it exists, else
 
-    - A new location is created if no location exists for current device, or
-    existing one is updated using coords from WHOIS record if it is estimated.
+    - A new location is created if current device has no location, or
+    if it does; it is updated using coords from WHOIS record if it is estimated.
 
     In case of multiple devices with same last_ip, the task will send a notification
     to the user to resolve the conflict manually.