From 77e841389c4ee734151960e486ad84f7687aec0f Mon Sep 17 00:00:00 2001 From: Tom Date: Fri, 24 Apr 2026 00:56:16 +0530 Subject: [PATCH 1/4] [test(geo)] Add regression tests for deactivated device WHOIS handling #1325 This commit adds regression tests to ensure that the WHOIS handling for deactivated devices works as expected. --- null | 1 + openwisp_controller/config/whois/service.py | 4 +++ .../config/whois/tests/tests.py | 35 +++++++++++++++++-- 3 files changed, 37 insertions(+), 3 deletions(-) create mode 100644 null diff --git a/null b/null new file mode 100644 index 000000000..992b1eeb8 --- /dev/null +++ b/null @@ -0,0 +1 @@ +fatal: no rebase in progress diff --git a/openwisp_controller/config/whois/service.py b/openwisp_controller/config/whois/service.py index 43a6ab3db..304a0d1fe 100644 --- a/openwisp_controller/config/whois/service.py +++ b/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_deactived(): + 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_deactived(): + return ip_address = self.device.last_ip if not self.is_valid_public_ip_address(ip_address): return diff --git a/openwisp_controller/config/whois/tests/tests.py b/openwisp_controller/config/whois/tests/tests.py index 05aaf644e..38241f09f 100644 --- a/openwisp_controller/config/whois/tests/tests.py +++ b/openwisp_controller/config/whois/tests/tests.py @@ -2,7 +2,7 @@ import importlib from datetime import timedelta from io import StringIO -from unittest import mock +from unittest import mock, mocke from uuid import uuid4 from django.conf import settings @@ -21,7 +21,9 @@ from selenium.webdriver.common.by import By from swapper import load_model +from openwisp_controller.config import settings as app_settings from openwisp_controller.config.signals import whois_fetched, whois_lookup_skipped +from openwisp_controller.config.whois.service import WHOISService from openwisp_utils.tests import SeleniumTestMixin, catch_signal from ....tests.utils import TestAdminMixin @@ -346,8 +348,8 @@ def test_last_ip_management_command_queries(self): name="default.test.device4", mac_address="66:33:44:55:66:77" ) args = ["--noinput"] - # 4 queries (3 for each device's last_ip update) and 1 for fetching devices - with self.assertNumQueries(4): + + with self.assertNumQueries(7): call_command("clear_last_ip", *args, stdout=out, stderr=StringIO()) @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True) @@ -1186,3 +1188,30 @@ def _assert_no_js_errors(): _assert_no_js_errors() except UnexpectedAlertPresentException: self.fail("XSS vulnerability detected in WHOIS details admin view.") + + +class TestWHOISDeactivated(TransactionTestCase): + def setUp(self): + self.device = self._create_device() + self.device.last_ip = "8.8.8.8" # public IP + self.device.save() + + @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): + self.device._is_deactivated = True + + service = WHOISService(self.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): + self.device._is_deactivated = False + + service = WHOISService(self.device) + service.process_ip_data_and_location() + + self.assertEqual(mock_task.call_count, 1) From a7b9842eb0d64007f9d86fcfbc05ffdfe3b2c3d3 Mon Sep 17 00:00:00 2001 From: Tom Date: Fri, 24 Apr 2026 01:17:08 +0530 Subject: [PATCH 2/4] [test(geo)] Add regression tests for deactivated device WHOIS handling #1325 This commit adds regression tests to ensure that the WHOIS handling for deactivated devices works as expected. --- openwisp_controller/config/whois/service.py | 4 ++-- openwisp_controller/config/whois/tests/tests.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/openwisp_controller/config/whois/service.py b/openwisp_controller/config/whois/service.py index 304a0d1fe..a98d44dd6 100644 --- a/openwisp_controller/config/whois/service.py +++ b/openwisp_controller/config/whois/service.py @@ -210,7 +210,7 @@ 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_deactived(): + if self.device.is_deactivated(): return new_ip = self.device.last_ip initial_ip = self.device._initial_last_ip @@ -231,7 +231,7 @@ def update_whois_info(self): when the data is older than ``OPENWISP_CONTROLLER_WHOIS_REFRESH_THRESHOLD_DAYS``. """ - if self.device.is_deactived(): + if self.device.is_deactivated(): return ip_address = self.device.last_ip if not self.is_valid_public_ip_address(ip_address): diff --git a/openwisp_controller/config/whois/tests/tests.py b/openwisp_controller/config/whois/tests/tests.py index 38241f09f..bc1f064c9 100644 --- a/openwisp_controller/config/whois/tests/tests.py +++ b/openwisp_controller/config/whois/tests/tests.py @@ -2,7 +2,7 @@ import importlib from datetime import timedelta from io import StringIO -from unittest import mock, mocke +from unittest import mock from uuid import uuid4 from django.conf import settings From 8d528e7e86bb4e9973f9d5c94e08ff6e6126f50c Mon Sep 17 00:00:00 2001 From: Tom Date: Fri, 24 Apr 2026 15:52:32 +0530 Subject: [PATCH 3/4] Update tests.py --- openwisp_controller/config/whois/tests/tests.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/openwisp_controller/config/whois/tests/tests.py b/openwisp_controller/config/whois/tests/tests.py index bc1f064c9..0ab5d1f78 100644 --- a/openwisp_controller/config/whois/tests/tests.py +++ b/openwisp_controller/config/whois/tests/tests.py @@ -1190,11 +1190,15 @@ def _assert_no_js_errors(): self.fail("XSS vulnerability detected in WHOIS details admin view.") -class TestWHOISDeactivated(TransactionTestCase): +class TestWHOISDeactivated( + CreateWHOISMixin, WHOISTransactionMixin, TransactionTestCase +): def setUp(self): + super().setUp() self.device = self._create_device() - self.device.last_ip = "8.8.8.8" # public IP - self.device.save() + org_settings = self.device.organization.config_settings + org_settings.whois_enabled = True + org_settings.save() @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True) @mock.patch("openwisp_controller.config.whois.service.fetch_whois_details.delay") @@ -1210,6 +1214,7 @@ def test_process_ip_skips_when_deactivated(self, mock_task): @mock.patch("openwisp_controller.config.whois.service.fetch_whois_details.delay") def test_process_ip_runs_when_active(self, mock_task): self.device._is_deactivated = False + self.device.last_ip = "8.8.8.8" service = WHOISService(self.device) service.process_ip_data_and_location() From 6ace4369a147f982e9ace9f7934e998830c1f0f8 Mon Sep 17 00:00:00 2001 From: Tom Date: Fri, 24 Apr 2026 16:29:18 +0530 Subject: [PATCH 4/4] [fix] Skip WHOIS and estimated location for deactivated devices #1325 - add guard in WHOISService to skip processing for deactivated devices - add guard in EstimatedLocationService to prevent task scheduling - fix N+1 query by including _is_deactivated in clear_last_ip_command queryset - add regression tests ensuring tasks are skipped when device is deactivated - ensure deterministic tests by enabling whois on organization settings Fixes #1325 --- null | 1 - openwisp_controller/config/whois/commands.py | 2 +- .../config/whois/tests/tests.py | 97 ++++++++++++------- .../connection/tests/test_tasks.py | 5 +- .../geo/estimated_location/service.py | 2 + 5 files changed, 68 insertions(+), 39 deletions(-) delete mode 100644 null diff --git a/null b/null deleted file mode 100644 index 992b1eeb8..000000000 --- a/null +++ /dev/null @@ -1 +0,0 @@ -fatal: no rebase in progress diff --git a/openwisp_controller/config/whois/commands.py b/openwisp_controller/config/whois/commands.py index dafbe92e8..185ef9085 100644 --- a/openwisp_controller/config/whois/commands.py +++ b/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( diff --git a/openwisp_controller/config/whois/tests/tests.py b/openwisp_controller/config/whois/tests/tests.py index 0ab5d1f78..f5529ebb2 100644 --- a/openwisp_controller/config/whois/tests/tests.py +++ b/openwisp_controller/config/whois/tests/tests.py @@ -21,9 +21,7 @@ from selenium.webdriver.common.by import By from swapper import load_model -from openwisp_controller.config import settings as app_settings from openwisp_controller.config.signals import whois_fetched, whois_lookup_skipped -from openwisp_controller.config.whois.service import WHOISService from openwisp_utils.tests import SeleniumTestMixin, catch_signal from ....tests.utils import TestAdminMixin @@ -348,8 +346,8 @@ def test_last_ip_management_command_queries(self): name="default.test.device4", mac_address="66:33:44:55:66:77" ) args = ["--noinput"] - - with self.assertNumQueries(7): + # 4 queries (3 for each device's last_ip update) and 1 for fetching devices + with self.assertNumQueries(4): call_command("clear_last_ip", *args, stdout=out, stderr=StringIO()) @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True) @@ -1079,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): @@ -1188,35 +1245,3 @@ def _assert_no_js_errors(): _assert_no_js_errors() except UnexpectedAlertPresentException: self.fail("XSS vulnerability detected in WHOIS details admin view.") - - -class TestWHOISDeactivated( - CreateWHOISMixin, WHOISTransactionMixin, TransactionTestCase -): - def setUp(self): - super().setUp() - self.device = self._create_device() - org_settings = self.device.organization.config_settings - org_settings.whois_enabled = True - org_settings.save() - - @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): - self.device._is_deactivated = True - - service = WHOISService(self.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): - self.device._is_deactivated = False - self.device.last_ip = "8.8.8.8" - - service = WHOISService(self.device) - service.process_ip_data_and_location() - - self.assertEqual(mock_task.call_count, 1) diff --git a/openwisp_controller/connection/tests/test_tasks.py b/openwisp_controller/connection/tests/test_tasks.py index 8ce18a694..440272a9c 100644 --- a/openwisp_controller/connection/tests/test_tasks.py +++ b/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 diff --git a/openwisp_controller/geo/estimated_location/service.py b/openwisp_controller/geo/estimated_location/service.py index a3d661f81..bda07cc5e 100644 --- a/openwisp_controller/geo/estimated_location/service.py +++ b/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",