From 980320e24b30c22d16ab097eec24404ac5ea8eda Mon Sep 17 00:00:00 2001 From: Jens L Date: Mon, 27 Feb 2023 22:42:36 +0100 Subject: [PATCH] tests/e2e: use example blueprints for testing (#4805) * tests/e2e: use blueprints for testing Signed-off-by: Jens Langhammer * add identification stage assignment Signed-off-by: Jens Langhammer * add recovery flow tests Signed-off-by: Jens Langhammer --------- Signed-off-by: Jens Langhammer --- .../flows-enrollment-email-verification.yaml | 9 -- .../flows-recovery-email-verification.yaml | 2 +- tests/e2e/test_flows_enroll.py | 131 +++--------------- tests/e2e/test_flows_recovery.py | 126 +++++++++++++++++ 4 files changed, 144 insertions(+), 124 deletions(-) create mode 100644 tests/e2e/test_flows_recovery.py diff --git a/blueprints/example/flows-enrollment-email-verification.yaml b/blueprints/example/flows-enrollment-email-verification.yaml index 3bd8b1dca..30648e9e6 100644 --- a/blueprints/example/flows-enrollment-email-verification.yaml +++ b/blueprints/example/flows-enrollment-email-verification.yaml @@ -79,15 +79,6 @@ entries: model: authentik_stages_email.emailstage attrs: use_global_settings: true - host: localhost - port: 25 - username: "" - use_tls: false - use_ssl: false - timeout: 10 - from_address: system@authentik.local - token_expiry: 30 - subject: authentik template: email/account_confirmation.html activate_user_on_success: true - identifiers: diff --git a/blueprints/example/flows-recovery-email-verification.yaml b/blueprints/example/flows-recovery-email-verification.yaml index 13e0da8aa..1f4c6ca43 100644 --- a/blueprints/example/flows-recovery-email-verification.yaml +++ b/blueprints/example/flows-recovery-email-verification.yaml @@ -43,7 +43,7 @@ entries: model: authentik_policies_expression.expressionpolicy attrs: expression: | - return request.context.get('is_restored', False) + return bool(request.context.get('is_restored', True)) - identifiers: name: default-recovery-email id: default-recovery-email diff --git a/tests/e2e/test_flows_enroll.py b/tests/e2e/test_flows_enroll.py index 9e5880952..5f9865c5e 100644 --- a/tests/e2e/test_flows_enroll.py +++ b/tests/e2e/test_flows_enroll.py @@ -4,6 +4,7 @@ from time import sleep from typing import Any, Optional from unittest.case import skipUnless +from django.test import override_settings from docker.types import Healthcheck from selenium.webdriver.common.by import By from selenium.webdriver.support import expected_conditions as ec @@ -11,14 +12,8 @@ from selenium.webdriver.support.wait import WebDriverWait from authentik.blueprints.tests import apply_blueprint from authentik.core.models import User -from authentik.core.tests.utils import create_test_flow -from authentik.flows.models import FlowDesignation, FlowStageBinding -from authentik.lib.generators import generate_id -from authentik.stages.email.models import EmailStage, EmailTemplates +from authentik.flows.models import Flow from authentik.stages.identification.models import IdentificationStage -from authentik.stages.prompt.models import FieldTypes, Prompt, PromptStage -from authentik.stages.user_login.models import UserLoginStage -from authentik.stages.user_write.models import UserWriteStage from tests.e2e.utils import SeleniumTestCase, retry @@ -44,61 +39,18 @@ class TestFlowsEnroll(SeleniumTestCase): "default/flow-default-authentication-flow.yaml", "default/flow-default-invalidation-flow.yaml", ) + @apply_blueprint( + "example/flows-enrollment-2-stage.yaml", + ) def test_enroll_2_step(self): """Test 2-step enroll flow""" - # First stage fields - username_prompt = Prompt.objects.create( - name=generate_id(), - field_key="username", - label="Username", - order=0, - type=FieldTypes.TEXT, - ) - password = Prompt.objects.create( - name=generate_id(), - field_key="password", - label="Password", - order=1, - type=FieldTypes.PASSWORD, - ) - password_repeat = Prompt.objects.create( - name=generate_id(), - field_key="password_repeat", - label="Password (repeat)", - order=2, - type=FieldTypes.PASSWORD, - ) - - # Second stage fields - name_field = Prompt.objects.create( - name=generate_id(), field_key="name", label="Name", order=0, type=FieldTypes.TEXT - ) - email = Prompt.objects.create( - name=generate_id(), field_key="email", label="E-Mail", order=1, type=FieldTypes.EMAIL - ) - - # Stages - first_stage = PromptStage.objects.create(name=generate_id()) - first_stage.fields.set([username_prompt, password, password_repeat]) - first_stage.save() - second_stage = PromptStage.objects.create(name=generate_id()) - second_stage.fields.set([name_field, email]) - second_stage.save() - user_write = UserWriteStage.objects.create(name=generate_id()) - user_login = UserLoginStage.objects.create(name=generate_id()) - - flow = create_test_flow(FlowDesignation.ENROLLMENT) - # Attach enrollment flow to identification stage - ident_stage: IdentificationStage = IdentificationStage.objects.first() - ident_stage.enrollment_flow = flow + ident_stage: IdentificationStage = IdentificationStage.objects.get( + name="default-authentication-identification" + ) + ident_stage.enrollment_flow = Flow.objects.get(slug="default-enrollment-flow") ident_stage.save() - FlowStageBinding.objects.create(target=flow, stage=first_stage, order=0) - FlowStageBinding.objects.create(target=flow, stage=second_stage, order=1) - FlowStageBinding.objects.create(target=flow, stage=user_write, order=2) - FlowStageBinding.objects.create(target=flow, stage=user_login, order=3) - self.driver.get(self.live_server_url) self.initial_stages() @@ -119,68 +71,19 @@ class TestFlowsEnroll(SeleniumTestCase): "default/flow-default-authentication-flow.yaml", "default/flow-default-invalidation-flow.yaml", ) + @apply_blueprint( + "example/flows-enrollment-email-verification.yaml", + ) + @override_settings(EMAIL_PORT=1025) def test_enroll_email(self): """Test enroll with Email verification""" - # First stage fields - username_prompt = Prompt.objects.create( - name=generate_id(), - field_key="username", - label="Username", - order=0, - type=FieldTypes.TEXT, - ) - password = Prompt.objects.create( - name=generate_id(), - field_key="password", - label="Password", - order=1, - type=FieldTypes.PASSWORD, - ) - password_repeat = Prompt.objects.create( - name=generate_id(), - field_key="password_repeat", - label="Password (repeat)", - order=2, - type=FieldTypes.PASSWORD, - ) - - # Second stage fields - name_field = Prompt.objects.create( - name=generate_id(), field_key="name", label="Name", order=0, type=FieldTypes.TEXT - ) - email = Prompt.objects.create( - name=generate_id(), field_key="email", label="E-Mail", order=1, type=FieldTypes.EMAIL - ) - - # Stages - first_stage = PromptStage.objects.create(name=generate_id()) - first_stage.fields.set([username_prompt, password, password_repeat]) - first_stage.save() - second_stage = PromptStage.objects.create(name=generate_id()) - second_stage.fields.set([name_field, email]) - second_stage.save() - email_stage = EmailStage.objects.create( - name=generate_id(), - host="localhost", - port=1025, - template=EmailTemplates.ACCOUNT_CONFIRM, - ) - user_write = UserWriteStage.objects.create(name=generate_id()) - user_login = UserLoginStage.objects.create(name=generate_id()) - - flow = create_test_flow(FlowDesignation.ENROLLMENT) - # Attach enrollment flow to identification stage - ident_stage: IdentificationStage = IdentificationStage.objects.first() - ident_stage.enrollment_flow = flow + ident_stage: IdentificationStage = IdentificationStage.objects.get( + name="default-authentication-identification" + ) + ident_stage.enrollment_flow = Flow.objects.get(slug="default-enrollment-flow") ident_stage.save() - FlowStageBinding.objects.create(target=flow, stage=first_stage, order=0) - FlowStageBinding.objects.create(target=flow, stage=second_stage, order=1) - FlowStageBinding.objects.create(target=flow, stage=user_write, order=2) - FlowStageBinding.objects.create(target=flow, stage=email_stage, order=3) - FlowStageBinding.objects.create(target=flow, stage=user_login, order=4) - self.driver.get(self.live_server_url) self.initial_stages() diff --git a/tests/e2e/test_flows_recovery.py b/tests/e2e/test_flows_recovery.py new file mode 100644 index 000000000..f548b3a28 --- /dev/null +++ b/tests/e2e/test_flows_recovery.py @@ -0,0 +1,126 @@ +"""Test recovery flow""" +from sys import platform +from time import sleep +from typing import Any, Optional +from unittest.case import skipUnless + +from django.test import override_settings +from docker.types import Healthcheck +from selenium.webdriver.common.by import By +from selenium.webdriver.support import expected_conditions as ec +from selenium.webdriver.support.wait import WebDriverWait + +from authentik.blueprints.tests import apply_blueprint +from authentik.core.models import User +from authentik.core.tests.utils import create_test_admin_user +from authentik.flows.models import Flow +from authentik.lib.generators import generate_id +from authentik.stages.identification.models import IdentificationStage +from tests.e2e.utils import SeleniumTestCase, retry + + +@skipUnless(platform.startswith("linux"), "requires local docker") +class TestFlowsRecovery(SeleniumTestCase): + """Test Recovery flow""" + + def get_container_specs(self) -> Optional[dict[str, Any]]: + return { + "image": "mailhog/mailhog:v1.0.1", + "detach": True, + "network_mode": "host", + "auto_remove": True, + "healthcheck": Healthcheck( + test=["CMD", "wget", "--spider", "http://localhost:8025"], + interval=5 * 100 * 1000000, + start_period=1 * 100 * 1000000, + ), + } + + def initial_stages(self, user: User): + """Fill out initial stages""" + # Identification stage, click recovery + flow_executor = self.get_shadow_root("ak-flow-executor") + identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor) + wait = WebDriverWait(identification_stage, self.wait_timeout) + + wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "#recovery"))) + identification_stage.find_element(By.CSS_SELECTOR, "#recovery").click() + + # First prompt stage + flow_executor = self.get_shadow_root("ak-flow-executor") + identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor) + wait = WebDriverWait(identification_stage, self.wait_timeout) + + wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "input[name=uidField]"))) + identification_stage.find_element(By.CSS_SELECTOR, "input[name=uidField]").send_keys( + user.username + ) + identification_stage.find_element(By.CSS_SELECTOR, ".pf-c-button").click() + + @retry() + @apply_blueprint( + "default/flow-default-authentication-flow.yaml", + "default/flow-default-invalidation-flow.yaml", + ) + @apply_blueprint( + "example/flows-recovery-email-verification.yaml", + ) + @override_settings(EMAIL_PORT=1025) + def test_recover_email(self): + """Test recovery with Email verification""" + # Attach recovery flow to identification stage + ident_stage: IdentificationStage = IdentificationStage.objects.get( + name="default-authentication-identification" + ) + ident_stage.recovery_flow = Flow.objects.filter(slug="default-recovery-flow").first() + ident_stage.save() + + user = create_test_admin_user() + + self.driver.get(self.live_server_url) + self.initial_stages(user) + + # Email stage + flow_executor = self.get_shadow_root("ak-flow-executor") + email_stage = self.get_shadow_root("ak-stage-email", flow_executor) + + wait = WebDriverWait(email_stage, self.wait_timeout) + + # Wait for the success message so we know the email is sent + wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, ".pf-c-form p"))) + + # Open Mailhog + self.driver.get("http://localhost:8025") + + # Click on first message + self.wait.until(ec.presence_of_element_located((By.CLASS_NAME, "msglist-message"))) + self.driver.find_element(By.CLASS_NAME, "msglist-message").click() + self.driver.switch_to.frame(self.driver.find_element(By.CLASS_NAME, "tab-pane")) + self.driver.find_element(By.ID, "confirm").click() + self.driver.close() + self.driver.switch_to.window(self.driver.window_handles[0]) + + sleep(2) + # We can now enter the new password + flow_executor = self.get_shadow_root("ak-flow-executor") + prompt_stage = self.get_shadow_root("ak-stage-prompt", flow_executor) + wait = WebDriverWait(prompt_stage, self.wait_timeout) + + new_password = generate_id() + + wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "input[name=password]"))) + prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(new_password) + prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password_repeat]").send_keys( + new_password + ) + prompt_stage.find_element(By.CSS_SELECTOR, ".pf-c-button").click() + + # We're now logged in + wait = WebDriverWait(self.get_shadow_root("ak-interface-user"), self.wait_timeout) + + wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, ".pf-c-page__header"))) + self.driver.get(self.if_user_url("/settings")) + + self.assert_user(user) + user.refresh_from_db() + self.assertTrue(user.check_password(new_password))