Egor fureunoir Gorbunov
943aa02cd3
Fixes: 1) Correct entrypoint scripts by removing redundant `python` reference in `uv run` commands; 2) Resolve incorrect imports and adjust class renaming in vibes_auth tests; 3) Address typing errors and minor omissions in existing code. Extra: 1) Improve formatting in settings and middleware files; 2) Update messaging test class names for clarity; 3) Cleanup unused imports and extra whitespaces, ensuring cleaner codebase.
87 lines
3.1 KiB
Python
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 MessagingTests(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"))
|