[fix] Prevent HTTP 500 on missing device credentials

openwisp_firmware_upgrader/admin.py

@@ -668,7 +668,21 @@ def __init__(self, device, *args, **kwargs):
             device, device_firmware=self.instance
         )
 
+    def _has_credentials_in_form(self):
+        if not self.data:
+            return False
+        total = int(self.data.get("deviceconnection_set-TOTAL_FORMS", 0))
+        for i in range(total):
+            prefix = f"deviceconnection_set-{i}"
+            has_cred = self.data.get(f"{prefix}-credentials")
+            is_deleted = self.data.get(f"{prefix}-DELETE")
+            if has_cred and not is_deleted:
+                return True
+        return False
+
     def full_clean(self):
+        if self._has_credentials_in_form():
+            self.instance._skip_connection_check = True
         super().full_clean()
         if not self.is_bound:
             return

openwisp_firmware_upgrader/base/models.py

@@ -61,12 +61,18 @@ class Meta:
     def validate_upgrade_options(self):
         if not self.upgrade_options:
             return
-        if not getattr(self.upgrader_class, "SCHEMA"):
+        try:
+            upgrader_class = self.upgrader_class
+        except ObjectDoesNotExist:
+            raise ValidationError(
+                _("No related connection or credentials found for this device.")
+            )
+        if not getattr(upgrader_class, "SCHEMA", None):
             raise ValidationError(
                 _("Using upgrade options is not allowed with this upgrader.")
             )
         try:
-            self.upgrader_class.validate_upgrade_options(self.upgrade_options)
+            upgrader_class.validate_upgrade_options(self.upgrade_options)
         except jsonschema.ValidationError:
             raise ValidationError("The upgrade options are invalid")
         except FirmwareUpgradeOptionsException as error:
@@ -403,7 +409,11 @@ def clean(self):
                     )
                 }
             )
-        if self.device.deviceconnection_set.count() < 1:
+        if (
+            self.image_has_changed
+            and not getattr(self, "_skip_connection_check", False)
+            and self.device.deviceconnection_set.count() < 1
+        ):
             raise ValidationError(
                 _(
                     "This device does not have a related connection object defined "
@@ -1006,6 +1016,13 @@ def upgrade(self, recoverable=True):
             self.device.devicefirmware.installed = True
             self.device.devicefirmware.save(upgrade=False)
 
+    def validate_upgrade_options(self):
+        try:
+            super().validate_upgrade_options()
+        except ValidationError:
+            if self._state.adding:
+                raise
+
     def save(self, *args, **kwargs):
         super().save(*args, **kwargs)
         # when an operation is completed

openwisp_firmware_upgrader/tests/test_admin.py

@@ -394,6 +394,70 @@ def test_device_firmware_upgrade_without_device_connection(
         mocked_func.assert_not_called()
         self.assertEqual(response.status_code, 200)
 
+    def test_save_device_after_credentials_deleted(self, *args):
+        """Regression test for #250."""
+        self._login()
+        device_fw = self._create_device_firmware()
+        device = device_fw.device
+        device_conn = device.deviceconnection_set.first()
+        device_params = self._get_device_params(
+            device, device_conn, device_fw.image, device_fw
+        )
+        device.deviceconnection_set.all().delete()
+        device_params["deviceconnection_set-TOTAL_FORMS"] = 0
+        device_params["deviceconnection_set-INITIAL_FORMS"] = 0
+        del device_params["deviceconnection_set-0-credentials"]
+        del device_params["deviceconnection_set-0-id"]
+        del device_params["deviceconnection_set-0-update_strategy"]
+        del device_params["deviceconnection_set-0-enabled"]
+        response = self.client.post(
+            reverse(f"admin:{self.config_app_label}_device_change", args=[device.id]),
+            data=device_params,
+            follow=True,
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertNotContains(response, "Please correct the error")
+
+    def test_change_image_and_add_credentials_together(self, *args):
+        """Regression test for #250."""
+        self._login()
+        device_fw = self._create_device_firmware()
+        device = device_fw.device
+        device_conn = device.deviceconnection_set.first()
+        credentials_id = device_conn.credentials_id
+        update_strategy = device_conn.update_strategy
+        # create a new image to switch to
+        build = self._create_build(version="0.2")
+        new_image = self._create_firmware_image(build=build)
+        # delete existing connection
+        device.deviceconnection_set.all().delete()
+        # build form data: new image + new credentials in same submit
+        device_params = self._get_device_params(
+            device, device_conn, new_image, device_fw
+        )
+        device_params.update(
+            {
+                "devicefirmware-0-image": str(new_image.id),
+                "deviceconnection_set-TOTAL_FORMS": 1,
+                "deviceconnection_set-INITIAL_FORMS": 0,
+                "deviceconnection_set-0-credentials": str(credentials_id),
+                "deviceconnection_set-0-update_strategy": update_strategy,
+                "deviceconnection_set-0-enabled": True,
+            }
+        )
+        if "deviceconnection_set-0-id" in device_params:
+            del device_params["deviceconnection_set-0-id"]
+        response = self.client.post(
+            reverse(f"admin:{self.config_app_label}_device_change", args=[device.id]),
+            data=device_params,
+            follow=True,
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertNotContains(response, "Please correct the error")
+        device_fw.refresh_from_db()
+        self.assertEqual(device_fw.image, new_image)
+        self.assertEqual(device.deviceconnection_set.count(), 1)
+
     def test_deactivated_firmware_image_inline(self):
         self._login()
         device = self._create_config(organization=self._get_org()).device

openwisp_firmware_upgrader/tests/test_api.py

@@ -1309,7 +1309,7 @@ def test_device_firmware_detail_400(self):
 
         with self.subTest("Test device and image model validation"):
             url = reverse("upgrader:api_devicefirmware_detail", args=[device1.pk])
-            with self.assertNumQueries(18):
+            with self.assertNumQueries(17):
                 # Try to make a request when the
                 # device model does not match the image model
                 data = {"image": image1a.pk}
@@ -1490,7 +1490,7 @@ def test_device_firmware_detail_update(self):
         self.assertEqual(DeviceFirmware.objects.count(), 2)
         self.assertEqual(UpgradeOperation.objects.count(), 0)
 
-        with self.assertNumQueries(27):
+        with self.assertNumQueries(26):
             data = {"image": image2a.pk}
             r = self.client.put(
                 f"{url}?format=api", data, content_type="application/json"
@@ -1529,7 +1529,7 @@ def test_device_firmware_detail_partial_update(self):
         self.assertEqual(DeviceFirmware.objects.count(), 2)
         self.assertEqual(UpgradeOperation.objects.count(), 0)
 
-        with self.assertNumQueries(27):
+        with self.assertNumQueries(26):
             data = {"image": image2a.pk}
             r = self.client.patch(
                 f"{url}?format=api", data, content_type="application/json"

openwisp_firmware_upgrader/tests/test_models.py

@@ -177,6 +177,15 @@ def test_device_fw_no_connection(self):
         else:
             self.fail("ValidationError not raised")
 
+    def test_device_fw_save_after_credentials_removed(self):
+        """Regression test for #250."""
+        device_fw = self._create_device_firmware()
+        device_fw.device.deviceconnection_set.all().delete()
+        device_fw.full_clean()
+        uo_count = UpgradeOperation.objects.count()
+        device_fw.save(upgrade=False)
+        self.assertEqual(UpgradeOperation.objects.count(), uo_count)
+
     def test_invalid_board(self):
         image = FIRMWARE_IMAGE_MAP[self.TPLINK_4300_IMAGE]
         boards = image["boards"]
@@ -243,6 +252,20 @@ def test_upgrade_operation_invalid_upgrade_options(self):
                 ['The "-n" and "-o" options cannot be used together'],
             )
 
+    def test_upgrade_operation_credentials_removed(self):
+        """Regression test for #250."""
+        device_fw = self._create_device_firmware()
+        device = device_fw.device
+        uo = UpgradeOperation(
+            device=device,
+            image=device_fw.image,
+            upgrade_options={"n": True},
+        )
+        device.deviceconnection_set.all().delete()
+        with self.assertRaises(ValidationError) as ctx:
+            uo.full_clean()
+        self.assertIn("connection", str(ctx.exception).lower())
+
     def test_upgrade_operation_log_line(self):
         device_fw = self._create_device_firmware()
         uo = UpgradeOperation(device=device_fw.device, image=device_fw.image)