[tests] Added tests

Author: pandafy

openwisp_radius/api/views.py

@@ -316,7 +316,7 @@ def post(self, request, *args, **kwargs):
         self.update_user_details(user)
         context = {"view": self, "request": request}
         serializer = self.serializer_class(instance=token, context=context)
-        response = RadiusUserSerializer(user).data
+        response = RadiusUserSerializer(user, context=context).data
         response.update(serializer.data)
         status_code = 200 if user.is_active else 401
         # If identity verification is required, check if user is verified
@@ -336,24 +336,24 @@ def validate_membership(self, user):
             if get_organization_radius_settings(
                 self.organization, "registration_enabled"
             ):
-                if self._needs_identity_verification(
-                    org=self.organization
-                ) and not self.is_identity_verified_strong(user, self.organization):
-                    raise PermissionDenied
                 try:
                     org_user = OrganizationUser(
                         user=user, organization=self.organization
                     )
                     org_user.full_clean()
                     org_user.save()
+                    RegisteredUser.objects.get_or_create(
+                        user=user,
+                        organization=self.organization,
+                        defaults={"method": ""},
+                    )
                 except ValidationError as error:
                     raise serializers.ValidationError(
                         {"non_field_errors": error.message_dict.pop("__all__")}
                     )
             else:
                 message = _(
-                    "{organization} does not allow self registration "
-                    "of new accounts."
+                    "{organization} does not allow self registration of new accounts."
                 ).format(organization=self.organization.name)
                 raise PermissionDenied(message)
 
@@ -411,8 +411,8 @@ def post(self, request, *args, **kwargs):
                     user.phone_number = (
                         phone_token.phone_number if phone_token else user.phone_number
                     )
-                response = RadiusUserSerializer(user).data
                 context = {"view": self, "request": request}
+                response = RadiusUserSerializer(user, context=context).data
                 token_data = rest_auth_settings.api_settings.TOKEN_SERIALIZER(
                     token, context=context
                 ).data
@@ -621,11 +621,13 @@ class CreatePhoneTokenView(
     )
 
     @swagger_auto_schema(
-        operation_description=("""
+        operation_description=(
+            """
             **Requires the user auth token (Bearer Token).**
             Used for SMS verification, sends a code via SMS to the
             phone number of the user.
-            """),
+            """
+        ),
         request_body=no_body,
         responses={201: ""},
     )
@@ -699,12 +701,14 @@ class GetPhoneTokenStatusView(DispatchOrgMixin, GenericAPIView):
     serializer_class = serializers.Serializer
 
     @swagger_auto_schema(
-        operation_description=("""
+        operation_description=(
+            """
             **Requires the user auth token (Bearer Token).**
             Used for SMS verification, allows checking whether an active
             SMS token was already requested for the mobile phone number
             of the logged in account.
-            """),
+            """
+        ),
         responses={200: '`{"active":"true/false"}`'},
     )
     def get(self, request, *args, **kwargs):
@@ -772,6 +776,7 @@ def post(self, request, *args, **kwargs):
                 },
             )
             reg_user.is_verified = True
+            reg_user.method = "mobile_phone"
             # Update username if phone_number is used as username
             if user.username == user.phone_number:
                 user.username = phone_token.phone_number
@@ -797,11 +802,13 @@ class ChangePhoneNumberView(ThrottledAPIMixin, CreatePhoneTokenView):
     serializer_class = ChangePhoneNumberSerializer
 
     @swagger_auto_schema(
-        operation_description=("""
+        operation_description=(
+            """
             **Requires the user auth token (Bearer Token).**
             Allows users to change their phone number, will flag the
             user as inactive and send them a verification code via SMS.
-            """),
+            """
+        ),
         responses={200: ""},
     )
     def post(self, request, *args, **kwargs):

openwisp_radius/base/models.py

@@ -1058,12 +1058,13 @@ def save_user(self, user):
         OrganizationUser = swapper.load_model("openwisp_users", "OrganizationUser")
         RegisteredUser = swapper.load_model("openwisp_radius", "RegisteredUser")
         user.save()
+        radius_settings = self.organization.radius_settings
         registered_user, created = RegisteredUser.get_or_create_for_user_and_org(
             user=user,
             organization=self.organization,
             defaults={
                 "method": "manual",
-                "is_verified": self.organization.radius_settings.needs_identity_verification,
+                "is_verified": radius_settings.needs_identity_verification,
             },
         )
         if (

openwisp_radius/tests/test_admin.py

@@ -1407,6 +1407,20 @@ def test_inline_registered_user(self):
             register_registration_method("github", "GitHub", strong_identity=False)
             self.assertIn("github", RegisteredUser._weak_verification_methods)
 
+    def test_user_admin_shows_multiple_registered_user_records(self):
+        user = self._create_user(username="multiuser", email="multi@test.org")
+        org2 = self._create_org(name="org2", slug="org2")
+        RegisteredUser.objects.create(
+            user=user, organization=self.default_org, is_verified=True
+        )
+        RegisteredUser.objects.create(user=user, organization=org2, is_verified=False)
+        RegisteredUser.objects.create(user=user, organization=None, is_verified=True)
+        user_url = reverse(f"admin:{User._meta.app_label}_user_change", args=[user.pk])
+        response = self.client.get(user_url)
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "id_registered_users-TOTAL_FORMS")
+        self.assertIn('value="3"', response.rendered_content)
+
     def test_get_is_verified_user_admin_list(self):
         unknown = User.objects.first()
         self.assertIsNotNone(unknown)

openwisp_radius/tests/test_api/test_api.py

@@ -41,6 +41,7 @@
 RadiusBatch = load_model("RadiusBatch")
 RadiusUserGroup = load_model("RadiusUserGroup")
 RadiusGroup = load_model("RadiusGroup")
+RegisteredUser = load_model("RegisteredUser")
 OrganizationRadiusSettings = load_model("OrganizationRadiusSettings")
 Organization = swapper.load_model("openwisp_users", "Organization")
 OrganizationUser = swapper.load_model("openwisp_users", "OrganizationUser")
@@ -60,7 +61,7 @@ def _radius_batch_post_request(self, data, username="admin", password="tester"):
         login_payload = {"username": username, "password": password}
         login_url = reverse("radius:user_auth_token", args=[self.default_org.slug])
         login_response = self.client.post(login_url, data=login_payload)
-        header = f'Bearer {login_response.json()["key"]}'
+        header = f"Bearer {login_response.json()['key']}"
         url = reverse("radius:batch")
         return self.client.post(url, data, HTTP_AUTHORIZATION=header)
 
@@ -381,6 +382,48 @@ def test_radius_user_serializer(self):
                 },
             )
 
+        with self.subTest("org-specific takes precedence over global"):
+            # Create user with both a global (unverified) and
+            # org-specific (verified) record
+            user2 = self._create_user(username="user2", email="user2@test.com")
+            self._create_org_user(user=user2, organization=self.default_org)
+            RegisteredUser.objects.create(
+                user=user2, organization=None, is_verified=False
+            )
+            RegisteredUser.objects.create(
+                user=user2,
+                organization=self.default_org,
+                is_verified=True,
+                method="mobile_phone",
+            )
+            url = reverse("radius:user_auth_token", args=[self.default_org.slug])
+            r = self.client.post(url, {"username": "user2", "password": "tester"})
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.data["is_verified"], True)
+            self.assertEqual(r.data["method"], "mobile_phone")
+
+        with self.subTest("global record as fallback when no org-specific"):
+            # Create user with only a global (verified) record
+            user3 = self._create_user(username="user3", email="user3@test.com")
+            self._create_org_user(user=user3, organization=self.default_org)
+            RegisteredUser.objects.create(
+                user=user3, organization=None, is_verified=True, method="email"
+            )
+            url = reverse("radius:user_auth_token", args=[self.default_org.slug])
+            r = self.client.post(url, {"username": "user3", "password": "tester"})
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.data["is_verified"], True)
+            self.assertEqual(r.data["method"], "email")
+
+        with self.subTest("returns None when no RegisteredUser records exist"):
+            user4 = self._create_user(username="user4", email="user4@test.com")
+            self._create_org_user(user=user4, organization=self.default_org)
+            url = reverse("radius:user_auth_token", args=[self.default_org.slug])
+            r = self.client.post(url, {"username": "user4", "password": "tester"})
+            self.assertEqual(r.status_code, 200)
+            self.assertIsNone(r.data["is_verified"])
+            self.assertIsNone(r.data["method"])
+
     # The fallback value is set on project startup, hence it also requires mocking.
     @mock.patch.object(
         OrganizationRadiusSettings._meta.get_field("first_name"),

openwisp_radius/tests/test_api/test_freeradius_api.py

@@ -206,6 +206,88 @@ def test_authorize_unverified_user(self):
         self.assertEqual(response.status_code, 200)
         self.assertIsNone(response.data)
 
+    def test_authorize_verified_user(self):
+        org_user = self._get_org_user()
+        user = org_user.user
+        org_settings = OrganizationRadiusSettings.objects.get(
+            organization=self._get_org()
+        )
+        org_settings.needs_identity_verification = True
+        org_settings.save()
+
+        with self.subTest("org-specific verified record passes authorization"):
+            RegisteredUser.objects.create(
+                user=user, organization=self._get_org(), is_verified=True
+            )
+            response = self._authorize_user(auth_header=self.auth_header)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.data, {"control:Auth-Type": "Accept"})
+
+        with self.subTest("global verified record passes authorization (fallback)"):
+            RegisteredUser.objects.filter(user=user).delete()
+            RegisteredUser.objects.create(
+                user=user, organization=None, is_verified=True
+            )
+            response = self._authorize_user(auth_header=self.auth_header)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.data, {"control:Auth-Type": "Accept"})
+
+    def test_multi_org_user_different_verification_states(self):
+        org1 = self._get_org()
+        org_settings = OrganizationRadiusSettings.objects.get(organization=org1)
+        org_settings.needs_identity_verification = True
+        org_settings.save()
+        org2 = self._create_org(name="org2", slug="org2")
+        org2_settings = OrganizationRadiusSettings.objects.get_or_create(
+            organization=org2
+        )[0]
+        org2_settings.needs_identity_verification = True
+        org2_settings.full_clean()
+        org2_settings.save()
+        user = self._get_user_with_org()
+        self._create_org_user(organization=org2, user=user)
+        RegisteredUser.objects.create(user=user, organization=org1, is_verified=True)
+        auth_header_org1 = f"Bearer {org1.pk} {org1.radius_settings.token}"
+        response = self._authorize_user(
+            username=user.username, auth_header=auth_header_org1
+        )
+        self.assertEqual(response.data["control:Auth-Type"], "Accept")
+
+        auth_header_org2 = f"Bearer {org2.pk} {org2.radius_settings.token}"
+        response = self._authorize_user(
+            username=user.username, auth_header=auth_header_org2
+        )
+        self.assertIsNone(response.data)
+
+    def test_global_fallback_for_orgs_without_specific_records(self):
+        org1 = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        org2_settings = OrganizationRadiusSettings.objects.get_or_create(
+            organization=org2
+        )[0]
+        org2_settings.needs_identity_verification = True
+        org2_settings.full_clean()
+        org2_settings.save()
+        user = self._get_user_with_org()
+        self._create_org_user(organization=org2, user=user)
+        RegisteredUser.objects.create(user=user, organization=None, is_verified=True)
+        org_settings = OrganizationRadiusSettings.objects.get(organization=org1)
+        org_settings.needs_identity_verification = True
+        org_settings.save()
+        user.registered_users.exclude(organization=None).delete()
+
+        auth_header_org1 = f"Bearer {org1.pk} {org1.radius_settings.token}"
+        response = self._authorize_user(
+            username=user.username, auth_header=auth_header_org1
+        )
+        self.assertEqual(response.data["control:Auth-Type"], "Accept")
+
+        auth_header_org2 = f"Bearer {org2.pk} {org2.radius_settings.token}"
+        response = self._authorize_user(
+            username=user.username, auth_header=auth_header_org2
+        )
+        self.assertEqual(response.data["control:Auth-Type"], "Accept")
+
     def test_authorize_radius_token_unverified_user(self):
         user = self._get_org_user()
         org_settings = OrganizationRadiusSettings.objects.get(
@@ -258,7 +340,7 @@ def test_postauth_radius_token_accept_201(self):
     def test_postauth_accept_201_querystring(self):
         self.assertEqual(RadiusPostAuth.objects.all().count(), 0)
         params = self._get_postauth_params()
-        post_url = f'{reverse("radius:postauth")}{self.token_querystring}'
+        post_url = f"{reverse('radius:postauth')}{self.token_querystring}"
         response = self.client.post(post_url, params)
         params["password"] = ""
         self.assertEqual(RadiusPostAuth.objects.filter(**params).count(), 1)
@@ -2442,7 +2524,7 @@ def test_cache(self):
         )
         self._get_org_user()
         token_querystring = f"?token={rad.token}&uuid={str(self.org.pk)}"
-        post_url = f'{reverse("radius:authorize")}{token_querystring}'
+        post_url = f"{reverse('radius:authorize')}{token_querystring}"
         # Clear cache before sending request
         cache.clear()
         self.client.post(post_url, {"username": "tester", "password": "tester"})