added a test unit to the token-service

Backend/token_service/Dockerfile

@@ -8,7 +8,7 @@ ENV LANG=C.UTF-8
 RUN apk update && apk add --no-cache python3 py3-pip \
 postgresql16 postgresql16-client \
 bash supervisor curl openssl bash \
-build-base libffi-dev python3-dev
+build-base libffi-dev python3-dev nano
 # Set work directory
 RUN mkdir /run/postgresql && \
     chown postgres:postgres /run/postgresql && \

Backend/token_service/requirements.txt

@@ -22,4 +22,5 @@ django-environ
 daphne
 twisted[http2,tls]
 pytest
-pytest-django
+pytest-django
+pytest-mock

Backend/token_service/token_service/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+DJANGO_SETTINGS_MODULE = token_service.settings
+python_files = tests.py test_*.py *_tests.py

Backend/token_service/token_service/token_app/views.py

@@ -24,8 +24,6 @@ class CustomTokenObtainPairView(TokenObtainPairView):
     """
     serializer_class = CustomTokenObtainPairSerializer
 
-    @staticmethod
-    @method_decorator(csrf_exempt)
     def handle_token_request(ch, method, properties, body):
         """
             Method to handle the token request.

Backend/token_service/token_service/token_service/test/conftest.py

@@ -0,0 +1,102 @@
+import os
+import django
+from django.conf import settings
+import pytest
+from datetime import timedelta
+
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'token_service.settings')
+
+if not settings.configured:
+    settings.configure(
+        DEBUG=True,
+        DATABASES={
+            "default": {
+                "ENGINE": "django.db.backends.postgresql",
+                "NAME": "postgres",
+                "USER": "root",
+                "PASSWORD": "root",
+                "PORT": "5432",
+                "HOST": "localhost",
+                "ATOMIC_REQUESTS": True,
+            }
+        },
+        INSTALLED_APPS=[
+            'django.contrib.admin',
+            'django.contrib.auth',
+            'django.contrib.contenttypes',
+            'django.contrib.sessions',
+            'django.contrib.messages',
+            'django.contrib.staticfiles',
+            'rest_framework',
+            'corsheaders',
+            'token_app',
+        ],
+        MIDDLEWARE=[
+            'django.middleware.security.SecurityMiddleware',
+            'django.contrib.sessions.middleware.SessionMiddleware',
+            'django.middleware.common.CommonMiddleware',
+            'django.middleware.csrf.CsrfViewMiddleware',
+            'django.contrib.auth.middleware.AuthenticationMiddleware',
+            'django.contrib.messages.middleware.MessageMiddleware',
+            'django.middleware.clickjacking.XFrameOptionsMiddleware',
+        ],
+        ROOT_URLCONF='token_service.urls',
+        TEMPLATES=[
+            {
+                'BACKEND': 'django.template.backends.django.DjangoTemplates',
+                'DIRS': [],
+                '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='token_service.wsgi.application',
+        SECRET_KEY='django-insecure-woftd2en2**zr(b%#*2vit2v%s@(k54gb^c(ots0abo7(wsmo%',
+        ALLOWED_HOSTS=['localhost', '127.0.0.1', '[::1]', 'user-session', 'user-session:8004'],
+        RABBITMQ_HOST='localhost',
+        RABBITMQ_USER='user',
+        RABBITMQ_PASS='pass',
+        RABBITMQ_PORT='5672',
+        REST_FRAMEWORK={
+            'DEFAULT_AUTHENTICATION_CLASSES': (
+                'rest_framework_simplejwt.authentication.JWTAuthentication',
+            ),
+        },
+        SIMPLE_JWT={
+            'ACCESS_TOKEN_LIFETIME': timedelta(minutes=60),
+            'REFRESH_TOKEN_LIFETIME': timedelta(days=1),
+            'ROTATE_REFRESH_TOKENS': False,
+            'BLACKLIST_AFTER_ROTATION': True,
+            'AUTH_HEADER_TYPES': ('Bearer',),
+            'AUTH_TOKEN_CLASSES': ('rest_framework_simplejwt.tokens.AccessToken',),
+            'TOKEN_OBTAIN_SERIALIZER': 'user_auth.serializers.CustomTokenObtainPairSerializer',
+        },
+        LANGUAGE_CODE='en-us',
+        TIME_ZONE='UTC',
+        USE_I18N=True,
+        USE_L10N=True,
+        USE_TZ=True,
+        STATIC_URL='/static/',
+        DEFAULT_AUTO_FIELD='django.db.models.BigAutoField',
+        CORS_ORIGIN_ALLOW_ALL=True,
+    )
+
+django.setup()
+
+@pytest.fixture(scope='session', autouse=True)
+def django_db_setup():
+    settings.DATABASES['default'] = {
+        'ENGINE': 'django.db.backends.postgresql',
+        'NAME': 'postgres',
+        'USER': 'postgres',
+        'PASSWORD': 'postgres',
+        'HOST': 'localhost',
+        'PORT': '5432',
+        'ATOMIC_REQUESTS': True,
+    }

Backend/token_service/token_service/token_service/test/tests.py

@@ -0,0 +1,65 @@
+import pytest
+import json
+from unittest.mock import patch, MagicMock
+from rest_framework_simplejwt.tokens import RefreshToken
+from token_app.views import CustomTokenObtainPairView
+
+@pytest.fixture
+def api_client():
+    from rest_framework.test import APIClient
+    return APIClient()
+
+@pytest.fixture
+def user_data():
+    return {
+        'username': 'testuser',
+    }
+
+@pytest.fixture
+def request_factory(): 
+    """
+    This is a fixture that returns a RequestFactory object.
+    This is a Django object that allows you to create mock requests.
+    This is useful for testing views.
+    """
+    from django.test import RequestFactory
+    return RequestFactory()
+
+def test_handle_token_request(mocker, user_data):
+    # Mock the publish_message function
+    publish_message_mock = mocker.patch('token_app.views.publish_message')
+
+    # Prepare the message body
+    message_body = json.dumps(user_data)
+
+    # Mock RefreshToken to return a fixed value
+    with patch('rest_framework_simplejwt.tokens.RefreshToken.for_user') as mock_for_user:
+        mock_refresh_token = MagicMock()
+        mock_refresh_token.__str__.return_value = 'fixed-refresh-token'
+        mock_refresh_token.access_token = 'fixed-access-token'
+        mock_for_user.return_value = mock_refresh_token
+
+        # Call the handle_token_request method directly
+        CustomTokenObtainPairView.handle_token_request(None, None, None, message_body)
+
+    # Assert the publish_message was called with correct arguments
+    expected_refresh_token = 'fixed-refresh-token'
+    expected_access_token = 'fixed-access-token'
+
+    publish_message_mock.assert_called_once_with(
+        "user_token_response_queue",
+        json.dumps({"refresh": expected_refresh_token, "access": expected_access_token})
+    )
+
+def test_start_consumer(mocker):
+    # Mock the consume_message function
+    consume_message_mock = mocker.patch('token_app.views.consume_message')
+
+    # Create a test instance of the view
+    view = CustomTokenObtainPairView()
+
+    # Call the start_consumer method
+    view.start_consumer()
+
+    # Assert the consume_message was called with correct arguments
+    consume_message_mock.assert_called_once_with("user_token_request_queue", view.handle_token_request)