authentik.core.tests.test_tasks

Test tasks

 1"""Test tasks"""
 2
 3from time import mktime
 4
 5from django.utils.timezone import now
 6from guardian.shortcuts import get_anonymous_user
 7from rest_framework.test import APITestCase
 8
 9from authentik.core.models import (
10    USER_ATTRIBUTE_EXPIRES,
11    USER_ATTRIBUTE_GENERATED,
12    Token,
13    TokenIntents,
14    User,
15)
16from authentik.core.tasks import (
17    clean_expired_models,
18    clean_temporary_users,
19)
20from authentik.core.tests.utils import create_test_admin_user
21from authentik.lib.generators import generate_id
22
23
24class TestTasks(APITestCase):
25    """Test token API"""
26
27    def setUp(self) -> None:
28        super().setUp()
29        self.user = User.objects.create(username="testuser")
30        self.admin = create_test_admin_user()
31        self.client.force_login(self.user)
32
33    def test_token_expire(self):
34        """Test Token expire task"""
35        token: Token = Token.objects.create(
36            expires=now(), user=get_anonymous_user(), intent=TokenIntents.INTENT_API
37        )
38        key = token.key
39        clean_expired_models.send()
40        token.refresh_from_db()
41        self.assertNotEqual(key, token.key)
42
43    def test_clean_temporary_users(self):
44        """Test clean_temporary_users task"""
45        username = generate_id
46        User.objects.create(
47            username=username,
48            attributes={
49                USER_ATTRIBUTE_GENERATED: True,
50                USER_ATTRIBUTE_EXPIRES: mktime(now().timetuple()),
51            },
52        )
53        clean_temporary_users.send()
54        self.assertFalse(User.objects.filter(username=username))
class TestTasks(rest_framework.test.APITestCase): 
25class TestTasks(APITestCase):
26    """Test token API"""
27
28    def setUp(self) -> None:
29        super().setUp()
30        self.user = User.objects.create(username="testuser")
31        self.admin = create_test_admin_user()
32        self.client.force_login(self.user)
33
34    def test_token_expire(self):
35        """Test Token expire task"""
36        token: Token = Token.objects.create(
37            expires=now(), user=get_anonymous_user(), intent=TokenIntents.INTENT_API
38        )
39        key = token.key
40        clean_expired_models.send()
41        token.refresh_from_db()
42        self.assertNotEqual(key, token.key)
43
44    def test_clean_temporary_users(self):
45        """Test clean_temporary_users task"""
46        username = generate_id
47        User.objects.create(
48            username=username,
49            attributes={
50                USER_ATTRIBUTE_GENERATED: True,
51                USER_ATTRIBUTE_EXPIRES: mktime(now().timetuple()),
52            },
53        )
54        clean_temporary_users.send()
55        self.assertFalse(User.objects.filter(username=username))

Test token API

def setUp(self) -> None: 
28    def setUp(self) -> None:
29        super().setUp()
30        self.user = User.objects.create(username="testuser")
31        self.admin = create_test_admin_user()
32        self.client.force_login(self.user)

Hook method for setting up the test fixture before exercising it.

def test_token_expire(self): 
34    def test_token_expire(self):
35        """Test Token expire task"""
36        token: Token = Token.objects.create(
37            expires=now(), user=get_anonymous_user(), intent=TokenIntents.INTENT_API
38        )
39        key = token.key
40        clean_expired_models.send()
41        token.refresh_from_db()
42        self.assertNotEqual(key, token.key)

Test Token expire task

def test_clean_temporary_users(self): 
44    def test_clean_temporary_users(self):
45        """Test clean_temporary_users task"""
46        username = generate_id
47        User.objects.create(
48            username=username,
49            attributes={
50                USER_ATTRIBUTE_GENERATED: True,
51                USER_ATTRIBUTE_EXPIRES: mktime(now().timetuple()),
52            },
53        )
54        clean_temporary_users.send()
55        self.assertFalse(User.objects.filter(username=username))

Test clean_temporary_users task