[feature] Estimated location: added notification and admin warning #1035

Closes #1035

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

Author: DragnEmperor

docs/user/estimated-location.rst

@@ -54,3 +54,15 @@ but only if the existing location is estimated.
 If two devices share the same IP address and are assigned to the same
 location, and the last IP of one of the devices is updated, the system
 will create a new estimated location for that device.
+
+Visibility of Estimated Status
+------------------------------
+
+The estimated status of a location is visible on the location page if the
+feature is enabled for the organization. The location admin page also
+includes indicators for the estimated status.
+
+- The name of the location will have suffix **(Estimated Location :
+  <ip_address>)**.
+- A warning on top of the page.
+- **Is Estimated** field.

openwisp_controller/config/whois/tasks.py

@@ -46,7 +46,9 @@ class WHOISCeleryRetryTask(OpenwispCeleryTask):
     def on_failure(self, exc, task_id, args, kwargs, einfo):
         """Notify the user about the failure of the WHOIS task."""
         device_pk = kwargs.get("device_pk")
-        send_whois_task_notification(device_pk=device_pk, notify_type="device_error")
+        send_whois_task_notification(
+            device_pk=device_pk, notify_type="whois_device_error"
+        )
         logger.error(f"WHOIS lookup failed. Details: {exc}")
         return super().on_failure(exc, task_id, args, kwargs, einfo)
 

openwisp_controller/config/whois/test_whois.py

@@ -614,7 +614,7 @@ def assert_logging_on_exception(
                     "Failed to fetch WHOIS details for device",
                     notification.message,
                 )
-                self.assertIn(device.last_ip, notification.description)
+                self.assertIn(device.last_ip, notification.rendered_description)
 
             mock_info.reset_mock()
             mock_warn.reset_mock()

openwisp_controller/config/whois/utils.py

@@ -1,43 +1,57 @@
-from django.utils.translation import gettext as _
+from django.utils.translation import gettext_lazy as _
 from openwisp_notifications.signals import notify
 from swapper import load_model
 
 from .. import settings as app_settings
 
 MESSAGE_MAP = {
-    "device_error": {
+    "whois_device_error": {
+        "type": "generic_message",
+        "level": "error",
         "message": _(
             "Failed to fetch WHOIS details for device"
             " [{notification.target}]({notification.target_link})"
         ),
         "description": _("WHOIS details could not be fetched for ip: {ip_address}."),
-        "level": "error",
     },
-    "location_error": {
+    "estimated_location_error": {
+        "level": "error",
+        "type": "estimated_location_info",
         "message": _(
             "Unable to create estimated location for device "
             "[{notification.target}]({notification.target_link}). "
             "Please assign/create a location manually."
         ),
         "description": _("Multiple devices found for IP: {ip_address}"),
-        "level": "error",
+    },
+    "estimated_location_created": {
+        "type": "estimated_location_info",
+        "description": _("Estimated Location {notification.verb} for IP: {ip_address}"),
+    },
+    "estimated_location_updated": {
+        "type": "estimated_location_info",
+        "message": _(
+            "Estimated location [{notification.actor}]({notification.actor_link})"
+            " for device"
+            " [{notification.target}]({notification.target_link})"
+            " updated successfully."
+        ),
+        "description": _("Estimated Location updated for IP: {ip_address}"),
     },
 }
 
 
-def send_whois_task_notification(device_pk, notify_type):
+def send_whois_task_notification(device_pk, notify_type, actor=None):
     Device = load_model("config", "Device")
 
     device = Device.objects.get(pk=device_pk)
     notify_details = MESSAGE_MAP[notify_type]
     notify.send(
-        sender=device,
-        type="generic_message",
+        sender=actor or device,
         target=device,
         action_object=device,
-        level=notify_details["level"],
-        message=notify_details["message"],
-        description=notify_details["description"].format(ip_address=device.last_ip),
+        ip_address=device.last_ip,
+        **notify_details,
     )
 
 

openwisp_controller/geo/admin.py

@@ -99,14 +99,30 @@ class LocationAdmin(MultitenantAdminMixin, AbstractLocationAdmin):
     form = LocationForm
     inlines = [FloorPlanInline]
     list_select_related = ("organization",)
-    readonly_fields = ("is_estimated",)
+    change_form_template = "admin/geo/location/change_form.html"
 
     def get_fields(self, request, obj=None):
         fields = super().get_fields(request, obj)
         if obj and not check_estimate_location_configured(obj.organization_id):
-            fields.remove("is_estimated")
+            if "is_estimated" in fields:
+                fields.remove("is_estimated")
         return fields
 
+    def get_readonly_fields(self, request, obj=None):
+        fields = super().get_readonly_fields(request, obj)
+        if obj and check_estimate_location_configured(obj.organization_id):
+            fields = fields + ("is_estimated",)
+        return fields
+
+    def change_view(self, request, object_id, form_url="", extra_context=None):
+        obj = self.get_object(request, object_id)
+        extra_context = {
+            "estimated_configured": check_estimate_location_configured(
+                obj.organization_id
+            )
+        }
+        return super().change_view(request, object_id, form_url, extra_context)
+
 
 LocationAdmin.list_display.insert(1, "organization")
 LocationAdmin.list_filter.insert(0, MultitenantOrgFilter)

openwisp_controller/geo/apps.py

@@ -8,6 +8,8 @@
 from openwisp_utils.admin_theme import register_dashboard_chart
 from openwisp_utils.admin_theme.menu import register_menu_group
 
+from .estimated_location.handlers import register_estimated_location_notification_types
+
 
 class GeoConfig(LociConfig):
     name = "openwisp_controller.geo"
@@ -21,6 +23,7 @@ def ready(self):
         super().ready()
         self.register_dashboard_charts()
         self.register_menu_groups()
+        register_estimated_location_notification_types()
         if getattr(settings, "TESTING", False):
             self._add_params_to_test_config()
 

openwisp_controller/geo/base/models.py

@@ -22,9 +22,17 @@ class BaseLocation(OrgMixin, AbstractLocation):
     class Meta(AbstractLocation.Meta):
         abstract = True
 
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._initial_is_estimated = self.is_estimated
+
     def clean(self):
-        if self.is_estimated and not check_estimate_location_configured(
-            self.organization_id
+        # Raise validation error if `is_estimated` is True but estimated feature is
+        # disabled.
+        if (
+            (self._state.adding or self._initial_is_estimated != self.is_estimated)
+            and self.is_estimated
+            and not check_estimate_location_configured(self.organization_id)
         ):
             raise ValidationError(
                 {

openwisp_controller/geo/estimated_location/handlers.py

@@ -0,0 +1,41 @@
+from django.utils.translation import gettext_lazy as _
+from openwisp_notifications.types import register_notification_type
+from swapper import load_model
+
+from openwisp_controller.config import settings as config_app_settings
+
+
+def register_estimated_location_notification_types():
+    """
+    Register the notification types used by the Estimated Location module.
+    This is necessary to ensure that the notifications are properly configured.
+    """
+    if not config_app_settings.WHOIS_CONFIGURED:
+        return
+
+    Device = load_model("config", "Device")
+    Location = load_model("geo", "Location")
+
+    register_notification_type(
+        "estimated_location_info",
+        {
+            "verbose_name": _("Estimated Location INFO"),
+            "verb": _("created"),
+            "level": "info",
+            "email_subject": _(
+                "Estimated Location: Created for device {notification.target}"
+            ),
+            "message": _(
+                "Estimated location [{notification.actor}]({notification.actor_link})"
+                " for device"
+                " [{notification.target}]({notification.target_link})"
+                " {notification.verb} successfully."
+            ),
+            "target_link": (
+                "openwisp_controller.geo.estimated_location.utils"
+                ".get_device_location_notification_target_url"
+            ),
+            "email_notification": False,
+        },
+        models=[Device, Location],
+    )

openwisp_controller/geo/estimated_location/tasks.py

@@ -44,6 +44,11 @@ def _create_estimated_location(device_location, location_defaults):
                 f"Estimated location saved successfully for {device_pk}"
                 f" for IP: {ip_address}"
             )
+            send_whois_task_notification(
+                device_pk=device_pk,
+                notify_type="estimated_location_created",
+                actor=location,
+            )
 
     def _update_or_create_estimated_location(
         device_location, whois_obj, attached_devices_exists=False
@@ -72,6 +77,11 @@ def _update_or_create_estimated_location(
                         f"Estimated location saved successfully for {device_pk}"
                         f" for IP: {ip_address}"
                     )
+                    send_whois_task_notification(
+                        device_pk=device_pk,
+                        notify_type="estimated_location_updated",
+                        actor=current_location,
+                    )
             elif not current_location:
                 # If there is no current location, we create a new one.
                 _create_estimated_location(device_location, location_defaults)
@@ -97,7 +107,7 @@ def _handle_attach_existing_location(
         # the user to resolve the conflict manually.
         if devices_with_location.count() > 1:
             send_whois_task_notification(
-                device_pk=device_pk, notify_type="location_error"
+                device_pk=device_pk, notify_type="estimated_location_error"
             )
             logger.error(
                 "Multiple devices with locations found with same "
@@ -123,6 +133,11 @@ def _handle_attach_existing_location(
             f"Estimated location saved successfully for {device_pk}"
             f" for IP: {ip_address}"
         )
+        send_whois_task_notification(
+            device_pk=device_pk,
+            notify_type="estimated_location_updated",
+            actor=existing_location,
+        )
 
     whois_obj = WHOISInfo.objects.filter(ip_address=ip_address).first()
     device = (
@@ -146,3 +161,8 @@ def _handle_attach_existing_location(
         _handle_attach_existing_location(
             device, device_location, whois_obj, attached_devices_exists
         )
+    else:
+        logger.info(
+            f"Non Estimated location already set for {device_pk}. Update"
+            f" location manually as per IP: {ip_address}"
+        )