authentik.outposts.tests.test_api

Test outpost service connection API

  1"""Test outpost service connection API"""
  2
  3from django.urls import reverse
  4from rest_framework.test import APITestCase
  5
  6from authentik.blueprints.tests import reconcile_app
  7from authentik.core.models import PropertyMapping
  8from authentik.core.tests.utils import create_test_admin_user, create_test_flow
  9from authentik.lib.generators import generate_id
 10from authentik.outposts.api.outposts import OutpostSerializer
 11from authentik.outposts.apps import MANAGED_OUTPOST
 12from authentik.outposts.models import Outpost, OutpostType, default_outpost_config
 13from authentik.providers.ldap.models import LDAPProvider
 14from authentik.providers.proxy.models import ProxyProvider
 15
 16
 17class TestOutpostServiceConnectionsAPI(APITestCase):
 18    """Test outpost service connection API"""
 19
 20    def setUp(self) -> None:
 21        super().setUp()
 22        self.mapping = PropertyMapping.objects.create(
 23            name=generate_id(), expression="""return {'foo': 'bar'}"""
 24        )
 25        self.user = create_test_admin_user()
 26        self.client.force_login(self.user)
 27
 28    @reconcile_app("authentik_outposts")
 29    def test_managed_name_change(self):
 30        """Test name change for embedded outpost"""
 31        embedded_outpost = Outpost.objects.filter(managed=MANAGED_OUTPOST).first()
 32        self.assertIsNotNone(embedded_outpost)
 33        response = self.client.patch(
 34            reverse("authentik_api:outpost-detail", kwargs={"pk": embedded_outpost.pk}),
 35            {"name": "foo"},
 36        )
 37        self.assertEqual(response.status_code, 400)
 38        self.assertJSONEqual(
 39            response.content, {"name": ["Embedded outpost's name cannot be changed"]}
 40        )
 41
 42    @reconcile_app("authentik_outposts")
 43    def test_managed_without_managed(self):
 44        """Test name change for embedded outpost"""
 45        embedded_outpost = Outpost.objects.filter(managed=MANAGED_OUTPOST).first()
 46        self.assertIsNotNone(embedded_outpost)
 47        embedded_outpost.managed = ""
 48        embedded_outpost.save()
 49        response = self.client.patch(
 50            reverse("authentik_api:outpost-detail", kwargs={"pk": embedded_outpost.pk}),
 51            {"name": "foo"},
 52        )
 53        self.assertEqual(response.status_code, 200)
 54        embedded_outpost.refresh_from_db()
 55        self.assertEqual(embedded_outpost.managed, MANAGED_OUTPOST)
 56
 57    def test_outpost_validation(self):
 58        """Test Outpost validation"""
 59        valid = OutpostSerializer(
 60            data={
 61                "name": generate_id(),
 62                "type": OutpostType.PROXY,
 63                "config": default_outpost_config(),
 64                "providers": [
 65                    ProxyProvider.objects.create(
 66                        name=generate_id(), authorization_flow=create_test_flow()
 67                    ).pk
 68                ],
 69            }
 70        )
 71        self.assertTrue(valid.is_valid())
 72        invalid = OutpostSerializer(
 73            data={
 74                "name": generate_id(),
 75                "type": OutpostType.PROXY,
 76                "config": default_outpost_config(),
 77                "providers": [
 78                    LDAPProvider.objects.create(
 79                        name=generate_id(), authorization_flow=create_test_flow()
 80                    ).pk
 81                ],
 82            }
 83        )
 84        self.assertFalse(invalid.is_valid())
 85        self.assertIn("providers", invalid.errors)
 86
 87    def test_types(self):
 88        """Test OutpostServiceConnections's types endpoint"""
 89        response = self.client.get(
 90            reverse("authentik_api:outpostserviceconnection-types"),
 91        )
 92        self.assertEqual(response.status_code, 200)
 93
 94    def test_outpost_config(self):
 95        """Test Outpost's config field"""
 96        provider = ProxyProvider.objects.create(
 97            name=generate_id(), authorization_flow=create_test_flow()
 98        )
 99        invalid = OutpostSerializer(
100            data={"name": generate_id(), "providers": [provider.pk], "config": ""}
101        )
102        self.assertFalse(invalid.is_valid())
103        self.assertIn("config", invalid.errors)
104        valid = OutpostSerializer(
105            data={
106                "name": generate_id(),
107                "providers": [provider.pk],
108                "config": default_outpost_config(generate_id()),
109                "type": OutpostType.PROXY,
110            }
111        )
112        self.assertTrue(valid.is_valid())
class TestOutpostServiceConnectionsAPI(rest_framework.test.APITestCase): 
 18class TestOutpostServiceConnectionsAPI(APITestCase):
 19    """Test outpost service connection API"""
 20
 21    def setUp(self) -> None:
 22        super().setUp()
 23        self.mapping = PropertyMapping.objects.create(
 24            name=generate_id(), expression="""return {'foo': 'bar'}"""
 25        )
 26        self.user = create_test_admin_user()
 27        self.client.force_login(self.user)
 28
 29    @reconcile_app("authentik_outposts")
 30    def test_managed_name_change(self):
 31        """Test name change for embedded outpost"""
 32        embedded_outpost = Outpost.objects.filter(managed=MANAGED_OUTPOST).first()
 33        self.assertIsNotNone(embedded_outpost)
 34        response = self.client.patch(
 35            reverse("authentik_api:outpost-detail", kwargs={"pk": embedded_outpost.pk}),
 36            {"name": "foo"},
 37        )
 38        self.assertEqual(response.status_code, 400)
 39        self.assertJSONEqual(
 40            response.content, {"name": ["Embedded outpost's name cannot be changed"]}
 41        )
 42
 43    @reconcile_app("authentik_outposts")
 44    def test_managed_without_managed(self):
 45        """Test name change for embedded outpost"""
 46        embedded_outpost = Outpost.objects.filter(managed=MANAGED_OUTPOST).first()
 47        self.assertIsNotNone(embedded_outpost)
 48        embedded_outpost.managed = ""
 49        embedded_outpost.save()
 50        response = self.client.patch(
 51            reverse("authentik_api:outpost-detail", kwargs={"pk": embedded_outpost.pk}),
 52            {"name": "foo"},
 53        )
 54        self.assertEqual(response.status_code, 200)
 55        embedded_outpost.refresh_from_db()
 56        self.assertEqual(embedded_outpost.managed, MANAGED_OUTPOST)
 57
 58    def test_outpost_validation(self):
 59        """Test Outpost validation"""
 60        valid = OutpostSerializer(
 61            data={
 62                "name": generate_id(),
 63                "type": OutpostType.PROXY,
 64                "config": default_outpost_config(),
 65                "providers": [
 66                    ProxyProvider.objects.create(
 67                        name=generate_id(), authorization_flow=create_test_flow()
 68                    ).pk
 69                ],
 70            }
 71        )
 72        self.assertTrue(valid.is_valid())
 73        invalid = OutpostSerializer(
 74            data={
 75                "name": generate_id(),
 76                "type": OutpostType.PROXY,
 77                "config": default_outpost_config(),
 78                "providers": [
 79                    LDAPProvider.objects.create(
 80                        name=generate_id(), authorization_flow=create_test_flow()
 81                    ).pk
 82                ],
 83            }
 84        )
 85        self.assertFalse(invalid.is_valid())
 86        self.assertIn("providers", invalid.errors)
 87
 88    def test_types(self):
 89        """Test OutpostServiceConnections's types endpoint"""
 90        response = self.client.get(
 91            reverse("authentik_api:outpostserviceconnection-types"),
 92        )
 93        self.assertEqual(response.status_code, 200)
 94
 95    def test_outpost_config(self):
 96        """Test Outpost's config field"""
 97        provider = ProxyProvider.objects.create(
 98            name=generate_id(), authorization_flow=create_test_flow()
 99        )
100        invalid = OutpostSerializer(
101            data={"name": generate_id(), "providers": [provider.pk], "config": ""}
102        )
103        self.assertFalse(invalid.is_valid())
104        self.assertIn("config", invalid.errors)
105        valid = OutpostSerializer(
106            data={
107                "name": generate_id(),
108                "providers": [provider.pk],
109                "config": default_outpost_config(generate_id()),
110                "type": OutpostType.PROXY,
111            }
112        )
113        self.assertTrue(valid.is_valid())

Test outpost service connection API

def setUp(self) -> None: 
21    def setUp(self) -> None:
22        super().setUp()
23        self.mapping = PropertyMapping.objects.create(
24            name=generate_id(), expression="""return {'foo': 'bar'}"""
25        )
26        self.user = create_test_admin_user()
27        self.client.force_login(self.user)

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

@reconcile_app('authentik_outposts')
def test_managed_name_change(self): 
29    @reconcile_app("authentik_outposts")
30    def test_managed_name_change(self):
31        """Test name change for embedded outpost"""
32        embedded_outpost = Outpost.objects.filter(managed=MANAGED_OUTPOST).first()
33        self.assertIsNotNone(embedded_outpost)
34        response = self.client.patch(
35            reverse("authentik_api:outpost-detail", kwargs={"pk": embedded_outpost.pk}),
36            {"name": "foo"},
37        )
38        self.assertEqual(response.status_code, 400)
39        self.assertJSONEqual(
40            response.content, {"name": ["Embedded outpost's name cannot be changed"]}
41        )

Test name change for embedded outpost

@reconcile_app('authentik_outposts')
def test_managed_without_managed(self): 
43    @reconcile_app("authentik_outposts")
44    def test_managed_without_managed(self):
45        """Test name change for embedded outpost"""
46        embedded_outpost = Outpost.objects.filter(managed=MANAGED_OUTPOST).first()
47        self.assertIsNotNone(embedded_outpost)
48        embedded_outpost.managed = ""
49        embedded_outpost.save()
50        response = self.client.patch(
51            reverse("authentik_api:outpost-detail", kwargs={"pk": embedded_outpost.pk}),
52            {"name": "foo"},
53        )
54        self.assertEqual(response.status_code, 200)
55        embedded_outpost.refresh_from_db()
56        self.assertEqual(embedded_outpost.managed, MANAGED_OUTPOST)

Test name change for embedded outpost

def test_outpost_validation(self): 
58    def test_outpost_validation(self):
59        """Test Outpost validation"""
60        valid = OutpostSerializer(
61            data={
62                "name": generate_id(),
63                "type": OutpostType.PROXY,
64                "config": default_outpost_config(),
65                "providers": [
66                    ProxyProvider.objects.create(
67                        name=generate_id(), authorization_flow=create_test_flow()
68                    ).pk
69                ],
70            }
71        )
72        self.assertTrue(valid.is_valid())
73        invalid = OutpostSerializer(
74            data={
75                "name": generate_id(),
76                "type": OutpostType.PROXY,
77                "config": default_outpost_config(),
78                "providers": [
79                    LDAPProvider.objects.create(
80                        name=generate_id(), authorization_flow=create_test_flow()
81                    ).pk
82                ],
83            }
84        )
85        self.assertFalse(invalid.is_valid())
86        self.assertIn("providers", invalid.errors)

Test Outpost validation

def test_types(self): 
88    def test_types(self):
89        """Test OutpostServiceConnections's types endpoint"""
90        response = self.client.get(
91            reverse("authentik_api:outpostserviceconnection-types"),
92        )
93        self.assertEqual(response.status_code, 200)

Test OutpostServiceConnections's types endpoint

def test_outpost_config(self): 
 95    def test_outpost_config(self):
 96        """Test Outpost's config field"""
 97        provider = ProxyProvider.objects.create(
 98            name=generate_id(), authorization_flow=create_test_flow()
 99        )
100        invalid = OutpostSerializer(
101            data={"name": generate_id(), "providers": [provider.pk], "config": ""}
102        )
103        self.assertFalse(invalid.is_valid())
104        self.assertIn("config", invalid.errors)
105        valid = OutpostSerializer(
106            data={
107                "name": generate_id(),
108                "providers": [provider.pk],
109                "config": default_outpost_config(generate_id()),
110                "type": OutpostType.PROXY,
111            }
112        )
113        self.assertTrue(valid.is_valid())

Test Outpost's config field