[tests] Added tests

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

Author: DragnEmperor

openwisp_controller/config/whois/service.py

@@ -125,7 +125,7 @@ def trigger_whois_lookup(self):
             )
         elif whois_info_exists and self.is_whois_enabled:
             transaction.on_commit(
-                manage_fuzzy_locations.delay(
+                lambda: manage_fuzzy_locations.delay(
                     self.device.pk, self.device.last_ip, add_existing=True
                 )
             )

openwisp_controller/config/whois/tasks.py

@@ -136,6 +136,14 @@ def fetch_whois_details(self, device_pk, initial_ip_address, new_ip_address):
         whois_obj.full_clean()
         whois_obj.save()
         logger.info(f"Successfully fetched WHOIS details for {new_ip_address}.")
+        location_address = whois_obj.formatted_address
+        manage_fuzzy_locations.delay(
+            device_pk,
+            new_ip_address,
+            data.location.latitude,
+            data.location.longitude,
+            location_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
@@ -192,15 +200,22 @@ def manage_fuzzy_locations(
     # if attaching an existing location, the current device should not have
     # a location already set.
     # TODO: Do we do this if device location exists but is marked fuzzy?
-    if add_existing and not device_location.location:
-        existing_device_with_location = (
-            Device.objects.select_related("device_location")
-            .filter(last_ip=ip_address, device_location__location__isnull=False)
+    current_location = device_location.location
+    if add_existing and (not current_location or current_location.fuzzy):
+        existing_device_location = (
+            Device.objects.select_related("devicelocation")
+            .filter(last_ip=ip_address, devicelocation__location__isnull=False)
+            .exclude(pk=device_pk)
             .first()
         )
-        if existing_device_with_location:
-            location = existing_device_with_location.device_location.location
-            device_location.location = location
+        if existing_device_location:
+            existing_location = existing_device_location.devicelocation.location
+            if current_location and current_location.pk != existing_location.pk:
+                # If current device already has a location, delete it
+                # and set the existing location.
+                current_location.delete()
+
+            device_location.location = existing_location
             device_location.full_clean()
             device_location.save()
     elif latitude and longitude:

openwisp_controller/config/whois/test_whois.py

@@ -1,6 +1,7 @@
 import importlib
 from unittest import mock
 
+from django.contrib.gis.geos import GEOSGeometry
 from django.core.exceptions import ImproperlyConfigured, ValidationError
 from django.db.models.signals import post_delete, post_save
 from django.test import TestCase, TransactionTestCase, override_settings
@@ -14,6 +15,7 @@
 from .utils import CreateWHOISMixin
 
 Device = load_model("config", "Device")
+Location = load_model("geo", "Location")
 WHOISInfo = load_model("config", "WHOISInfo")
 OrganizationConfigSettings = load_model("config", "OrganizationConfigSettings")
 Notification = load_model("openwisp_notifications", "Notification")
@@ -233,37 +235,45 @@ def setUp(self):
         super().setUp()
         self.admin = self._get_admin()
 
-    @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
-    @mock.patch("openwisp_controller.config.whois.tasks.fetch_whois_details.delay")
-    def test_whois_task_called(self, mocked_task):
+    @staticmethod
+    def _mocked_client_response():
+        mock_response = mock.MagicMock()
+        mock_response.city.name = "Mountain View"
+        mock_response.country.name = "United States"
+        mock_response.continent.name = "North America"
+        mock_response.postal.code = "94043"
+        mock_response.traits.autonomous_system_organization = "Google LLC"
+        mock_response.traits.autonomous_system_number = 15169
+        mock_response.traits.network = "172.217.22.0/24"
+        mock_response.location.time_zone = "America/Los_Angeles"
+        mock_response.location.latitude = 2
+        mock_response.location.longitude = 23
+        return mock_response
+
+    def _task_called(self, mocked_task, task_name="WHOIS lookup"):
         org = self._get_org()
         connect_whois_handlers()
 
-        with self.subTest("task called when last_ip is public"):
+        with self.subTest(f"{task_name} task called when last_ip is public"):
             device = self._create_device(last_ip="172.217.22.14")
             mocked_task.assert_called()
         mocked_task.reset_mock()
 
-        with self.subTest("task called when last_ip is changed and is public"):
+        with self.subTest(
+            f"{task_name} task called when last_ip is changed and is public"
+        ):
             device.last_ip = "172.217.22.10"
             device.save()
             mocked_task.assert_called()
         mocked_task.reset_mock()
 
-        with self.subTest("task not called when last_ip is private"):
+        with self.subTest(f"{task_name} task not called when last_ip is private"):
             device.last_ip = "10.0.0.1"
             device.save()
             mocked_task.assert_not_called()
         mocked_task.reset_mock()
 
-        with self.subTest("task not called when last_ip has related WHOISInfo"):
-            device.last_ip = "172.217.22.10"
-            self._create_whois_info(ip_address=device.last_ip)
-            device.save()
-            mocked_task.assert_not_called()
-        mocked_task.reset_mock()
-
-        with self.subTest("task not called when WHOIS is disabled"):
+        with self.subTest(f"{task_name} task not called when WHOIS is disabled"):
             Device.objects.all().delete()
             org.config_settings.whois_enabled = False
             # Invalidates old org config settings cache
@@ -272,7 +282,9 @@ def test_whois_task_called(self, mocked_task):
             mocked_task.assert_not_called()
         mocked_task.reset_mock()
 
-        with self.subTest("task called via DeviceChecksumView when WHOIS is enabled"):
+        with self.subTest(
+            f"{task_name} task called via DeviceChecksumView when WHOIS is enabled"
+        ):
             org.config_settings.whois_enabled = True
             # Invalidates old org config settings cache
             org.config_settings.save(update_fields=["whois_enabled"])
@@ -290,7 +302,7 @@ def test_whois_task_called(self, mocked_task):
         mocked_task.reset_mock()
 
         with self.subTest(
-            "task called via DeviceChecksumView when a device has no WHOIS record"
+            f"{task_name} task called via DeviceChecksumView when a device has no WHOIS record"
         ):
             WHOISInfo.objects.all().delete()
             response = self.client.get(
@@ -302,6 +314,40 @@ def test_whois_task_called(self, mocked_task):
             mocked_task.assert_called()
         mocked_task.reset_mock()
 
+    @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
+    @mock.patch("openwisp_controller.config.whois.tasks.fetch_whois_details.delay")
+    def test_whois_task_called(self, mocked_lookup_task):
+        self._task_called(mocked_lookup_task)
+
+        Device.objects.all().delete()  # Clear existing devices
+        device = self._create_device()
+        with self.subTest("WHOIS lookup task not called when last_ip has related WhoIsInfo"):
+            device.last_ip = "172.217.22.14"
+            self._create_whois_info(ip_address=device.last_ip)
+            device.save()
+            mocked_lookup_task.assert_not_called()
+        mocked_lookup_task.reset_mock()
+
+    @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
+    @mock.patch("openwisp_controller.config.whois.tasks.manage_fuzzy_locations.delay")
+    @mock.patch(_WHOIS_GEOIP_CLIENT)
+    def test_fuzzy_location_task_called(
+        self, mocked_client, mocked_fuzzy_location_task
+    ):
+        mocked_client.return_value.city.return_value = self._mocked_client_response()
+        self._task_called(mocked_fuzzy_location_task, task_name="Fuzzy location")
+
+        Device.objects.all().delete()
+        device = self._create_device()
+        with self.subTest(
+            "Fuzzy location task called when last_ip has related WhoIsInfo"
+        ):
+            device.last_ip = "172.217.22.14"
+            self._create_whois_info(ip_address=device.last_ip)
+            device.save()
+            mocked_fuzzy_location_task.assert_called()
+        mocked_fuzzy_location_task.reset_mock()
+
     @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
     @mock.patch("openwisp_controller.config.whois.tasks.fetch_whois_details.delay")
     def test_whois_multiple_orgs(self, mocked_task):
@@ -406,16 +452,7 @@ def _verify_whois_details(instance, ip_address):
             )
 
         # mocking the response from the geoip2 client
-        mock_response = mock.MagicMock()
-        mock_response.city.name = "Mountain View"
-        mock_response.country.name = "United States"
-        mock_response.continent.name = "North America"
-        mock_response.postal.code = "94043"
-        mock_response.traits.autonomous_system_organization = "Google LLC"
-        mock_response.traits.autonomous_system_number = 15169
-        mock_response.traits.network = "172.217.22.0/24"
-        mock_response.location.time_zone = "America/Los_Angeles"
-        mock_client.return_value.city.return_value = mock_response
+        mock_client.return_value.city.return_value = self._mocked_client_response()
 
         with self.subTest("Test WHOIS create when device is created"):
             device = self._create_device(last_ip="172.217.22.14")
@@ -455,6 +492,109 @@ def _verify_whois_details(instance, ip_address):
             # WHOIS related to the device's last_ip should be deleted
             self.assertEqual(WHOISInfo.objects.filter(ip_address=ip_address).count(), 0)
 
+    @mock.patch.object(app_settings, "WHOIS_CONFIGURED", True)
+    @mock.patch(_WHOIS_GEOIP_CLIENT)
+    def test_fuzzy_location_creation_and_update(self, mock_client):
+        connect_whois_handlers()
+
+        def _verify_location_details(device, mocked_response):
+            location = device.devicelocation.location
+            mocked_location = mocked_response.location
+            formatted_address = ", ".join(
+                [
+                    mocked_response.city.name,
+                    mocked_response.country.name,
+                    mocked_response.continent.name,
+                    mocked_response.postal.code,
+                ]
+            )
+            self.assertEqual(location.address, formatted_address)
+            self.assertEqual(
+                location.geometry,
+                GEOSGeometry(
+                    f"POINT({mocked_location.longitude} {mocked_location.latitude})",
+                    srid=4326,
+                ),
+            )
+
+        mocked_response = self._mocked_client_response()
+        mock_client.return_value.city.return_value = mocked_response
+
+        with self.subTest("Test Fuzzy location created when device is created"):
+            device = self._create_device(last_ip="172.217.22.14")
+
+            location = device.devicelocation.location
+            self.assertEqual(location.fuzzy, True)
+            self.assertEqual(location.is_mobile, False)
+            self.assertEqual(location.type, "outdoor")
+            _verify_location_details(device, mocked_response)
+
+        with self.subTest("Test Fuzzy location updated when last ip is updated"):
+            device.last_ip = "172.217.22.10"
+            mocked_response.location.latitude = 100
+            mocked_response.location.longitude = 200
+            mocked_response.city.name = "New City"
+            mock_client.return_value.city.return_value = mocked_response
+            device.save()
+            device.refresh_from_db()
+
+            location = device.devicelocation.location
+            self.assertEqual(location.fuzzy, True)
+            self.assertEqual(location.is_mobile, False)
+            self.assertEqual(location.type, "outdoor")
+            _verify_location_details(device, mocked_response)
+
+        with self.subTest(
+            "Test Location not updated if it is not fuzzy when last ip is updated"
+        ):
+            device.last_ip = "172.217.22.11"
+            device.devicelocation.location.fuzzy = False
+            mock_client.return_value.city.return_value = self._mocked_client_response()
+            device.devicelocation.location.save()
+            device.save()
+            device.refresh_from_db()
+
+            location = device.devicelocation.location
+            self.assertEqual(location.fuzzy, False)
+            self.assertEqual(location.is_mobile, False)
+            self.assertEqual(location.type, "outdoor")
+            _verify_location_details(device, mocked_response)
+
+        with self.subTest(
+            "Test Location shared among devices with same last_ip when new device's location does not exist"
+        ):
+            Device.objects.all().delete()
+            device1 = self._create_device(last_ip="172.217.22.10")
+            device2 = self._create_device(
+                name="11:22:33:44:55:66",
+                mac_address="11:22:33:44:55:66",
+                last_ip="172.217.22.10",
+            )
+
+            self.assertEqual(
+                device1.devicelocation.location.pk, device2.devicelocation.location.pk
+            )
+
+        with self.subTest(
+            "Test Location shared among devices with same last_ip when new device's location exist"
+        ):
+            Device.objects.all().delete()
+            device1 = self._create_device(last_ip="172.217.22.10")
+            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",
+            )
+            old_location = device2.devicelocation.location
+            device2.last_ip = "172.217.22.10"
+            device2.save()
+            device2.refresh_from_db()
+
+            self.assertEqual(
+                device1.devicelocation.location.pk, device2.devicelocation.location.pk
+            )
+            self.assertEqual(Location.objects.filter(pk=old_location.pk).count(), 0)
+
     # we need to allow the task to propagate exceptions to ensure
     # `on_failure` method is called and notifications are executed
     @override_settings(CELERY_TASK_EAGER_PROPAGATES=False)

openwisp_controller/geo/base/models.py

@@ -1,4 +1,5 @@
 from django.contrib.gis.db import models
+from django.utils.translation import gettext_lazy as _
 from django_loci.base.models import (
     AbstractFloorPlan,
     AbstractLocation,
@@ -10,6 +11,14 @@
 
 
 class BaseLocation(OrgMixin, AbstractLocation):
+    fuzzy = models.BooleanField(
+        default=False,
+        help_text=_(
+            "If true, the location is considered fuzzy and "
+            "may not have precise coordinates."
+        ),
+    )
+
     class Meta(AbstractLocation.Meta):
         abstract = True
 

openwisp_controller/geo/migrations/0004_location_fuzzy.py

@@ -0,0 +1,24 @@
+# Generated by Django 5.2.1 on 2025-06-25 19:22
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("geo", "0003_alter_devicelocation_floorplan_location"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="location",
+            name="fuzzy",
+            field=models.BooleanField(
+                default=False,
+                help_text=(
+                    "If true, the location is considered fuzzy and "
+                    "may not have precise coordinates."
+                ),
+            ),
+        ),
+    ]