[fix] Skip WHOIS and estimated location for deactivated devices #1325

openwisp_controller/config/whois/commands.py

@@ -39,7 +39,7 @@ def clear_last_ip_command(stdout, stderr, interactive=True):
         )
         # include the FK field 'organization' in .only() so the related
         # `organization__config_settings` traversal is not deferred
-        .only("last_ip", "organization", "key")
+        .only("last_ip", "organization", "key", "_is_deactivated")
     )
     # Filter out devices that have WHOIS information for their last IP
     devices = devices.exclude(last_ip=None).exclude(

openwisp_controller/config/whois/service.py

@@ -210,6 +210,8 @@ def process_ip_data_and_location(self, force_lookup=False):
         Trigger WHOIS lookup based on the conditions of `_need_whois_lookup`.
         Tasks are triggered on commit to ensure redundant data is not created.
         """
+        if self.device.is_deactivated():
+            return
         new_ip = self.device.last_ip
         initial_ip = self.device._initial_last_ip
         if force_lookup or self._need_whois_lookup(new_ip):
@@ -229,6 +231,8 @@ def update_whois_info(self):
         when the data is older than
         ``OPENWISP_CONTROLLER_WHOIS_REFRESH_THRESHOLD_DAYS``.
         """
+        if self.device.is_deactivated():
+            return
         ip_address = self.device.last_ip
         if not self.is_valid_public_ip_address(ip_address):
             return

openwisp_controller/config/whois/tests/tests.py

@@ -1077,6 +1077,65 @@ def test_get_whois_info_when_not_configured(self):
         result = get_whois_info(pk=device.pk)
         self.assertIsNone(result)
 
+    @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
+    @mock.patch("openwisp_controller.config.whois.service.fetch_whois_details.delay")
+    def test_process_ip_skips_when_deactivated(self, mock_task):
+        # Public IP that would normally trigger a lookup, so the only reason
+        # the task is not enqueued is the deactivation guard under test.
+        device = self._create_device()
+        org_settings = device.organization.config_settings
+        org_settings.whois_enabled = True
+        org_settings.save()
+        device._initial_last_ip = None
+        device.last_ip = "8.8.8.8"
+        device.save()
+        mock_task.reset_mock()
+        device.deactivate()
+        service = WHOISService(device)
+        service.process_ip_data_and_location()
+        self.assertEqual(mock_task.call_count, 0)
+
+    @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
+    @mock.patch("openwisp_controller.config.whois.service.fetch_whois_details.delay")
+    def test_process_ip_runs_when_active(self, mock_task):
+        device = self._create_device()
+        org_settings = device.organization.config_settings
+        org_settings.whois_enabled = True
+        org_settings.save()
+        device._initial_last_ip = None
+        device.last_ip = "8.8.8.8"
+        device.save()
+        mock_task.reset_mock()
+        service = WHOISService(device)
+        service.process_ip_data_and_location()
+        # transaction.on_commit executes immediately in TransactionTestCase,
+        # so the task is triggered synchronously here
+        mock_task.assert_called_once_with(
+            device_pk=device.pk,
+            initial_ip_address=device._initial_last_ip,
+        )
+
+    @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
+    @mock.patch("openwisp_controller.config.whois.service.fetch_whois_details.delay")
+    def test_update_whois_skips_when_deactivated(self, mock_task):
+        device = self._create_device()
+        org_settings = device.organization.config_settings
+        org_settings.whois_enabled = True
+        org_settings.save()
+        ip = "8.8.8.8"
+        device._initial_last_ip = None
+        device.last_ip = ip
+        device.save()
+        threshold = app_settings.WHOIS_REFRESH_THRESHOLD_DAYS + 1
+        expired_time = timezone.now() - timedelta(days=threshold)
+        whois_obj = self._create_whois_info(ip_address=ip)
+        WHOISInfo.objects.filter(pk=whois_obj.pk).update(modified=expired_time)
+        device.deactivate()
+        mock_task.reset_mock()
+        service = WHOISService(device)
+        service.update_whois_info()
+        mock_task.assert_not_called()
+
 
 @tag("selenium_tests")
 class TestWHOISSelenium(CreateWHOISMixin, SeleniumTestMixin, StaticLiveServerTestCase):

openwisp_controller/connection/tests/test_tasks.py

@@ -142,8 +142,11 @@ def test_launch_command_exception(self, *args):
 class TestTransactionTasks(
     TestRegistrationMixin, CreateConnectionsMixin, TransactionTestCase
 ):
+    @mock.patch("openwisp_controller.connection.tasks.launch_command.delay")
     @mock.patch.object(tasks.update_config, "delay")
-    def test_update_config_hostname_changed_on_reregister(self, mocked_update_config):
+    def test_update_config_hostname_changed_on_reregister(
+        self, mocked_update_config, mocked_launch_command
+    ):
         device = self._create_device_config()
         self._create_device_connection(device=device)
         # Trigger re-registration with new hostname

openwisp_controller/geo/estimated_location/service.py

@@ -63,6 +63,8 @@ def is_estimated_location_enabled(self):
         return self.check_estimated_location_enabled(self.device.organization_id)
 
     def trigger_estimated_location_task(self, ip_address):
+        if self.device.is_deactivated():
+            return
         try:
             current_app.send_task(
                 "whois_estimated_location_task",