[changes] Added PointField, functions refactoring

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

Author: DragnEmperor

openwisp_controller/config/base/whois.py

@@ -1,8 +1,8 @@
 from ipaddress import ip_address, ip_network
 
+from django.contrib.gis.db.models import PointField
 from django.core.cache import cache
 from django.core.exceptions import ValidationError
-from django.core.validators import MaxValueValidator, MinValueValidator
 from django.db import models, transaction
 from django.utils.translation import gettext_lazy as _
 from jsonfield import JSONField
@@ -52,17 +52,23 @@ class AbstractWHOISInfo(TimeStampedEditableModel):
         blank=True,
         help_text=_("CIDR"),
     )
-    latitude = models.FloatField(
+    # latitude = models.FloatField(
+    #     null=True,
+    #     blank=True,
+    #     help_text=_("Latitude"),
+    #     validators=[MinValueValidator(-90.0), MaxValueValidator(90.0)],
+    # )
+    # longitude = models.FloatField(
+    #     null=True,
+    #     blank=True,
+    #     help_text=_("Longitude"),
+    #     validators=[MinValueValidator(-180.0), MaxValueValidator(180.0)],
+    # )
+    coordinates = PointField(
         null=True,
         blank=True,
-        help_text=_("Latitude"),
-        validators=[MinValueValidator(-90.0), MaxValueValidator(90.0)],
-    )
-    longitude = models.FloatField(
-        null=True,
-        blank=True,
-        help_text=_("Longitude"),
-        validators=[MinValueValidator(-180.0), MaxValueValidator(180.0)],
+        help_text=_("Coordinates"),
+        srid=4326,
     )
 
     class Meta:
@@ -87,6 +93,20 @@ def clean(self):
                 raise ValidationError(
                     {"cidr": _("Invalid CIDR format: %(error)s") % {"error": str(e)}}
                 )
+
+        if self.coordinates:
+            if not (-90 <= self.coordinates.y <= 90):
+                raise ValidationError(
+                    {"coordinates": _("Latitude must be between -90 and 90 degrees.")}
+                )
+            if not (-180 <= self.coordinates.x <= 180):
+                raise ValidationError(
+                    {
+                        "coordinates": _(
+                            "Longitude must be between -180 and 180 degrees."
+                        )
+                    }
+                )
         return super().clean()
 
     @staticmethod
@@ -126,3 +146,28 @@ def formatted_address(self):
                 ],
             )
         )
+
+    @property
+    def _location_name(self):
+        """
+        Used to get location name based on the address and IP.
+        """
+        address = self.formatted_address
+        if address:
+            parts = [part.strip() for part in address.split(",")[:2] if part.strip()]
+            location = ", ".join(parts)
+            return f"{location} (Estimated Location: {self.ip_address})"
+        return f"Estimated Location: {self.ip_address}"
+
+    def _get_defaults_for_estimated_location(self):
+        """
+        Used to get default values for creating or updating
+        an estimated location based on the WHOIS information.
+        """
+        return {
+            "name": self._location_name,
+            "type": "outdoor",
+            "is_mobile": False,
+            "geometry": self.coordinates,
+            "address": self.formatted_address,
+        }

openwisp_controller/config/migrations/0062_organizationconfigsettings_approximate_location_enabled_and_more.py

@@ -1,7 +1,7 @@
 # Generated by Django 5.2.1 on 2025-07-10 18:09
 
-import django.core.validators
-from django.db import migrations, models
+import django.contrib.gis.db.models.fields
+from django.db import migrations
 
 import openwisp_utils.fields
 
@@ -27,28 +27,9 @@ class Migration(migrations.Migration):
         ),
         migrations.AddField(
             model_name="whoisinfo",
-            name="latitude",
-            field=models.FloatField(
-                blank=True,
-                help_text="Latitude",
-                null=True,
-                validators=[
-                    django.core.validators.MinValueValidator(-90.0),
-                    django.core.validators.MaxValueValidator(90.0),
-                ],
-            ),
-        ),
-        migrations.AddField(
-            model_name="whoisinfo",
-            name="longitude",
-            field=models.FloatField(
-                blank=True,
-                help_text="Longitude",
-                null=True,
-                validators=[
-                    django.core.validators.MinValueValidator(-180.0),
-                    django.core.validators.MaxValueValidator(180.0),
-                ],
+            name="coordinates",
+            field=django.contrib.gis.db.models.fields.PointField(
+                blank=True, help_text="Coordinates", null=True, srid=4326
             ),
         ),
     ]

openwisp_controller/config/whois/service.py

@@ -1,4 +1,4 @@
-from ipaddress import ip_address
+from ipaddress import ip_address as ip_addr
 
 from django.core.cache import cache
 from django.db import transaction
@@ -30,7 +30,7 @@ def is_valid_public_ip_address(ip):
         Check if given IP address is a valid public IP address.
         """
         try:
-            return ip and ip_address(ip).is_global
+            return ip and ip_addr(ip).is_global
         except ValueError:
             return False
 

openwisp_controller/config/whois/tasks.py

@@ -2,6 +2,7 @@
 
 import requests
 from celery import shared_task
+from django.contrib.gis.geos import Point
 from django.utils.translation import gettext as _
 from geoip2 import errors
 from geoip2 import webservice as geoip2_webservice
@@ -109,16 +110,15 @@ def fetch_whois_details(self, device_pk, initial_ip_address, new_ip_address):
             "continent": data.continent.name or "",
             "postal": str(data.postal.code or ""),
         }
-
+        coordinates = Point(data.location.longitude, data.location.latitude, srid=4326)
         whois_obj = WHOISInfo(
             isp=data.traits.autonomous_system_organization,
             asn=data.traits.autonomous_system_number,
             timezone=data.location.time_zone,
             address=address,
             cidr=data.traits.network,
             ip_address=new_ip_address,
-            longitude=data.location.longitude,
-            latitude=data.location.latitude,
+            coordinates=coordinates,
         )
         whois_obj.full_clean()
         whois_obj.save()

openwisp_controller/config/whois/test_whois.py

@@ -2,6 +2,7 @@
 from unittest import mock
 
 from django.contrib.staticfiles.testing import StaticLiveServerTestCase
+from django.contrib.gis.geos import Point
 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, tag
@@ -15,8 +16,7 @@
 from ...tests.utils import TestAdminMixin
 from .. import settings as app_settings
 from .handlers import connect_whois_handlers
-from .tests_utils import CreateWHOISMixin
-from utils import WHOISTransactionMixin
+from .tests_utils import CreateWHOISMixin, WHOISTransactionMixin
 
 Device = load_model("config", "Device")
 WHOISInfo = load_model("config", "WHOISInfo")
@@ -302,32 +302,20 @@ def test_whois_model_fields_validation(self):
         with self.assertRaises(ValidationError):
             self._create_whois_info(asn="InvalidASN")
 
-        # Common validation checks for latitude and longitude
-        latitudes = [
-            (100.0, "Ensure this value is less than or equal to 90.0."),
-            (-100.0, "Ensure this value is greater than or equal to -90.0."),
+        # Common validation checks for longitude and latitude
+        coordinates_cases = [
+            (150.0, 100.0, "Latitude must be between -90 and 90 degrees."),
+            (150.0, -100.0, "Latitude must be between -90 and 90 degrees."),
+            (200.0, 80.0, "Longitude must be between -180 and 180 degrees."),
+            (-200.0, -80.0, "Longitude must be between -180 and 180 degrees."),
         ]
-        for value, expected_msg in latitudes:
+        for longitude, latitude, expected_msg in coordinates_cases:
             with self.assertRaises(ValidationError) as context_manager:
-                self._create_whois_info(latitude=value)
+                point = Point(longitude, latitude, srid=4326)
+                self._create_whois_info(coordinates=point)
             try:
                 self.assertEqual(
-                    context_manager.exception.message_dict["latitude"][0],
-                    expected_msg,
-                )
-            except AssertionError:
-                self.fail("ValidationError message not equal to expected message.")
-
-        longitudes = [
-            (200.0, "Ensure this value is less than or equal to 180.0."),
-            (-200.0, "Ensure this value is greater than or equal to -180.0."),
-        ]
-        for value, expected_msg in longitudes:
-            with self.assertRaises(ValidationError) as context_manager:
-                self._create_whois_info(longitude=value)
-            try:
-                self.assertEqual(
-                    context_manager.exception.message_dict["longitude"][0],
+                    context_manager.exception.message_dict["coordinates"][0],
                     expected_msg,
                 )
             except AssertionError:
@@ -467,8 +455,8 @@ def _verify_whois_details(instance, ip_address):
                 instance.formatted_address,
                 "Mountain View, United States, North America, 94043",
             )
-            self.assertEqual(instance.latitude, 50)
-            self.assertEqual(instance.longitude, 150)
+            self.assertEqual(instance.coordinates.x, 150.0)
+            self.assertEqual(instance.coordinates.y, 50.0)
 
         # mocking the response from the geoip2 client
         mock_client.return_value.city.return_value = self._mocked_client_response()

openwisp_controller/config/whois/tests_utils.py

@@ -1,3 +1,7 @@
+from unittest import mock
+
+from django.contrib.gis.geos import Point
+from django.urls import reverse
 from swapper import load_model
 
 from ..tests.utils import CreateConfigMixin
@@ -21,6 +25,7 @@ def _create_whois_info(self, **kwargs):
             isp="Google LLC",
             timezone="America/Los_Angeles",
             cidr="172.217.22.0/24",
+            coordinates=Point(150, 50, srid=4326),
         )
 
         options.update(kwargs)
@@ -34,3 +39,104 @@ def setUp(self):
         OrganizationConfigSettings.objects.create(
             organization=self._get_org(), whois_enabled=True
         )
+
+
+class WHOISTransactionMixin:
+    @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 = 50
+        mock_response.location.longitude = 150
+        return mock_response
+
+    def _task_called(self, mocked_task, task_name="WHOIS lookup"):
+        org = self._get_org()
+
+        with self.subTest(f"{task_name} task called when last_ip is public"):
+            with mock.patch("django.core.cache.cache.set") as mocked_set:
+                device = self._create_device(last_ip="172.217.22.14")
+                mocked_task.assert_called()
+                mocked_set.assert_called_once()
+        mocked_task.reset_mock()
+
+        with self.subTest(
+            f"{task_name} task called when last_ip is changed and is public"
+        ):
+            with mock.patch("django.core.cache.cache.get") as mocked_get:
+                device.last_ip = "172.217.22.10"
+                device.save()
+                mocked_task.assert_called()
+                # The cache `get` is called twice, once for `whois_enabled` and
+                # once for `estimated_location_enabled`
+                mocked_get.assert_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()
+            mocked_task.assert_not_called()
+        mocked_task.reset_mock()
+
+        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
+            org.config_settings.save(update_fields=["whois_enabled"])
+            device = self._create_device(last_ip="172.217.22.14")
+            mocked_task.assert_not_called()
+        mocked_task.reset_mock()
+
+        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"])
+            # config is required for checksum view to work
+            self._create_config(device=device)
+            # setting remote address field to a public IP to trigger WHOIS task
+            # since the view uses this header for tracking the device's IP
+            response = self.client.get(
+                reverse("controller:device_checksum", args=[device.pk]),
+                {"key": device.key},
+                REMOTE_ADDR="172.217.22.10",
+            )
+            self.assertEqual(response.status_code, 200)
+            mocked_task.assert_called()
+        mocked_task.reset_mock()
+
+        with self.subTest(
+            f"{task_name} task called via DeviceChecksumView for no WHOIS record"
+        ):
+            WHOISInfo.objects.all().delete()
+            response = self.client.get(
+                reverse("controller:device_checksum", args=[device.pk]),
+                {"key": device.key},
+                REMOTE_ADDR=device.last_ip,
+            )
+            self.assertEqual(response.status_code, 200)
+            mocked_task.assert_called()
+        mocked_task.reset_mock()
+
+        with self.subTest(
+            f"{task_name} task not called via DeviceChecksumView when WHOIS is disabled"
+        ):
+            WHOISInfo.objects.all().delete()
+            org.config_settings.whois_enabled = False
+            org.config_settings.save(update_fields=["whois_enabled"])
+            response = self.client.get(
+                reverse("controller:device_checksum", args=[device.pk]),
+                {"key": device.key},
+                REMOTE_ADDR=device.last_ip,
+            )
+            self.assertEqual(response.status_code, 200)
+            mocked_task.assert_not_called()
+        mocked_task.reset_mock()

openwisp_controller/config/whois/utils.py

@@ -1,6 +1,3 @@
-from unittest import mock
-
-from django.urls import reverse
 from django.utils.translation import gettext as _
 from openwisp_notifications.signals import notify
 from swapper import load_model
@@ -33,6 +30,8 @@
 
 
 def send_whois_task_notification(device_pk, notify_type):
+    Device = load_model("config", "Device")
+
     device = Device.objects.get(pk=device_pk)
     notify_details = MESSAGE_MAP[notify_type]
     notify.send(