Coverage for chats\middleware.py: 100.0%

1from urllib.parse import parse_qs 

2 

3from channels.middleware import BaseMiddleware 

4from channels.db import database_sync_to_async 

5from rest_framework_simplejwt.tokens import AccessToken 

6from rest_framework_simplejwt.exceptions import TokenError 

7from rest_framework_simplejwt.settings import api_settings 

8 

9from users.models import Users 

10from users.services.auth_service import AuthService 

11 

12# Funcion asincrona para obtener el usaurio a partir de su id 

13@database_sync_to_async 

14def get_user(user_id, token=None): 

15 try: 

16 user = Users.objects.get(id=user_id) 

17 except Users.DoesNotExist: 

18 return None 

19 

20 if not user.is_authenticated: 

21 return None 

22 

23 if token is not None and AuthService.token_is_revoked(user, token): 

24 return None 

25 

26 return user 

27 

28# Middleware de autenticacion para los WebSocket 

29class JWTAuthMiddleware(BaseMiddleware): 

30 

31 async def __call__(self, scope, receive, send): 

32 query_string = scope.get('query_string', b'').decode() 

33 token = parse_qs(query_string).get('token', [None])[0] 

34 

35 scope['user'] = None 

36 

37 if token: 

38 try: 

39 access_token = AccessToken(token) 

40 user_id = access_token[api_settings.USER_ID_CLAIM] 

41 scope['user'] = await get_user(user_id, access_token) 

42 except (TokenError, KeyError): 

43 scope['user'] = None 

44 

45 return await super().__call__(scope, receive, send)