[feature] Allowed system-wide shared category #204

Closes #204

openwisp_firmware_upgrader/admin.py

@@ -9,6 +9,7 @@
 from django.contrib import admin, messages
 from django.contrib.admin.helpers import ACTION_CHECKBOX_NAME
 from django.core.serializers.json import DjangoJSONEncoder
+from django.db.models import Q
 from django.shortcuts import redirect
 from django.template.response import TemplateResponse
 from django.templatetags.static import static
@@ -348,10 +349,10 @@ class Media:
         css = {'all': ['firmware-upgrader/css/device-firmware.css']}
 
     def _get_image_queryset(self, device):
-        # restrict firmware images to organization of the current device
         qs = (
             FirmwareImage.objects.filter(
-                build__category__organization_id=device.organization_id
+                Q(build__category__organization_id=device.organization_id)
+                | Q(build__category__organization__isnull=True)
             )
             .order_by('-created')
             .select_related('build', 'build__category')

openwisp_firmware_upgrader/api/serializers.py

@@ -106,6 +106,7 @@ def validate(self, data):
         if (
             image
             and device
+            and image.build.category.organization is not None
             and image.build.category.organization != device.organization
         ):
             raise ValidationError(

openwisp_firmware_upgrader/api/views.py

@@ -1,5 +1,6 @@
 import swapper
 from django.core.exceptions import ValidationError
+from django.db.models import Q
 from django.http import Http404
 from django_filters.rest_framework import DjangoFilterBackend
 from rest_framework import filters, generics, pagination, serializers, status
@@ -293,7 +294,8 @@ def _get_image_queryset(self, device_firmware=None, device=None):
             return
         image_qs = (
             FirmwareImage.objects.filter(
-                build__category__organization_id=device.organization_id
+                Q(build__category__organization_id=device.organization_id)
+                | Q(build__category__organization__isnull=True)
             )
             .order_by('-created')
             .select_related('build', 'build__category')

openwisp_firmware_upgrader/base/models.py

@@ -12,7 +12,7 @@
 from private_storage.fields import PrivateFileField
 
 from openwisp_controller.connection.exceptions import NoWorkingDeviceConnectionError
-from openwisp_users.mixins import OrgMixin
+from openwisp_users.mixins import ShareableOrgMixin
 from openwisp_utils.base import TimeStampedEditableModel
 
 from .. import settings as app_settings
@@ -69,7 +69,7 @@ def clean(self):
         self.validate_upgrade_options()
 
 
-class AbstractCategory(OrgMixin, TimeStampedEditableModel):
+class AbstractCategory(ShareableOrgMixin, TimeStampedEditableModel):
     name = models.CharField(max_length=64, db_index=True)
     description = models.TextField(blank=True)
 
@@ -295,7 +295,10 @@ class Meta:
     def clean(self):
         if not hasattr(self, 'image') or not hasattr(self, 'device'):
             return
-        if self.image.build.category.organization != self.device.organization:
+        if (
+            self.image.build.category.organization is not None
+            and self.image.build.category.organization != self.device.organization
+        ):
             raise ValidationError(
                 {
                     'image': _(

openwisp_firmware_upgrader/migrations/0011_alter_category_organization.py

@@ -0,0 +1,25 @@
+# Generated by Django 4.2.16 on 2025-02-20 00:43
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("openwisp_users", "0020_populate_password_updated_field"),
+        ("firmware_upgrader", "0010_alter_firmwareimage_file"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="category",
+            name="organization",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                to="openwisp_users.organization",
+                verbose_name="organization",
+            ),
+        ),
+    ]

openwisp_firmware_upgrader/tests/base.py

@@ -71,11 +71,13 @@ def _create_build(self, **kwargs):
     def _create_firmware_image(self, **kwargs):
         opts = dict(type=self.TPLINK_4300_IMAGE)
         opts.update(kwargs)
-        build_opts = {}
+        category_opts = {}
         if 'organization' in opts:
-            build_opts['organization'] = opts.pop('organization')
+            category_opts['organization'] = opts.pop('organization')
         if 'build' not in opts:
-            opts['build'] = self._get_build(**build_opts)
+            opts['build'] = self._get_build(
+                category=self._create_category(**category_opts)
+            )
         if 'file' not in opts:
             opts['file'] = self._get_simpleuploadedfile()
         fw = FirmwareImage(**opts)

openwisp_firmware_upgrader/tests/test_admin.py

@@ -278,6 +278,24 @@ def test_image_queryset_no_model_nor_device_firmware(self):
         self.assertIn(yuncore, form.fields['image'].queryset)
         self.assertNotIn(img_org2, form.fields['image'].queryset)
 
+    def test_image_queryset_shared_firmware(self):
+        (
+            device,
+            device_fw,
+            _,
+            _,
+            _,
+        ) = self._prepare_image_qs_test_env()
+        shared_image = self._create_firmware_image(
+            build=self._create_build(
+                category=self._create_category(organization=None, name='Shared')
+            )
+        )
+        form = DeviceFirmwareForm(device=device)
+        self.assertEqual(form.fields['image'].queryset.count(), 3)
+        self.assertIn(device_fw.image, form.fields['image'].queryset)
+        self.assertIn(shared_image, form.fields['image'].queryset)
+
     def test_admin_menu_groups(self):
         # Test menu group (openwisp-utils menu group) for Build, Category,
         # BatchUpgradeOperation models

openwisp_firmware_upgrader/tests/test_api.py

@@ -332,6 +332,62 @@ def test_category_unauthorized(self):
             r = client.get(url)
         self.assertEqual(r.status_code, 401)
 
+    def test_shared_category(self):
+        self._create_admin(username='admin', password='tester')
+        list_view_path = reverse('upgrader:api_category_list')
+
+        with self.subTest('Test superuser can create shared category'):
+            self._login(username='admin', password='tester')
+            response = self.client.post(
+                list_view_path, {'name': 'Shared', 'organization': ''}
+            )
+            self.assertEqual(response.status_code, 201)
+            self.assertEqual(Category.objects.count(), 1)
+            category = Category.objects.first()
+            self.assertEqual(category.organization, None)
+
+        category_details_path = reverse(
+            'upgrader:api_category_detail', args=[category.pk]
+        )
+
+        with self.subTest('Test superuser can view shared category'):
+            response = self.client.get(list_view_path)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(
+                response.data['results'], [self._serialize_category(category)]
+            )
+
+        with self.subTest('Test superuser can update shared object'):
+            response = self.client.patch(
+                category_details_path,
+                {'description': 'Shared category'},
+                content_type='application/json',
+            )
+            self.assertEqual(response.status_code, 200)
+            category.refresh_from_db()
+            self.assertEqual(category.description, 'Shared category')
+
+        self._logout()
+        self._login()
+        with self.subTest('Test org admin cannot create shared category'):
+            response = self.client.post(
+                list_view_path, {'name': 'Shared 2', 'organization': ''}
+            )
+            self.assertEqual(response.status_code, 400)
+
+        with self.subTest('Test org admin can view shared categories'):
+            response = self.client.get(list_view_path)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(
+                response.data['results'], [self._serialize_category(category)]
+            )
+
+        with self.subTest('Test org admin cannot update shared category'):
+            response = self.client.patch(
+                category_details_path, {'description': 'Changed by org admin'}
+            )
+            self.assertEqual(response.status_code, 403)
+
     def test_category_list(self):
         self._create_category()
         self._create_category(name='New category')
@@ -1046,6 +1102,28 @@ def test_device_firmware_detail_create(self):
         self.assertNotIn(f'{image2b}</option>', repsonse)
         self.assertNotIn(f'{image2}</option>', repsonse)
 
+    def test_device_firmware_detail_created_shared_image(self):
+        env = self._create_upgrade_env(device_firmware=False)
+        shared_image = self._create_firmware_image(
+            type=self.TPLINK_4300_IMAGE, organization=None
+        )
+        path = reverse('upgrader:api_devicefirmware_detail', args=[env['d1'].pk])
+        self.assertEqual(DeviceFirmware.objects.count(), 0)
+        self.assertEqual(UpgradeOperation.objects.count(), 0)
+
+        with self.assertNumQueries(25):
+            data = {'image': shared_image.pk}
+            r = self.client.put(
+                f'{path}?format=api', data, content_type='application/json'
+            )
+        self.assertEqual(r.status_code, 201)
+        self.assertEqual(DeviceFirmware.objects.count(), 1)
+        self.assertEqual(UpgradeOperation.objects.count(), 1)
+        device_fw = DeviceFirmware.objects.first()
+        self.assertEqual(device_fw.image, shared_image)
+        repsonse = r.content.decode()
+        self.assertIn(f'{shared_image}</option>', repsonse)
+
     def test_device_firmware_detail_update(self):
         env = self._create_upgrade_env()
         image1a = env['image1a']

openwisp_firmware_upgrader/tests/test_models.py

@@ -91,19 +91,28 @@ def test_fw_auto_type(self):
         fw = self._create_firmware_image(type='')
         self.assertEqual(fw.type, self.TPLINK_4300_IMAGE)
 
-    def test_device_firmware_image_invalid_org(self):
+    def test_device_firmware_multitenancy(self):
         device_fw = self._create_device_firmware()
         org2 = self._create_org(name='org2')
-        build2 = self._create_build(organization=org2)
-        img2 = self._create_firmware_image(build=build2)
+        shared_image = self._create_firmware_image(organization=None)
+        org2_image = self._create_firmware_image(organization=org2)
 
-        device_fw.image = img2
-        try:
-            device_fw.full_clean()
-        except ValidationError as e:
-            self.assertIn('image', e.message_dict)
-        else:
-            self.fail('ValidationError not raised')
+        with self.subTest('Test with firmware from another organization'):
+            device_fw.image = org2_image
+            with self.assertRaises(ValidationError) as context:
+                device_fw.full_clean()
+            self.assertEqual(
+                context.exception.message_dict['image'][0],
+                "The organization of the image doesn't match the organization of the device",
+            )
+
+        with self.subTest('Test with shared firmware'):
+            device_fw.image = shared_image
+            try:
+                device_fw.full_clean()
+                device_fw.save()
+            except Exception as error:
+                self.fail('Test failed with error: {}'.format(error))
 
     def test_device_fw_image_changed(self, *args):
         with mock.patch(