Hello everyone.
I am doing a project that consists in creating an app, and I am using Django Rest Framework to create my APIs. I am getting this error, and just don’t know why it is appearing.
I am doing everything using DjangoCassandraEngine, because my databse is in Cassandra.
I am creating the authentication, and I want to use Refresh Tokens so that the user does not need to login again and again.
Here is my models.py
from django_cassandra_engine.models import DjangoCassandraModel
from cassandra.cqlengine import columns
import uuid
import datetime
import bcrypt
class Conversations(DjangoCassandraModel):
id_conversation = columns.UUID(primary_key=True, default=uuid.uuid4)
id_user = columns.UUID(required=True)
message_ids = columns.List(columns.UUID)
title = columns.Text()
created_at = columns.DateTime(default=datetime.datetime.utcnow)
class Meta:
db_table = "conversations"
class Message(DjangoCassandraModel):
id_message = columns.UUID(primary_key=True, default=uuid.uuid4)
id_conversation = columns.UUID(required=True)
sender = columns.Text(required=True)
content = columns.Text(required=True)
created_at = columns.DateTime(default=datetime.datetime.utcnow)
class Meta:
db_table = "messages"
class Users(DjangoCassandraModel):
id_user = columns.UUID(primary_key=True, default=uuid.uuid4)
email = columns.Text(required=True, index=True)
username = columns.Text(required=True, index=True)
password = columns.Text(required=True)
is_active = columns.Boolean(default=True)
is_staff = columns.Boolean(default=False)
is_superuser = columns.Boolean(default=False)
USERNAME_FIELD = 'email'
REQUIRED_FIELDS = ['username', 'password']
class Meta:
db_table = 'users'
def set_password(self, password):
self.password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
def check_password(self, password):
return bcrypt.checkpw(password.encode('utf-8'), self.password.encode('utf-8'))
serializers.py
from rest_framework import serializers
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
import bcrypt
from .models import Users, Message
import uuid
class MessageSerializer(serializers.ModelSerializer):
class Meta:
model = Message
fields = ['id_message', 'id_conversation', 'sender', 'content', 'created_at']
class UserSerializer(serializers.ModelSerializer):
id_user = serializers.UUIDField()
email = serializers.CharField()
username = serializers.CharField()
is_active = serializers.BooleanField()
is_staff = serializers.BooleanField()
is_superuser = serializers.BooleanField()
class Meta:
model = Users
fields = ['id_user', 'email', 'username', 'is_active', 'is_staff', 'is_superuser']
class RegisterSerializer(serializers.Serializer):
email = serializers.EmailField()
username = serializers.CharField(max_length=255)
password = serializers.CharField(write_only=True)
def validate_email(self, value):
if list(Users.objects.filter(email=value)):
raise serializers.ValidationError("This email is already in use.")
return value
def validate_username(self, value):
if list(Users.objects.filter(username=value)):
raise serializers.ValidationError("This username is already in use.")
return value
def create(self, validated_data):
"""Create a user and encrypt the password before saving"""
hashed_password = bcrypt.hashpw(validated_data['password'].encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
user = Users.create(
id_user=uuid.uuid4(),
email=validated_data['email'].lower(),
username=validated_data['username'],
password=hashed_password
)
return user
class LoginSerializer(serializers.Serializer):
email = serializers.EmailField()
password = serializers.CharField(write_only=True)
def validate(self, data):
email = data.get('email')
password = data.get('password')
user = Users.objects.filter(email=email).first()
if not user:
raise ValueError("Invalid email or password!")
if not user.password or not bcrypt.checkpw(password.encode('utf-8'), user.password.encode('utf-8')):
raise serializers.ValidationError("Invalid email or password!")
data['user'] = user
return data
settings.py
SIMPLE_JWT = {
"ACCESS_TOKEN_LIFETIME": timedelta(minutes=5),
"REFRESH_TOKEN_LIFETIME": timedelta(days=7),
"ROTATE_REFRESH_TOKENS": True,
"BLACKLIST_AFTER_ROTATION": True,
"AUTH_HEADER_TYPES": ("Bearer",),
"USER_ID_FIELD": "id_user",
"USER_ID_CLAIM": "user_id",
"AUTH_TOKEN_CLASSES": ("rest_framework_simplejwt.tokens.AccessToken",),
# 'USERNAME_FIELD': 'email',
"ALGORITHM": "HS256",
}
tests.py
from django.test import TestCase
from rest_framework.test import APIClient
from rest_framework import status
import uuid
# Create your tests here.
class UserAuthTests(TestCase):
def setUp(self):
"""Initial Test Configuration."""
self.client_api = APIClient()
self.register_url = "/api/register/"
self.login_url = "/api/login/"
self.token_refresh_url = "/api/token/refresh/"
self.protected_url = "/api/protected/"
self.user_data = {
"email": "test@example.com",
"username": "TestUser",
"password": "testpass123"
}
def test_register_user(self):
"""Test to check if a user can register."""
response = self.client_api.post(self.register_url, self.user_data, format="json")
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertIn("user", response.data)
def test_login_user(self):
"""Login test and obtain JWT."""
# First, register a user
self.client_api.post(self.register_url, self.user_data, format='json')
# Now, try to Login
response = self.client_api.post(self.login_url, {
"email": self.user_data['email'],
"password": self.user_data['password']
}, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("access_token", response.data)
self.assertIn("refresh_token", response.data)
self.access_token = response.data['access_token']
self.refresh_token = response.data['refresh_token']
def test_access_protected_route(self):
"""Test to verify if a protected route demand authentication."""
self.test_login_user()
print(f"DEBUG: Access Token = {self.access_token}")
if not self.access_token:
self.fail("Access token was not generated correctly!")
self.client_api.credentials(HTTP_AUTHORIZATION=f"Bearer {self.access_token}")
response = self.client_api.get(self.protected_url) # Try to access a protected route
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_refresh_token(self):
"""Test to verify if the refresh token works."""
# Login to get the RefreshToken
self.test_login_user()
response = self.client_api.post(self.token_refresh_url, {
"refresh": self.refresh_token
}, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("access", response.data)
This is the error I am facing.
django.core.exceptions.FieldError: Cannot resolve keyword 'id_user' into field.
Choices are: date_joined, email, first_name, groups, id, is_active, is_staff, is_superuser, last_login, last_name, logentry, password, user_permissions, username
======================================================================
FAIL: test_access_protected_route