complete token items

OneSila/core/factories/multi_tenant.py

@@ -12,10 +12,14 @@ def __init__(self, token, info):
         self.info = info
         self.token = token
 
+    def set_token_instance(self):
+        self.token_instance = MultiTenantUserLoginToken.objects.get_by_token(self.token)
+
     def set_user(self):
-        self.user = MultiTenantUser.objects.get_by_token(self.token)
+        self.user = self.token_instance.multi_tenant_user
 
     def run(self):
+        self.set_token_instance()
         self.set_user()
 
 

OneSila/core/managers.py

@@ -2,10 +2,32 @@
 from django.db.models import Manager as DjangoManager
 from django.db import IntegrityError
 from django.db.models import QuerySet
+from django.core.exceptions import ValidationError
 
 from django.contrib.auth.models import BaseUserManager
 
 
+class MultiTenantUserLoginTokenQuerySet(DjangoQueryset):
+    def get_by_token(self, token):
+        try:
+            instance = self.get(token=token)
+        except self.model.DoesNotExist:
+            raise ValidationError('Unknown token')
+
+        if not instance.is_valid:
+            raise ValidationError('Token is no longer valid')
+
+        return instance
+
+
+class MultiTenantUserLoginTokenManager(DjangoManager):
+    def get_queryset(self):
+        return MultiTenantUserLoginTokenQuerySet(self.model, using=self._db)
+
+    def get_by_token(self, token):
+        return self.get_queryset().get_by_token(token)
+
+
 class MultiTenantQuerySet(DjangoQueryset):
     def filter(self, *, multi_tenant_company, **kwargs):
         return super().filter(multi_tenant_company=multi_tenant_company, **kwargs)

OneSila/core/models/multi_tenant.py

@@ -13,7 +13,7 @@
 
 from core.typing import LanguageType
 from core.helpers import get_languages
-from core.managers import MultiTenantManager
+from core.managers import MultiTenantManager, MultiTenantUserLoginTokenManager
 from core.validators import phone_regex, validate_image_extension, \
     no_dots_in_filename
 
@@ -161,6 +161,8 @@ class MultiTenantUserLoginToken(models.Model):
 
     multi_tenant_user = models.ForeignKey(MultiTenantUser, on_delete=models.CASCADE)
 
+    objects = MultiTenantUserLoginTokenManager()
+
     def save(self, *args, **kwargs):
         print(self.created_at)
         self.set_token()
@@ -172,5 +174,12 @@ def set_token(self):
             self.token = shake_256(shortuuid.uuid().encode('utf-8')).hexdigest(10)
 
     def set_expires_at(self, save=False):
+        # This strange construction is to avoid a second save signal.
         expires_at = self.created_at + timezone.timedelta(minutes=self.EXPIRES_AFTER_MIN)
         self.__class__.objects.filter(id=self.id).update(expires_at=expires_at)
+
+    def is_valid(self, now=timezone.now()):
+        if self.expires_at > now:
+            return False
+
+        return True

OneSila/core/schema/multi_tenant/mutations/fields.py

@@ -8,12 +8,11 @@
 from .mutation_classes import MyMultiTenantCompanyCreateMutation, \
     MyMultiTentantCompanyUpdateMutation, UpdateMeMutation, \
     InviteUserMutation, AcceptInvitationMutation, EnableUserMutation, \
-    DisableUserMutation, LoginTokenMutation, RecoveryTokenMutation, \
-    AuthenticateTokenMutation
+    DisableUserMutation, LoginTokenMutation, RecoveryTokenMutation
 
 import functools
 import strawberry
-from .resolvers import resolve_register_user
+from .resolvers import resolve_register_user, resolve_authenticate_token
 
 
 def register_my_multi_tenant_company():
@@ -61,9 +60,5 @@ def login_token():
     return RecoveryTokenMutation(MultiTenantLoginLinkInput, extensions=extensions)
 
 
-def authenticate_token():
-    extensions = []
-    return AuthenticateTokenMutation(MultiTenantUserAuthenticateTokenInput, extensions=extensions)
-
-
 register_user = functools.partial(strawberry.mutation, resolver=resolve_register_user)
+authenticate_token = functools.partial(strawberry.mutation, resolver=resolve_authenticate_token)

OneSila/core/schema/multi_tenant/mutations/mutation_classes.py

@@ -54,26 +54,6 @@ def create_token(self, *, user):
         return fac.token
 
 
-class AuthenticateTokenMutation(CleanupDataMixin, DjangoCreateMutation):
-    """
-    This takes the token supplied, and logs in the user
-    if the token exists and is still valid
-    """
-
-    def login(self):
-        user = self.user
-
-    def create(self, data: dict[str, Any], *, info: Info):
-        token = data.get('token')
-        fac = AuthenticateTokenFactory(token, info)
-        fac.run()
-
-        self.user = fac.user
-        self.login()
-
-        return fac.user
-
-
 class InviteUserMutation(CleanupDataMixin, GetMultiTenantCompanyMixin, DjangoCreateMutation):
     def create(self, data: dict[str, Any], *, info: Info):
         multi_tenant_company = self.get_multi_tenant_company(info)

OneSila/core/schema/multi_tenant/mutations/resolvers.py

@@ -1,22 +1,14 @@
 from strawberry_django.resolvers import django_resolver
 from asgiref.sync import async_to_sync
-from core.factories.multi_tenant import RegisterUserFactory
+from core.factories.multi_tenant import RegisterUserFactory, AuthenticateTokenFactory
 from strawberry_django.utils.requests import get_request
 from channels import auth as channels_auth
 from django.contrib import auth
 
 from .mutation_classes import CleanupDataMixin
 
 
-@django_resolver
-def resolve_register_user(info, username: str, password: str, language: str):
-    request = get_request(info)
-
-    fac = RegisterUserFactory(username=username, password=password, language=language)
-    fac.run()
-
-    user = fac.user
-
+def login(request, user):
     try:
         # We dont have any need for the classic django-auth outside of the
         # tests.  So this is why this try except clause exists.
@@ -34,4 +26,27 @@ def resolve_register_user(info, username: str, password: str, language: str):
             # When Django-channels is not installed,
             # this code will be non-existing
             pass
-    return user
+
+
+@django_resolver
+def resolve_register_user(info, username: str, password: str, language: str):
+    request = get_request(info)
+
+    fac = RegisterUserFactory(username=username, password=password, language=language)
+    fac.run()
+
+    login(request=request, user=fac.user)
+
+    return fac.user
+
+
+@django_resolver
+def resolve_authenticate_token(info, token: str):
+    request = get_request(info)
+
+    fac = AuthenticateTokenFactory(token, info)
+    fac.run()
+
+    login(request=request, user=fac.user)
+
+    return fac.user

OneSila/core/tests/tests_schemas/mutations.py

@@ -19,6 +19,14 @@
   }
 """
 
+AUTHENTICATE_TOKEN = """
+mutation($token: String!){
+  authenticateToken(token: $token){
+    username
+  }
+}
+"""
+
 
 LOGOUT_MUTATION = """
   mutation logout {

OneSila/core/tests/tests_schemas/tests_mutations.py

@@ -1,11 +1,72 @@
 from django.test import TestCase, TransactionTestCase
 from model_bakery import baker
 from strawberry.relay import to_base64
-from core.models.multi_tenant import MultiTenantUser, MultiTenantCompany
+from core.models.multi_tenant import MultiTenantUser, MultiTenantCompany, \
+    MultiTenantUserLoginToken
 from core.tests.tests_schemas.tests_queries import TransactionTestCaseMixin
 
 from .mutations import REGISTER_USER_MUTATION, LOGIN_MUTATION, LOGOUT_MUTATION, \
-    ME_QUERY
+    ME_QUERY, AUTHENTICATE_TOKEN
+
+
+class AuthenticateTokenTestCase(TransactionTestCaseMixin, TransactionTestCase):
+    def setUp(self):
+        self.multi_tenant_company = baker.make(MultiTenantCompany)
+        self.user = baker.make(MultiTenantUser, multi_tenant_company=self.multi_tenant_company)
+
+    def test_recovery_login(self):
+        mutations = """
+        mutation($username: String!){
+          recoveryToken(data:{username: $username}){
+            expiresAt
+          }
+        }
+        """
+
+        resp = self.stawberry_anonymous_test_client(
+            query=mutations,
+            variables={"username": self.user.username}
+        )
+
+        self.assertTrue(resp.errors is None)
+        self.assertTrue(resp.data is not None)
+
+        token = MultiTenantUserLoginToken.objects.filter(multi_tenant_user=self.user).last()
+
+        resp = self.stawberry_anonymous_test_client(
+            query=AUTHENTICATE_TOKEN,
+            variables={"token": token.token}
+        )
+
+        self.assertTrue(resp.errors is None)
+        self.assertTrue(resp.data is not None)
+
+    def test_token_login(self):
+        mutations = """
+        mutation($username: String!){
+          loginToken(data:{username: $username}){
+            expiresAt
+          }
+        }
+        """
+
+        resp = self.stawberry_anonymous_test_client(
+            query=mutations,
+            variables={"username": self.user.username}
+        )
+
+        self.assertTrue(resp.errors is None)
+        self.assertTrue(resp.data is not None)
+
+        token = MultiTenantUserLoginToken.objects.filter(multi_tenant_user=self.user).last()
+
+        resp = self.stawberry_anonymous_test_client(
+            query=AUTHENTICATE_TOKEN,
+            variables={"token": token.token}
+        )
+
+        self.assertTrue(resp.errors is None)
+        self.assertTrue(resp.data is not None)
 
 
 class AccountsTestCase(TransactionTestCaseMixin, TransactionTestCase):
@@ -43,8 +104,6 @@ def test_register_flow(self):
             variables={"username": username, "password": password, 'language': language}
         )
 
-        print(resp)
-
         self.assertTrue(resp.errors is None)
         self.assertTrue(resp.data is not None)