[tests] Added multi-tentant tests for shared objects

Author: pandafy

openwisp_controller/config/tests/test_api.py

@@ -10,7 +10,7 @@
 
 from openwisp_controller.config.api.serializers import BaseConfigSerializer
 from openwisp_controller.tests.utils import TestAdminMixin
-from openwisp_users.tests.test_api import AuthenticationMixin
+from openwisp_users.tests.test_api import AuthenticationMixin, TestMultitenantApiMixin
 from openwisp_utils.tests import capture_any_output, catch_signal
 
 from .. import settings as app_settings
@@ -33,7 +33,7 @@
 OrganizationUser = load_model("openwisp_users", "OrganizationUser")
 
 
-class ApiTestMixin:
+class ApiTestMixin(AuthenticationMixin, TestMultitenantApiMixin):
     @property
     def _template_data(self):
         return {
@@ -103,7 +103,6 @@ class TestConfigApi(
     CreateConfigTemplateMixin,
     TestVpnX509Mixin,
     CreateDeviceGroupMixin,
-    AuthenticationMixin,
     TestCase,
 ):
     def setUp(self):
@@ -671,29 +670,59 @@ def test_template_create_of_vpn_type(self):
         self.assertEqual(Template.objects.count(), 1)
         self.assertEqual(r.status_code, 201)
 
-    def test_template_create_with_shared_vpn(self):
-        org1 = self._get_org()
-        test_user = self._create_operator(organizations=[org1])
-        self.client.force_login(test_user)
-        vpn1 = self._create_vpn(name="vpn1", organization=None)
-        path = reverse("config_api:template_list")
-        data = self._template_data
-        data["type"] = "vpn"
-        data["vpn"] = vpn1.id
-        data["organization"] = org1.pk
-        r = self.client.post(path, data, content_type="application/json")
-        self.assertEqual(r.status_code, 201)
-        self.assertEqual(Template.objects.count(), 1)
-        self.assertEqual(r.data["vpn"], vpn1.id)
-
-    def test_template_creation_with_no_org_by_operator(self):
-        path = reverse("config_api:template_list")
-        data = self._template_data
+    def test_operator_access_shared_template(self):
         test_user = self._create_operator(organizations=[self._get_org()])
-        self.client.force_login(test_user)
-        r = self.client.post(path, data, content_type="application/json")
-        self.assertEqual(r.status_code, 400)
-        self.assertIn("This field may not be null.", str(r.content))
+        token = self._obtain_auth_token(test_user)
+        self._create_template(organization=None)
+        self._test_org_user_access_shared_object(
+            listview_name='config_api:template_list',
+            detailview_name='config_api:template_detail',
+            create_payload={'name': 'test', 'organization': ''},
+            update_payload={'name': 'updated-test'},
+            expected_count=1,
+            token=token,
+            expected_status_codes={
+                'create': 400,
+                'list': 200,
+                'retrieve': 403,
+                'update': 403,
+                'delete': 403,
+                'head': 403,
+                'option': 200,
+            },
+        )
+
+    def test_org_admin_create_template_with_shared_vpn(self):
+        org = self._get_org()
+        vpn = self._create_vpn(organization=None)
+        create_payload = self._template_data
+        create_payload.update(
+            {
+                'organization': org.pk,
+                'type': 'vpn',
+                'vpn': vpn.pk,
+            }
+        )
+        update_payload = create_payload.copy()
+        update_payload['name'] = 'updated-test'
+        test_user = self._create_operator(organizations=[org])
+        self._test_org_user_access_shared_object(
+            listview_name='config_api:template_list',
+            detailview_name='config_api:template_detail',
+            create_payload=create_payload,
+            update_payload=update_payload,
+            expected_count=1,
+            expected_status_codes={
+                'create': 201,
+                'list': 200,
+                'retrieve': 200,
+                'update': 200,
+                'delete': 204,
+                'head': 403,
+                'option': 200,
+            },
+            token=self._obtain_auth_token(test_user),
+        )
 
     def test_template_create_with_empty_config(self):
         path = reverse("config_api:template_list")
@@ -855,19 +884,50 @@ def test_vpn_create_api(self):
         self.assertEqual(r.status_code, 201)
         self.assertEqual(Vpn.objects.count(), 1)
 
-    def test_vpn_create_with_shared_objects(self):
-        org1 = self._get_org()
-        shared_ca = self._create_ca(name="shared_ca", organization=None)
-        test_user = self._create_administrator(organizations=[org1])
-        self.client.force_login(test_user)
+    def test_org_admin_access_vpn_with_shared_objects(self):
+        org = self._get_org()
+        shared_ca = self._create_ca(name='shared_ca', organization=None)
+        create_payload = self._vpn_data
+        create_payload.update(
+            {
+                'organization': org.pk,
+                'ca': shared_ca.pk,
+            }
+        )
+        update_payload = create_payload.copy()
+        update_payload['name'] = 'updated-test-vpn'
+        administrator = self._create_administrator(organizations=[org])
+        self._test_access_shared_object(
+            listview_name='config_api:vpn_list',
+            detailview_name='config_api:vpn_detail',
+            create_payload=create_payload,
+            update_payload=update_payload,
+            expected_count=1,
+            expected_status_codes={
+                'create': 201,
+                'list': 200,
+                'retrieve': 200,
+                'update': 200,
+                'delete': 204,
+                'head': 200,
+                'option': 200,
+            },
+            token=self._obtain_auth_token(administrator),
+        )
+
+    def test_org_admin_create_shared_vpn(self):
+        shared_ca = self._create_ca(name='shared_ca', organization=None)
         data = self._vpn_data
-        data["organization"] = org1.pk
-        data["ca"] = shared_ca.pk
-        path = reverse("config_api:vpn_list")
-        r = self.client.post(path, data, content_type="application/json")
-        self.assertEqual(Vpn.objects.count(), 1)
-        self.assertEqual(r.status_code, 201)
-        self.assertEqual(r.data["ca"], shared_ca.pk)
+        data['ca'] = shared_ca.pk
+        # API does not allow creating shared VPN by org admin,
+        # therefore we create an object to test the detail view.
+        self._create_vpn(organization=None, ca=shared_ca)
+        self._test_org_user_access_shared_object(
+            listview_name='config_api:vpn_list',
+            detailview_name='config_api:vpn_detail',
+            create_payload=data,
+            expected_count=1,
+        )
 
     def test_vpn_list_api(self):
         org = self._get_org()

openwisp_controller/connection/api/serializers.py

@@ -64,6 +64,7 @@ class Meta:
 
 class CredentialSerializer(BaseSerializer):
     params = serializers.JSONField()
+    include_shared = True
 
     class Meta:
         model = Credentials

openwisp_controller/connection/tests/test_api.py

@@ -12,7 +12,7 @@
 from swapper import load_model
 
 from openwisp_controller.tests.utils import TestAdminMixin
-from openwisp_users.tests.test_api import AuthenticationMixin
+from openwisp_users.tests.test_api import APITestCase, AuthenticationMixin
 
 from .. import settings as app_settings
 from ..api.views import ListViewPagination
@@ -387,9 +387,7 @@ def test_create_command_without_connection(self):
         )
 
 
-class TestConnectionApi(
-    TestAdminMixin, AuthenticationMixin, TestCase, CreateConnectionsMixin
-):
+class TestConnectionApi(TestAdminMixin, CreateConnectionsMixin, APITestCase):
     def setUp(self):
         super().setUp()
         self._login()

openwisp_controller/pki/api/serializers.py

@@ -73,13 +73,18 @@ class Meta:
         ]
         read_only_fields = ["created", "modified"]
         extra_kwargs = {
-            "organization": {"required": True},
-            "common_name": {"default": "", "required": False},
-            "key_length": {"initial": "2048"},
-            "digest": {"initial": "sha256"},
-            "passphrase": {"write_only": True},
-            "validity_start": {"default": default_validity_start()},
-            "validity_end": {"default": default_ca_validity_end()},
+            # In DRF 3.16+, nullable fields that are part of a unique constraint
+            # automatically get `default: None`, which can cause validation issues.
+            # Setting the default to `serializers.empty` ensures DRF does not treat
+            # these fields as both required and having a default value, avoiding
+            # conflicts.
+            'organization': {'required': True, 'default': serializers.empty},
+            'common_name': {'default': '', 'required': False},
+            'key_length': {'initial': '2048'},
+            'digest': {'initial': 'sha256'},
+            'passphrase': {'write_only': True},
+            'validity_start': {'default': default_validity_start()},
+            'validity_end': {'default': default_ca_validity_end()},
         }
 
 

openwisp_controller/pki/tests/test_api.py

@@ -1,12 +1,10 @@
-from django.test import TestCase
 from django.urls import reverse
 from packaging.version import parse as parse_version
 from rest_framework import VERSION as REST_FRAMEWORK_VERSION
 from swapper import load_model
 
 from openwisp_controller.tests.utils import TestAdminMixin
-from openwisp_users.tests.test_api import AuthenticationMixin
-from openwisp_users.tests.utils import TestOrganizationMixin
+from openwisp_users.tests.test_api import APITestCase
 from openwisp_utils.tests import AssertNumQueriesSubTestMixin, capture_any_output
 
 from .utils import TestPkiMixin
@@ -16,12 +14,7 @@
 
 
 class TestPkiApi(
-    AssertNumQueriesSubTestMixin,
-    TestAdminMixin,
-    TestPkiMixin,
-    TestOrganizationMixin,
-    AuthenticationMixin,
-    TestCase,
+    AssertNumQueriesSubTestMixin, TestAdminMixin, TestPkiMixin, APITestCase
 ):
     def setUp(self):
         super().setUp()
@@ -168,6 +161,31 @@ def test_ca_post_renew_api(self):
         self.assertNotEqual(ca1.serial_number, old_serial_num)
         self.assertNotEqual(r.data["serial_number"], old_serial_num)
 
+    def test_org_admin_access_shared_ca(self):
+        # API wouldn't allow creating the object,
+        # therefore, we create one here to test list view.
+        self._create_ca()
+
+        create_payload = self._ca_data
+        update_payload = create_payload.copy()
+        update_payload['name'] = 'updated-name'
+        self._test_org_user_access_shared_object(
+            listview_name='pki_api:ca_list',
+            detailview_name='pki_api:ca_detail',
+            create_payload=create_payload,
+            update_payload=update_payload,
+            expected_count=1,
+            expected_status_codes={
+                'create': 400,
+                'list': 200,
+                'retrieve': 403,
+                'update': 403,
+                'delete': 403,
+                'head': 403,
+                'option': 200,
+            },
+        )
+
     def test_cert_post_api(self):
         path = reverse("pki_api:cert_list")
         data = self._cert_data
@@ -307,6 +325,65 @@ def test_post_cert_revoke_api(self):
         self.assertTrue(cert1.revoked)
         self.assertTrue(r.data["revoked"])
 
+    def test_org_admin_access_shared_cert(self):
+        # API wouldn't allow creating the object,
+        # therefore, we create one here to test list view.
+        shared_ca = self._create_ca(organization=None)
+        self._create_cert(ca=shared_ca)
+
+        create_payload = self._cert_data
+        create_payload['ca'] = shared_ca.pk
+        update_payload = create_payload.copy()
+        update_payload['name'] = 'update-name'
+        self._test_org_user_access_shared_object(
+            listview_name='pki_api:cert_list',
+            detailview_name='pki_api:cert_detail',
+            create_payload=create_payload,
+            update_payload=update_payload,
+            expected_count=1,
+            expected_status_codes={
+                'create': 400,
+                'list': 200,
+                'retrieve': 403,
+                'update': 403,
+                'delete': 403,
+                'head': 403,
+                'option': 200,
+            },
+        )
+
+    def test_org_admin_access_cert_with_shared_ca(self):
+        org = self._get_org()
+        shared_ca = self._create_ca(organization=None)
+        create_payload = self._cert_data
+        create_payload.update(
+            {
+                'organization': org.pk,
+                'ca': shared_ca.pk,
+            }
+        )
+        update_payload = {
+            'name': 'updated-name',
+            'organization': org.pk,
+            'notes': 'new-notes',
+        }
+        self._test_org_user_access_shared_object(
+            listview_name='pki_api:cert_list',
+            detailview_name='pki_api:cert_detail',
+            create_payload=create_payload,
+            update_payload=update_payload,
+            expected_count=1,
+            expected_status_codes={
+                'create': 201,
+                'list': 200,
+                'retrieve': 200,
+                'update': 200,
+                'delete': 204,
+                'head': 403,
+                'option': 200,
+            },
+        )
+
     @capture_any_output()
     def test_bearer_authentication(self):
         self.client.logout()