schon/engine/vibes_auth/tests/test_messaging.py
Egor fureunoir Gorbunov 3228a89d4b Features:
1) Add new `test_graphene` test module for expanded testing coverage;
2) Introduce `test_drf` module in `engine/blog` for improved API testing;

Fixes:
1) Remove unnecessary `--extra testing` flag from Dockerfile to streamline dependencies;
2) Update `uv.lock` with newer versions of dependencies (`certifi`, `coverage`, `django-constance`) for enhanced security and functionality;

Extra:
1) Remove deprecated packages (`bandit`, `cfgv`, `distlib`) from `uv.lock` for cleanup;
2) Adjust `uv.lock` content and formatting to be consistent with updated dependencies.
2025-11-13 15:24:44 +03:00

87 lines
3.1 KiB
Python

import asyncio
from unittest.mock import patch
from django.contrib.auth.models import AnonymousUser
from django.test import TestCase
from engine.vibes_auth.models import User
from engine.vibes_auth.messaging import auth as auth_module
class MessagingAuthTests(TestCase):
def test_extract_jwt_from_subprotocols_cases(self):
fn = auth_module._extract_jwt_from_subprotocols
self.assertIsNone(fn(None))
self.assertIsNone(fn([]))
self.assertEqual(fn(["bearer", "abc.token"]), "abc.token")
self.assertEqual(fn(["Bearer", "abc"]), "abc")
self.assertEqual(fn(["single-token"]), "single-token")
self.assertIsNone(fn([""]))
self.assertIsNone(fn(["Bearer", ""]))
def test_jwt_middleware_sets_anonymous_without_token(self):
captured = {}
async def inner_app(scope_dict, _receive, _send):
captured["is_anon"] = isinstance(scope_dict["user"], AnonymousUser) or scope_dict["user"].is_anonymous
middleware = auth_module.JWTAuthMiddleware(inner_app)
scope = {"type": "websocket", "subprotocols": []}
async def dummy_receive():
return {"type": "websocket.disconnect"}
async def dummy_send(_message):
return None
asyncio.run(middleware(scope, dummy_receive, dummy_send))
self.assertTrue(captured.get("is_anon"))
def test_jwt_middleware_sets_user_with_valid_token(self):
user = User.objects.create_user(email="user@example.com", password="Str0ngPass!word")
class FakeAuth:
def authenticate(self, _request):
return user, "token"
captured = {}
async def inner_app(scope_dict, _receive, _send):
captured["user_id"] = getattr(scope_dict["user"], "pk", None)
middleware = auth_module.JWTAuthMiddleware(inner_app)
scope = {"type": "websocket", "subprotocols": ["bearer", "abc.def"]}
async def dummy_receive():
return {"type": "websocket.disconnect"}
async def dummy_send(_message):
return None
with patch.object(auth_module, "JWTAuthentication", FakeAuth):
asyncio.run(middleware(scope, dummy_receive, dummy_send))
self.assertEqual(captured.get("user_id"), user.pk)
def test_jwt_middleware_handles_bad_token_gracefully(self):
class FakeAuth:
def authenticate(self, _request):
raise Exception("bad token")
captured = {}
async def inner_app(scope_dict, _receive, _send):
captured["is_anon"] = isinstance(scope_dict["user"], AnonymousUser) or scope_dict["user"].is_anonymous
middleware = auth_module.JWTAuthMiddleware(inner_app)
scope = {"type": "websocket", "subprotocols": ["bearer", "bad.token"]}
async def dummy_receive():
return {"type": "websocket.disconnect"}
async def dummy_send(_message):
return None
with patch.object(auth_module, "JWTAuthentication", FakeAuth):
asyncio.run(middleware(scope, dummy_receive, dummy_send))
self.assertTrue(captured.get("is_anon"))