"""Prompt tests"""
from unittest.mock import MagicMock, patch

from django.test import RequestFactory
from django.urls import reverse
from rest_framework.exceptions import ErrorDetail

from authentik.core.tests.utils import create_test_admin_user
from authentik.flows.markers import StageMarker
from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding
from authentik.flows.planner import FlowPlan
from authentik.flows.tests import FlowTestCase
from authentik.flows.views.executor import SESSION_KEY_PLAN
from authentik.lib.generators import generate_id
from authentik.policies.expression.models import ExpressionPolicy
from authentik.stages.prompt.models import FieldTypes, Prompt, PromptStage
from authentik.stages.prompt.stage import PLAN_CONTEXT_PROMPT, PromptChallengeResponse


class TestPromptStage(FlowTestCase):
    """Prompt tests"""

    def setUp(self):
        super().setUp()
        self.user = create_test_admin_user()
        self.factory = RequestFactory()
        self.flow = Flow.objects.create(
            name="test-prompt",
            slug="test-prompt",
            designation=FlowDesignation.AUTHENTICATION,
        )
        username_prompt = Prompt.objects.create(
            field_key="username_prompt",
            label="USERNAME_LABEL",
            type=FieldTypes.USERNAME,
            required=True,
            placeholder="USERNAME_PLACEHOLDER",
        )
        text_prompt = Prompt.objects.create(
            field_key="text_prompt",
            label="TEXT_LABEL",
            type=FieldTypes.TEXT,
            required=True,
            placeholder="TEXT_PLACEHOLDER",
        )
        email_prompt = Prompt.objects.create(
            field_key="email_prompt",
            label="EMAIL_LABEL",
            type=FieldTypes.EMAIL,
            required=True,
            placeholder="EMAIL_PLACEHOLDER",
        )
        password_prompt = Prompt.objects.create(
            field_key="password_prompt",
            label="PASSWORD_LABEL",
            type=FieldTypes.PASSWORD,
            required=True,
            placeholder="PASSWORD_PLACEHOLDER",
        )
        password2_prompt = Prompt.objects.create(
            field_key="password2_prompt",
            label="PASSWORD_LABEL",
            type=FieldTypes.PASSWORD,
            required=True,
            placeholder="PASSWORD_PLACEHOLDER",
        )
        number_prompt = Prompt.objects.create(
            field_key="number_prompt",
            label="NUMBER_LABEL",
            type=FieldTypes.NUMBER,
            required=True,
            placeholder="NUMBER_PLACEHOLDER",
        )
        hidden_prompt = Prompt.objects.create(
            field_key="hidden_prompt",
            type=FieldTypes.HIDDEN,
            required=True,
            placeholder="HIDDEN_PLACEHOLDER",
        )
        static_prompt = Prompt.objects.create(
            field_key="static_prompt",
            type=FieldTypes.STATIC,
            required=True,
            placeholder="static",
        )
        self.stage = PromptStage.objects.create(name="prompt-stage")
        self.stage.fields.set(
            [
                username_prompt,
                text_prompt,
                email_prompt,
                password_prompt,
                password2_prompt,
                number_prompt,
                hidden_prompt,
                static_prompt,
            ]
        )

        self.prompt_data = {
            username_prompt.field_key: "test-username",
            text_prompt.field_key: "test-input",
            email_prompt.field_key: "<EMAIL>",
            password_prompt.field_key: "test",
            password2_prompt.field_key: "test",
            number_prompt.field_key: 3,
            hidden_prompt.field_key: hidden_prompt.placeholder,
            static_prompt.field_key: static_prompt.placeholder,
        }

        self.binding = FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)

    def test_render(self):
        """Test render of form, check if all prompts are rendered correctly"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        session = self.client.session
        session[SESSION_KEY_PLAN] = plan
        session.save()

        response = self.client.get(
            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
        )
        self.assertEqual(response.status_code, 200)
        for prompt in self.stage.fields.all():
            self.assertIn(prompt.field_key, response.content.decode())
            self.assertIn(prompt.label, response.content.decode())
            self.assertIn(prompt.placeholder, response.content.decode())

    def test_valid_challenge_with_policy(self) -> PromptChallengeResponse:
        """Test challenge_response validation"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        expr = (
            "return request.context['prompt_data']['password_prompt'] "
            "== request.context['prompt_data']['password2_prompt']"
        )
        expr_policy = ExpressionPolicy.objects.create(name="validate-form", expression=expr)
        self.stage.validation_policies.set([expr_policy])
        self.stage.save()
        challenge_response = PromptChallengeResponse(
            None, stage=self.stage, plan=plan, data=self.prompt_data
        )
        self.assertEqual(challenge_response.is_valid(), True)
        return challenge_response

    def test_invalid_challenge(self) -> PromptChallengeResponse:
        """Test challenge_response validation"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        expr = "False"
        expr_policy = ExpressionPolicy.objects.create(name="validate-form", expression=expr)
        self.stage.validation_policies.set([expr_policy])
        self.stage.save()
        challenge_response = PromptChallengeResponse(
            None, stage=self.stage, plan=plan, data=self.prompt_data
        )
        self.assertEqual(challenge_response.is_valid(), False)
        return challenge_response

    def test_valid_challenge_request(self):
        """Test a request with valid challenge_response data"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        session = self.client.session
        session[SESSION_KEY_PLAN] = plan
        session.save()

        challenge_response = self.test_valid_challenge_with_policy()

        with patch("authentik.flows.views.executor.FlowExecutorView.cancel", MagicMock()):
            response = self.client.post(
                reverse(
                    "authentik_api:flow-executor",
                    kwargs={"flow_slug": self.flow.slug},
                ),
                challenge_response.validated_data,
            )
        self.assertEqual(response.status_code, 200)
        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))

        # Check that valid data has been saved
        session = self.client.session
        plan: FlowPlan = session[SESSION_KEY_PLAN]
        data = plan.context[PLAN_CONTEXT_PROMPT]
        for prompt in self.stage.fields.all():
            prompt: Prompt
            self.assertEqual(data[prompt.field_key], self.prompt_data[prompt.field_key])

    def test_invalid_password(self):
        """Test challenge_response validation"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        self.prompt_data["password2_prompt"] = "qwerqwerqr"
        challenge_response = PromptChallengeResponse(
            None, stage=self.stage, plan=plan, data=self.prompt_data
        )
        self.assertEqual(challenge_response.is_valid(), False)
        self.assertEqual(
            challenge_response.errors,
            {"non_field_errors": [ErrorDetail(string="Passwords don't match.", code="invalid")]},
        )

    def test_invalid_username(self):
        """Test challenge_response validation"""
        user = create_test_admin_user()
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        self.prompt_data["username_prompt"] = user.username
        challenge_response = PromptChallengeResponse(
            None, stage=self.stage, plan=plan, data=self.prompt_data
        )
        self.assertEqual(challenge_response.is_valid(), False)
        self.assertEqual(
            challenge_response.errors,
            {"username_prompt": [ErrorDetail(string="Username is already taken.", code="invalid")]},
        )

    def test_static_hidden_overwrite(self):
        """Test that static and hidden fields ignore any value sent to them"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        plan.context[PLAN_CONTEXT_PROMPT] = {"hidden_prompt": "hidden"}
        self.prompt_data["hidden_prompt"] = "foo"
        self.prompt_data["static_prompt"] = "foo"
        challenge_response = PromptChallengeResponse(
            None, stage=self.stage, plan=plan, data=self.prompt_data
        )
        self.assertEqual(challenge_response.is_valid(), True)
        self.assertNotEqual(challenge_response.validated_data["hidden_prompt"], "foo")
        self.assertEqual(challenge_response.validated_data["hidden_prompt"], "hidden")
        self.assertNotEqual(challenge_response.validated_data["static_prompt"], "foo")

    def test_prompt_placeholder(self):
        """Test placeholder and expression"""
        context = {
            "foo": generate_id(),
        }
        prompt: Prompt = Prompt(
            field_key="text_prompt_expression",
            label="TEXT_LABEL",
            type=FieldTypes.TEXT,
            placeholder="return prompt_context['foo']",
            placeholder_expression=True,
        )
        self.assertEqual(
            prompt.get_placeholder(context, self.user, self.factory.get("/")), context["foo"]
        )
        context["text_prompt_expression"] = generate_id()
        self.assertEqual(
            prompt.get_placeholder(context, self.user, self.factory.get("/")),
            context["text_prompt_expression"],
        )
        self.assertNotEqual(
            prompt.get_placeholder(context, self.user, self.factory.get("/")), context["foo"]
        )

    def test_prompt_placeholder_error(self):
        """Test placeholder and expression"""
        context = {}
        prompt: Prompt = Prompt(
            field_key="text_prompt_expression",
            label="TEXT_LABEL",
            type=FieldTypes.TEXT,
            placeholder="something invalid dunno",
            placeholder_expression=True,
        )
        self.assertEqual(
            prompt.get_placeholder(context, self.user, self.factory.get("/")),
            "something invalid dunno",
        )

    def test_prompt_placeholder_disabled(self):
        """Test placeholder and expression"""
        context = {}
        prompt: Prompt = Prompt(
            field_key="text_prompt_expression",
            label="TEXT_LABEL",
            type=FieldTypes.TEXT,
            placeholder="return prompt_context['foo']",
            placeholder_expression=False,
        )
        self.assertEqual(
            prompt.get_placeholder(context, self.user, self.factory.get("/")), prompt.placeholder
        )

    def test_invalid_save(self):
        """Ensure field can't be saved with invalid type"""
        prompt: Prompt = Prompt(
            field_key="text_prompt_expression",
            label="TEXT_LABEL",
            type="foo",
            placeholder="foo",
            placeholder_expression=False,
            sub_text="test",
            order=123,
        )
        with self.assertRaises(ValueError):
            prompt.save()


def field_type_tester_factory(field_type: FieldTypes, required: bool):
    """Test field for field_type"""

    def tester(self: TestPromptStage):
        prompt: Prompt = Prompt(
            field_key="text_prompt_expression",
            label="TEXT_LABEL",
            type=field_type,
            placeholder="foo",
            placeholder_expression=False,
            sub_text="test",
            order=123,
            required=required,
        )
        self.assertIsNotNone(prompt.field("foo"))

    return tester


for _required in (True, False):
    for _type in FieldTypes:
        test_name = f"test_field_type_{_type}"
        if _required:
            test_name += "_required"
        setattr(TestPromptStage, test_name, field_type_tester_factory(_type, _required))

"""Prompt tests"""
from unittest.mock import MagicMock, patch

from django.test import RequestFactory
from django.urls import reverse
from rest_framework.exceptions import ErrorDetail

from authentik.core.tests.utils import create_test_admin_user
from authentik.flows.markers import StageMarker
from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding
from authentik.flows.planner import FlowPlan
from authentik.flows.tests import FlowTestCase
from authentik.flows.views.executor import SESSION_KEY_PLAN
from authentik.lib.generators import generate_id
from authentik.policies.expression.models import ExpressionPolicy
from authentik.stages.prompt.models import FieldTypes, Prompt, PromptStage
from authentik.stages.prompt.stage import PLAN_CONTEXT_PROMPT, PromptChallengeResponse


class TestPromptStage(FlowTestCase):
    """Prompt tests"""

    def setUp(self):
        super().setUp()
        self.user = create_test_admin_user()
        self.factory = RequestFactory()
        self.flow = Flow.objects.create(
            name="test-prompt",
            slug="test-prompt",
            designation=FlowDesignation.AUTHENTICATION,
        )
        username_prompt = Prompt.objects.create(
            field_key="username_prompt",
            label="USERNAME_LABEL",
            type=FieldTypes.USERNAME,
            required=True,
            placeholder="USERNAME_PLACEHOLDER",
        )
        text_prompt = Prompt.objects.create(
            field_key="text_prompt",
            label="TEXT_LABEL",
            type=FieldTypes.TEXT,
            required=True,
            placeholder="TEXT_PLACEHOLDER",
        )
        email_prompt = Prompt.objects.create(
            field_key="email_prompt",
            label="EMAIL_LABEL",
            type=FieldTypes.EMAIL,
            required=True,
            placeholder="EMAIL_PLACEHOLDER",
        )
        password_prompt = Prompt.objects.create(
            field_key="password_prompt",
            label="PASSWORD_LABEL",
            type=FieldTypes.PASSWORD,
            required=True,
            placeholder="PASSWORD_PLACEHOLDER",
        )
        password2_prompt = Prompt.objects.create(
            field_key="password2_prompt",
            label="PASSWORD_LABEL",
            type=FieldTypes.PASSWORD,
            required=True,
            placeholder="PASSWORD_PLACEHOLDER",
        )
        number_prompt = Prompt.objects.create(
            field_key="number_prompt",
            label="NUMBER_LABEL",
            type=FieldTypes.NUMBER,
            required=True,
            placeholder="NUMBER_PLACEHOLDER",
        )
        hidden_prompt = Prompt.objects.create(
            field_key="hidden_prompt",
            type=FieldTypes.HIDDEN,
            required=True,
            placeholder="HIDDEN_PLACEHOLDER",
        )
        static_prompt = Prompt.objects.create(
            field_key="static_prompt",
            type=FieldTypes.STATIC,
            required=True,
            placeholder="static",
        )
        self.stage = PromptStage.objects.create(name="prompt-stage")
        self.stage.fields.set(
            [
                username_prompt,
                text_prompt,
                email_prompt,
                password_prompt,
                password2_prompt,
                number_prompt,
                hidden_prompt,
                static_prompt,
            ]
        )

        self.prompt_data = {
            username_prompt.field_key: "test-username",
            text_prompt.field_key: "test-input",
            email_prompt.field_key: "<EMAIL>",
            password_prompt.field_key: "test",
            password2_prompt.field_key: "test",
            number_prompt.field_key: 3,
            hidden_prompt.field_key: hidden_prompt.placeholder,
            static_prompt.field_key: static_prompt.placeholder,
        }

        self.binding = FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)

    def test_render(self):
        """Test render of form, check if all prompts are rendered correctly"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        session = self.client.session
        session[SESSION_KEY_PLAN] = plan
        session.save()

        response = self.client.get(
            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
        )
        self.assertEqual(response.status_code, 200)
        for prompt in self.stage.fields.all():
            self.assertIn(prompt.field_key, response.content.decode())
            self.assertIn(prompt.label, response.content.decode())
            self.assertIn(prompt.placeholder, response.content.decode())

    def test_valid_challenge_with_policy(self) -> PromptChallengeResponse:
        """Test challenge_response validation"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        expr = (
            "return request.context['prompt_data']['password_prompt'] "
            "== request.context['prompt_data']['password2_prompt']"
        )
        expr_policy = ExpressionPolicy.objects.create(name="validate-form", expression=expr)
        self.stage.validation_policies.set([expr_policy])
        self.stage.save()
        challenge_response = PromptChallengeResponse(
            None, stage=self.stage, plan=plan, data=self.prompt_data
        )
        self.assertEqual(challenge_response.is_valid(), True)
        return challenge_response

    def test_invalid_challenge(self) -> PromptChallengeResponse:
        """Test challenge_response validation"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        expr = "False"
        expr_policy = ExpressionPolicy.objects.create(name="validate-form", expression=expr)
        self.stage.validation_policies.set([expr_policy])
        self.stage.save()
        challenge_response = PromptChallengeResponse(
            None, stage=self.stage, plan=plan, data=self.prompt_data
        )
        self.assertEqual(challenge_response.is_valid(), False)
        return challenge_response

    def test_valid_challenge_request(self):
        """Test a request with valid challenge_response data"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        session = self.client.session
        session[SESSION_KEY_PLAN] = plan
        session.save()

        challenge_response = self.test_valid_challenge_with_policy()

        with patch("authentik.flows.views.executor.FlowExecutorView.cancel", MagicMock()):
            response = self.client.post(
                reverse(
                    "authentik_api:flow-executor",
                    kwargs={"flow_slug": self.flow.slug},
                ),
                challenge_response.validated_data,
            )
        self.assertEqual(response.status_code, 200)
        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))

        # Check that valid data has been saved
        session = self.client.session
        plan: FlowPlan = session[SESSION_KEY_PLAN]
        data = plan.context[PLAN_CONTEXT_PROMPT]
        for prompt in self.stage.fields.all():
            prompt: Prompt
            self.assertEqual(data[prompt.field_key], self.prompt_data[prompt.field_key])

    def test_invalid_password(self):
        """Test challenge_response validation"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        self.prompt_data["password2_prompt"] = "qwerqwerqr"
        challenge_response = PromptChallengeResponse(
            None, stage=self.stage, plan=plan, data=self.prompt_data
        )
        self.assertEqual(challenge_response.is_valid(), False)
        self.assertEqual(
            challenge_response.errors,
            {"non_field_errors": [ErrorDetail(string="Passwords don't match.", code="invalid")]},
        )

    def test_invalid_username(self):
        """Test challenge_response validation"""
        user = create_test_admin_user()
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        self.prompt_data["username_prompt"] = user.username
        challenge_response = PromptChallengeResponse(
            None, stage=self.stage, plan=plan, data=self.prompt_data
        )
        self.assertEqual(challenge_response.is_valid(), False)
        self.assertEqual(
            challenge_response.errors,
            {"username_prompt": [ErrorDetail(string="Username is already taken.", code="invalid")]},
        )

    def test_static_hidden_overwrite(self):
        """Test that static and hidden fields ignore any value sent to them"""
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        plan.context[PLAN_CONTEXT_PROMPT] = {"hidden_prompt": "hidden"}
        self.prompt_data["hidden_prompt"] = "foo"
        self.prompt_data["static_prompt"] = "foo"
        challenge_response = PromptChallengeResponse(
            None, stage=self.stage, plan=plan, data=self.prompt_data
        )
        self.assertEqual(challenge_response.is_valid(), True)
        self.assertNotEqual(challenge_response.validated_data["hidden_prompt"], "foo")
        self.assertEqual(challenge_response.validated_data["hidden_prompt"], "hidden")
        self.assertNotEqual(challenge_response.validated_data["static_prompt"], "foo")

    def test_prompt_placeholder(self):
        """Test placeholder and expression"""
        context = {
            "foo": generate_id(),
        }
        prompt: Prompt = Prompt(
            field_key="text_prompt_expression",
            label="TEXT_LABEL",
            type=FieldTypes.TEXT,
            placeholder="return prompt_context['foo']",
            placeholder_expression=True,
        )
        self.assertEqual(
            prompt.get_placeholder(context, self.user, self.factory.get("/")), context["foo"]
        )
        context["text_prompt_expression"] = generate_id()
        self.assertEqual(
            prompt.get_placeholder(context, self.user, self.factory.get("/")),
            context["text_prompt_expression"],
        )
        self.assertNotEqual(
            prompt.get_placeholder(context, self.user, self.factory.get("/")), context["foo"]
        )

    def test_prompt_placeholder_error(self):
        """Test placeholder and expression"""
        context = {}
        prompt: Prompt = Prompt(
            field_key="text_prompt_expression",
            label="TEXT_LABEL",
            type=FieldTypes.TEXT,
            placeholder="something invalid dunno",
            placeholder_expression=True,
        )
        self.assertEqual(
            prompt.get_placeholder(context, self.user, self.factory.get("/")),
            "something invalid dunno",
        )

    def test_prompt_placeholder_disabled(self):
        """Test placeholder and expression"""
        context = {}
        prompt: Prompt = Prompt(
            field_key="text_prompt_expression",
            label="TEXT_LABEL",
            type=FieldTypes.TEXT,
            placeholder="return prompt_context['foo']",
            placeholder_expression=False,
        )
        self.assertEqual(
            prompt.get_placeholder(context, self.user, self.factory.get("/")), prompt.placeholder
        )

    def test_invalid_save(self):
        """Ensure field can't be saved with invalid type"""
        prompt: Prompt = Prompt(
            field_key="text_prompt_expression",
            label="TEXT_LABEL",
            type="foo",
            placeholder="foo",
            placeholder_expression=False,
            sub_text="test",
            order=123,
        )
        with self.assertRaises(ValueError):
            prompt.save()


def field_type_tester_factory(field_type: FieldTypes, required: bool):
    """Test field for field_type"""

    def tester(self: TestPromptStage):
        prompt: Prompt = Prompt(
            field_key="text_prompt_expression",
            label="TEXT_LABEL",
            type=field_type,
            placeholder="foo",
            placeholder_expression=False,
            sub_text="test",
            order=123,
            required=required,
        )
        self.assertIsNotNone(prompt.field("foo"))

    return tester


for _required in (True, False):
    for _type in FieldTypes:
        test_name = f"test_field_type_{_type}"
        if _required:
            test_name += "_required"
        setattr(TestPromptStage, test_name, field_type_tester_factory(_type, _required))

Prompt tests Prompt tests Test render of form, check if all prompts are rendered correctly Test challenge_response validation Test challenge_response validation Test a request with valid challenge_response data # Check that valid data has been saved Test challenge_response validation Test challenge_response validation Test that static and hidden fields ignore any value sent to them Test placeholder and expression Test placeholder and expression Test placeholder and expression Ensure field can't be saved with invalid type Test field for field_type