feat(backend): password recovery #33

.env.example

@@ -1,17 +1,18 @@
-SECRET_KEY=23984794924798@39485973@h76h@dfhighudfgh
+SECRET_KEY=your_secret_key
 DJANGO_EMAIL_BACKEND=django.core.mail.backends.smtp.EmailBackend
 ALLOWED_HOSTS=127.0.0.1,localhost,yourdomain.com
 
 EMAIL_HOST=smtp.example.com
 EMAIL_PORT=587
 EMAIL_USE_TLS=True
-EMAIL_HOST_USER=user@example.com
-EMAIL_HOST_PASSWORD=secret123
+EMAIL_HOST_USER=your_email_here
+EMAIL_HOST_PASSWORD=your_password_here
 EMAIL_TIMEOUT=10
+DEFAULT_FROM_EMAIL=your_email_for_sending
 
-DATABASE_NAME="postgres"
-DATABASE_USER="postgres"
-DATABASE_PASSWORD="password"
-DATABASE_PORT="5432"
-DATABASE_HOST="db"
-DATABASE_ENGINE="postgresql"
+DATABASE_NAME=your_database_name
+DATABASE_USER=your_database_user_name
+DATABASE_PASSWORD=your_database_user_password
+DATABASE_PORT=database_port
+DATABASE_HOST=database_host
+DATABASE_ENGINE=postgresql

.gitignore

@@ -11,6 +11,7 @@ coverage.xml
 .python-version
 .ruff_cache
 .pytest_cache
+*.log
 db.sqlite3
 PATH
 itlabormarket_project_plan.md
@@ -23,3 +24,5 @@ error.html
 **/node_modules/
 staticfiles/
 static/
+*.session
+.mypy_cache

app/__init__.py

@@ -0,0 +1,3 @@
+from .celery import app as celery_app
+
+__all__ = ("celery_app",)

app/celery.py

@@ -0,0 +1,8 @@
+import os
+
+from celery import Celery
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "app.settings")
+app = Celery("app")
+app.config_from_object("django.conf:settings", namespace="CELERY")
+app.autodiscover_tasks()

app/services/auth/password_reset/admin.py

@@ -0,0 +1,3 @@
+# from django.contrib import admin
+
+# Register your models here.

app/services/auth/password_reset/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class PasswordResetConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "app.services.auth.password_reset"

app/services/auth/password_reset/configs.py

@@ -0,0 +1,4 @@
+# Token expiration time in seconds
+PASSWORD_RESET_TIMEOUT = 3600
+# Maximum number of retries
+MAX_RETRIES = 3

app/services/auth/password_reset/migrations/0001_initial.py

@@ -0,0 +1,27 @@
+# Generated by Django 5.2.7 on 2025-12-24 16:23
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='PasswordResetToken',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('token_hash', models.CharField()),
+                ('expires_at', models.DateTimeField()),
+                ('is_used', models.BooleanField(default=False)),
+                ('user_id', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+        ),
+    ]

app/services/auth/password_reset/models.py

@@ -0,0 +1,19 @@
+from django.db import models
+
+from app.services.auth.users.models import User
+
+
+class PasswordResetToken(models.Model):
+    user_id = models.ForeignKey(User, on_delete=models.CASCADE)
+    token_hash = models.CharField(null=True)
+    expires_at = models.DateTimeField()
+    is_used = models.BooleanField(default=False)
+
+    def mark_as_used(self):
+        self.is_used = True
+        self.save()
+
+    @classmethod
+    def mark_all_as_used(cls, user: User):
+        queryset = cls.objects.filter(is_used=False, user_id=user)
+        return queryset.update(is_used=True)

app/services/auth/password_reset/tasks.py

@@ -0,0 +1,59 @@
+import logging
+from typing import Sequence
+
+from django.core.mail import send_mail
+
+from app.celery import app
+from app.services.auth.password_reset import configs
+
+logger = logging.getLogger(__name__)
+
+
+@app.task(
+    bind=True,
+    max_retries=configs.MAX_RETRIES,
+    default_retry_delay=60,
+    autoretry_for=(Exception,),
+    retry_backoff=True,
+    retry_jitter=True,
+)
+def send_mail_task(
+    self,
+    subject: str,
+    message: str,
+    html_message: str,
+    from_email: str,
+    recipient_list: Sequence[str],
+    fail_silently: bool = False,
+) -> None:
+    """
+    Асинхронная отправка email через Celery с автоматическими повторами.
+
+    Использует настройки Django EMAIL_* и конфиг `configs.MAX_RETRIES`
+    для повторных попыток.
+    Поддерживает plain text и HTML версии письма.
+    """
+    current_retry = self.request.retries
+    max_retries = self.max_retries
+    try:
+        send_mail(
+            subject=subject,
+            message=message,
+            html_message=html_message,
+            from_email=from_email,
+            recipient_list=recipient_list,
+            fail_silently=fail_silently,
+        )
+        logger.info(f"Password reset email sent to '{recipient_list}'")
+    except Exception as e:
+        logger.warning(
+            f"Attempt {current_retry + 1} failed to send password "
+            f"reset email to '{recipient_list}': {str(e)}"
+        )
+        if current_retry >= max_retries - 1:
+            logger.error(
+                f"Failed to send password reset email to '{recipient_list}' "
+                f"after [{max_retries}] attempts"
+            )
+        if not fail_silently:
+            raise

app/services/auth/password_reset/tests.py

@@ -0,0 +1,98 @@
+from datetime import timedelta
+from unittest.mock import patch
+
+from django.contrib.auth import get_user_model
+from django.test import Client
+from django.urls import reverse, reverse_lazy
+from django.utils import timezone
+from inertia.test import InertiaTestCase  # type: ignore
+
+from app.services.auth.password_reset import configs
+from app.services.auth.password_reset.models import PasswordResetToken
+from app.services.auth.password_reset.views import PasswordResetView
+
+User = get_user_model()
+
+
+class PasswordResetTestCase(InertiaTestCase):
+    def setUp(self):
+        super().setUp()
+        self.client = Client()
+        self.url = reverse_lazy("password_reset")
+        user_data = {
+            "email": "testuser@example.com",
+            "password": "password123",
+        }
+        self.user = User.objects.create_user(**user_data)
+        current_time = timezone.now()
+        time_out = timedelta(configs.PASSWORD_RESET_TIMEOUT)
+        expires_at = current_time + time_out
+        token_data = {
+            "user_id": self.user,
+            "token_hash": hash("test_token"),
+            "expires_at": expires_at,
+        }
+        self.token = PasswordResetToken.objects.create(**token_data)
+
+
+@patch.object(PasswordResetView, "send_reset_email")
+class PasswordResetTests(PasswordResetTestCase):
+    def test_post_valid_email(self, mock_get_stats):
+        mock_get_stats.return_value = None
+        response = self.client.post(self.url, {"email": "testuser@example.com"})
+        assert response.props["status_code"] == 200
+        token = PasswordResetToken.objects.filter(user_id=self.user.id).first()
+        assert token.token_hash
+        assert token.expires_at > timezone.now()
+
+    def test_post_invalid_email(self, mock_get_stats):
+        mock_get_stats.return_value = None
+        response = self.client.post(self.url, {"email": "invalid@example.com"})
+        assert response.props["status_code"] == 200
+        assert response.props["status"] == "ok"
+
+
+class PasswordResetConfirmTests(PasswordResetTestCase):
+    def test_get_valid_token(self):
+        url = reverse("password_reset_confirm") + "?token=test_token"
+        self.client.get(url)
+        self.assertIncludesProps({"status_code": 200})
+
+    def test_get_expired_token(self):
+        PasswordResetToken.objects.create(
+            user_id=self.user,
+            token_hash=hash("expired_token"),
+            expires_at=timezone.now() - timedelta(hours=1),
+        )
+
+        url = reverse_lazy("password_reset_confirm") + "?token=expired_token"
+        self.client.get(url)
+        self.assertIncludesProps({"status_code": 400})
+
+    def test_post_valid_password(self):
+        url = reverse_lazy("password_reset_confirm")
+        data = {
+            "token": self.token.token_hash,
+            "newPassword": "StrongPass123!",
+        }
+        response = self.client.post(url, data)
+        assert response.props["status_code"] == 200
+
+        self.user.refresh_from_db()
+        self.token.refresh_from_db()
+        assert self.user.check_password("StrongPass123!")
+        assert self.token.is_used
+
+    def test_post_weak_password(self):
+        url = reverse_lazy("password_reset_confirm")
+        data = {"token": self.token.token_hash, "newPassword": "weak"}
+
+        self.client.post(url, data)
+        self.assertIncludesProps({"status_code": 422})
+
+    def test_post_invalid_token(self):
+        url = reverse_lazy("password_reset_confirm")
+        data = {"token": "invalid_token", "newPassword": "StrongPass123!"}
+
+        self.client.post(url, data)
+        self.assertIncludesProps({"status_code": 400})

app/services/auth/password_reset/urls.py

@@ -0,0 +1,16 @@
+from django.urls import path
+
+from . import views
+
+urlpatterns = [
+    path(
+        "forgot/",
+        views.PasswordResetView.as_view(),
+        name="password_reset",
+    ),
+    path(
+        "reset/",
+        views.PasswordResetConfirmView.as_view(),
+        name="password_reset_confirm",
+    ),
+]