admin: add more tests

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>
This commit is contained in:
Jens Langhammer 2021-11-15 20:05:13 +01:00
parent d5341c2284
commit e4095dfffe
4 changed files with 30 additions and 3 deletions

View file

@ -36,7 +36,7 @@ class TaskSerializer(PassiveSerializer):
are pickled in cache. In that case, just delete the info"""
try:
return super().to_representation(instance)
except AttributeError:
except AttributeError: # pragma: no cover
if isinstance(self.instance, list):
for inst in self.instance:
inst.delete()

View file

@ -23,6 +23,6 @@ class WorkerView(APIView):
"""Get currently connected worker count."""
count = len(CELERY_APP.control.ping(timeout=0.5))
# In debug we run with `CELERY_TASK_ALWAYS_EAGER`, so tasks are ran on the main process
if settings.DEBUG:
if settings.DEBUG: # pragma: no cover
count += 1
return Response({"count": count})

View file

@ -8,6 +8,7 @@ from authentik import __version__
from authentik.core.models import Group, User
from authentik.core.tasks import clean_expired_models
from authentik.events.monitored_tasks import TaskResultStatus
from authentik.managed.tasks import managed_reconcile
class TestAdminAPI(TestCase):
@ -94,5 +95,7 @@ class TestAdminAPI(TestCase):
def test_system(self):
"""Test system API"""
# pyright: reportGeneralTypeIssues=false
managed_reconcile() # pylint: disable=no-value-for-parameter
response = self.client.get(reverse("authentik_api:admin_system"))
self.assertEqual(response.status_code, 200)

View file

@ -3,8 +3,13 @@ from django.core.cache import cache
from django.test import TestCase
from requests_mock import Mocker
from authentik.admin.tasks import VERSION_CACHE_KEY, update_latest_version
from authentik.admin.tasks import (
VERSION_CACHE_KEY,
clear_update_notifications,
update_latest_version,
)
from authentik.events.models import Event, EventAction
from authentik.lib.config import CONFIG
RESPONSE_VALID = {
"$schema": "https://version.goauthentik.io/schema.json",
@ -56,3 +61,22 @@ class TestAdminTasks(TestCase):
action=EventAction.UPDATE_AVAILABLE, context__new_version="0.0.0"
).exists()
)
def test_version_disabled(self):
"""Test Update checker while its disabled"""
with CONFIG.patch("disable_update_check", True):
update_latest_version.delay().get()
self.assertEqual(cache.get(VERSION_CACHE_KEY), "0.0.0")
def test_clear_update_notifications(self):
"""Test clear of previous notification"""
Event.objects.create(
action=EventAction.UPDATE_AVAILABLE, context={"new_version": "99999999.9999999.9999999"}
)
Event.objects.create(action=EventAction.UPDATE_AVAILABLE, context={"new_version": "1.1.1"})
clear_update_notifications()
self.assertFalse(
Event.objects.filter(
action=EventAction.UPDATE_AVAILABLE, context__new_version="1.1"
).exists()
)