This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
authentik/passbook/flows/tests/test_views.py

354 lines
14 KiB
Python
Raw Normal View History

"""flow views tests"""
from unittest.mock import MagicMock, PropertyMock, patch
from django.http import HttpRequest, HttpResponse
from django.shortcuts import reverse
from django.test import Client, TestCase
2020-09-11 21:21:11 +00:00
from django.utils.encoding import force_str
from passbook.flows.exceptions import EmptyFlowException, FlowNonApplicableException
from passbook.flows.markers import ReevaluateMarker, StageMarker
from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
from passbook.flows.planner import FlowPlan
from passbook.flows.views import NEXT_ARG_NAME, SESSION_KEY_PLAN
from passbook.lib.config import CONFIG
from passbook.policies.dummy.models import DummyPolicy
from passbook.policies.http import AccessDeniedResponse
from passbook.policies.models import PolicyBinding
from passbook.policies.types import PolicyResult
from passbook.stages.dummy.models import DummyStage
POLICY_RETURN_FALSE = PropertyMock(return_value=PolicyResult(False))
POLICY_RETURN_TRUE = MagicMock(return_value=PolicyResult(True))
def to_stage_response(request: HttpRequest, source: HttpResponse):
"""Mock for to_stage_response that returns the original response, so we can check
inheritance and member attributes"""
return source
TO_STAGE_RESPONSE_MOCK = MagicMock(side_effect=to_stage_response)
class TestFlowExecutor(TestCase):
"""Test views logic"""
def setUp(self):
self.client = Client()
def test_existing_plan_diff_flow(self):
"""Check that a plan for a different flow cancels the current plan"""
flow = Flow.objects.create(
name="test-existing-plan-diff",
slug="test-existing-plan-diff",
designation=FlowDesignation.AUTHENTICATION,
)
stage = DummyStage.objects.create(name="dummy")
plan = FlowPlan(
flow_pk=flow.pk.hex + "a", stages=[stage], markers=[StageMarker()]
)
session = self.client.session
session[SESSION_KEY_PLAN] = plan
session.save()
cancel_mock = MagicMock()
with patch("passbook.flows.views.FlowExecutorView.cancel", cancel_mock):
response = self.client.get(
reverse(
"passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}
),
)
2020-09-14 19:53:32 +00:00
self.assertEqual(response.status_code, 200)
self.assertEqual(cancel_mock.call_count, 2)
@patch(
2020-09-30 17:34:22 +00:00
"passbook.flows.views.to_stage_response",
TO_STAGE_RESPONSE_MOCK,
)
@patch(
2020-09-30 17:34:22 +00:00
"passbook.policies.engine.PolicyEngine.result",
POLICY_RETURN_FALSE,
)
def test_invalid_non_applicable_flow(self):
"""Tests that a non-applicable flow returns the correct error message"""
flow = Flow.objects.create(
name="test-non-applicable",
slug="test-non-applicable",
designation=FlowDesignation.AUTHENTICATION,
)
CONFIG.update_from_dict({"domain": "testserver"})
response = self.client.get(
reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}),
)
2020-09-14 19:53:32 +00:00
self.assertEqual(response.status_code, 200)
self.assertIsInstance(response, AccessDeniedResponse)
self.assertInHTML(FlowNonApplicableException.__doc__, response.rendered_content)
@patch(
2020-09-30 17:34:22 +00:00
"passbook.flows.views.to_stage_response",
TO_STAGE_RESPONSE_MOCK,
)
def test_invalid_empty_flow(self):
"""Tests that an empty flow returns the correct error message"""
flow = Flow.objects.create(
name="test-empty",
slug="test-empty",
designation=FlowDesignation.AUTHENTICATION,
)
CONFIG.update_from_dict({"domain": "testserver"})
response = self.client.get(
reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}),
)
2020-09-14 19:53:32 +00:00
self.assertEqual(response.status_code, 200)
self.assertIsInstance(response, AccessDeniedResponse)
self.assertInHTML(EmptyFlowException.__doc__, response.rendered_content)
def test_invalid_flow_redirect(self):
"""Tests that an invalid flow still redirects"""
flow = Flow.objects.create(
name="test-empty",
slug="test-empty",
designation=FlowDesignation.AUTHENTICATION,
)
CONFIG.update_from_dict({"domain": "testserver"})
dest = "/unique-string"
2020-05-11 19:27:46 +00:00
url = reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug})
response = self.client.get(url + f"?{NEXT_ARG_NAME}={dest}")
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
2020-09-30 17:34:22 +00:00
force_str(response.content),
{"type": "redirect", "to": dest},
)
def test_multi_stage_flow(self):
"""Test a full flow with multiple stages"""
flow = Flow.objects.create(
name="test-full",
slug="test-full",
designation=FlowDesignation.AUTHENTICATION,
)
FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy1"), order=0
)
FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy2"), order=1
)
exec_url = reverse(
"passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}
)
# First Request, start planning, renders form
response = self.client.get(exec_url)
self.assertEqual(response.status_code, 200)
# Check that two stages are in plan
session = self.client.session
plan: FlowPlan = session[SESSION_KEY_PLAN]
self.assertEqual(len(plan.stages), 2)
# Second request, submit form, one stage left
response = self.client.post(exec_url)
# Second request redirects to the same URL
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, exec_url)
# Check that two stages are in plan
session = self.client.session
plan: FlowPlan = session[SESSION_KEY_PLAN]
self.assertEqual(len(plan.stages), 1)
def test_reevaluate_remove_last(self):
"""Test planner with re-evaluate (last stage is removed)"""
flow = Flow.objects.create(
name="test-default-context",
slug="test-default-context",
designation=FlowDesignation.AUTHENTICATION,
)
false_policy = DummyPolicy.objects.create(result=False, wait_min=1, wait_max=2)
binding = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy1"), order=0
)
binding2 = FlowStageBinding.objects.create(
target=flow,
stage=DummyStage.objects.create(name="dummy2"),
order=1,
evaluate_on_call=True,
)
PolicyBinding.objects.create(policy=false_policy, target=binding2, order=0)
# Here we patch the dummy policy to evaluate to true so the stage is included
with patch(
"passbook.policies.dummy.models.DummyPolicy.passes", POLICY_RETURN_TRUE
):
exec_url = reverse(
"passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}
)
# First request, run the planner
response = self.client.get(exec_url)
self.assertEqual(response.status_code, 200)
plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
self.assertEqual(plan.stages[0], binding.stage)
self.assertEqual(plan.stages[1], binding2.stage)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], ReevaluateMarker)
# Second request, this passes the first dummy stage
response = self.client.post(exec_url)
self.assertEqual(response.status_code, 302)
# third request, this should trigger the re-evaluate
# We do this request without the patch, so the policy results in false
response = self.client.post(exec_url)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse("passbook_core:overview"))
def test_reevaluate_remove_middle(self):
"""Test planner with re-evaluate (middle stage is removed)"""
flow = Flow.objects.create(
name="test-default-context",
slug="test-default-context",
designation=FlowDesignation.AUTHENTICATION,
)
false_policy = DummyPolicy.objects.create(result=False, wait_min=1, wait_max=2)
binding = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy1"), order=0
)
binding2 = FlowStageBinding.objects.create(
target=flow,
stage=DummyStage.objects.create(name="dummy2"),
order=1,
evaluate_on_call=True,
)
binding3 = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy3"), order=2
)
PolicyBinding.objects.create(policy=false_policy, target=binding2, order=0)
# Here we patch the dummy policy to evaluate to true so the stage is included
with patch(
"passbook.policies.dummy.models.DummyPolicy.passes", POLICY_RETURN_TRUE
):
exec_url = reverse(
"passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}
)
# First request, run the planner
response = self.client.get(exec_url)
self.assertEqual(response.status_code, 200)
plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
self.assertEqual(plan.stages[0], binding.stage)
self.assertEqual(plan.stages[1], binding2.stage)
self.assertEqual(plan.stages[2], binding3.stage)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], ReevaluateMarker)
self.assertIsInstance(plan.markers[2], StageMarker)
# Second request, this passes the first dummy stage
response = self.client.post(exec_url)
self.assertEqual(response.status_code, 302)
plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
self.assertEqual(plan.stages[0], binding2.stage)
self.assertEqual(plan.stages[1], binding3.stage)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], StageMarker)
# third request, this should trigger the re-evaluate
# We do this request without the patch, so the policy results in false
response = self.client.post(exec_url)
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
2020-09-11 21:21:11 +00:00
force_str(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)
def test_reevaluate_remove_consecutive(self):
"""Test planner with re-evaluate (consecutive stages are removed)"""
flow = Flow.objects.create(
name="test-default-context",
slug="test-default-context",
designation=FlowDesignation.AUTHENTICATION,
)
false_policy = DummyPolicy.objects.create(result=False, wait_min=1, wait_max=2)
binding = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy1"), order=0
)
binding2 = FlowStageBinding.objects.create(
target=flow,
stage=DummyStage.objects.create(name="dummy2"),
order=1,
evaluate_on_call=True,
)
binding3 = FlowStageBinding.objects.create(
target=flow,
stage=DummyStage.objects.create(name="dummy3"),
order=2,
evaluate_on_call=True,
)
binding4 = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy4"), order=2
)
PolicyBinding.objects.create(policy=false_policy, target=binding2, order=0)
PolicyBinding.objects.create(policy=false_policy, target=binding3, order=0)
# Here we patch the dummy policy to evaluate to true so the stage is included
with patch(
"passbook.policies.dummy.models.DummyPolicy.passes", POLICY_RETURN_TRUE
):
exec_url = reverse(
"passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}
)
# First request, run the planner
response = self.client.get(exec_url)
self.assertEqual(response.status_code, 200)
2020-09-11 21:21:11 +00:00
self.assertIn("dummy1", force_str(response.content))
plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
self.assertEqual(plan.stages[0], binding.stage)
self.assertEqual(plan.stages[1], binding2.stage)
self.assertEqual(plan.stages[2], binding3.stage)
self.assertEqual(plan.stages[3], binding4.stage)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], ReevaluateMarker)
self.assertIsInstance(plan.markers[2], ReevaluateMarker)
self.assertIsInstance(plan.markers[3], StageMarker)
# Second request, this passes the first dummy stage
response = self.client.post(exec_url)
self.assertEqual(response.status_code, 302)
# third request, this should trigger the re-evaluate
# A get request will evaluate the policies and this will return stage 4
# but it won't save it, hence we cant' check the plan
response = self.client.get(exec_url)
self.assertEqual(response.status_code, 200)
2020-09-11 21:21:11 +00:00
self.assertIn("dummy4", force_str(response.content))
# fourth request, this confirms the last stage (dummy4)
# We do this request without the patch, so the policy results in false
response = self.client.post(exec_url)
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(
2020-09-11 21:21:11 +00:00
force_str(response.content),
{"type": "redirect", "to": reverse("passbook_core:overview")},
)