authentik.blueprints.tests.test_v1_api

  1"""Test blueprints v1 api"""
  2
  3from json import dumps, loads
  4from tempfile import NamedTemporaryFile, mkdtemp
  5
  6from django.core.files.uploadedfile import SimpleUploadedFile
  7from django.urls import reverse
  8from rest_framework.test import APITestCase
  9from yaml import dump
 10
 11from authentik.core.tests.utils import create_test_admin_user
 12from authentik.flows.models import Flow
 13from authentik.lib.config import CONFIG
 14from authentik.lib.generators import generate_id
 15from authentik.stages.invitation.models import InvitationStage
 16from authentik.stages.user_write.models import UserWriteStage
 17
 18TMP = mkdtemp("authentik-blueprints")
 19
 20
 21class TestBlueprintsV1API(APITestCase):
 22    """Test Blueprints API"""
 23
 24    def setUp(self) -> None:
 25        self.user = create_test_admin_user()
 26        self.client.force_login(self.user)
 27
 28    @CONFIG.patch("blueprints_dir", TMP)
 29    def test_api_available(self):
 30        """Test valid file"""
 31        with NamedTemporaryFile(mode="w+", suffix=".yaml", dir=TMP) as file:
 32            file.write(
 33                dump(
 34                    {
 35                        "version": 1,
 36                        "entries": [],
 37                    }
 38                )
 39            )
 40            file.flush()
 41            res = self.client.get(reverse("authentik_api:blueprintinstance-available"))
 42            self.assertEqual(res.status_code, 200)
 43            response = loads(res.content.decode())
 44            self.assertEqual(len(response), 1)
 45            self.assertEqual(
 46                response[0]["hash"],
 47                (
 48                    "e52bb445b03cd36057258dc9f0ce0fbed8278498ee1470e45315293e5f026d1bd1f9b352"
 49                    "6871c0003f5c07be5c3316d9d4a08444bd8fed1b3f03294e51e44522"
 50                ),
 51            )
 52
 53    def test_api_oci(self):
 54        """Test validation with OCI path"""
 55        res = self.client.post(
 56            reverse("authentik_api:blueprintinstance-list"),
 57            data={"name": "foo", "path": "oci://foo/bar"},
 58        )
 59        self.assertEqual(res.status_code, 201)
 60
 61    def test_api_blank(self):
 62        """Test blank"""
 63        res = self.client.post(
 64            reverse("authentik_api:blueprintinstance-list"),
 65            data={
 66                "name": "foo",
 67            },
 68        )
 69        self.assertEqual(res.status_code, 400)
 70        self.assertJSONEqual(
 71            res.content.decode(), {"non_field_errors": ["Either path or content must be set."]}
 72        )
 73
 74    def test_api_content(self):
 75        """Test blank"""
 76        res = self.client.post(
 77            reverse("authentik_api:blueprintinstance-list"),
 78            data={
 79                "name": "foo",
 80                "content": '{"version": 3}',
 81            },
 82        )
 83        self.assertEqual(res.status_code, 400)
 84        self.assertJSONEqual(
 85            res.content.decode(),
 86            {"content": ["Failed to validate blueprint", "- Invalid blueprint version"]},
 87        )
 88
 89    def test_api_import_with_context(self):
 90        """Test that the import endpoint applies the supplied context to the real blueprint"""
 91        slug = f"invitation-enrollment-{generate_id()}"
 92        flow_name = f"Invitation Enrollment {generate_id()}"
 93        stage_name = f"invitation-stage-{generate_id()}"
 94        user_type = "internal"
 95        continue_without_invitation = True
 96
 97        res = self.client.post(
 98            reverse("authentik_api:blueprintinstance-import-"),
 99            data={
100                "path": "example/flows-invitation-enrollment-minimal.yaml",
101                "context": dumps(
102                    {
103                        "flow_slug": slug,
104                        "flow_name": flow_name,
105                        "stage_name": stage_name,
106                        "continue_flow_without_invitation": continue_without_invitation,
107                        "user_type": user_type,
108                    }
109                ),
110            },
111            format="multipart",
112        )
113        self.assertEqual(res.status_code, 200)
114        self.assertTrue(res.json()["success"])
115
116        flow = Flow.objects.get(slug=slug)
117        self.assertEqual(flow.name, flow_name)
118        self.assertEqual(flow.title, flow_name)
119
120        invitation_stage = InvitationStage.objects.get(name=stage_name)
121        self.assertEqual(
122            invitation_stage.continue_flow_without_invitation,
123            continue_without_invitation,
124        )
125
126        user_write_stage = UserWriteStage.objects.get(
127            name=f"invitation-enrollment-user-write-{slug}"
128        )
129        self.assertEqual(user_write_stage.user_type, user_type)
130        self.assertEqual(user_write_stage.user_path_template, f"users/{user_type}")
131
132    def test_api_import_blank_path(self):
133        """Validator returns empty path unchanged (covers api.py:53)."""
134        with NamedTemporaryFile(mode="w+", suffix=".yaml") as file:
135            file.write(dump({"version": 1, "entries": []}))
136            file.flush()
137            file.seek(0)
138            res = self.client.post(
139                reverse("authentik_api:blueprintinstance-import-"),
140                data={"path": "", "file": file},
141                format="multipart",
142            )
143        self.assertEqual(res.status_code, 200)
144
145    def test_api_import_invalid_blueprint_returns_result_payload(self):
146        """Invalid blueprint content returns a result payload instead of a 400 response."""
147        file = SimpleUploadedFile("invalid-blueprint.yaml", b'{"version": 3}')
148
149        res = self.client.post(
150            reverse("authentik_api:blueprintinstance-import-"),
151            data={"file": file},
152            format="multipart",
153        )
154
155        self.assertEqual(res.status_code, 200)
156        self.assertFalse(res.json()["success"])
157        self.assertGreater(len(res.json()["logs"]), 0)
158
159    def test_api_import_unknown_path(self):
160        """Path not in available blueprints is rejected (covers api.py:56)."""
161        res = self.client.post(
162            reverse("authentik_api:blueprintinstance-import-"),
163            data={"path": "does/not/exist.yaml"},
164            format="multipart",
165        )
166        self.assertEqual(res.status_code, 400)
167        self.assertIn("Blueprint file does not exist", res.content.decode())
168
169    def test_api_import_blank_context(self):
170        """Blank context is normalized to empty dict (covers api.py:62)."""
171        res = self.client.post(
172            reverse("authentik_api:blueprintinstance-import-"),
173            data={
174                "path": "example/flows-invitation-enrollment-minimal.yaml",
175                "context": "",
176            },
177            format="multipart",
178        )
179        self.assertEqual(res.status_code, 200)
180
181    def test_api_import_invalid_json_context(self):
182        """Malformed JSON context raises ValidationError (covers api.py:65-66)."""
183        res = self.client.post(
184            reverse("authentik_api:blueprintinstance-import-"),
185            data={
186                "path": "example/flows-invitation-enrollment-minimal.yaml",
187                "context": "{not json",
188            },
189            format="multipart",
190        )
191        self.assertEqual(res.status_code, 400)
192        self.assertIn("Context must be valid JSON", res.content.decode())
193
194    def test_api_import_non_object_context(self):
195        """JSON context that isn't an object is rejected (covers api.py:68)."""
196        res = self.client.post(
197            reverse("authentik_api:blueprintinstance-import-"),
198            data={
199                "path": "example/flows-invitation-enrollment-minimal.yaml",
200                "context": "[1, 2, 3]",
201            },
202            format="multipart",
203        )
204        self.assertEqual(res.status_code, 400)
205        self.assertIn("Context must be a JSON object", res.content.decode())