Effortlessly create, manage, and send newsletters tailored to your business.
# Generated by Django 4.2.5 on 2023-09-28 22:37

from django.db import migrations, models
import uuid


class Migration(migrations.Migration):
    dependencies = [
        ("authorization", "0001_initial"),
    ]

    operations = [
        migrations.AddField(
            model_name="user",
            name="temp_id",
            field=models.UUIDField(default=uuid.uuid4),
        ),
    ]

newsltr/authorization/migrations/0002_user_temp_id.py
Style77/newsltr
# Generated by Django 4.2.5 on 2023-09-28 22:39

from django.db import migrations, models
import uuid


class Migration(migrations.Migration):
    dependencies = [
        ("authorization", "0002_user_temp_id"),
    ]

    operations = [
        migrations.RemoveField(
            model_name="user",
            name="id",
        ),
        migrations.AlterField(
            model_name="user",
            name="temp_id",
            field=models.UUIDField(
                default=uuid.uuid4, primary_key=True, serialize=False
            ),
        ),
    ]

newsltr/authorization/migrations/0003_remove_user_id_alter_user_temp_id.py
Style77/newsltr
# Generated by Django 4.2.5 on 2023-09-28 22:39

from django.db import migrations


class Migration(migrations.Migration):
    dependencies = [
        ("authorization", "0003_remove_user_id_alter_user_temp_id"),
    ]

    operations = [
        migrations.RenameField(
            model_name="user",
            old_name="temp_id",
            new_name="id",
        ),
    ]

newsltr/authorization/migrations/0004_rename_temp_id_user_id.py
Style77/newsltr
# Generated by Django 4.2.5 on 2023-09-28 22:39

from django.db import migrations, models
import uuid


class Migration(migrations.Migration):
    dependencies = [
        ("authorization", "0004_rename_temp_id_user_id"),
    ]

    operations = [
        migrations.AlterField(
            model_name="user",
            name="id",
            field=models.UUIDField(
                default=uuid.uuid4,
                editable=False,
                primary_key=True,
                serialize=False,
                unique=True,
            ),
        ),
    ]

newsltr/authorization/migrations/0005_alter_user_id.py
Style77/newsltr
# Generated by Django 4.2.5 on 2023-09-29 12:33

from django.db import migrations


class Migration(migrations.Migration):
    dependencies = [
        ("authorization", "0005_alter_user_id"),
    ]

    operations = [
        migrations.AlterModelManagers(
            name="user",
            managers=[],
        ),
    ]

newsltr/authorization/migrations/0006_alter_user_managers.py
Style77/newsltr
# Generated by Django 4.2.5 on 2023-09-29 13:31

from django.db import migrations, models
import uuid


class Migration(migrations.Migration):
    dependencies = [
        ("authorization", "0006_alter_user_managers"),
    ]

    operations = [
        migrations.AlterField(
            model_name="user",
            name="id",
            field=models.CharField(
                default=uuid.uuid4,
                editable=False,
                max_length=36,
                primary_key=True,
                serialize=False,
                unique=True,
            ),
        ),
    ]

newsltr/authorization/migrations/0007_alter_user_id.py
Style77/newsltr
import uuid

from django.contrib.auth.models import AbstractUser, BaseUserManager
from django.db import models
from django.utils.translation import gettext_lazy as _


class UserManager(BaseUserManager):
    def _create_user(self, email, password, **extra_fields):
        """
        Create and save a user with the given email, and password.
        """
        if not email:
            raise ValueError("The given email must be set")

        if not password:
            raise ValueError("The given password must be set")

        if not extra_fields.get("first_name") or not extra_fields.get("last_name"):
            raise ValueError("The given name must be set")

        extra_fields["first_name"] = extra_fields["first_name"].title()
        extra_fields["last_name"] = extra_fields["last_name"].title()

        email = self.normalize_email(email)
        user = self.model(email=email, **extra_fields)
        user.set_password(password)
        user.save(using=self._db)
        return user

    def create_user(self, email, password=None, **extra_fields):
        extra_fields.setdefault("is_staff", False)
        extra_fields.setdefault("is_superuser", False)
        return self._create_user(email, password, **extra_fields)

    def create_superuser(self, email, password=None, **extra_fields):
        extra_fields.setdefault("is_staff", True)
        extra_fields.setdefault("is_superuser", True)
        extra_fields.setdefault("is_active", True)

        if extra_fields.get("is_staff") is not True:
            raise ValueError("Superuser must have is_staff=True.")
        if extra_fields.get("is_superuser") is not True:
            raise ValueError("Superuser must have is_superuser=True.")

        return self._create_user(email, password, **extra_fields)


class User(AbstractUser):
    USERNAME_FIELD = "email"
    REQUIRED_FIELDS = ["first_name", "last_name"]

    id = models.CharField(
        primary_key=True, default=uuid.uuid4, editable=False, unique=True, max_length=36
    )
    email = models.EmailField(_("email address"), blank=False, unique=True)
    username = None

    objects = UserManager()

newsltr/authorization/models.py
Style77/newsltr
from django.contrib.auth import get_user_model
from rest_framework import serializers, status
from djoser.serializers import UserCreatePasswordRetypeSerializer, UserSerializer
from djoser.conf import settings
from rest_framework_simplejwt.serializers import (
    TokenObtainSerializer,
    update_last_login,
)
from rest_framework_simplejwt.exceptions import AuthenticationFailed
from rest_framework_simplejwt.tokens import RefreshToken


User = get_user_model()


class CustomUserCreateSerliazier(UserCreatePasswordRetypeSerializer):
    default_error_messages = {
        "name_invalid": "Enter a valid value.",
        "name_blank": "This field may not be blank.",
        "name_max_length": "Ensure this field has no more than 30 characters.",
        "name_required": "This field is required.",
    }

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.fields["first_name"] = serializers.CharField(
            required=True,
            style={"input_type": "text"},
        )
        self.fields["last_name"] = serializers.CharField(
            required=True,
            style={"input_type": "text"},
        )

    def validate(self, attrs):
        if attrs["first_name"] == "" or attrs["last_name"] == "":
            raise serializers.ValidationError(
                self.error_messages["name_blank"],
                code="name_blank",
            )
        if len(attrs["first_name"]) > 30 or len(attrs["last_name"]) > 30:
            raise serializers.ValidationError(
                self.error_messages["name_max_length"],
                code="name_max_length",
            )
        if (
            attrs["first_name"].isalpha() is False
            or attrs["last_name"].isalpha() is False
        ):
            raise serializers.ValidationError(
                self.error_messages["name_invalid"],
                code="name_invalid",
            )
        if attrs["first_name"] is None or attrs["last_name"] is None:
            raise serializers.ValidationError(
                self.error_messages["name_required"],
                code="name_required",
            )

        return super().validate(attrs)


class InActiveUser(AuthenticationFailed):
    status_code = status.HTTP_406_NOT_ACCEPTABLE
    default_detail = "User is not active, please confirm your email"
    default_code = "user_is_inactive"


class CustomUserSerializer(UserSerializer):
    class Meta:
        model = User
        fields = tuple(User.REQUIRED_FIELDS) + (
            settings.USER_ID_FIELD,
            settings.LOGIN_FIELD,
            "first_name",
            "last_name",
            "date_joined",
            "last_login",
        )
        read_only_fields = (settings.LOGIN_FIELD, "date_joined", "last_login")


class CustomTokenObtainPairSerializer(TokenObtainSerializer):
    username_field = settings.LOGIN_FIELD

    @classmethod
    def get_token(cls, user):
        return RefreshToken.for_user(user)

    def validate(self, attrs):
        data = super().validate(attrs)
        if not self.user.is_active:
            raise InActiveUser()

        refresh = self.get_token(self.user)

        data["refresh"] = str(refresh)
        data["access"] = str(refresh.access_token)

        if settings.UPDATE_LAST_LOGIN:
            update_last_login(None, self.user)

        return data

newsltr/authorization/serializers.py
Style77/newsltr
from django.contrib.auth import get_user_model
from django.db import IntegrityError
from unittest import mock
from django.urls import reverse

__all__ = [
    "get_user_model",
    "IntegrityError",
    "mock",
    "TEST_DATA",
]


TEST_DATA = {
    "first_name": "Test",
    "last_name": "Test",
    "email": "test@test.com",
    "password": "#Test1234",
}


def create_user(first_name=None, last_name=None, email=None, password=None, **kwargs):
    data = {
        "first_name": first_name or TEST_DATA["first_name"],
        "last_name": last_name or TEST_DATA["last_name"],
        "email": email or TEST_DATA["email"],
        "password": password or TEST_DATA["password"],
    }
    data.update(kwargs)
    user = get_user_model().objects.create_user(**data)
    user.raw_password = data["password"]
    return user


def login_user(client, email, password):
    response = client.post(
        reverse("jwt-create"), {"email": email, "password": password}
    )
    token = response.data["access"]
    client.credentials(HTTP_AUTHORIZATION="Bearer " + token)


def perform_create_mock(x):
    raise IntegrityError

newsltr/authorization/tests/common.py
Style77/newsltr
from django.contrib.auth.tokens import default_token_generator
from djet import assertions
from rest_framework import status
from rest_framework.reverse import reverse
from rest_framework.test import APITestCase
from authorization.tests.common import create_user

import djoser.signals
import djoser.utils
import djoser.views
from djoser.conf import settings as default_settings


class ActivationViewTest(
    APITestCase, assertions.EmailAssertionsMixin, assertions.StatusCodeAssertionsMixin
):
    def setUp(self):
        self.base_url = reverse("user-activation")
        self.signal_sent = False

    def signal_receiver(self, *args, **kwargs):
        self.signal_sent = True

    def test_post_activate_user_and_not_login(self):
        user = create_user()
        user.is_active = False
        user.save()
        data = {
            "uid": djoser.utils.encode_uid(user.pk),
            "token": default_token_generator.make_token(user),
        }

        response = self.client.post(self.base_url, data)
        user.refresh_from_db()

        self.assert_status_equal(response, status.HTTP_204_NO_CONTENT)
        self.assertTrue(user.is_active)

    def test_post_respond_with_bad_request_when_wrong_uid(self):
        user = create_user()
        data = {"uid": "wrong-uid", "token": default_token_generator.make_token(user)}

        response = self.client.post(self.base_url, data)

        self.assert_status_equal(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(list(response.data.keys()), ["uid"])
        self.assertEqual(
            response.data["uid"],
            [default_settings.CONSTANTS.messages.INVALID_UID_ERROR],
        )

    def test_post_respond_with_bad_request_when_stale_token(self):
        user = create_user()
        djoser.signals.user_activated.connect(self.signal_receiver)
        data = {
            "uid": djoser.utils.encode_uid(user.pk),
            "token": default_token_generator.make_token(user),
        }

        response = self.client.post(self.base_url, data)

        self.assert_status_equal(response, status.HTTP_403_FORBIDDEN)
        self.assertEqual(list(response.data.keys()), ["detail"])
        self.assertEqual(
            response.data["detail"],
            default_settings.CONSTANTS.messages.STALE_TOKEN_ERROR,
        )
        self.assertFalse(self.signal_sent)

    def test_post_respond_with_bad_request_when_wrong_token(self):
        user = create_user()
        djoser.signals.user_activated.connect(self.signal_receiver)
        data = {"uid": djoser.utils.encode_uid(user.pk), "token": "wrong-token"}

        response = self.client.post(self.base_url, data)

        self.assert_status_equal(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(list(response.data.keys()), ["token"])
        self.assertEqual(
            response.data["token"],
            [default_settings.CONSTANTS.messages.INVALID_TOKEN_ERROR],
        )
        self.assertFalse(self.signal_sent)

newsltr/authorization/tests/test_activation.py
Style77/newsltr
from django.test import TestCase
from rest_framework_simplejwt.serializers import TokenVerifySerializer

from djoser.social.token.jwt import TokenStrategy
from django.contrib.auth import get_user_model

from authorization.tests.common import TEST_DATA, create_user

User = get_user_model()


class JWTStrategyTestCase(TestCase):
    def test_obtain_provides_valid_token_for_given_user(self):
        user = create_user(**TEST_DATA)

        res = TokenStrategy.obtain(user)
        self.assertEqual(res["user"], user)

        data = {"token": res["access"]}
        serializer = TokenVerifySerializer(data=data)
        self.assertTrue(serializer.is_valid())

newsltr/authorization/tests/test_jwt.py
Style77/newsltr
from rest_framework import status
from rest_framework.test import APITestCase
from rest_framework.reverse import reverse
from djet import assertions
from django.contrib.auth import get_user_model

from authorization.tests.common import TEST_DATA, create_user, mock, perform_create_mock
from djoser.conf import settings as default_settings

User = get_user_model()


class UserCreateViewTest(
    APITestCase,
    assertions.StatusCodeAssertionsMixin,
    assertions.EmailAssertionsMixin,
    assertions.InstanceAssertionsMixin,
):
    def setUp(self):
        self.base_url = reverse("user-list")

    def test_post_create_user_without_login(self):
        data = {
            "first_name": TEST_DATA["first_name"],
            "last_name": TEST_DATA["last_name"],
            "email": TEST_DATA["email"],
            "password": TEST_DATA["password"],
            "re_password": TEST_DATA["password"],
        }

        response = self.client.post(self.base_url, data)
        self.assert_status_equal(response, status.HTTP_201_CREATED)
        self.assertTrue("password" not in response.data)
        self.assert_instance_exists(User, email=TEST_DATA["email"])

        user = User.objects.get(email=TEST_DATA["email"])
        self.assertTrue(user.check_password(TEST_DATA["password"]))
        self.assert_emails_in_mailbox(1)
        self.assert_email_exists(to=[data["email"]])
        self.assertFalse(user.is_active)

    def test_post_not_create_new_user_if_email_exists(self):
        create_user()
        response = self.client.post(self.base_url, TEST_DATA)
        self.assert_status_equal(response, status.HTTP_400_BAD_REQUEST)

    def test_post_not_register_if_password_common(self):
        data = {
            "first_name": TEST_DATA["first_name"],
            "last_name": TEST_DATA["last_name"],
            "email": TEST_DATA["email"],
            "password": "12345678",
            "re_password": "12345678",
        }
        response = self.client.post(self.base_url, data)

        self.assert_status_equal(response, status.HTTP_400_BAD_REQUEST)
        response.render()
        self.assertEqual(
            str(response.data["password"][0]), "This password is too common."
        )
        self.assertEqual(response.data["password"][0].code, "password_too_common")

    def test_post_not_register_if_password_mismatch(self):
        data = {
            "first_name": TEST_DATA["first_name"],
            "last_name": TEST_DATA["last_name"],
            "email": TEST_DATA["email"],
            "password": "#Secret1234",
            "re_password": "#Wrong1234",
        }
        response = self.client.post(self.base_url, data)

        self.assert_status_equal(response, status.HTTP_400_BAD_REQUEST)
        response.render()
        self.assertEqual(
            str(response.data["non_field_errors"][0]),
            default_settings.CONSTANTS.messages.PASSWORD_MISMATCH_ERROR,
        )

    @mock.patch(
        "djoser.serializers.UserCreateSerializer.perform_create",
        side_effect=perform_create_mock,
    )
    def test_post_return_400_for_integrity_error(self, perform_create):
        data = {
            "integrity_error": "true",
            "first_name": TEST_DATA["first_name"],
            "last_name": TEST_DATA["last_name"],
            "email": "john@beatles.com",
            "password": "Secret1234",
            "re_password": "Secret1234",
        }
        response = self.client.post(self.base_url, data)

        self.assert_status_equal(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(
            response.data,
            [default_settings.CONSTANTS.messages.CANNOT_CREATE_USER_ERROR],
        )

newsltr/authorization/tests/test_user_create.py
Style77/newsltr
from django.contrib.auth import get_user_model

from rest_framework import status
from rest_framework.reverse import reverse
from rest_framework.test import APITestCase
from djet import assertions

from djoser.views import UserViewSet

from authorization.tests.common import create_user, login_user, TEST_DATA


User = get_user_model()


class UserMeDeleteViewTestCase(
    APITestCase,
    assertions.StatusCodeAssertionsMixin,
    assertions.InstanceAssertionsMixin,
):
    viewset = UserViewSet

    def test_delete_user_if_logged_in(self):
        user = create_user()
        self.assert_instance_exists(User, email=TEST_DATA["email"])
        data = {"current_password": TEST_DATA["password"]}

        self.client.force_authenticate(user=user)
        response = self.client.delete(reverse("user-me"), data=data)

        self.assert_status_equal(response, status.HTTP_204_NO_CONTENT)
        self.assert_instance_does_not_exist(User, email=TEST_DATA["email"])

    def test_not_delete_if_fails_password_validation(self):
        user = create_user()
        self.assert_instance_exists(User, email=TEST_DATA["email"])
        data = {"current_password": "incorrect"}

        self.client.force_authenticate(user=user)
        response = self.client.delete(reverse("user-me"), data=data)

        self.assert_status_equal(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(response.data, {"current_password": ["Invalid password."]})


class UserViewSetDeletionTest(
    APITestCase,
    assertions.StatusCodeAssertionsMixin,
    assertions.EmailAssertionsMixin,
    assertions.InstanceAssertionsMixin,
):
    def test_delete_user_if_logged_in(self):
        user = create_user()
        self.assert_instance_exists(User, email=TEST_DATA["email"])
        data = {"current_password": TEST_DATA["password"]}

        login_user(self.client, user.email, TEST_DATA["password"])

        response = self.client.delete(
            reverse("user-detail", kwargs={User._meta.pk.name: user.pk}),
            data=data,
        )

        self.assert_status_equal(response, status.HTTP_204_NO_CONTENT)
        self.assert_instance_does_not_exist(User, email=TEST_DATA["email"])

    def test_not_delete_if_fails_password_validation(self):
        user = create_user()
        self.assert_instance_exists(User, email=TEST_DATA["email"])
        data = {"current_password": "incorrect"}

        login_user(self.client, user.email, TEST_DATA["password"])

        response = self.client.delete(
            reverse("user-detail", kwargs={User._meta.pk.name: user.pk}),
            data=data,
        )

        self.assert_status_equal(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(response.data, {"current_password": ["Invalid password."]})

newsltr/authorization/tests/test_user_delete.py
Style77/newsltr
from djet import assertions
from djoser import signals
from rest_framework import status
from rest_framework.reverse import reverse
from rest_framework.test import APITestCase
from authorization.tests.common import create_user, login_user, TEST_DATA


class BaseUserViewSetListTest(APITestCase, assertions.StatusCodeAssertionsMixin):
    def setUp(self):
        super().setUp()
        self.user = create_user()
        self.superuser = create_user(
            first_name="super",
            last_name="user",
            email="superuser@example.com",
            password="Superuser123",
            is_superuser=True,
            is_staff=True,
        )


class UserViewSetListTest(BaseUserViewSetListTest):
    def test_unauthenticated_user_cannot_get_user_detail(self):
        response = self.client.get(reverse("user-detail", args=[self.user.pk]))

        self.assert_status_equal(response, status.HTTP_401_UNAUTHORIZED)

    def test_user_can_get_own_details(self):
        login_user(self.client, self.user.email, TEST_DATA["password"])
        response = self.client.get(reverse("user-detail", args=[self.user.pk]))

        self.assert_status_equal(response, status.HTTP_200_OK)

    def test_user_cannot_get_other_user_detail(self):
        login_user(self.client, self.user.email, TEST_DATA["password"])
        response = self.client.get(reverse("user-detail", args=[self.superuser.pk]))

        self.assert_status_equal(response, status.HTTP_404_NOT_FOUND)

    def test_superuser_can_get_other_user_detail(self):
        login_user(self.client, self.superuser.email, "Superuser123")
        response = self.client.get(reverse("user-detail", args=[self.user.pk]))

        self.assert_status_equal(response, status.HTTP_200_OK)


class UserViewSetEditTest(APITestCase, assertions.StatusCodeAssertionsMixin):
    def setUp(self):
        self.signal_sent = False

    def signal_receiver(self, *args, **kwargs):
        self.signal_sent = True

    def test_patch_edits_user_attribute(self):
        user = create_user()
        signals.user_updated.connect(self.signal_receiver)
        login_user(self.client, user.email, TEST_DATA["password"])
        response = self.client.patch(
            path=reverse("user-detail", args=(user.pk,)),
            data={"first_name": "Test2"},
        )

        self.assert_status_equal(response, status.HTTP_200_OK)
        self.assertTrue("first_name" in response.data)

        user.refresh_from_db()
        self.assertTrue(self.signal_sent)
        self.assertTrue(user.first_name == "Test2")

    def test_patch_cant_edit_others_attribute(self):
        user = create_user()
        another_user = create_user(
            **{
                "first_name": "Paul",
                "last_name": "Thomson",
                "password": "Password123",
                "email": "paul@beatles.com",
            }
        )
        login_user(self.client, user.email, TEST_DATA["password"])
        response = self.client.patch(
            path=reverse("user-detail", args=(another_user.pk,)),
            data={"email": "new@gmail.com"},
        )

        self.assert_status_equal(response, status.HTTP_404_NOT_FOUND)

        another_user.refresh_from_db()
        self.assertTrue(another_user.email == "paul@beatles.com")

    def test_put_edits_user_attribute(self):
        user = create_user()
        login_user(self.client, user.email, TEST_DATA["password"])

        response = self.client.patch(
            path=reverse("user-detail", args=(user.pk,)),
            data={"password": "NewPassword123"},
        )

        self.assert_status_equal(response, status.HTTP_200_OK)

        user.refresh_from_db()
        self.assertTrue(user.email == TEST_DATA["email"])

    def test_put_cant_edit_others_attribute(self):
        user = create_user()
        another_user_data = {
            "first_name": "Paul",
            "last_name": "Thomson",
            "password": "secret",
            "email": "paul@beatles.com",
        }
        another_user = create_user(**another_user_data)
        another_user_data["password"] = "changed_secret"
        login_user(self.client, user.email, TEST_DATA["password"])

        response = self.client.patch(
            path=reverse("user-detail", args=(another_user.pk,)), data=another_user_data
        )

        self.assert_status_equal(response, status.HTTP_404_NOT_FOUND)

        another_user.refresh_from_db()
        self.assertTrue(another_user.email == "paul@beatles.com")

newsltr/authorization/tests/test_user_detail.py
Style77/newsltr
from djet import assertions
from rest_framework import status
from rest_framework.reverse import reverse
from rest_framework.test import APITestCase
from authorization.tests.common import create_user, login_user, TEST_DATA


class UserViewSetListTest(APITestCase, assertions.StatusCodeAssertionsMixin):
    def setUp(self):
        self.base_url = reverse("user-list")
        self.user = create_user()
        self.superuser = create_user(
            first_name="Super",
            last_name="User",
            email="superuser@example.com",
            password="Superuser123",
            is_superuser=True,
            is_staff=True,
        )

    def test_unauthenticated_user_cannot_list_users(self):
        response = self.client.get(self.base_url)

        self.assert_status_equal(response, status.HTTP_401_UNAUTHORIZED)

    def test_user_cannot_list_other_users(self):
        login_user(self.client, self.user.email, TEST_DATA["password"])
        response = self.client.get(self.base_url)

        self.assert_status_equal(response, status.HTTP_200_OK)
        self.assertEqual(len(response.json()), 1)

    def test_superuser_can_list_all_users(self):
        login_user(self.client, self.superuser, "Superuser123")
        response = self.client.get(self.base_url)

        self.assert_status_equal(response, status.HTTP_200_OK)
        self.assertEqual(len(response.json()), 2)

newsltr/authorization/tests/test_user_list.py
Style77/newsltr
from django.conf import settings
from django.contrib.auth import get_user_model
from django.test import override_settings
from djet import assertions
from rest_framework import serializers, status
from rest_framework.reverse import reverse
from rest_framework.test import APITestCase
from authorization.tests.common import create_user, login_user, TEST_DATA

User = get_user_model()


class UserViewSetMeTest(
    APITestCase,
    assertions.EmailAssertionsMixin,
    assertions.InstanceAssertionsMixin,
    assertions.StatusCodeAssertionsMixin,
):
    class DummyCurrentUserSerializer(serializers.ModelSerializer):
        class Meta:
            model = User
            fields = ("is_staff",)

    def setUp(self):
        self.base_url = reverse("user-me")
        data = {
            "first_name": TEST_DATA["first_name"],
            "last_name": TEST_DATA["last_name"],
            "email": TEST_DATA["email"],
            "password": TEST_DATA["password"],
        }
        self.user = create_user(**data)
        login_user(self.client, data["email"], data["password"])

    def test_get_return_user(self):
        response = self.client.get(self.base_url)

        self.assert_status_equal(response, status.HTTP_200_OK)
        self.assertEqual(
            set(response.data.keys()),
            set([User.USERNAME_FIELD, User._meta.pk.name] + User.REQUIRED_FIELDS),
        )

    def test_patch_email_change(self):
        data = {"email": TEST_DATA["email"]}
        response = self.client.patch(self.base_url, data=data)

        self.assert_status_equal(response, status.HTTP_200_OK)
        self.user.refresh_from_db()
        self.assertEqual(data["email"], self.user.email)
        self.assertTrue(self.user.is_active)
        # self.assert_emails_in_mailbox(1)
        # self.assert_email_exists(to=[data["email"]])

    @override_settings(
        DJOSER=dict(
            settings.DJOSER,
            **{"SERIALIZERS": {"current_user": DummyCurrentUserSerializer}},
        )
    )
    def test_serializer(self):
        """Test that the endpoints use the proper serializer.

        How it works: it adds an additional field to the current_user
        serializer and then checks that the field shows in the response.
        """
        response = self.client.get(self.base_url)

        self.user.refresh_from_db()
        self.assertEqual(response.data["is_staff"], self.user.is_staff)


class UserViewSetMeDeleteTest(
    APITestCase,
    assertions.InstanceAssertionsMixin,
    assertions.StatusCodeAssertionsMixin,
):
    def setUp(self):
        self.base_url = reverse("user-me")
        self.data = {
            "first_name": TEST_DATA["first_name"],
            "last_name": TEST_DATA["last_name"],
            "email": TEST_DATA["email"],
            "password": TEST_DATA["password"],
        }

    def test_delete_user_if_logged_in(self):
        user = create_user()
        self.assert_instance_exists(User, email=self.data["email"])
        data = {"current_password": self.data["password"]}
        login_user(self.client, user.email, self.data["password"])

        response = self.client.delete(self.base_url, data=data)
        self.assert_status_equal(response, status.HTTP_204_NO_CONTENT)
        self.assert_instance_does_not_exist(User, email=self.data["email"])

    def test_not_delete_if_fails_password_validation(self):
        user = create_user()
        self.assert_instance_exists(User, email=self.data["email"])
        data = {"current_password": "incorrect"}
        login_user(self.client, user.email, self.data["password"])

        response = self.client.delete(self.base_url, data=data)

        self.assert_status_equal(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(response.data, {"current_password": ["Invalid password."]})

newsltr/authorization/tests/test_user_me.py
Style77/newsltr
from django.urls import path, include

# Override the default djoser.urls.jwt and add my custom view for token obtain
from authorization.views import EmailTokenObtainPairView
from djoser.urls.jwt import urlpatterns as jwt_urlpatterns

jwt_urlpatterns.pop(0)

urlpatterns = [
    path(r"auth/", include("djoser.urls")),
    path(r"auth/", include("djoser.urls.jwt")),
    path(r"auth/", include("djoser.social.urls")),
    path(r"auth/jwt/create/", EmailTokenObtainPairView.as_view(), name="jwt-create"),
]

newsltr/authorization/urls.py
Style77/newsltr
def user_authentication_rule(user):
    return user is not None

newsltr/authorization/utils.py
Style77/newsltr
from rest_framework import status
from rest_framework.response import Response

from rest_framework_simplejwt.views import TokenViewBase
from rest_framework_simplejwt.exceptions import (
    AuthenticationFailed,
    TokenError,
    InvalidToken,
)
from authorization.serializers import CustomTokenObtainPairSerializer, InActiveUser


class EmailTokenObtainPairView(TokenViewBase):
    serializer_class = CustomTokenObtainPairSerializer

    def post(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        try:
            serializer.is_valid(raise_exception=True)
        except AuthenticationFailed:
            raise InActiveUser()
        except TokenError:
            raise InvalidToken()

        return Response(serializer.validated_data, status=status.HTTP_200_OK)

newsltr/authorization/views.py
Style77/newsltr
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys


def main():
    """Run administrative tasks."""
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "newsltr.settings")
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    execute_from_command_line(sys.argv)


if __name__ == "__main__":
    main()

newsltr/manage.py
Style77/newsltr
"""
ASGI config for newsltr project.

It exposes the ASGI callable as a module-level variable named ``application``.

For more information on this file, see
https://docs.djangoproject.com/en/4.2/howto/deployment/asgi/
"""

import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "newsltr.settings")

application = get_asgi_application()

newsltr/newsltr/asgi.py
Style77/newsltr
import os
import ssl
from celery import Celery
from django.apps import apps

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "newsltr.settings")

app = Celery(
    "newsltr",
    broker_use_ssl={"ssl_cert_reqs": ssl.CERT_NONE},
    redis_backend_use_ssl={"ssl_cert_reqs": ssl.CERT_NONE},
)

app.config_from_object("django.conf:settings", namespace="CELERY")
app.conf.enable_utc = True

app.autodiscover_tasks(lambda: [n.name for n in apps.get_app_configs()])

newsltr/newsltr/celery.py
Style77/newsltr
from django.contrib.auth.tokens import default_token_generator
from templated_mail.mail import BaseEmailMessage

from djoser import utils
from djoser.conf import settings


class ActivationEmail(BaseEmailMessage):
    template_name = "email/activation.html"

    def get_context_data(self):
        # ActivationEmail can be deleted
        context = super().get_context_data()

        user = context.get("user")
        print(user.first_name)
        context["uid"] = utils.encode_uid(user.pk)
        context["token"] = default_token_generator.make_token(user)
        context["url"] = settings.ACTIVATION_URL.format(**context)
        context["first_name"] = user.first_name
        print(context)
        return context


class ConfirmationEmail(BaseEmailMessage):
    template_name = "email/confirmation.html"


class PasswordResetEmail(BaseEmailMessage):
    template_name = "email/password_reset.html"

    def get_context_data(self):
        # PasswordResetEmail can be deleted
        context = super().get_context_data()

        user = context.get("user")
        context["uid"] = utils.encode_uid(user.pk)
        context["token"] = default_token_generator.make_token(user)
        context["url"] = settings.PASSWORD_RESET_CONFIRM_URL.format(**context)
        return context


class PasswordChangedConfirmationEmail(BaseEmailMessage):
    template_name = "email/password_changed_confirmation.html"


class UsernameChangedConfirmationEmail(BaseEmailMessage):
    template_name = "email/username_changed_confirmation.html"


class UsernameResetEmail(BaseEmailMessage):
    template_name = "email/username_reset.html"

    def get_context_data(self):
        context = super().get_context_data()

        user = context.get("user")
        context["uid"] = utils.encode_uid(user.pk)
        context["token"] = default_token_generator.make_token(user)
        context["url"] = settings.USERNAME_RESET_CONFIRM_URL.format(**context)
        return context

newsltr/newsltr/email.py
Style77/newsltr
"""
Django settings for newsltr project.

Generated by 'django-admin startproject' using Django 4.2.5.

For more information on this file, see
https://docs.djangoproject.com/en/4.2/topics/settings/

For the full list of settings and their values, see
https://docs.djangoproject.com/en/4.2/ref/settings/
"""

import os
from pathlib import Path
from datetime import timedelta
from dotenv import load_dotenv
import dj_database_url

DEVELOPMENT = os.getenv("DJANGO_DEVELOPMENT", True)
if DEVELOPMENT:
    load_dotenv(".env")


# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent


# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/4.2/howto/deployment/checklist/

# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = os.getenv("DJANGO_SECRET_KEY")

# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = DEVELOPMENT

if DEVELOPMENT:
    ALLOWED_HOSTS = ["0.0.0.0", "localhost", "127.0.0.1"]
else:
    RENDER_EXTERNAL_HOSTNAME = os.getenv("RENDER_EXTERNAL_HOSTNAME")
    if RENDER_EXTERNAL_HOSTNAME:
        ALLOWED_HOSTS = [RENDER_EXTERNAL_HOSTNAME]
    else:
        ALLOWED_HOSTS = []

# Application definition

# Core Django Apps
CORE_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
]

# Third-party Packages
THIRD_PARTY_APPS = [
    "corsheaders",
    "rest_framework",
    "djoser",
    "djoser.social",
]

# Health Check Apps
HEALTH_CHECK_APPS = [
    "health_check",
    "health_check.db",
    "health_check.cache",
    "health_check.storage",
    "health_check.contrib.migrations",
    # "health_check.contrib.celery",
    # "health_check.contrib.celery_ping",
    "health_check.contrib.psutil",
    # "health_check.contrib.rabbitmq",
    "health_check.contrib.redis",
]

# Custom Apps
CUSTOM_APPS = ["authorization.apps.AuthorizationConfig"]

# Celery Apps
CELERY_APPS = [
    "django_celery_results",
    "django_celery_beat",
]

# Combine all categories
INSTALLED_APPS = (
    CORE_APPS + THIRD_PARTY_APPS + HEALTH_CHECK_APPS + CUSTOM_APPS + CELERY_APPS
)

# Redis

REDIS_URL = os.getenv("REDIS_URL")

# Celery

CELERY_BROKER_URL = REDIS_URL
CELERY_RESULT_BACKEND = "django-db"
CELERY_ACCEPT_CONTENT = ["application/json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_IMPORTS = ()

CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers.DatabaseScheduler"

# Email

EMAIL_BACKEND = (
    "anymail.backends.sendinblue.EmailBackend"
    if os.getenv("SENDINBLUE_API_KEY")
    else "django.core.mail.backends.smtp.EmailBackend"
)
EMAIL_HOST = os.getenv("EMAIL_HOST")
EMAIL_HOST_USER = os.getenv("EMAIL_HOST_USER")
EMAIL_HOST_PASSWORD = os.getenv("EMAIL_HOST_PASSWORD")
EMAIL_PORT = 587
EMAIL_USE_TLS = True
EMAIL_TIMEOUT = 300  # in seconds
DEFAULT_FROM_EMAIL = f"Newsltr <{'newsltr@newsltr.io' if os.getenv('SENDINBLUE_API_KEY') else EMAIL_HOST}>"

ANYMAIL = {
    "SENDINBLUE_API_KEY": os.getenv("SENDINBLUE_API_KEY"),
}

CELERY_EMAIL_CHUNK_SIZE = 10
# CELERY_IMPORTS = ("newsltr.tasks",)
# Rest Framework

AUTH_USER_MODEL = "authorization.User"

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": (
        "rest_framework_simplejwt.authentication.JWTAuthentication",
    ),
    "DEFAULT_PERMISSION_CLASSES": [
        "rest_framework.permissions.IsAuthenticated",
    ],
}

if DEVELOPMENT:
    INSTALLED_APPS += ["drf_spectacular", "drf_spectacular_sidecar"]

    REST_FRAMEWORK["DEFAULT_SCHEMA_CLASS"] = "drf_spectacular.openapi.AutoSchema"

    SPECTACULAR_SETTINGS = {
        "TITLE": "Newsltr",
        "DESCRIPTION": "Newsltr API Documentation",
        "VERSION": "1.0.0",
        "SERVE_INCLUDE_SCHEMA": False,
        "SWAGGER_UI_DIST": "SIDECAR",  # shorthand to use the sidecar instead
        "SWAGGER_UI_FAVICON_HREF": "SIDECAR",
        "REDOC_DIST": "SIDECAR",
        # OTHER SETTINGS
    }


SIMPLE_JWT = {
    "ACCESS_TOKEN_LIFETIME": timedelta(minutes=15),
    "REFRESH_TOKEN_LIFETIME": timedelta(days=7),
    # To include additional security let refresh_tokens be blacklisted
    # after using one of them
    # "ROTATE_REFRESH_TOKENS": True,
    # "BLACKLIST_AFTER_ROTATION": True,
    "AUTH_HEADER_TYPES": ("Bearer",),
    "AUTH_HEADER_NAME": "HTTP_AUTHORIZATION",
    "USER_ID_FIELD": "id",
    "USER_ID_CLAIM": "user_id",
    "TOKEN_OBTAIN_SERIALIZER": "authorization.serializers.CustomTokenObtainSerializer",
    "USER_AUTHENTICATION_RULE": "authorization.utils.user_authentication_rule",
}

DJOSER = {
    "LOGIN_FIELD": "email",
    "UPDATE_LAST_LOGIN": True,
    "SEND_ACTIVATION_EMAIL": True,
    "ACTIVATION_URL": "activate/{uid}/{token}",
    "PASSWORD_RESET_CONFIRM_URL": "password/reset/confirm/{uid}/{token}",
    "USERNAME_RESET_CONFIRM_URL": "email/reset/confirm/{uid}/{token}",
    "SEND_CONFIRMATION_EMAIL": True,
    "USER_CREATE_PASSWORD_RETYPE": True,
    "SET_PASSWORD_RETYPE": True,
    "PASSWORD_RESET_CONFIRM_RETYPE": True,
    "TOKEN_MODEL": None,
    "HIDE_USERS": True,
    "SERIALIZERS": {
        "user_create_password_retype": "authorization.serializers.CustomUserCreateSerliazier",
        "user": "authorization.serializers.CustomUserSerializer",
    },
    "EMAIL": {
        "activation": "newsltr.email.ActivationEmail",
        "confirmation": "newsltr.email.ConfirmationEmail",
        "password_reset": "newsltr.email.PasswordResetEmail",
        "password_changed_confirmation": "newsltr.email.PasswordChangedConfirmationEmail",
        "username_changed_confirmation": "newsltr.email.UsernameChangedConfirmationEmail",
        "username_reset": "newsltr.email.UsernameResetEmail",
    },
}

AUTHENTICATION_BACKENDS = (
    "social_core.backends.google.GoogleOAuth2",
    "django.contrib.auth.backends.AllowAllUsersModelBackend",
    # "django.contrib.auth.backends.ModelBackend",
)

SOCIAL_AUTH_GOOGLE_OAUTH2_KEY = os.getenv("GOOGLE_OAUTH2_KEY")
SOCIAL_AUTH_GOOGLE_OAUTH2_SECRET = os.getenv("GOOGLE_OAUTH2_SECRET")
SOCIAL_AUTH_GOOGLE_OAUTH2_SCOPE = [
    "https://www.googleapis.com/auth/userinfo.email",
    "https://www.googleapis.com/auth/userinfo.profile",
    "openid",
]
SOCIAL_AUTH_GOOGLE_OAUTH2_EXTRA_DATA = ["first_name", "last_name"]

MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "corsheaders.middleware.CorsMiddleware",
    "social_django.middleware.SocialAuthExceptionMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django.middleware.clickjacking.XFrameOptionsMiddleware",
]

# Cors

CORS_ALLOWED_ORIGINS = ["http://localhost:3000", "http://127.0.0.1:3000"]
CORS_ORIGIN_WHITELIST = [
    "http://localhost:3000",
    "http://127.0.0.1:3000",
]
CORS_ALLOW_CREDENTIALS = True

ROOT_URLCONF = "newsltr.urls"

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TEMPLATES = [
    {
        "BACKEND": "django.template.backends.django.DjangoTemplates",
        "DIRS": [os.path.join(BASE_DIR, "templates")],
        "APP_DIRS": True,
        "OPTIONS": {
            "context_processors": [
                "django.template.context_processors.debug",
                "django.template.context_processors.request",
                "django.contrib.auth.context_processors.auth",
                "django.contrib.messages.context_processors.messages",
            ],
        },
    },
]

WSGI_APPLICATION = "newsltr.wsgi.application"


# Database
# https://docs.djangoproject.com/en/4.2/ref/settings/#databases

DATABASES = dict()
DATABASES["default"] = dj_database_url.config(
    conn_max_age=600,
    conn_health_checks=True,
)

CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": REDIS_URL,
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
        },
    }
}


# Password validation
# https://docs.djangoproject.com/en/4.2/ref/settings/#auth-password-validators

AUTH_PASSWORD_VALIDATORS = [
    {
        "NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator",
    },
    {
        "NAME": "django.contrib.auth.password_validation.MinimumLengthValidator",
    },
    {
        "NAME": "django.contrib.auth.password_validation.CommonPasswordValidator",
    },
    {
        "NAME": "django.contrib.auth.password_validation.NumericPasswordValidator",
    },
]


# Internationalization
# https://docs.djangoproject.com/en/4.2/topics/i18n/

LANGUAGE_CODE = "en-us"

TIME_ZONE = "UTC"

USE_I18N = True

USE_TZ = True


# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/4.2/howto/static-files/

STATIC_URL = "static/"

# Default primary key field type
# https://docs.djangoproject.com/en/4.2/ref/settings/#default-auto-field

DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"

newsltr/newsltr/settings.py
Style77/newsltr
from django.urls import path, include
from django.contrib import admin

from . import settings

urlpatterns = [
    path(r"api/v1/", include("authorization.urls")),
    path(r"api/v1/health/", include("health_check.urls")),
]

if settings.DEVELOPMENT:
    from drf_spectacular.views import (
        SpectacularAPIView,
        SpectacularRedocView,
        SpectacularSwaggerView,
    )  # noqa

    urlpatterns += [
        path("api/v1/schema/", SpectacularAPIView.as_view(), name="schema"),
        # Optional UI:
        path(
            "api/v1/schema/swagger-ui/",
            SpectacularSwaggerView.as_view(url_name="schema"),
            name="swagger-ui",
        ),
        path(
            "api/v1/schema/redoc/",
            SpectacularRedocView.as_view(url_name="schema"),
            name="redoc",
        ),
        path(r"admin/", admin.site.urls),
    ]

newsltr/newsltr/urls.py
Style77/newsltr
"""
WSGI config for newsltr project.

It exposes the WSGI callable as a module-level variable named ``application``.

For more information on this file, see
https://docs.djangoproject.com/en/4.2/howto/deployment/wsgi/
"""

import os

from django.core.wsgi import get_wsgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "newsltr.settings")

application = get_wsgi_application()

newsltr/newsltr/wsgi.py
Style77/newsltr