authentik.flows.tests

Test helpers

 1"""Test helpers"""
 2
 3from collections.abc import Callable, Generator
 4from contextlib import contextmanager
 5from json import loads
 6from typing import Any
 7from unittest.mock import MagicMock, patch
 8
 9from django.http.response import HttpResponse
10from django.urls.base import reverse
11from rest_framework.test import APITestCase
12
13from authentik.core.models import User
14from authentik.flows.models import Flow
15from authentik.flows.planner import FlowPlan
16from authentik.flows.views.executor import SESSION_KEY_PLAN
17
18
19class FlowTestCase(APITestCase):
20    """Helpers for testing flows and stages."""
21
22    def assertStageResponse(
23        self,
24        response: HttpResponse,
25        flow: Flow | None = None,
26        user: User | None = None,
27        **kwargs,
28    ) -> dict[str, Any]:
29        """Assert various attributes of a stage response"""
30        self.assertEqual(response.status_code, 200)
31        raw_response = loads(response.content.decode())
32        self.assertIsNotNone(raw_response["component"])
33        if flow:
34            self.assertIn("flow_info", raw_response)
35            self.assertTrue(
36                raw_response["flow_info"]["cancel_url"].startswith(
37                    reverse("authentik_flows:cancel")
38                )
39            )
40            # We don't check the flow title since it will most likely go
41            # through ChallengeStageView.format_title() so might not match 1:1
42            # self.assertEqual(raw_response["flow_info"]["title"], flow.title)
43            self.assertIsNotNone(raw_response["flow_info"]["title"])
44        if user:
45            self.assertEqual(raw_response["pending_user"], user.username)
46            self.assertEqual(raw_response["pending_user_avatar"], user.avatar)
47        for key, expected in kwargs.items():
48            self.assertEqual(raw_response[key], expected)
49        return raw_response
50
51    def get_flow_plan(self) -> FlowPlan | None:
52        return self.client.session.get(SESSION_KEY_PLAN)
53
54    def set_flow_plan(self, plan: FlowPlan):
55        session = self.client.session
56        session[SESSION_KEY_PLAN] = plan
57        session.save()
58
59    def assertStageRedirects(self, response: HttpResponse, to: str) -> dict[str, Any]:
60        """Wrapper around assertStageResponse that checks for a redirect"""
61        return self.assertStageResponse(response, component="xak-flow-redirect", to=to)
62
63    @contextmanager
64    def assertFlowFinishes(self) -> Generator[Callable[[], FlowPlan]]:
65        """Capture the flow plan before the flow finishes and return it"""
66        try:
67            with patch("authentik.flows.views.executor.FlowExecutorView.cancel", MagicMock()):
68                yield lambda: self.client.session.get(SESSION_KEY_PLAN)
69        finally:
70            pass
class FlowTestCase(rest_framework.test.APITestCase): 
20class FlowTestCase(APITestCase):
21    """Helpers for testing flows and stages."""
22
23    def assertStageResponse(
24        self,
25        response: HttpResponse,
26        flow: Flow | None = None,
27        user: User | None = None,
28        **kwargs,
29    ) -> dict[str, Any]:
30        """Assert various attributes of a stage response"""
31        self.assertEqual(response.status_code, 200)
32        raw_response = loads(response.content.decode())
33        self.assertIsNotNone(raw_response["component"])
34        if flow:
35            self.assertIn("flow_info", raw_response)
36            self.assertTrue(
37                raw_response["flow_info"]["cancel_url"].startswith(
38                    reverse("authentik_flows:cancel")
39                )
40            )
41            # We don't check the flow title since it will most likely go
42            # through ChallengeStageView.format_title() so might not match 1:1
43            # self.assertEqual(raw_response["flow_info"]["title"], flow.title)
44            self.assertIsNotNone(raw_response["flow_info"]["title"])
45        if user:
46            self.assertEqual(raw_response["pending_user"], user.username)
47            self.assertEqual(raw_response["pending_user_avatar"], user.avatar)
48        for key, expected in kwargs.items():
49            self.assertEqual(raw_response[key], expected)
50        return raw_response
51
52    def get_flow_plan(self) -> FlowPlan | None:
53        return self.client.session.get(SESSION_KEY_PLAN)
54
55    def set_flow_plan(self, plan: FlowPlan):
56        session = self.client.session
57        session[SESSION_KEY_PLAN] = plan
58        session.save()
59
60    def assertStageRedirects(self, response: HttpResponse, to: str) -> dict[str, Any]:
61        """Wrapper around assertStageResponse that checks for a redirect"""
62        return self.assertStageResponse(response, component="xak-flow-redirect", to=to)
63
64    @contextmanager
65    def assertFlowFinishes(self) -> Generator[Callable[[], FlowPlan]]:
66        """Capture the flow plan before the flow finishes and return it"""
67        try:
68            with patch("authentik.flows.views.executor.FlowExecutorView.cancel", MagicMock()):
69                yield lambda: self.client.session.get(SESSION_KEY_PLAN)
70        finally:
71            pass

Helpers for testing flows and stages.

def assertStageResponse( self, response: django.http.response.HttpResponse, flow: authentik.flows.models.Flow | None = None, user: authentik.core.models.User | None = None, **kwargs) -> dict[str, typing.Any]: 
23    def assertStageResponse(
24        self,
25        response: HttpResponse,
26        flow: Flow | None = None,
27        user: User | None = None,
28        **kwargs,
29    ) -> dict[str, Any]:
30        """Assert various attributes of a stage response"""
31        self.assertEqual(response.status_code, 200)
32        raw_response = loads(response.content.decode())
33        self.assertIsNotNone(raw_response["component"])
34        if flow:
35            self.assertIn("flow_info", raw_response)
36            self.assertTrue(
37                raw_response["flow_info"]["cancel_url"].startswith(
38                    reverse("authentik_flows:cancel")
39                )
40            )
41            # We don't check the flow title since it will most likely go
42            # through ChallengeStageView.format_title() so might not match 1:1
43            # self.assertEqual(raw_response["flow_info"]["title"], flow.title)
44            self.assertIsNotNone(raw_response["flow_info"]["title"])
45        if user:
46            self.assertEqual(raw_response["pending_user"], user.username)
47            self.assertEqual(raw_response["pending_user_avatar"], user.avatar)
48        for key, expected in kwargs.items():
49            self.assertEqual(raw_response[key], expected)
50        return raw_response

Assert various attributes of a stage response

def get_flow_plan(self) -> authentik.flows.planner.FlowPlan | None: 
52    def get_flow_plan(self) -> FlowPlan | None:
53        return self.client.session.get(SESSION_KEY_PLAN)
def set_flow_plan(self, plan: authentik.flows.planner.FlowPlan): 
55    def set_flow_plan(self, plan: FlowPlan):
56        session = self.client.session
57        session[SESSION_KEY_PLAN] = plan
58        session.save()
def assertStageRedirects( self, response: django.http.response.HttpResponse, to: str) -> dict[str, typing.Any]: 
60    def assertStageRedirects(self, response: HttpResponse, to: str) -> dict[str, Any]:
61        """Wrapper around assertStageResponse that checks for a redirect"""
62        return self.assertStageResponse(response, component="xak-flow-redirect", to=to)

Wrapper around assertStageResponse that checks for a redirect

@contextmanager
def assertFlowFinishes(self) -> Generator[Callable[[], authentik.flows.planner.FlowPlan]]: 
64    @contextmanager
65    def assertFlowFinishes(self) -> Generator[Callable[[], FlowPlan]]:
66        """Capture the flow plan before the flow finishes and return it"""
67        try:
68            with patch("authentik.flows.views.executor.FlowExecutorView.cancel", MagicMock()):
69                yield lambda: self.client.session.get(SESSION_KEY_PLAN)
70        finally:
71            pass

Capture the flow plan before the flow finishes and return it