Merge branch 'master' into issues/256-delete-org-blocked

Author: UltraBot05

openwisp_firmware_upgrader/admin.py

@@ -444,10 +444,39 @@ def change_view(self, request, object_id, extra_context=None, **kwargs):
             args=["00000000-0000-0000-0000-000000000000"],
         )
         extra_context["django_locale"] = get_language()
+        obj = self.get_object(request, object_id)
+        # for custom breadcrumbs
+        if obj and obj.batch_id:
+            batch_opts = BatchUpgradeOperation._meta
+            batch_admin_prefix = f"admin:{batch_opts.app_label}_{batch_opts.model_name}"
+            extra_context["batch"] = obj.batch
+            extra_context["batch_changelist_url"] = reverse(
+                f"{batch_admin_prefix}_changelist"
+            )
+            extra_context["batch_change_url"] = reverse(
+                f"{batch_admin_prefix}_change",
+                args=[obj.batch_id],
+            )
+            batch_admin = self.admin_site._registry.get(BatchUpgradeOperation)
+            extra_context["batch_has_view_permission"] = (
+                batch_admin.has_view_permission(request) if batch_admin else False
+            )
         return super().change_view(
             request, object_id, extra_context=extra_context, **kwargs
         )
 
+    def get_queryset(self, request):
+        return super().get_queryset(request).select_related("batch")
+
+    def get_object(self, request, object_id, from_field=None):
+        """Avoids duplicating queries in change_view custom logic"""
+        cache_attr = f"_cached_object_{object_id}_{from_field}"
+        if not hasattr(request, cache_attr):
+            setattr(
+                request, cache_attr, super().get_object(request, object_id, from_field)
+            )
+        return getattr(request, cache_attr)
+
     def get_fields(self, request, obj=None):
         fields = super().get_fields(request, obj).copy()
         if self._should_display_batch(obj, fields):

openwisp_firmware_upgrader/base/models.py

@@ -556,7 +556,7 @@ class Meta:
         verbose_name_plural = _("Mass upgrade operations")
 
     def __str__(self):
-        return f"Upgrade of {self.build} on {self.created}"
+        return f"{self.build} ({timezone.localtime(self.created).strftime('%Y-%m-%d %H:%M:%S')})"
 
     def clean(self):
         super().clean()
@@ -824,6 +824,9 @@ class AbstractUpgradeOperation(UpgradeOptionsMixin, TimeStampedEditableModel):
         null=True,
     )
 
+    def __str__(self):
+        return f"{self.device} ({timezone.localtime(self.created).strftime('%Y-%m-%d %H:%M:%S')})"
+
     class Meta:
         abstract = True
 

openwisp_firmware_upgrader/templates/admin/firmware_upgrader/upgrade_operation_change_form.html

@@ -1,6 +1,33 @@
 {% extends "admin/change_form.html" %}
 {% load i18n admin_urls static %}
 
+{% block breadcrumbs %}
+{% if batch %}
+<div class="breadcrumbs">
+  <a href="{% url 'admin:index' %}">{% trans 'Home' %}</a>
+  &rsaquo;
+  <a href="{% url 'admin:app_list' app_label=opts.app_label %}">
+    {{ opts.app_config.verbose_name }}
+  </a>
+  &rsaquo;
+  {% if batch_has_view_permission %}
+    <a href="{{ batch_changelist_url }}">{% trans 'Mass upgrade operations' %}</a>
+  {% else %}
+    {% trans 'Mass upgrade operations' %}
+  {% endif %}
+  &rsaquo;
+  {% if batch_has_view_permission %}
+    <a href="{{ batch_change_url }}">{{ batch }}</a>
+  {% else %}
+    {{ batch }}
+  {% endif %}
+  &rsaquo; {{ original }}
+</div>
+{% else %}
+{{ block.super }}
+{% endif %}
+{% endblock %}
+
 {% block extrahead %}
 {{ block.super }}
 <link rel="stylesheet" type="text/css" href="{% static "firmware-upgrader/css/upgrade-progress.css" %}">

openwisp_firmware_upgrader/tests/test_admin.py

@@ -863,6 +863,74 @@ def test_massive_upgrade_operation_page(self, *args):
         self.assertContains(response, "Failure rate")
         self.assertContains(response, "Abortion rate")
 
+    def test_upgrade_operation_change_breadcrumb_with_batch(self, *args):
+        self.test_upgrade_all()
+        uo = UpgradeOperation.objects.first()
+        url = reverse(f"admin:{self.app_label}_upgradeoperation_change", args=[uo.pk])
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        batch_changelist_url = reverse(
+            f"admin:{self.app_label}_batchupgradeoperation_changelist"
+        )
+        batch_change_url = reverse(
+            f"admin:{self.app_label}_batchupgradeoperation_change", args=[uo.batch.pk]
+        )
+        self.assertTrue(response.context["batch_has_view_permission"])
+        self.assertEqual(response.context["batch"], uo.batch)
+        self.assertContains(response, batch_changelist_url)
+        self.assertContains(response, batch_change_url)
+        self.assertContains(response, str(uo.batch))
+        generic_upgrade_changelist_url = reverse(
+            f"admin:{self.app_label}_upgradeoperation_changelist"
+        )
+        self.assertNotContains(response, f'href="{generic_upgrade_changelist_url}"')
+
+    def test_upgrade_operation_change_breadcrumb_without_batch(self, *args):
+        self._login()
+        device_fw = self._create_device_firmware()
+        device_fw.save(upgrade=True)
+        uo = device_fw.device.upgradeoperation_set.first()
+        self.assertIsNone(uo.batch_id)
+        url = reverse(f"admin:{self.app_label}_upgradeoperation_change", args=[uo.pk])
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        self.assertIsNone(response.context.get("batch"))
+        generic_upgrade_changelist_url = reverse(
+            f"admin:{self.app_label}_upgradeoperation_changelist"
+        )
+        self.assertContains(response, f'href="{generic_upgrade_changelist_url}"')
+
+    def test_upgrade_operation_change_breadcrumb_with_batch_no_permission(self, *args):
+        self.test_upgrade_all()
+        uo = UpgradeOperation.objects.first()
+        url = reverse(f"admin:{self.app_label}_upgradeoperation_change", args=[uo.pk])
+        with mock.patch(
+            "openwisp_firmware_upgrader.admin.BatchUpgradeOperationAdmin.has_view_permission",
+            return_value=False,
+        ):
+            response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        batch_changelist_url = reverse(
+            f"admin:{self.app_label}_batchupgradeoperation_changelist"
+        )
+        batch_change_url = reverse(
+            f"admin:{self.app_label}_batchupgradeoperation_change", args=[uo.batch.pk]
+        )
+        self.assertFalse(response.context["batch_has_view_permission"])
+        self.assertEqual(response.context["batch"], uo.batch)
+        breadcrumbs = (
+            response.content.decode()
+            .split('<div class="breadcrumbs">', 1)[1]
+            .split("</div>", 1)[0]
+        )
+        self.assertNotIn(f'href="{batch_changelist_url}"', breadcrumbs)
+        self.assertNotIn(f'href="{batch_change_url}"', breadcrumbs)
+        generic_upgrade_changelist_url = reverse(
+            f"admin:{self.app_label}_upgradeoperation_changelist"
+        )
+        self.assertNotIn(f'href="{generic_upgrade_changelist_url}"', breadcrumbs)
+        self.assertIn(str(uo.batch), breadcrumbs)
+
     def test_recent_upgrades(self, *args):
         self._login()
         env = self._create_upgrade_env()

openwisp_firmware_upgrader/tests/test_models.py

@@ -8,6 +8,7 @@
 from celery.exceptions import Retry
 from django.core.exceptions import ValidationError
 from django.test import TestCase, TransactionTestCase
+from django.utils import timezone
 
 from openwisp_utils.tests import capture_any_output
 
@@ -536,6 +537,21 @@ def test_remove_file_root_directory(self, mock_storage, mock_logger):
             "Deleted firmware file: %s", "firmware.bin"
         )
 
+    def test_batch_upgrade_operation_str(self):
+        build = self._create_build()
+        batch = BatchUpgradeOperation.objects.create(build=build)
+        expected = f"{build} ({timezone.localtime(batch.created).strftime('%Y-%m-%d %H:%M:%S')})"
+        self.assertEqual(str(batch), expected)
+
+    def test_upgrade_operation_str(self):
+        with mock.patch(
+            f"{self.app_label}.models.UpgradeOperation.upgrade", return_value=None
+        ):
+            self._create_device_firmware(upgrade=True)
+        uo = UpgradeOperation.objects.first()
+        expected = f"{uo.device} ({timezone.localtime(uo.created).strftime('%Y-%m-%d %H:%M:%S')})"
+        self.assertEqual(str(uo), expected)
+
 
 class TestModelsTransaction(TestUpgraderMixin, TransactionTestCase):
     _mock_updrade = "openwisp_firmware_upgrader.upgraders.openwrt.OpenWrt.upgrade"

openwisp_firmware_upgrader/tests/test_selenium.py

@@ -114,9 +114,10 @@ def test_restoring_deleted_device(self, *args):
             By.XPATH, '//*[@id="device_form"]/div/div[1]/input[1]'
         ).click()
         try:
-            WebDriverWait(self.web_driver, 5).until(
-                EC.url_to_be(f"{self.live_server_url}/admin/config/device/")
+            device_changelist_url = self.live_server_url + reverse(
+                f"admin:{self.config_app_label}_device_changelist"
             )
+            WebDriverWait(self.web_driver, 5).until(EC.url_to_be(device_changelist_url))
         except TimeoutException:
             self.fail("Deleted device was not restored")