Compare commits

...

5 Commits

Author SHA1 Message Date
Jens Langhammer
6281fe2a2a dedupe helper
Signed-off-by: Jens Langhammer <jens@goauthentik.io>
2026-03-24 01:24:12 +01:00
Jens Langhammer
61df26ce57 cleanup?
Signed-off-by: Jens Langhammer <jens@goauthentik.io>
2026-03-24 01:21:09 +01:00
Jens Langhammer
92eba575a8 Merge branch 'main' into enterprise/simplify-tests 2026-03-24 01:20:02 +01:00
Jens Langhammer
f113664e77 more cleanup
Signed-off-by: Jens Langhammer <jens@goauthentik.io>
2026-01-14 00:21:18 +01:00
Jens Langhammer
833fc3a407 enterprise: simplify tests
Signed-off-by: Jens Langhammer <jens@goauthentik.io>
2026-01-14 00:02:48 +01:00
14 changed files with 102 additions and 295 deletions

View File

@@ -1,5 +1,4 @@
from json import loads
from unittest.mock import MagicMock, patch
from django.urls import reverse
from rest_framework.test import APITestCase
@@ -14,9 +13,7 @@ from authentik.endpoints.connectors.agent.models import (
EnrollmentToken,
)
from authentik.endpoints.models import Device
from authentik.enterprise.license import LicenseKey
from authentik.enterprise.models import License
from authentik.enterprise.tests.test_license import expiry_valid
from authentik.enterprise.tests import enterprise_test
from authentik.lib.generators import generate_id
@@ -36,21 +33,9 @@ class TestAppleRegister(APITestCase):
self.user = create_test_user()
self.device_token = DeviceToken.objects.create(device=self.connection)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
@reconcile_app("authentik_crypto")
def test_register_device(self):
License.objects.create(key=generate_id())
response = self.client.post(
reverse("authentik_api:psso-register-device"),
data={
@@ -74,21 +59,9 @@ class TestAppleRegister(APITestCase):
},
)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
@reconcile_app("authentik_crypto")
def test_register_user(self):
License.objects.create(key=generate_id())
device_auth = DeviceAuthenticationToken.objects.create(
device=self.device,
device_token=self.device_token,

View File

@@ -1,6 +1,5 @@
from hashlib import sha256
from json import loads
from unittest.mock import MagicMock, patch
from urllib.parse import parse_qs, urlparse
from django.urls import reverse
@@ -13,9 +12,7 @@ from authentik.endpoints.connectors.agent.api.connectors import AgentDeviceConne
from authentik.endpoints.connectors.agent.models import AgentConnector, DeviceToken, EnrollmentToken
from authentik.endpoints.models import Device, DeviceAccessGroup
from authentik.enterprise.endpoints.connectors.agent.views.auth_interactive import QS_AGENT_IA_TOKEN
from authentik.enterprise.license import LicenseKey
from authentik.enterprise.models import License
from authentik.enterprise.tests.test_license import expiry_valid
from authentik.enterprise.tests import enterprise_test
from authentik.flows.tests import FlowTestCase
from authentik.lib.generators import generate_id
from authentik.policies.models import PolicyBinding
@@ -43,41 +40,17 @@ class TestConnectorAuthIA(FlowTestCase):
)
self.user = create_test_user()
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
def test_auth_ia_initiate(self):
License.objects.create(key=generate_id())
response = self.client.post(
reverse("authentik_api:agentconnector-auth-ia"),
HTTP_AUTHORIZATION=f"Bearer+agent {self.device_token.key}",
)
self.assertEqual(response.status_code, 200)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
@reconcile_app("authentik_crypto")
def test_auth_ia_fulfill(self):
License.objects.create(key=generate_id())
self.client.force_login(self.user)
response = self.client.post(
reverse("authentik_api:agentconnector-auth-ia"),
@@ -94,21 +67,9 @@ class TestConnectorAuthIA(FlowTestCase):
self.assertEqual(response.status_code, 200)
self.assertIn(b"Permission denied", response.content)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
@reconcile_app("authentik_crypto")
def test_auth_ia_fulfill_policy(self):
License.objects.create(key=generate_id())
device_group = DeviceAccessGroup.objects.create(name=generate_id())
self.device.access_group = device_group
self.device.save()

View File

@@ -5,11 +5,11 @@ from rest_framework.test import APITestCase
from authentik.core.models import Application, Group
from authentik.core.tests.utils import create_test_admin_user, create_test_user
from authentik.enterprise.lifecycle.models import LifecycleIteration, LifecycleRule, ReviewState
from authentik.enterprise.reports.tests.utils import patch_license
from authentik.enterprise.tests import enterprise_test
from authentik.lib.generators import generate_id
@patch_license
@enterprise_test()
class TestLifecycleRuleAPI(APITestCase):
def setUp(self):
@@ -179,7 +179,7 @@ class TestLifecycleRuleAPI(APITestCase):
self.assertFalse(LifecycleRule.objects.filter(pk=rule.pk).exists())
@patch_license
@enterprise_test()
class TestIterationAPI(APITestCase):
def setUp(self):
@@ -268,7 +268,7 @@ class TestIterationAPI(APITestCase):
self.assertIn("user_can_review", response.data["results"][0])
@patch_license
@enterprise_test()
class TestReviewAPI(APITestCase):
def setUp(self):

View File

@@ -2,7 +2,7 @@
from base64 import b64encode
from datetime import timedelta
from unittest.mock import MagicMock, PropertyMock, patch
from unittest.mock import PropertyMock, patch
from django.urls import reverse
from django.utils.timezone import now
@@ -12,9 +12,7 @@ from rest_framework.test import APITestCase
from authentik.blueprints.tests import apply_blueprint
from authentik.core.models import Application, Group, User
from authentik.core.tests.utils import create_test_admin_user
from authentik.enterprise.license import LicenseKey
from authentik.enterprise.models import License
from authentik.enterprise.tests.test_license import expiry_valid
from authentik.enterprise.tests import enterprise_test
from authentik.lib.generators import generate_id
from authentik.providers.scim.models import SCIMAuthenticationMode, SCIMMapping, SCIMProvider
from authentik.sources.oauth.models import OAuthSource, UserOAuthSourceConnection
@@ -146,20 +144,8 @@ class SCIMOAuthTests(APITestCase):
},
)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
def test_api_create(self):
License.objects.create(key=generate_id())
self.client.force_login(create_test_admin_user())
res = self.client.post(
reverse("authentik_api:scimprovider-list"),

View File

@@ -4,11 +4,11 @@ from rest_framework.test import APITestCase
from authentik.core.models import User
from authentik.core.tests.utils import create_test_admin_user
from authentik.enterprise.reports.tests.utils import patch_license
from authentik.enterprise.tests import enterprise_test
from authentik.events.models import Event
@patch_license
@enterprise_test()
class TestExportAPI(APITestCase):
def setUp(self) -> None:
self.user = create_test_admin_user()

View File

@@ -3,11 +3,11 @@ from django.test.testcases import TestCase
from authentik.core.tests.utils import create_test_user
from authentik.enterprise.reports.models import DataExport
from authentik.enterprise.reports.tests.utils import patch_license
from authentik.enterprise.tests import enterprise_test
from authentik.events.models import Event, EventAction
@patch_license
@enterprise_test()
class TestEventExport(TestCase):
def setUp(self) -> None:
self.user = create_test_user()

View File

@@ -2,10 +2,10 @@ from django.urls import reverse
from rest_framework.test import APITestCase
from authentik.core.tests.utils import create_test_user
from authentik.enterprise.reports.tests.utils import patch_license
from authentik.enterprise.tests import enterprise_test
@patch_license
@enterprise_test()
class TestExportPermissions(APITestCase):
def setUp(self) -> None:
self.user = create_test_user()

View File

@@ -1,10 +1,10 @@
from django.test.testcases import TestCase
from drf_spectacular.generators import SchemaGenerator
from authentik.enterprise.reports.tests.utils import patch_license
from authentik.enterprise.tests import enterprise_test
@patch_license
@enterprise_test()
class TestSchemaMatch(TestCase):
def setUp(self) -> None:
generator = SchemaGenerator()

View File

@@ -7,10 +7,10 @@ from authentik.admin.files.tests.utils import FileTestFileBackendMixin
from authentik.core.models import User
from authentik.core.tests.utils import create_test_user
from authentik.enterprise.reports.models import DataExport
from authentik.enterprise.reports.tests.utils import patch_license
from authentik.enterprise.tests import enterprise_test
@patch_license
@enterprise_test()
class TestUserExport(FileTestFileBackendMixin, TestCase):
def setUp(self) -> None:
super().setUp()

View File

@@ -1,6 +0,0 @@
from unittest.mock import MagicMock, patch
patch_license = patch(
"authentik.enterprise.models.LicenseUsageStatus.is_valid",
MagicMock(return_value=True),
)

View File

@@ -0,0 +1,56 @@
from collections.abc import Callable
from datetime import timedelta
from functools import wraps
from time import mktime
from unittest.mock import MagicMock, patch
from django.utils.timezone import now
from authentik.enterprise.license import LicenseKey
from authentik.enterprise.models import THRESHOLD_READ_ONLY_WEEKS, License
from authentik.lib.generators import generate_id
# Valid license expiry
expiry_valid = int(mktime((now() + timedelta(days=3000)).timetuple()))
# Valid license expiry, expires soon
expiry_soon = int(mktime((now() + timedelta(hours=10)).timetuple()))
# Invalid license expiry, recently expired
expiry_expired = int(mktime((now() - timedelta(hours=10)).timetuple()))
# Invalid license expiry, expired longer ago
expiry_expired_read_only = int(
mktime((now() - timedelta(weeks=THRESHOLD_READ_ONLY_WEEKS + 1)).timetuple())
)
def enterprise_test(
expiry: int = expiry_valid,
internal_users: int = 100,
external_users: int = 100,
create_key=True,
):
"""Install testing enterprise license"""
def wrapper_outer(func: Callable):
@wraps(func)
def wrapper(*args, **kwargs):
with patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry,
name=generate_id(),
internal_users=internal_users,
external_users=external_users,
)
),
):
if create_key:
License.objects.all().delete()
License.objects.create(key=generate_id())
return func(*args, **kwargs)
return wrapper
return wrapper_outer

View File

@@ -1,7 +1,6 @@
"""Enterprise license tests"""
from datetime import timedelta
from time import mktime
from unittest.mock import MagicMock, patch
from django.test import TestCase
@@ -18,35 +17,20 @@ from authentik.enterprise.models import (
LicenseUsage,
LicenseUsageStatus,
)
from authentik.lib.generators import generate_id
# Valid license expiry
expiry_valid = int(mktime((now() + timedelta(days=3000)).timetuple()))
# Valid license expiry, expires soon
expiry_soon = int(mktime((now() + timedelta(hours=10)).timetuple()))
# Invalid license expiry, recently expired
expiry_expired = int(mktime((now() - timedelta(hours=10)).timetuple()))
# Invalid license expiry, expired longer ago
expiry_expired_read_only = int(
mktime((now() - timedelta(weeks=THRESHOLD_READ_ONLY_WEEKS + 1)).timetuple())
from authentik.enterprise.tests import (
enterprise_test,
expiry_expired,
expiry_expired_read_only,
expiry_soon,
expiry_valid,
)
from authentik.lib.generators import generate_id
class TestEnterpriseLicense(TestCase):
"""Enterprise license tests"""
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
def test_valid(self):
"""Check license verification"""
lic = License.objects.create(key=generate_id())
@@ -58,18 +42,7 @@ class TestEnterpriseLicense(TestCase):
with self.assertRaises(ValidationError):
License.objects.create(key=generate_id())
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test(create_key=False)
def test_valid_multiple(self):
"""Check license verification"""
lic = License.objects.create(key=generate_id(), expiry=expiry_valid)
@@ -82,18 +55,7 @@ class TestEnterpriseLicense(TestCase):
self.assertEqual(total.exp, expiry_valid)
self.assertTrue(total.status().is_valid)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
@patch(
"authentik.enterprise.license.LicenseKey.get_internal_user_count",
MagicMock(return_value=1000),
@@ -108,7 +70,6 @@ class TestEnterpriseLicense(TestCase):
)
def test_limit_exceeded_read_only(self):
"""Check license verification"""
License.objects.create(key=generate_id())
usage = LicenseUsage.objects.create(
internal_user_count=100,
external_user_count=100,
@@ -118,18 +79,7 @@ class TestEnterpriseLicense(TestCase):
usage.save(update_fields=["record_date"])
self.assertEqual(LicenseKey.get_total().summary().status, LicenseUsageStatus.READ_ONLY)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
@patch(
"authentik.enterprise.license.LicenseKey.get_internal_user_count",
MagicMock(return_value=1000),
@@ -144,7 +94,6 @@ class TestEnterpriseLicense(TestCase):
)
def test_limit_exceeded_user_warning(self):
"""Check license verification"""
License.objects.create(key=generate_id())
usage = LicenseUsage.objects.create(
internal_user_count=100,
external_user_count=100,
@@ -156,18 +105,7 @@ class TestEnterpriseLicense(TestCase):
LicenseKey.get_total().summary().status, LicenseUsageStatus.LIMIT_EXCEEDED_USER
)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
@patch(
"authentik.enterprise.license.LicenseKey.get_internal_user_count",
MagicMock(return_value=1000),
@@ -182,7 +120,6 @@ class TestEnterpriseLicense(TestCase):
)
def test_limit_exceeded_admin_warning(self):
"""Check license verification"""
License.objects.create(key=generate_id())
usage = LicenseUsage.objects.create(
internal_user_count=100,
external_user_count=100,
@@ -194,39 +131,16 @@ class TestEnterpriseLicense(TestCase):
LicenseKey.get_total().summary().status, LicenseUsageStatus.LIMIT_EXCEEDED_ADMIN
)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_expired_read_only,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test(expiry=expiry_expired_read_only)
@patch(
"authentik.enterprise.license.LicenseKey.record_usage",
MagicMock(),
)
def test_expiry_read_only(self):
"""Check license verification"""
License.objects.create(key=generate_id())
self.assertEqual(LicenseKey.get_total().summary().status, LicenseUsageStatus.READ_ONLY)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_expired,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test(expiry=expiry_expired)
@patch(
"authentik.enterprise.license.LicenseKey.record_usage",
MagicMock(),
@@ -238,23 +152,11 @@ class TestEnterpriseLicense(TestCase):
License.objects.create(key=generate_id(), expiry=expiry_expired)
self.assertEqual(LicenseKey.get_total().summary().status, LicenseUsageStatus.EXPIRED)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_soon,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test(expiry=expiry_soon)
@patch(
"authentik.enterprise.license.LicenseKey.record_usage",
MagicMock(),
)
def test_expiry_soon(self):
"""Check license verification"""
License.objects.create(key=generate_id())
self.assertEqual(LicenseKey.get_total().summary().status, LicenseUsageStatus.EXPIRY_SOON)

View File

@@ -1,37 +1,20 @@
"""Enterprise metrics tests"""
from unittest.mock import MagicMock, patch
from django.test import TestCase
from prometheus_client import REGISTRY
from authentik.core.models import User
from authentik.core.tests.utils import create_test_user
from authentik.enterprise.license import LicenseKey
from authentik.enterprise.models import License
from authentik.enterprise.tests.test_license import expiry_valid
from authentik.lib.generators import generate_id
from authentik.enterprise.tests import enterprise_test
from authentik.root.monitoring import monitoring_set
class TestEnterpriseMetrics(TestCase):
"""Enterprise metrics tests"""
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
def test_usage_empty(self):
"""Test usage (no users)"""
License.objects.create(key=generate_id())
User.objects.all().delete()
create_test_user()
monitoring_set.send_robust(self)

View File

@@ -7,14 +7,13 @@ from django.urls import reverse
from django.utils.timezone import now
from authentik.core.tests.utils import create_test_admin_user, create_test_flow, create_test_user
from authentik.enterprise.license import LicenseKey
from authentik.enterprise.models import (
THRESHOLD_READ_ONLY_WEEKS,
License,
LicenseUsage,
LicenseUsageStatus,
)
from authentik.enterprise.tests.test_license import expiry_valid
from authentik.enterprise.tests import enterprise_test
from authentik.flows.models import (
FlowDesignation,
FlowStageBinding,
@@ -30,18 +29,7 @@ from authentik.stages.user_login.models import UserLoginStage
class TestReadOnly(FlowTestCase):
"""Test read_only"""
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
@patch(
"authentik.enterprise.license.LicenseKey.get_internal_user_count",
MagicMock(return_value=1000),
@@ -56,7 +44,6 @@ class TestReadOnly(FlowTestCase):
)
def test_login(self):
"""Test flow, ensure login is still possible with read only mode"""
License.objects.create(key=generate_id())
usage = LicenseUsage.objects.create(
internal_user_count=100,
external_user_count=100,
@@ -115,18 +102,7 @@ class TestReadOnly(FlowTestCase):
response = self.client.post(exec_url, {"password": user.username}, follow=True)
self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test(create_key=False)
@patch(
"authentik.enterprise.license.LicenseKey.get_internal_user_count",
MagicMock(return_value=1000),
@@ -163,18 +139,7 @@ class TestReadOnly(FlowTestCase):
)
self.assertEqual(response.status_code, 200)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
@patch(
"authentik.enterprise.license.LicenseKey.get_internal_user_count",
MagicMock(return_value=1000),
@@ -189,7 +154,6 @@ class TestReadOnly(FlowTestCase):
)
def test_manage_flows(self):
"""Test flow"""
License.objects.create(key=generate_id())
usage = LicenseUsage.objects.create(
internal_user_count=100,
external_user_count=100,
@@ -216,18 +180,7 @@ class TestReadOnly(FlowTestCase):
)
self.assertEqual(response.status_code, 400)
@patch(
"authentik.enterprise.license.LicenseKey.validate",
MagicMock(
return_value=LicenseKey(
aud="",
exp=expiry_valid,
name=generate_id(),
internal_users=100,
external_users=100,
)
),
)
@enterprise_test()
@patch(
"authentik.enterprise.license.LicenseKey.get_internal_user_count",
MagicMock(return_value=1000),
@@ -242,7 +195,6 @@ class TestReadOnly(FlowTestCase):
)
def test_manage_users(self):
"""Test that managing users is still possible"""
License.objects.create(key=generate_id())
usage = LicenseUsage.objects.create(
internal_user_count=100,
external_user_count=100,