tests/e2e: add login method

tests/e2e/test_flows_authenticators.py

@@ -38,30 +38,7 @@ class TestFlowsAuthenticator(SeleniumTestCase):
         )
 
         self.driver.get(f"{self.live_server_url}/flows/{flow.slug}/")
-
+        self.login()
-        flow_executor = self.get_shadow_root("ak-flow-executor")
-        identification_stage = self.get_shadow_root(
-            "ak-stage-identification", flow_executor
-        )
-
-        identification_stage.find_element(
-            By.CSS_SELECTOR, "input[name=uid_field]"
-        ).click()
-        identification_stage.find_element(
-            By.CSS_SELECTOR, "input[name=uid_field]"
-        ).send_keys(USER().username)
-        identification_stage.find_element(
-            By.CSS_SELECTOR, "input[name=uid_field]"
-        ).send_keys(Keys.ENTER)
-
-        flow_executor = self.get_shadow_root("ak-flow-executor")
-        password_stage = self.get_shadow_root("ak-stage-password", flow_executor)
-        password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
-            USER().username
-        )
-        password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
-            Keys.ENTER
-        )
 
         # Get expected token
         totp = TOTP(device.bin_key, device.step, device.t0, device.digits, device.drift)
@@ -89,30 +66,8 @@ class TestFlowsAuthenticator(SeleniumTestCase):
         flow: Flow = Flow.objects.get(slug="default-authentication-flow")
 
         self.driver.get(f"{self.live_server_url}/flows/{flow.slug}/")
+        self.login()
 
-        flow_executor = self.get_shadow_root("ak-flow-executor")
-        identification_stage = self.get_shadow_root(
-            "ak-stage-identification", flow_executor
-        )
-
-        identification_stage.find_element(
-            By.CSS_SELECTOR, "input[name=uid_field]"
-        ).click()
-        identification_stage.find_element(
-            By.CSS_SELECTOR, "input[name=uid_field]"
-        ).send_keys(USER().username)
-        identification_stage.find_element(
-            By.CSS_SELECTOR, "input[name=uid_field]"
-        ).send_keys(Keys.ENTER)
-
-        flow_executor = self.get_shadow_root("ak-flow-executor")
-        password_stage = self.get_shadow_root("ak-stage-password", flow_executor)
-        password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
-            USER().username
-        )
-        password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
-            Keys.ENTER
-        )
         self.wait_for_url(self.shell_url("/library"))
         self.assert_user(USER())
 
@@ -158,30 +113,8 @@ class TestFlowsAuthenticator(SeleniumTestCase):
         flow: Flow = Flow.objects.get(slug="default-authentication-flow")
 
         self.driver.get(f"{self.live_server_url}/flows/{flow.slug}/")
+        self.login()
 
-        flow_executor = self.get_shadow_root("ak-flow-executor")
-        identification_stage = self.get_shadow_root(
-            "ak-stage-identification", flow_executor
-        )
-
-        identification_stage.find_element(
-            By.CSS_SELECTOR, "input[name=uid_field]"
-        ).click()
-        identification_stage.find_element(
-            By.CSS_SELECTOR, "input[name=uid_field]"
-        ).send_keys(USER().username)
-        identification_stage.find_element(
-            By.CSS_SELECTOR, "input[name=uid_field]"
-        ).send_keys(Keys.ENTER)
-
-        flow_executor = self.get_shadow_root("ak-flow-executor")
-        password_stage = self.get_shadow_root("ak-stage-password", flow_executor)
-        password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
-            USER().username
-        )
-        password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
-            Keys.ENTER
-        )
         self.wait_for_url(self.shell_url("/library"))
         self.assert_user(USER())
 

tests/e2e/test_provider_oauth2_github.py

@@ -6,7 +6,6 @@ from unittest.case import skipUnless
 
 from docker.types import Healthcheck
 from selenium.webdriver.common.by import By
-from selenium.webdriver.common.keys import Keys
 from selenium.webdriver.support import expected_conditions as ec
 
 from authentik.core.models import Application
@@ -84,11 +83,7 @@ class TestProviderOAuth2Github(SeleniumTestCase):
 
         self.driver.get("http://localhost:3000")
         self.driver.find_element(By.CLASS_NAME, "btn-service--github").click()
-        self.driver.find_element(By.ID, "id_uid_field").click()
+        self.login()
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
-        self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
 
         self.wait_for_url("http://localhost:3000/?orgId=1")
         self.driver.get("http://localhost:3000/profile")
@@ -138,11 +133,7 @@ class TestProviderOAuth2Github(SeleniumTestCase):
 
         self.driver.get("http://localhost:3000")
         self.driver.find_element(By.CLASS_NAME, "btn-service--github").click()
-        self.driver.find_element(By.ID, "id_uid_field").click()
+        self.login()
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
-        self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
 
         sleep(1)
 
@@ -212,11 +203,7 @@ class TestProviderOAuth2Github(SeleniumTestCase):
 
         self.driver.get("http://localhost:3000")
         self.driver.find_element(By.CLASS_NAME, "btn-service--github").click()
-        self.driver.find_element(By.ID, "id_uid_field").click()
+        self.login()
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
-        self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
 
         self.wait.until(
             ec.presence_of_element_located((By.CSS_SELECTOR, "header > h1"))

tests/e2e/test_provider_oauth2_grafana.py

@@ -6,7 +6,6 @@ from unittest.case import skipUnless
 
 from docker.types import Healthcheck
 from selenium.webdriver.common.by import By
-from selenium.webdriver.common.keys import Keys
 from selenium.webdriver.support import expected_conditions as ec
 from structlog.stdlib import get_logger
 
@@ -143,11 +142,7 @@ class TestProviderOAuth2OAuth(SeleniumTestCase):
 
         self.driver.get("http://localhost:3000")
         self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
-        self.driver.find_element(By.ID, "id_uid_field").click()
+        self.login()
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
-        self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
         self.wait_for_url("http://localhost:3000/?orgId=1")
         self.driver.get("http://localhost:3000/profile")
         self.assertEqual(
@@ -204,11 +199,7 @@ class TestProviderOAuth2OAuth(SeleniumTestCase):
 
         self.driver.get("http://localhost:3000")
         self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
-        self.driver.find_element(By.ID, "id_uid_field").click()
+        self.login()
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
-        self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
         self.wait_for_url("http://localhost:3000/?orgId=1")
         self.driver.get("http://localhost:3000/profile")
         self.assertEqual(
@@ -273,11 +264,7 @@ class TestProviderOAuth2OAuth(SeleniumTestCase):
 
         self.driver.get("http://localhost:3000")
         self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
-        self.driver.find_element(By.ID, "id_uid_field").click()
+        self.login()
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
-        self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
 
         self.assertEqual(
             app.name,
@@ -350,11 +337,7 @@ class TestProviderOAuth2OAuth(SeleniumTestCase):
         PolicyBinding.objects.create(target=app, policy=negative_policy, order=0)
         self.driver.get("http://localhost:3000")
         self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
-        self.driver.find_element(By.ID, "id_uid_field").click()
+        self.login()
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
-        self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
 
         self.wait.until(
             ec.presence_of_element_located((By.CSS_SELECTOR, "header > h1"))

tests/e2e/test_provider_oauth2_oidc.py

@@ -8,7 +8,6 @@ from docker import DockerClient, from_env
 from docker.models.containers import Container
 from docker.types import Healthcheck
 from selenium.webdriver.common.by import By
-from selenium.webdriver.common.keys import Keys
 from selenium.webdriver.support import expected_conditions as ec
 from structlog.stdlib import get_logger
 
@@ -138,13 +137,7 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
         self.container = self.setup_client()
 
         self.driver.get("http://localhost:9009")
-
+        self.login()
-        self.driver.find_element(By.ID, "id_uid_field").click()
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
-        self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
-
         self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "pre")))
         body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
 
@@ -188,13 +181,7 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
         self.container = self.setup_client()
 
         self.driver.get("http://localhost:9009")
-
+        self.login()
-        self.driver.find_element(By.ID, "id_uid_field").click()
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
-        self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
-
         self.assertEqual(
             app.name,
             self.driver.find_element(By.ID, "application-name").text,
@@ -253,13 +240,7 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
 
         self.container = self.setup_client()
         self.driver.get("http://localhost:9009")
-
+        self.login()
-        self.driver.find_element(By.ID, "id_uid_field").click()
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
-        self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
-        self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
-
         self.wait.until(
             ec.presence_of_element_located((By.CSS_SELECTOR, "header > h1"))
         )

tests/e2e/utils.py

@@ -11,6 +11,8 @@ from typing import Any, Callable, Optional
 from django.apps import apps
 from django.contrib.staticfiles.testing import StaticLiveServerTestCase
 from django.db import connection, transaction
+from django.db.migrations.loader import MigrationLoader
+from django.db.migrations.operations.special import RunPython
 from django.db.utils import IntegrityError
 from django.test.testcases import TransactionTestCase
 from django.urls import reverse
@@ -24,6 +26,7 @@ from selenium.common.exceptions import (
 )
 from selenium.webdriver.common.by import By
 from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
+from selenium.webdriver.common.keys import Keys
 from selenium.webdriver.remote.webdriver import WebDriver
 from selenium.webdriver.remote.webelement import WebElement
 from selenium.webdriver.support.ui import WebDriverWait
@@ -127,6 +130,32 @@ class SeleniumTestCase(StaticLiveServerTestCase):
         )
         return element
 
+    def login(self):
+        """Do entire login flow and check user afterwards"""
+        flow_executor = self.get_shadow_root("ak-flow-executor")
+        identification_stage = self.get_shadow_root(
+            "ak-stage-identification", flow_executor
+        )
+
+        identification_stage.find_element(
+            By.CSS_SELECTOR, "input[name=uid_field]"
+        ).click()
+        identification_stage.find_element(
+            By.CSS_SELECTOR, "input[name=uid_field]"
+        ).send_keys(USER().username)
+        identification_stage.find_element(
+            By.CSS_SELECTOR, "input[name=uid_field]"
+        ).send_keys(Keys.ENTER)
+
+        flow_executor = self.get_shadow_root("ak-flow-executor")
+        password_stage = self.get_shadow_root("ak-stage-password", flow_executor)
+        password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
+            USER().username
+        )
+        password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
+            Keys.ENTER
+        )
+
     def assert_user(self, expected_user: User):
         """Check users/me API and assert it matches expected_user"""
         self.driver.get(self.url("authentik_api:user-me") + "?format=json")
@@ -168,7 +197,30 @@ class SeleniumTestCase(StaticLiveServerTestCase):
         ObjectManager().run()
 
 
-def retry(max_retires=3, exceptions=None):
+def apply_migration(app_name: str, migration_name: str):
+    """Re-apply migrations that create objects using RunPython before test cases"""
+
+    def wrapper_outter(func: Callable):
+        """Retry test multiple times"""
+
+        LOADER = MigrationLoader(connection)
+
+        @wraps(func)
+        def wrapper(self: TransactionTestCase, *args, **kwargs):
+            migration = LOADER.get_migration(app_name, migration_name)
+            with connection.schema_editor() as schema_editor:
+                for operation in migration.operations:
+                    if not isinstance(operation, RunPython):
+                        continue
+                    operation.code(apps, schema_editor)
+            return func(self, *args, **kwargs)
+
+        return wrapper
+
+    return wrapper_outter
+
+
+def retry(max_retires=1, exceptions=None):
     """Retry test multiple times. Default to catching Selenium Timeout Exception"""
 
     if not exceptions: