Egor fureunoir Gorbunov
943aa02cd3
Fixes: 1) Correct entrypoint scripts by removing redundant `python` reference in `uv run` commands; 2) Resolve incorrect imports and adjust class renaming in vibes_auth tests; 3) Address typing errors and minor omissions in existing code. Extra: 1) Improve formatting in settings and middleware files; 2) Update messaging test class names for clarity; 3) Cleanup unused imports and extra whitespaces, ensuring cleaner codebase.
144 lines
7 KiB
Python
144 lines
7 KiB
Python
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)
|
|
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)
|
|
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)
|