authentik.admin.tests.test_tasks

 1"""test admin tasks"""
 2
 3from django.apps import apps
 4from django.core.cache import cache
 5from django.test import TestCase
 6from requests_mock import Mocker
 7
 8from authentik.admin.tasks import (
 9    VERSION_CACHE_KEY,
10    update_latest_version,
11)
12from authentik.events.models import Event, EventAction
13from authentik.lib.config import CONFIG
14
15RESPONSE_VALID = {
16    "$schema": "https://version.goauthentik.io/schema.json",
17    "stable": {
18        "version": "99999999.9999999",
19        "changelog": "See https://goauthentik.io/test",
20        "changelog_url": "https://goauthentik.io/test",
21        "reason": "bugfix",
22    },
23}
24
25
26class TestAdminTasks(TestCase):
27    """test admin tasks"""
28
29    def test_version_valid_response(self):
30        """Test Update checker with valid response"""
31        with Mocker() as mocker, CONFIG.patch("disable_update_check", False):
32            mocker.get("https://version.goauthentik.io/version.json", json=RESPONSE_VALID)
33            update_latest_version.send()
34            self.assertEqual(cache.get(VERSION_CACHE_KEY), "99999999.9999999")
35            self.assertTrue(
36                Event.objects.filter(
37                    action=EventAction.UPDATE_AVAILABLE,
38                    context__new_version="99999999.9999999",
39                    context__message="New version 99999999.9999999 available!",
40                ).exists()
41            )
42            # test that a consecutive check doesn't create a duplicate event
43            update_latest_version.send()
44            self.assertEqual(
45                len(
46                    Event.objects.filter(
47                        action=EventAction.UPDATE_AVAILABLE,
48                        context__new_version="99999999.9999999",
49                        context__message="New version 99999999.9999999 available!",
50                    )
51                ),
52                1,
53            )
54
55    def test_version_error(self):
56        """Test Update checker with invalid response"""
57        with Mocker() as mocker:
58            mocker.get("https://version.goauthentik.io/version.json", status_code=400)
59            update_latest_version.send()
60            self.assertEqual(cache.get(VERSION_CACHE_KEY), "0.0.0")
61            self.assertFalse(
62                Event.objects.filter(
63                    action=EventAction.UPDATE_AVAILABLE, context__new_version="0.0.0"
64                ).exists()
65            )
66
67    def test_version_disabled(self):
68        """Test Update checker while its disabled"""
69        with CONFIG.patch("disable_update_check", True):
70            update_latest_version.send()
71            self.assertEqual(cache.get(VERSION_CACHE_KEY), "0.0.0")
72
73    def test_clear_update_notifications(self):
74        """Test clear of previous notification"""
75        admin_config = apps.get_app_config("authentik_admin")
76        Event.objects.create(
77            action=EventAction.UPDATE_AVAILABLE,
78            context={"new_version": "99999999.9999999.9999999"},
79        )
80        Event.objects.create(action=EventAction.UPDATE_AVAILABLE, context={"new_version": "1.1.1"})
81        Event.objects.create(action=EventAction.UPDATE_AVAILABLE, context={})
82        admin_config.clear_update_notifications()
83        self.assertFalse(
84            Event.objects.filter(
85                action=EventAction.UPDATE_AVAILABLE, context__new_version="1.1"
86            ).exists()
87        )