from base64 import urlsafe_b64encode from io import BytesIO from unittest.mock import patch from django.contrib.auth.tokens import PasswordResetTokenGenerator from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase from django.urls import reverse from rest_framework import status from rest_framework.test import APIClient from rest_framework_simplejwt.tokens import RefreshToken from engine.vibes_auth.models import User class DRFAuthViewsTests(TestCase): def setUp(self): super().setUp() self.client = APIClient() def test_token_obtain_pair_success(self): user = User.objects.create_user(email="user@example.com", password="Str0ngPass!word", is_active=True) url = reverse("vibes_auth:token_create") resp = self.client.post(url, {"email": user.email, "password": "Str0ngPass!word"}, format="json") self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertIn("access", data, data) self.assertTrue(data["access"], data) self.assertIn("refresh", data, data) self.assertTrue(data["refresh"], data) self.assertEqual(data["user"]["email"], user.email, data) def test_token_obtain_pair_invalid_credentials(self): User.objects.create_user(email="user@example.com", password="Str0ngPass!word", is_active=True) url = reverse("vibes_auth:token_create") resp = self.client.post(url, {"email": "user@example.com", "password": "wrong"}, format="json") self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED) def test_token_obtain_ratelimited(self): url = reverse("vibes_auth:token_create") for _ in range(0, 10): self.client.post(url, {"email": "user@example.com", "password": "wrong"}, format="json") resp = self.client.post(url, {"email": "user@example.com", "password": "wrong"}, format="json") self.assertEqual(resp.status_code, status.HTTP_429_TOO_MANY_REQUESTS) def test_token_refresh_and_verify_flow(self): user = User.objects.create_user(email="user@example.com", password="Str0ngPass!word", is_active=True) tokens = RefreshToken.for_user(user) refresh_url = reverse("vibes_auth:token_refresh") resp_refresh = self.client.post(refresh_url, {"refresh": str(tokens)}, format="json") self.assertEqual(resp_refresh.status_code, status.HTTP_200_OK) access = resp_refresh.json()["access"] verify_url = reverse("vibes_auth:token_verify") resp_verify = self.client.post(verify_url, {"token": access}, format="json") self.assertEqual(resp_verify.status_code, status.HTTP_200_OK) self.assertTrue(resp_verify.json()["token"]) self.assertEqual(resp_verify.json()["user"]["email"], user.email) def test_token_verify_invalid_token(self): verify_url = reverse("vibes_auth:token_verify") resp = self.client.post(verify_url, {"token": "malformed"}, format="json") self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("detail", resp.json(), resp.json()) def test_user_create_and_activate_flow(self): url_create = reverse("vibes_auth:users-list") payload = { "email": "new@example.com", "password": "Str0ngPass!word", "confirm_password": "Str0ngPass!word", "first_name": "New", "last_name": "User", } resp = self.client.post(url_create, payload, format="json") self.assertEqual(resp.status_code, status.HTTP_201_CREATED) user_uuid = resp.json()["uuid"] user = User.objects.get(uuid=user_uuid) self.assertFalse(user.is_active) activate_url = reverse("vibes_auth:users-activate") uidb64 = urlsafe_b64encode(str(user.uuid).encode()).decode() token_b64 = urlsafe_b64encode(str(user.activation_token).encode()).decode() resp_act = self.client.post(activate_url, {"uidb_64": uidb64, "token": token_b64}, format="json") self.assertEqual(resp_act.status_code, status.HTTP_200_OK) user.refresh_from_db() self.assertTrue(user.is_active and user.is_verified) def test_reset_password_triggers_task(self): user = User.objects.create_user(email="user@example.com", password="Str0ngPass!word", is_active=True) with patch("engine.vibes_auth.viewsets.send_reset_password_email_task.delay") as mocked_delay: url = reverse("vibes_auth:users-reset-password") resp = self.client.post(url, {"email": user.email}, format="json") self.assertEqual(resp.status_code, status.HTTP_200_OK) mocked_delay.assert_called_once() def test_confirm_password_reset_success(self): user = User.objects.create_user(email="user@example.com", password="OldPass!123", is_active=True) gen = PasswordResetTokenGenerator() token = gen.make_token(user) uidb64 = urlsafe_b64encode(str(user.uuid).encode()).decode() url = reverse("vibes_auth:users-confirm-password-reset") new_pass = "NewPass!12345" resp = self.client.post( url, { "uidb_64": uidb64, "token": token, "password": new_pass, "confirm_password": new_pass, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK, resp.json()) obtain_url = reverse("vibes_auth:token_create") r2 = self.client.post(obtain_url, {"email": user.email, "password": new_pass}, format="json") self.assertEqual(r2.status_code, status.HTTP_200_OK, resp.json()) def test_upload_avatar_permission_enforced(self): owner = User.objects.create_user(email="owner@example.com", password="Str0ngPass!word", is_active=True) stranger = User.objects.create_user(email="stranger@example.com", password="Str0ngPass!word", is_active=True) access = str(RefreshToken.for_user(stranger).access_token) # noinspection PyUnresolvedReferences self.client.credentials(HTTP_X_EVIBES_AUTH=f"Bearer {access}") url = reverse("vibes_auth:users-upload-avatar", kwargs={"pk": owner.pk}) file_content = BytesIO(b"fake image content") file = SimpleUploadedFile("avatar.png", file_content.getvalue(), content_type="image/png") resp = self.client.put(url, {"avatar": file}) self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) def test_merge_recently_viewed_permission_enforced(self): owner = User.objects.create_user(email="owner@example.com", password="Str0ngPass!word", is_active=True) stranger = User.objects.create_user(email="stranger@example.com", password="Str0ngPass!word", is_active=True) access = str(RefreshToken.for_user(stranger).access_token) # noinspection PyUnresolvedReferences self.client.credentials(HTTP_X_EVIBES_AUTH=f"Bearer {access}") url = reverse("vibes_auth:users-merge-recently-viewed", kwargs={"pk": owner.pk}) resp = self.client.put(url, {"product_uuids": []}, format="json") self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)