from base64 import urlsafe_b64encode from io import BytesIO from typing import Any, cast 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 = cast( User, cast(Any, 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": cast(Any, 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"], cast(Any, user).email, data) def test_token_obtain_pair_invalid_credentials(self): cast(Any, 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 = cast( User, cast(Any, 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"], cast(Any, 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(cast(Any, user).uuid).encode()).decode() token_b64 = urlsafe_b64encode( str(cast(Any, 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(cast(Any, user).is_active and cast(Any, user).is_verified) def test_reset_password_triggers_task(self): user: User = cast( User, cast(Any, 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": cast(Any, 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 = cast( User, cast(Any, 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(cast(Any, 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": cast(Any, 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 = cast( User, cast(Any, User.objects).create_user( email="owner@example.com", password="Str0ngPass!word", is_active=True ), ) stranger: User = cast( User, cast(Any, User.objects).create_user( email="stranger@example.com", password="Str0ngPass!word", is_active=True, ), ) access = str(RefreshToken.for_user(stranger).access_token) # noinspection PyUnresolvedReferences cast(Any, self.client).credentials(HTTP_X_EVIBES_AUTH=f"Bearer {access}") url = reverse( "vibes_auth:users-upload-avatar", kwargs={"pk": cast(Any, 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 = cast( User, cast(Any, User.objects).create_user( email="owner@example.com", password="Str0ngPass!word", is_active=True ), ) stranger: User = cast( User, cast(Any, User.objects).create_user( email="stranger@example.com", password="Str0ngPass!word", is_active=True, ), ) access = str(RefreshToken.for_user(stranger).access_token) # noinspection PyUnresolvedReferences cast(Any, self.client).credentials(HTTP_X_EVIBES_AUTH=f"Bearer {access}") url = reverse( "vibes_auth:users-merge-recently-viewed", kwargs={"pk": cast(Any, owner).pk}, ) resp = self.client.put(url, {"product_uuids": []}, format="json") self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)