authentik.tenants.tests.test_recovery

 1"""Test Tenant Recovery API"""
 2
 3from json import loads
 4
 5from django.urls import reverse
 6from django_tenants.utils import get_public_schema_name
 7
 8from authentik.core.models import Token, TokenIntents
 9from authentik.core.tests.utils import create_test_user
10from authentik.lib.config import CONFIG
11from authentik.lib.generators import generate_id
12from authentik.tenants.models import Tenant
13from authentik.tenants.tests.utils import TenantAPITestCase
14
15TENANTS_API_KEY = generate_id()
16HEADERS = {"Authorization": f"Bearer {TENANTS_API_KEY}"}
17
18
19class TestRecovery(TenantAPITestCase):
20    """Test recovery endpoints"""
21
22    def setUp(self):
23        super().setUp()
24        self.tenant = Tenant.objects.get(schema_name=get_public_schema_name())
25        self.user = create_test_user()
26
27    @CONFIG.patch("outposts.disable_embedded_outpost", True)
28    @CONFIG.patch("tenants.enabled", True)
29    @CONFIG.patch("tenants.api_key", TENANTS_API_KEY)
30    def test_recovery_admin_group(self):
31        """Test creation of admin group"""
32        response = self.client.post(
33            reverse(
34                "authentik_api:tenant-create-admin-group",
35                args=[self.tenant.tenant_uuid],
36            ),
37            data={"user": self.user.username},
38            headers=HEADERS,
39        )
40        self.assertEqual(response.status_code, 200)
41        self.assertTrue(self.user.is_superuser)
42
43    @CONFIG.patch("outposts.disable_embedded_outpost", True)
44    @CONFIG.patch("tenants.enabled", True)
45    @CONFIG.patch("tenants.api_key", TENANTS_API_KEY)
46    def test_recovery_admin_group_invalid(self):
47        """Test invalid creation of admin group"""
48        response = self.client.post(
49            reverse(
50                "authentik_api:tenant-create-admin-group",
51                args=[self.tenant.tenant_uuid],
52            ),
53            data={"user": "invalid"},
54            headers=HEADERS,
55        )
56        self.assertEqual(response.status_code, 404)
57
58    @CONFIG.patch("outposts.disable_embedded_outpost", True)
59    @CONFIG.patch("tenants.enabled", True)
60    @CONFIG.patch("tenants.api_key", TENANTS_API_KEY)
61    def test_create_key(self):
62        """Test creation of a new key"""
63        response = self.client.post(
64            reverse(
65                "authentik_api:tenant-create-recovery-key",
66                args=[self.tenant.tenant_uuid],
67            ),
68            data={"user": self.user.username, "duration_days": 365},
69            headers=HEADERS,
70        )
71        self.assertEqual(response.status_code, 200)
72        body = loads(response.content.decode())
73        token = Token.objects.get(intent=TokenIntents.INTENT_RECOVERY, user=self.user)
74        self.assertIn(token.key, body["url"])
75        self.assertEqual(len(Token.objects.filter(intent=TokenIntents.INTENT_RECOVERY)), 1)
76
77    @CONFIG.patch("outposts.disable_embedded_outpost", True)
78    @CONFIG.patch("tenants.enabled", True)
79    @CONFIG.patch("tenants.api_key", TENANTS_API_KEY)
80    def test_create_key_invalid(self):
81        """Test creation of a new key (invalid)"""
82        response = self.client.post(
83            reverse(
84                "authentik_api:tenant-create-recovery-key",
85                args=[self.tenant.tenant_uuid],
86            ),
87            data={"user": "invalid", "duration_days": 365},
88            headers=HEADERS,
89        )
90        self.assertEqual(response.status_code, 404)
91        self.assertEqual(len(Token.objects.filter(intent=TokenIntents.INTENT_RECOVERY)), 0)