providers/oauth2: fix redirect uri input

[BeryJu]

Details

REPLACE ME

---

Checklist

• [ ] Local tests pass (ak test authentik/)
• [ ] The code has been formatted (make lint-fix)

If an API change has been made

• [ ] The API schema has been updated (make gen-build)

If changes to the frontend have been made

• [ ] The code has been formatted (make web)

If applicable

• [ ] The documentation has been updated
• [ ] The documentation has been formatted (make website)

[netlify[bot]]

✅ Deploy Preview for authentik-storybook ready!

Name	Link
🔨 Latest commit	7e63ab19687c2369650bfa75e4223c8099a34411
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-storybook/deploys/673f593c7b4292000700fc1a
😎 Deploy Preview	https://deploy-preview-12122--authentik-storybook.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify site configuration.

[netlify[bot]]

✅ Deploy Preview for authentik-docs canceled.

Name	Link
🔨 Latest commit	7e63ab19687c2369650bfa75e4223c8099a34411
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-docs/deploys/673f593d95d7cc00082c15c6

[BeryJu]

/cherry-pick version-2024.10

[BeryJu]

/cherry-pick version-2024.8

[codecov[bot]]

:x: 6 Tests Failed:

Tests completed	Failed	Passed	Skipped
1601	6	1595	1

View the top 3 failed tests by shortest run time

> > ```python > tests.e2e.test_provider_proxy.TestProviderProxy::test_proxy_basic_auth > ``` > >

Stack Traces | 218s run time

> > > > > ```python > > self = > > call_result = > > exc_info = (, NoSuchElementException(), ) > > task_context = None, context = [<_contextvars.Context object at 0x7fe29940fdc0>] > > args = ('group_outpost_07171bf7-89f9-4076-a83c-da5c9861be49', {'type': 'event.update'}) > > kwargs = {}, __traceback_hide__ = True > > current_task = result=None> > > result = None > > > > async def main_wrap( > > self, > > call_result: "Future[_R]", > > exc_info: "OptExcInfo", > > task_context: "Optional[List[asyncio.Task[Any]]]", > > context: List[contextvars.Context], > > *args: _P.args, > > **kwargs: _P.kwargs, > > ) -> None: > > """ > > Wraps the awaitable with something that puts the result into the > > result/exception future. > > """ > > > > __traceback_hide__ = True # noqa: F841 > > > > if context is not None: > > _restore_context(context[0]) > > > > current_task = asyncio.current_task() > > if current_task is not None and task_context is not None: > > task_context.append(current_task) > > > > try: > > # If we have an exception, run the function inside the except block > > # after raising it so exc_info is correctly populated. > > if exc_info[1]: > > try: > > > raise exc_info[1] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > call_result = > > exc_info = (, NoSuchElementException(), ) > > task_context = None, context = [<_contextvars.Context object at 0x7fe298edaac0>] > > args = ('group_outpost_07171bf7-89f9-4076-a83c-da5c9861be49', {'type': 'event.update'}) > > kwargs = {}, __traceback_hide__ = True > > current_task = result=None> > > result = None > > > > async def main_wrap( > > self, > > call_result: "Future[_R]", > > exc_info: "OptExcInfo", > > task_context: "Optional[List[asyncio.Task[Any]]]", > > context: List[contextvars.Context], > > *args: _P.args, > > **kwargs: _P.kwargs, > > ) -> None: > > """ > > Wraps the awaitable with something that puts the result into the > > result/exception future. > > """ > > > > __traceback_hide__ = True # noqa: F841 > > > > if context is not None: > > _restore_context(context[0]) > > > > current_task = asyncio.current_task() > > if current_task is not None and task_context is not None: > > task_context.append(current_task) > > > > try: > > # If we have an exception, run the function inside the except block > > # after raising it so exc_info is correctly populated. > > if exc_info[1]: > > try: > > > raise exc_info[1] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > > return func(self, *args, **kwargs) > > > > tests/e2e/utils.py:287: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'system/providers-proxy.yaml' > > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, config = > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > config = apps.get_app_config(app_name) > > if isinstance(config, ManagedAppConfig): > > config._on_startup_callback(None) > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:43: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > @retry() > > @apply_blueprint( > > "default/flow-default-authentication-flow.yaml", > > "default/flow-default-invalidation-flow.yaml", > > ) > > @apply_blueprint( > > "default/flow-default-provider-authorization-implicit-consent.yaml", > > "default/flow-default-provider-invalidation.yaml", > > ) > > @apply_blueprint( > > "system/providers-oauth2.yaml", > > "system/providers-proxy.yaml", > > ) > > @reconcile_app("authentik_crypto") > > def test_proxy_basic_auth(self): > > """Test simple outpost setup with single provider""" > > cred = generate_id() > > attr = "basic-password" # nosec > > self.user.attributes["basic-username"] = cred > > self.user.attributes[attr] = cred > > self.user.save() > > > > proxy: ProxyProvider = ProxyProvider.objects.create( > > name=generate_id(), > > authorization_flow=Flow.objects.get( > > slug="default-provider-authorization-implicit-consent" > > ), > > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > > internal_host=f"http://{self.host}", > > external_host="http://localhost:9000", > > basic_auth_enabled=True, > > basic_auth_user_attribute="basic-username", > > basic_auth_password_attribute=attr, > > ) > > # Ensure OAuth2 Params are set > > proxy.set_oauth_defaults() > > proxy.save() > > # we need to create an application to actually access the proxy > > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > > outpost: Outpost = Outpost.objects.create( > > name=generate_id(), > > type=OutpostType.PROXY, > > ) > > outpost.providers.add(proxy) > > outpost.build_user_permissions(outpost.user) > > > > self.start_proxy(outpost) > > > > # Wait until outpost healthcheck succeeds > > healthcheck_retries = 0 > > while healthcheck_retries < 50: # noqa: PLR2004 > > if len(outpost.state) > 0: > > state = outpost.state[0] > > if state.last_seen: > > break > > healthcheck_retries += 1 > > sleep(0.5) > > sleep(5) > > > > self.driver.get("http://localhost:9000/api") > > > self.login() > > > > tests/e2e/test_provider_proxy.py:184: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > def login(self): > > """Do entire login flow and check user afterwards""" > > > flow_executor = self.get_shadow_root("ak-flow-executor") > > > > tests/e2e/utils.py:232: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > selector = 'ak-flow-executor' > > container = > > > > def get_shadow_root( > > self, selector: str, container: WebElement | WebDriver | None = None > > ) -> WebElement: > > """Get shadow root element's inner shadowRoot""" > > if not container: > > container = self.driver > > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > > > tests/e2e/utils.py:226: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > by = 'css selector', value = 'ak-flow-executor' > > > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > > """Find an element given a By strategy and locator. > > > > :Usage: > > :: > > > > element = driver.find_element(By.ID, 'foo') > > > > :rtype: WebElement > > """ > > by, value = self.locator_converter.convert(by, value) > > > > if isinstance(by, RelativeBy): > > elements = self.find_elements(by=by, value=value) > > if not elements: > > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > > return elements[0] > > > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > driver_command = 'findElement' > > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > > > def execute(self, driver_command: str, params: dict = None) -> dict: > > """Sends a command to be executed by a command.CommandExecutor. > > > > :Args: > > - driver_command: The name of the command to execute as a string. > > - params: A dictionary of named parameters to send with the command. > > > > :Returns: > > The command's JSON response loaded into a dictionary object. > > """ > > params = self._wrap_value(params) > > > > if self.session_id: > > if not params: > > params = {"sessionId": self.session_id} > > elif "sessionId" not in params: > > params["sessionId"] = self.session_id > > > > response = self.command_executor.execute(driver_command, params) > > if response: > > > self.error_handler.check_response(response) > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x56291555bfc3 \\u003Cunknown>\\n#17 0x56291556c944 \\u003Cunknown>\\n#18 0x7faa3b3baac3 \\u003Cunknown>\\n"}}'} > > > > def check_response(self, response: Dict[str, Any]) -> None: > > """Checks that a JSON response from the WebDriver does not have an > > error. > > > > :Args: > > - response - The JSON response from the WebDriver server as a dictionary > > object. > > > > :Raises: If the response contains an error message. > > """ > > status = response.get("status", None) > > if not status or status == ErrorCode.SUCCESS: > > return > > value = None > > message = response.get("message", "") > > screen: str = response.get("screen", "") > > stacktrace = None > > if isinstance(status, int): > > value_json = response.get("value", None) > > if value_json and isinstance(value_json, str): > > import json > > > > try: > > value = json.loads(value_json) > > if len(value) == 1: > > value = value["value"] > > status = value.get("error", None) > > if not status: > > status = value.get("status", ErrorCode.UNKNOWN_ERROR) > > message = value.get("value") or value.get("message") > > if not isinstance(message, str): > > value = message > > message = message.get("message") > > else: > > message = value.get("message", None) > > except ValueError: > > pass > > > > exception_class: Type[WebDriverException] > > e = ErrorCode() > > error_codes = [item for item in dir(e) if not item.startswith("__")] > > for error_code in error_codes: > > error_info = getattr(ErrorCode, error_code) > > if isinstance(error_info, list) and status in error_info: > > exception_class = getattr(ExceptionMapping, error_code, WebDriverException) > > break > > else: > > exception_class = WebDriverException > > > > if not value: > > value = response["value"] > > if isinstance(value, str): > > raise exception_class(value) > > if message == "" and "message" in value: > > message = value["message"] > > > > screen = None # type: ignore[assignment] > > if "screen" in value: > > screen = value["screen"] > > > > stacktrace = None > > st_value = value.get("stackTrace") or value.get("stacktrace") > > if st_value: > > if isinstance(st_value, str): > > stacktrace = st_value.split("\n") > > else: > > stacktrace = [] > > try: > > for frame in st_value: > > line = frame.get("lineNumber", "") > > file = frame.get("fileName", "") > > if line: > > file = f"{file}:{line}" > > meth = frame.get("methodName", "") > > if "className" in frame: > > meth = f"{frame['className']}.{meth}" > > msg = " at %s (%s)" > > msg = msg % (meth, file) > > stacktrace.append(msg) > > except TypeError: > > pass > > if exception_class == UnexpectedAlertPresentException: > > alert_text = None > > if "data" in value: > > alert_text = value["data"].get("text") > > elif "alert" in value: > > alert_text = value["alert"].get("text") > > raise exception_class(message, screen, stacktrace, alert_text) # type: ignore[call-arg] # mypy is not smart enough here > > > raise exception_class(message, screen, stacktrace) > > E selenium.common.exceptions.NoSuchElementException: Message: no such element: Unable to locate element: {"method":"css selector","selector":"ak-flow-executor"} > > E (Session info: chrome=122.0.6261.69); For documentation on this error, please visit: https://www.selenium..../webdriver/troubleshooting/errors#no-such-element-exception > > E Stacktrace: > > E #0 0x56291556d793 > > E #1 0x5629152611c6 > > E #2 0x5629152ac358 > > E #3 0x5629152ac411 > > E #4 0x5629152ef934 > > E #5 0x5629152ce3fd > > E #6 0x5629152ecdd9 > > E #7 0x5629152ce173 > > E #8 0x56291529f2d3 > > E #9 0x56291529fc9e > > E #10 0x5629155318cb > > E #11 0x562915535745 > > E #12 0x56291551e2e1 > > E #13 0x5629155362d2 > > E #14 0x56291550217f > > E #15 0x56291555bdc8 > > E #16 0x56291555bfc3 > > E #17 0x56291556c944 > > E #18 0x7faa3b3baac3 > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > > > During handling of the above exception, another exception occurred: > > > > self = > > call_result = > > exc_info = (, NoSuchElementException(), ) > > task_context = None, context = [<_contextvars.Context object at 0x7fe2991fa900>] > > args = ('group_outpost_1c1da534-9f88-4fa6-947c-b894686fddcf', {'type': 'event.update'}) > > kwargs = {}, __traceback_hide__ = True > > current_task = result=None> > > result = None > > > > async def main_wrap( > > self, > > call_result: "Future[_R]", > > exc_info: "OptExcInfo", > > task_context: "Optional[List[asyncio.Task[Any]]]", > > context: List[contextvars.Context], > > *args: _P.args, > > **kwargs: _P.kwargs, > > ) -> None: > > """ > > Wraps the awaitable with something that puts the result into the > > result/exception future. > > """ > > > > __traceback_hide__ = True # noqa: F841 > > > > if context is not None: > > _restore_context(context[0]) > > > > current_task = asyncio.current_task() > > if current_task is not None and task_context is not None: > > task_context.append(current_task) > > > > try: > > # If we have an exception, run the function inside the except block > > # after raising it so exc_info is correctly populated. > > if exc_info[1]: > > try: > > > raise exc_info[1] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > call_result = > > exc_info = (, NoSuchElementException(), ) > > task_context = None, context = [<_contextvars.Context object at 0x7fe29c438280>] > > args = ('group_outpost_1c1da534-9f88-4fa6-947c-b894686fddcf', {'type': 'event.update'}) > > kwargs = {}, __traceback_hide__ = True > > current_task = result=None> > > result = None > > > > async def main_wrap( > > self, > > call_result: "Future[_R]", > > exc_info: "OptExcInfo", > > task_context: "Optional[List[asyncio.Task[Any]]]", > > context: List[contextvars.Context], > > *args: _P.args, > > **kwargs: _P.kwargs, > > ) -> None: > > """ > > Wraps the awaitable with something that puts the result into the > > result/exception future. > > """ > > > > __traceback_hide__ = True # noqa: F841 > > > > if context is not None: > > _restore_context(context[0]) > > > > current_task = asyncio.current_task() > > if current_task is not None and task_context is not None: > > task_context.append(current_task) > > > > try: > > # If we have an exception, run the function inside the except block > > # after raising it so exc_info is correctly populated. > > if exc_info[1]: > > try: > > > raise exc_info[1] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > > return func(self, *args, **kwargs) > > > > tests/e2e/utils.py:287: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'system/providers-proxy.yaml' > > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, config = > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > config = apps.get_app_config(app_name) > > if isinstance(config, ManagedAppConfig): > > config._on_startup_callback(None) > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:43: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > @retry() > > @apply_blueprint( > > "default/flow-default-authentication-flow.yaml", > > "default/flow-default-invalidation-flow.yaml", > > ) > > @apply_blueprint( > > "default/flow-default-provider-authorization-implicit-consent.yaml", > > "default/flow-default-provider-invalidation.yaml", > > ) > > @apply_blueprint( > > "system/providers-oauth2.yaml", > > "system/providers-proxy.yaml", > > ) > > @reconcile_app("authentik_crypto") > > def test_proxy_basic_auth(self): > > """Test simple outpost setup with single provider""" > > cred = generate_id() > > attr = "basic-password" # nosec > > self.user.attributes["basic-username"] = cred > > self.user.attributes[attr] = cred > > self.user.save() > > > > proxy: ProxyProvider = ProxyProvider.objects.create( > > name=generate_id(), > > authorization_flow=Flow.objects.get( > > slug="default-provider-authorization-implicit-consent" > > ), > > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > > internal_host=f"http://{self.host}", > > external_host="http://localhost:9000", > > basic_auth_enabled=True, > > basic_auth_user_attribute="basic-username", > > basic_auth_password_attribute=attr, > > ) > > # Ensure OAuth2 Params are set > > proxy.set_oauth_defaults() > > proxy.save() > > # we need to create an application to actually access the proxy > > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > > outpost: Outpost = Outpost.objects.create( > > name=generate_id(), > > type=OutpostType.PROXY, > > ) > > outpost.providers.add(proxy) > > outpost.build_user_permissions(outpost.user) > > > > self.start_proxy(outpost) > > > > # Wait until outpost healthcheck succeeds > > healthcheck_retries = 0 > > while healthcheck_retries < 50: # noqa: PLR2004 > > if len(outpost.state) > 0: > > state = outpost.state[0] > > if state.last_seen: > > break > > healthcheck_retries += 1 > > sleep(0.5) > > sleep(5) > > > > self.driver.get("http://localhost:9000/api") > > > self.login() > > > > tests/e2e/test_provider_proxy.py:184: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > def login(self): > > """Do entire login flow and check user afterwards""" > > > flow_executor = self.get_shadow_root("ak-flow-executor") > > > > tests/e2e/utils.py:232: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > selector = 'ak-flow-executor' > > container = > > > > def get_shadow_root( > > self, selector: str, container: WebElement | WebDriver | None = None > > ) -> WebElement: > > """Get shadow root element's inner shadowRoot""" > > if not container: > > container = self.driver > > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > > > tests/e2e/utils.py:226: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > by = 'css selector', value = 'ak-flow-executor' > > > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > > """Find an element given a By strategy and locator. > > > > :Usage: > > :: > > > > element = driver.find_element(By.ID, 'foo') > > > > :rtype: WebElement > > """ > > by, value = self.locator_converter.convert(by, value) > > > > if isinstance(by, RelativeBy): > > elements = self.find_elements(by=by, value=value) > > if not elements: > > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > > return elements[0] > > > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > driver_command = 'findElement' > > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > > > def execute(self, driver_command: str, params: dict = None) -> dict: > > """Sends a command to be executed by a command.CommandExecutor. > > > > :Args: > > - driver_command: The name of the command to execute as a string. > > - params: A dictionary of named parameters to send with the command. > > > > :Returns: > > The command's JSON response loaded into a dictionary object. > > """ > > params = self._wrap_value(params) > > > > if self.session_id: > > if not params: > > params = {"sessionId": self.session_id} > > elif "sessionId" not in params: > > params["sessionId"] = self.session_id > > > > response = self.command_executor.execute(driver_command, params) > > if response: > > > self.error_handler.check_response(response) > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x55f3840c3fc3 \\u003Cunknown>\\n#17 0x55f3840d4944 \\u003Cunknown>\\n#18 0x7fd34322fac3 \\u003Cunknown>\\n"}}'} > > > > def check_response(self, response: Dict[str, Any]) -> None: > > """Checks that a JSON response from the WebDriver does not have an > > error. > > > > :Args: > > - response - The JSON response from the WebDriver server as a dictionary > > object. > > > > :Raises: If the response contains an error message. > > """ > > status = response.get("status", None) > > if not status or status == ErrorCode.SUCCESS: > > return > > value = None > > message = response.get("message", "") > > screen: str = response.get("screen", "") > > stacktrace = None > > if isinstance(status, int): > > value_json = response.get("value", None) > > if value_json and isinstance(value_json, str): > > import json > > > > try: > > value = json.loads(value_json) > > if len(value) == 1: > > value = value["value"] > > status = value.get("error", None) > > if not status: > > status = value.get("status", ErrorCode.UNKNOWN_ERROR) > > message = value.get("value") or value.get("message") > > if not isinstance(message, str): > > value = message > > message = message.get("message") > > else: > > message = value.get("message", None) > > except ValueError: > > pass > > > > exception_class: Type[WebDriverException] > > e = ErrorCode() > > error_codes = [item for item in dir(e) if not item.startswith("__")] > > for error_code in error_codes: > > error_info = getattr(ErrorCode, error_code) > > if isinstance(error_info, list) and status in error_info: > > exception_class = getattr(ExceptionMapping, error_code, WebDriverException) > > break > > else: > > exception_class = WebDriverException > > > > if not value: > > value = response["value"] > > if isinstance(value, str): > > raise exception_class(value) > > if message == "" and "message" in value: > > message = value["message"] > > > > screen = None # type: ignore[assignment] > > if "screen" in value: > > screen = value["screen"] > > > > stacktrace = None > > st_value = value.get("stackTrace") or value.get("stacktrace") > > if st_value: > > if isinstance(st_value, str): > > stacktrace = st_value.split("\n") > > else: > > stacktrace = [] > > try: > > for frame in st_value: > > line = frame.get("lineNumber", "") > > file = frame.get("fileName", "") > > if line: > > file = f"{file}:{line}" > > meth = frame.get("methodName", "") > > if "className" in frame: > > meth = f"{frame['className']}.{meth}" > > msg = " at %s (%s)" > > msg = msg % (meth, file) > > stacktrace.append(msg) > > except TypeError: > > pass > > if exception_class == UnexpectedAlertPresentException: > > alert_text = None > > if "data" in value: > > alert_text = value["data"].get("text") > > elif "alert" in value: > > alert_text = value["alert"].get("text") > > raise exception_class(message, screen, stacktrace, alert_text) # type: ignore[call-arg] # mypy is not smart enough here > > > raise exception_class(message, screen, stacktrace) > > E selenium.common.exceptions.NoSuchElementException: Message: no such element: Unable to locate element: {"method":"css selector","selector":"ak-flow-executor"} > > E (Session info: chrome=122.0.6261.69); For documentation on this error, please visit: https://www.selenium..../webdriver/troubleshooting/errors#no-such-element-exception > > E Stacktrace: > > E #0 0x55f3840d5793 > > E #1 0x55f383dc91c6 > > E #2 0x55f383e14358 > > E #3 0x55f383e14411 > > E #4 0x55f383e57934 > > E #5 0x55f383e363fd > > E #6 0x55f383e54dd9 > > E #7 0x55f383e36173 > > E #8 0x55f383e072d3 > > E #9 0x55f383e07c9e > > E #10 0x55f3840998cb > > E #11 0x55f38409d745 > > E #12 0x55f3840862e1 > > E #13 0x55f38409e2d2 > > E #14 0x55f38406a17f > > E #15 0x55f3840c3dc8 > > E #16 0x55f3840c3fc3 > > E #17 0x55f3840d4944 > > E #18 0x7fd34322fac3 > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > > > During handling of the above exception, another exception occurred: > > > > self = > > test_case = > > subTest = False > > > > @contextlib.contextmanager > > def testPartExecutor(self, test_case, subTest=False): > > old_success = self.success > > self.success = True > > try: > > > yield > > > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:58: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > result = > > > > def run(self, result=None): > > if result is None: > > result = self.defaultTestResult() > > startTestRun = getattr(result, 'startTestRun', None) > > stopTestRun = getattr(result, 'stopTestRun', None) > > if startTestRun is not None: > > startTestRun() > > else: > > stopTestRun = None > > > > result.startTest(self) > > try: > > testMethod = getattr(self, self._testMethodName) > > if (getattr(self.__class__, "__unittest_skip__", False) or > > getattr(testMethod, "__unittest_skip__", False)): > > # If the class or method was skipped. > > skip_why = (getattr(self.__class__, '__unittest_skip_why__', '') > > or getattr(testMethod, '__unittest_skip_why__', '')) > > _addSkip(result, self, skip_why) > > return result > > > > expecting_failure = ( > > getattr(self, "__unittest_expecting_failure__", False) or > > getattr(testMethod, "__unittest_expecting_failure__", False) > > ) > > outcome = _Outcome(result) > > start_time = time.perf_counter() > > try: > > self._outcome = outcome > > > > with outcome.testPartExecutor(self): > > self._callSetUp() > > if outcome.success: > > outcome.expecting_failure = expecting_failure > > with outcome.testPartExecutor(self): > > > self._callTestMethod(testMethod) > > > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:634: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > method = > > > > > def _callTestMethod(self, method): > > > if method() is not None: > > > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:589: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > return func(self, *args, **kwargs) > > > > except tuple(exceptions) as exc: > > count += 1 > > if count > max_retires: > > logger.debug("Exceeded retry count", exc=exc, test=self) > > > > raise exc > > logger.debug("Retrying on error", exc=exc, test=self) > > self.tearDown() > > self._post_teardown() > > self._pre_setup() > > self.setUp() > > > return wrapper(self, *args, **kwargs) > > > > tests/e2e/utils.py:300: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > return func(self, *args, **kwargs) > > > > except tuple(exceptions) as exc: > > count += 1 > > if count > max_retires: > > logger.debug("Exceeded retry count", exc=exc, test=self) > > > > raise exc > > logger.debug("Retrying on error", exc=exc, test=self) > > self.tearDown() > > self._post_teardown() > > self._pre_setup() > > self.setUp() > > > return wrapper(self, *args, **kwargs) > > > > tests/e2e/utils.py:300: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > return func(self, *args, **kwargs) > > > > except tuple(exceptions) as exc: > > count += 1 > > if count > max_retires: > > logger.debug("Exceeded retry count", exc=exc, test=self) > > > > > raise exc > > > > tests/e2e/utils.py:294: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > > return func(self, *args, **kwargs) > > > > tests/e2e/utils.py:287: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'system/providers-proxy.yaml' > > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, config = > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > config = apps.get_app_config(app_name) > > if isinstance(config, ManagedAppConfig): > > config._on_startup_callback(None) > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:43: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > @retry() > > @apply_blueprint( > > "default/flow-default-authentication-flow.yaml", > > "default/flow-default-invalidation-flow.yaml", > > ) > > @apply_blueprint( > > "default/flow-default-provider-authorization-implicit-consent.yaml", > > "default/flow-default-provider-invalidation.yaml", > > ) > > @apply_blueprint( > > "system/providers-oauth2.yaml", > > "system/providers-proxy.yaml", > > ) > > @reconcile_app("authentik_crypto") > > def test_proxy_basic_auth(self): > > """Test simple outpost setup with single provider""" > > cred = generate_id() > > attr = "basic-password" # nosec > > self.user.attributes["basic-username"] = cred > > self.user.attributes[attr] = cred > > self.user.save() > > > > proxy: ProxyProvider = ProxyProvider.objects.create( > > name=generate_id(), > > authorization_flow=Flow.objects.get( > > slug="default-provider-authorization-implicit-consent" > > ), > > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > > internal_host=f"http://{self.host}", > > external_host="http://localhost:9000", > > basic_auth_enabled=True, > > basic_auth_user_attribute="basic-username", > > basic_auth_password_attribute=attr, > > ) > > # Ensure OAuth2 Params are set > > proxy.set_oauth_defaults() > > proxy.save() > > # we need to create an application to actually access the proxy > > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > > outpost: Outpost = Outpost.objects.create( > > name=generate_id(), > > type=OutpostType.PROXY, > > ) > > outpost.providers.add(proxy) > > outpost.build_user_permissions(outpost.user) > > > > self.start_proxy(outpost) > > > > # Wait until outpost healthcheck succeeds > > healthcheck_retries = 0 > > while healthcheck_retries < 50: # noqa: PLR2004 > > if len(outpost.state) > 0: > > state = outpost.state[0] > > if state.last_seen: > > break > > healthcheck_retries += 1 > > sleep(0.5) > > sleep(5) > > > > self.driver.get("http://localhost:9000/api") > > > self.login() > > > > tests/e2e/test_provider_proxy.py:184: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > def login(self): > > """Do entire login flow and check user afterwards""" > > > flow_executor = self.get_shadow_root("ak-flow-executor") > > > > tests/e2e/utils.py:232: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > selector = 'ak-flow-executor' > > container = > > > > def get_shadow_root( > > self, selector: str, container: WebElement | WebDriver | None = None > > ) -> WebElement: > > """Get shadow root element's inner shadowRoot""" > > if not container: > > container = self.driver > > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > > > tests/e2e/utils.py:226: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > by = 'css selector', value = 'ak-flow-executor' > > > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > > """Find an element given a By strategy and locator. > > > > :Usage: > > :: > > > > element = driver.find_element(By.ID, 'foo') > > > > :rtype: WebElement > > """ > > by, value = self.locator_converter.convert(by, value) > > > > if isinstance(by, RelativeBy): > > elements = self.find_elements(by=by, value=value) > > if not elements: > > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > > return elements[0] > > > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > driver_command = 'findElement' > > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > > > def execute(self, driver_command: str, params: dict = None) -> dict: > > """Sends a command to be executed by a command.CommandExecutor. > > > > :Args: > > - driver_command: The name of the command to execute as a string. > > - params: A dictionary of named parameters to send with the command. > > > > :Returns: > > The command's JSON response loaded into a dictionary object. > > """ > > params = self._wrap_value(params) > > > > if self.session_id: > > if not params: > > params = {"sessionId": self.session_id} > > elif "sessionId" not in params: > > params["sessionId"] = self.session_id > > > > response = self.command_executor.execute(driver_command, params) > > if response: > > > self.error_handler.check_response(response) > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x563af4b50fc3 \\u003Cunknown>\\n#17 0x563af4b61944 \\u003Cunknown>\\n#18 0x7f7bb16abac3 \\u003Cunknown>\\n"}}'} > > > > def check_response(self, response: Dict[str, Any]) -> None: > > """Checks that a JSON response from the WebDriver does not have an > > error. > > > > :Args: > > - response - The JSON response from the WebDriver server as a dictionary > > object. > > > > :Raises: If the response contains an error message. > > """ > > status = response.get("status", None) > > if not status or status == ErrorCode.SUCCESS: > > return > > value = None > > message = response.get("message", "") > > screen: str = response.get("screen", "") > > stacktrace = None > > if isinstance(status, int): > > value_json = response.get("value", None) > > if value_json and isinstance(value_json, str): > > import json > > > > try: > > value = json.loads(value_json) > > if len(value) == 1: > > value = value["value"] > > status = value.get("error", None) > > if not status: > > status = value.get("status", ErrorCode.UNKNOWN_ERROR) > > message = value.get("value") or value.get("message") > > if not isinstance(message, str): > > value = message > > message = message.get("message") > > else: > > message = value.get("message", None) > > except ValueError: > > pass > > > > exception_class: Type[WebDriverException] > > e = ErrorCode() > > error_codes = [item for item in dir(e) if not item.startswith("__")] > > for error_code in error_codes: > > error_info = getattr(ErrorCode, error_code) > > if isinstance(error_info, list) and status in error_info: > > exception_class = getattr(ExceptionMapping, error_code, WebDriverException) > > break > > else: > > exception_class = WebDriverException > > > > if not value: > > value = response["value"] > > if isinstance(value, str): > > raise exception_class(value) > > if message == "" and "message" in value: > > message = value["message"] > > > > screen = None # type: ignore[assignment] > > if "screen" in value: > > screen = value["screen"] > > > > stacktrace = None > > st_value = value.get("stackTrace") or value.get("stacktrace") > > if st_value: > > if isinstance(st_value, str): > > stacktrace = st_value.split("\n") > > else: > > stacktrace = [] > > try: > > for frame in st_value: > > line = frame.get("lineNumber", "") > > file = frame.get("fileName", "") > > if line: > > file = f"{file}:{line}" > > meth = frame.get("methodName", "") > > if "className" in frame: > > meth = f"{frame['className']}.{meth}" > > msg = " at %s (%s)" > > msg = msg % (meth, file) > > stacktrace.append(msg) > > except TypeError: > > pass > > if exception_class == UnexpectedAlertPresentException: > > alert_text = None > > if "data" in value: > > alert_text = value["data"].get("text") > > elif "alert" in value: > > alert_text = value["alert"].get("text") > > raise exception_class(message, screen, stacktrace, alert_text) # type: ignore[call-arg] # mypy is not smart enough here > > > raise exception_class(message, screen, stacktrace) > > E selenium.common.exceptions.NoSuchElementException: Message: no such element: Unable to locate element: {"method":"css selector","selector":"ak-flow-executor"} > > E (Session info: chrome=122.0.6261.69); For documentation on this error, please visit: https://www.selenium..../webdriver/troubleshooting/errors#no-such-element-exception > > E Stacktrace: > > E #0 0x563af4b62793 > > E #1 0x563af48561c6 > > E #2 0x563af48a1358 > > E #3 0x563af48a1411 > > E #4 0x563af48e4934 > > E #5 0x563af48c33fd > > E #6 0x563af48e1dd9 > > E #7 0x563af48c3173 > > E #8 0x563af48942d3 > > E #9 0x563af4894c9e > > E #10 0x563af4b268cb > > E #11 0x563af4b2a745 > > E #12 0x563af4b132e1 > > E #13 0x563af4b2b2d2 > > E #14 0x563af4af717f > > E #15 0x563af4b50dc8 > > E #16 0x563af4b50fc3 > > E #17 0x563af4b61944 > > E #18 0x7f7bb16abac3 > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > ``` > >

tests.e2e.test_provider_proxy.TestProviderProxy::test_proxy_simple

Stack Traces | 218s run time

> > ```python > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fe28a537240>] > args = ('group_outpost_4ed06ef0-138f-4197-9ec5-3daf18d9f53b', {'type': 'event.update'}) > kwargs = {}, __traceback_hide__ = True > current_task = result=None> > result = None > > async def main_wrap( > self, > call_result: "Future[_R]", > exc_info: "OptExcInfo", > task_context: "Optional[List[asyncio.Task[Any]]]", > context: List[contextvars.Context], > *args: _P.args, > **kwargs: _P.kwargs, > ) -> None: > """ > Wraps the awaitable with something that puts the result into the > result/exception future. > """ > > __traceback_hide__ = True # noqa: F841 > > if context is not None: > _restore_context(context[0]) > > current_task = asyncio.current_task() > if current_task is not None and task_context is not None: > task_context.append(current_task) > > try: > # If we have an exception, run the function inside the except block > # after raising it so exc_info is correctly populated. > if exc_info[1]: > try: > > raise exc_info[1] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fe2990b8940>] > args = ('group_outpost_4ed06ef0-138f-4197-9ec5-3daf18d9f53b', {'type': 'event.update'}) > kwargs = {}, __traceback_hide__ = True > current_task = result=None> > result = None > > async def main_wrap( > self, > call_result: "Future[_R]", > exc_info: "OptExcInfo", > task_context: "Optional[List[asyncio.Task[Any]]]", > context: List[contextvars.Context], > *args: _P.args, > **kwargs: _P.kwargs, > ) -> None: > """ > Wraps the awaitable with something that puts the result into the > result/exception future. > """ > > __traceback_hide__ = True # noqa: F841 > > if context is not None: > _restore_context(context[0]) > > current_task = asyncio.current_task() > if current_task is not None and task_context is not None: > task_context.append(current_task) > > try: > # If we have an exception, run the function inside the except block > # after raising it so exc_info is correctly populated. > if exc_info[1]: > try: > > raise exc_info[1] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > > return func(self, *args, **kwargs) > > tests/e2e/utils.py:287: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > @wraps(func) > def wrapper(*args, **kwargs): > for file in files: > content = BlueprintInstance(path=file).retrieve() > Importer.from_string(content).apply() > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:25: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > @wraps(func) > def wrapper(*args, **kwargs): > for file in files: > content = BlueprintInstance(path=file).retrieve() > Importer.from_string(content).apply() > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:25: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, file = 'system/providers-proxy.yaml' > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > @wraps(func) > def wrapper(*args, **kwargs): > for file in files: > content = BlueprintInstance(path=file).retrieve() > Importer.from_string(content).apply() > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:25: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, config = > > @wraps(func) > def wrapper(*args, **kwargs): > config = apps.get_app_config(app_name) > if isinstance(config, ManagedAppConfig): > config._on_startup_callback(None) > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:43: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > @retry() > @apply_blueprint( > "default/flow-default-authentication-flow.yaml", > "default/flow-default-invalidation-flow.yaml", > ) > @apply_blueprint( > "default/flow-default-provider-authorization-implicit-consent.yaml", > "default/flow-default-provider-invalidation.yaml", > ) > @apply_blueprint( > "system/providers-oauth2.yaml", > "system/providers-proxy.yaml", > ) > @reconcile_app("authentik_crypto") > def test_proxy_simple(self): > """Test simple outpost setup with single provider""" > # set additionalHeaders to test later > self.user.attributes["additionalHeaders"] = {"X-Foo": "bar"} > self.user.save() > > proxy: ProxyProvider = ProxyProvider.objects.create( > name=generate_id(), > authorization_flow=Flow.objects.get( > slug="default-provider-authorization-implicit-consent" > ), > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > internal_host=f"http://{self.host}", > external_host="http://localhost:9000", > ) > # Ensure OAuth2 Params are set > proxy.set_oauth_defaults() > proxy.save() > # we need to create an application to actually access the proxy > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > outpost: Outpost = Outpost.objects.create( > name=generate_id(), > type=OutpostType.PROXY, > ) > outpost.providers.add(proxy) > outpost.build_user_permissions(outpost.user) > > self.start_proxy(outpost) > > # Wait until outpost healthcheck succeeds > healthcheck_retries = 0 > while healthcheck_retries < 50: # noqa: PLR2004 > if len(outpost.state) > 0: > state = outpost.state[0] > if state.last_seen: > break > healthcheck_retries += 1 > sleep(0.5) > sleep(5) > > self.driver.get("http://localhost:9000/api") > > self.login() > > tests/e2e/test_provider_proxy.py:103: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > def login(self): > """Do entire login flow and check user afterwards""" > > flow_executor = self.get_shadow_root("ak-flow-executor") > > tests/e2e/utils.py:232: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > selector = 'ak-flow-executor' > container = > > def get_shadow_root( > self, selector: str, container: WebElement | WebDriver | None = None > ) -> WebElement: > """Get shadow root element's inner shadowRoot""" > if not container: > container = self.driver > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > tests/e2e/utils.py:226: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > by = 'css selector', value = 'ak-flow-executor' > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > """Find an element given a By strategy and locator. > > :Usage: > :: > > element = driver.find_element(By.ID, 'foo') > > :rtype: WebElement > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x5628a2d9dfc3 \\u003Cunknown>\\n#17 0x5628a2dae944 \\u003Cunknown>\\n#18 0x7f253d591ac3 \\u003Cunknown>\\n"}}'} > > def check_response(self, response: Dict[str, Any]) -> None: > """Checks that a JSON response from the WebDriver does not have an > error. > > :Args: > - response - The JSON response from the WebDriver server as a dictionary > object. > > :Raises: If the response contains an error message. > """ > status = response.get("status", None) > if not status or status == ErrorCode.SUCCESS: > return > value = None > message = response.get("message", "") > screen: str = response.get("screen", "") > stacktrace = None > if isinstance(status, int): > value_json = response.get("value", None) > if value_json and isinstance(value_json, str): > import json > > try: > value = json.loads(value_json) > if len(value) == 1: > value = value["value"] > status = value.get("error", None) > if not status: > status = value.get("status", ErrorCode.UNKNOWN_ERROR) > message = value.get("value") or value.get("message") > if not isinstance(message, str): > value = message > message = message.get("message") > else: > message = value.get("message", None) > except ValueError: > pass > > exception_class: Type[WebDriverException] > e = ErrorCode() > error_codes = [item for item in dir(e) if not item.startswith("__")] > for error_code in error_codes: > error_info = getattr(ErrorCode, error_code) > if isinstance(error_info, list) and status in error_info: > exception_class = getattr(ExceptionMapping, error_code, WebDriverException) > break > else: > exception_class = WebDriverException > > if not value: > value = response["value"] > if isinstance(value, str): > raise exception_class(value) > if message == "" and "message" in value: > message = value["message"] > > screen = None # type: ignore[assignment] > if "screen" in value: > screen = value["screen"] > > stacktrace = None > st_value = value.get("stackTrace") or value.get("stacktrace") > if st_value: > if isinstance(st_value, str): > stacktrace = st_value.split("\n") > else: > stacktrace = [] > try: > for frame in st_value: > line = frame.get("lineNumber", "") > file = frame.get("fileName", "") > if line: > file = f"{file}:{line}" > meth = frame.get("methodName", "") > if "className" in frame: > meth = f"{frame['className']}.{meth}" > msg = " at %s (%s)" > msg = msg % (meth, file) > stacktrace.append(msg) > except TypeError: > pass > if exception_class == UnexpectedAlertPresentException: > alert_text = None > if "data" in value: > alert_text = value["data"].get("text") > elif "alert" in value: > alert_text = value["alert"].get("text") > raise exception_class(message, screen, stacktrace, alert_text) # type: ignore[call-arg] # mypy is not smart enough here > > raise exception_class(message, screen, stacktrace) > E selenium.common.exceptions.NoSuchElementException: Message: no such element: Unable to locate element: {"method":"css selector","selector":"ak-flow-executor"} > E (Session info: chrome=122.0.6261.69); For documentation on this error, please visit: https://www.selenium..../webdriver/troubleshooting/errors#no-such-element-exception > E Stacktrace: > E #0 0x5628a2daf793 > E #1 0x5628a2aa31c6 > E #2 0x5628a2aee358 > E #3 0x5628a2aee411 > E #4 0x5628a2b31934 > E #5 0x5628a2b103fd > E #6 0x5628a2b2edd9 > E #7 0x5628a2b10173 > E #8 0x5628a2ae12d3 > E #9 0x5628a2ae1c9e > E #10 0x5628a2d738cb > E #11 0x5628a2d77745 > E #12 0x5628a2d602e1 > E #13 0x5628a2d782d2 > E #14 0x5628a2d4417f > E #15 0x5628a2d9ddc8 > E #16 0x5628a2d9dfc3 > E #17 0x5628a2dae944 > E #18 0x7f253d591ac3 > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > During handling of the above exception, another exception occurred: > > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fe2992c9540>] > args = ('group_outpost_43d27139-1b49-468d-abbe-7c4c88ec5f50', {'type': 'event.update'}) > kwargs = {}, __traceback_hide__ = True > current_task = result=None> > result = None > > async def main_wrap( > self, > call_result: "Future[_R]", > exc_info: "OptExcInfo", > task_context: "Optional[List[asyncio.Task[Any]]]", > context: List[contextvars.Context], > *args: _P.args, > **kwargs: _P.kwargs, > ) -> None: > """ > Wraps the awaitable with something that puts the result into the > result/exception future. > """ > > __traceback_hide__ = True # noqa: F841 > > if context is not None: > _restore_context(context[0]) > > current_task = asyncio.current_task() > if current_task is not None and task_context is not None: > task_context.append(current_task) > > try: > # If we have an exception, run the function inside the except block > # after raising it so exc_info is correctly populated. > if exc_info[1]: > try: > > raise exc_info[1] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fe298db6f40>] > args = ('group_outpost_43d27139-1b49-468d-abbe-7c4c88ec5f50', {'type': 'event.update'}) > kwargs = {}, __traceback_hide__ = True > current_task = result=None> > result = None > > async def main_wrap( > self, > call_result: "Future[_R]", > exc_info: "OptExcInfo", > task_context: "Optional[List[asyncio.Task[Any]]]", > context: List[contextvars.Context], > *args: _P.args, > **kwargs: _P.kwargs, > ) -> None: > """ > Wraps the awaitable with something that puts the result into the > result/exception future. > """ > > __traceback_hide__ = True # noqa: F841 > > if context is not None: > _restore_context(context[0]) > > current_task = asyncio.current_task() > if current_task is not None and task_context is not None: > task_context.append(current_task) > > try: > # If we have an exception, run the function inside the except block > # after raising it so exc_info is correctly populated. > if exc_info[1]: > try: > > raise exc_info[1] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > > return func(self, *args, **kwargs) > > tests/e2e/utils.py:287: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > @wraps(func) > def wrapper(*args, **kwargs): > for file in files: > content = BlueprintInstance(path=file).retrieve() > Importer.from_string(content).apply() > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:25: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > @wraps(func) > def wrapper(*args, **kwargs): > for file in files: > content = BlueprintInstance(path=file).retrieve() > Importer.from_string(content).apply() > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:25: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, file = 'system/providers-proxy.yaml' > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > @wraps(func) > def wrapper(*args, **kwargs): > for file in files: > content = BlueprintInstance(path=file).retrieve() > Importer.from_string(content).apply() > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:25: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, config = > > @wraps(func) > def wrapper(*args, **kwargs): > config = apps.get_app_config(app_name) > if isinstance(config, ManagedAppConfig): > config._on_startup_callback(None) > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:43: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > @retry() > @apply_blueprint( > "default/flow-default-authentication-flow.yaml", > "default/flow-default-invalidation-flow.yaml", > ) > @apply_blueprint( > "default/flow-default-provider-authorization-implicit-consent.yaml", > "default/flow-default-provider-invalidation.yaml", > ) > @apply_blueprint( > "system/providers-oauth2.yaml", > "system/providers-proxy.yaml", > ) > @reconcile_app("authentik_crypto") > def test_proxy_simple(self): > """Test simple outpost setup with single provider""" > # set additionalHeaders to test later > self.user.attributes["additionalHeaders"] = {"X-Foo": "bar"} > self.user.save() > > proxy: ProxyProvider = ProxyProvider.objects.create( > name=generate_id(), > authorization_flow=Flow.objects.get( > slug="default-provider-authorization-implicit-consent" > ), > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > internal_host=f"http://{self.host}", > external_host="http://localhost:9000", > ) > # Ensure OAuth2 Params are set > proxy.set_oauth_defaults() > proxy.save() > # we need to create an application to actually access the proxy > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > outpost: Outpost = Outpost.objects.create( > name=generate_id(), > type=OutpostType.PROXY, > ) > outpost.providers.add(proxy) > outpost.build_user_permissions(outpost.user) > > self.start_proxy(outpost) > > # Wait until outpost healthcheck succeeds > healthcheck_retries = 0 > while healthcheck_retries < 50: # noqa: PLR2004 > if len(outpost.state) > 0: > state = outpost.state[0] > if state.last_seen: > break > healthcheck_retries += 1 > sleep(0.5) > sleep(5) > > self.driver.get("http://localhost:9000/api") > > self.login() > > tests/e2e/test_provider_proxy.py:103: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > def login(self): > """Do entire login flow and check user afterwards""" > > flow_executor = self.get_shadow_root("ak-flow-executor") > > tests/e2e/utils.py:232: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > selector = 'ak-flow-executor' > container = > > def get_shadow_root( > self, selector: str, container: WebElement | WebDriver | None = None > ) -> WebElement: > """Get shadow root element's inner shadowRoot""" > if not container: > container = self.driver > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > tests/e2e/utils.py:226: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > by = 'css selector', value = 'ak-flow-executor' > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > """Find an element given a By strategy and locator. > > :Usage: > :: > > element = driver.find_element(By.ID, 'foo') > > :rtype: WebElement > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x56014f9b9fc3 \\u003Cunknown>\\n#17 0x56014f9ca944 \\u003Cunknown>\\n#18 0x7fd67db3dac3 \\u003Cunknown>\\n"}}'} > > def check_response(self, response: Dict[str, Any]) -> None: > """Checks that a JSON response from the WebDriver does not have an > error. > > :Args: > - response - The JSON response from the WebDriver server as a dictionary > object. > > :Raises: If the response contains an error message. > """ > status = response.get("status", None) > if not status or status == ErrorCode.SUCCESS: > return > value = None > message = response.get("message", "") > screen: str = response.get("screen", "") > stacktrace = None > if isinstance(status, int): > value_json = response.get("value", None) > if value_json and isinstance(value_json, str): > import json > > try: > value = json.loads(value_json) > if len(value) == 1: > value = value["value"] > status = value.get("error", None) > if not status: > status = value.get("status", ErrorCode.UNKNOWN_ERROR) > message = value.get("value") or value.get("message") > if not isinstance(message, str): > value = message > message = message.get("message") > else: > message = value.get("message", None) > except ValueError: > pass > > exception_class: Type[WebDriverException] > e = ErrorCode() > error_codes = [item for item in dir(e) if not item.startswith("__")] > for error_code in error_codes: > error_info = getattr(ErrorCode, error_code) > if isinstance(error_info, list) and status in error_info: > exception_class = getattr(ExceptionMapping, error_code, WebDriverException) > break > else: > exception_class = WebDriverException > > if not value: > value = response["value"] > if isinstance(value, str): > raise exception_class(value) > if message == "" and "message" in value: > message = value["message"] > > screen = None # type: ignore[assignment] > if "screen" in value: > screen = value["screen"] > > stacktrace = None > st_value = value.get("stackTrace") or value.get("stacktrace") > if st_value: > if isinstance(st_value, str): > stacktrace = st_value.split("\n") > else: > stacktrace = [] > try: > for frame in st_value: > line = frame.get("lineNumber", "") > file = frame.get("fileName", "") > if line: > file = f"{file}:{line}" > meth = frame.get("methodName", "") > if "className" in frame: > meth = f"{frame['className']}.{meth}" > msg = " at %s (%s)" > msg = msg % (meth, file) > stacktrace.append(msg) > except TypeError: > pass > if exception_class == UnexpectedAlertPresentException: > alert_text = None > if "data" in value: > alert_text = value["data"].get("text") > elif "alert" in value: > alert_text = value["alert"].get("text") > raise exception_class(message, screen, stacktrace, alert_text) # type: ignore[call-arg] # mypy is not smart enough here > > raise exception_class(message, screen, stacktrace) > E selenium.common.exceptions.NoSuchElementException: Message: no such element: Unable to locate element: {"method":"css selector","selector":"ak-flow-executor"} > E (Session info: chrome=122.0.6261.69); For documentation on this error, please visit: https://www.selenium..../webdriver/troubleshooting/errors#no-such-element-exception > E Stacktrace: > E #0 0x56014f9cb793 > E #1 0x56014f6bf1c6 > E #2 0x56014f70a358 > E #3 0x56014f70a411 > E #4 0x56014f74d934 > E #5 0x56014f72c3fd > E #6 0x56014f74add9 > E #7 0x56014f72c173 > E #8 0x56014f6fd2d3 > E #9 0x56014f6fdc9e > E #10 0x56014f98f8cb > E #11 0x56014f993745 > E #12 0x56014f97c2e1 > E #13 0x56014f9942d2 > E #14 0x56014f96017f > E #15 0x56014f9b9dc8 > E #16 0x56014f9b9fc3 > E #17 0x56014f9ca944 > E #18 0x7fd67db3dac3 > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > During handling of the above exception, another exception occurred: > > self = > test_case = > subTest = False > > @contextlib.contextmanager > def testPartExecutor(self, test_case, subTest=False): > old_success = self.success > self.success = True > try: > > yield > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:58: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > result = > > def run(self, result=None): > if result is None: > result = self.defaultTestResult() > startTestRun = getattr(result, 'startTestRun', None) > stopTestRun = getattr(result, 'stopTestRun', None) > if startTestRun is not None: > startTestRun() > else: > stopTestRun = None > > result.startTest(self) > try: > testMethod = getattr(self, self._testMethodName) > if (getattr(self.__class__, "__unittest_skip__", False) or > getattr(testMethod, "__unittest_skip__", False)): > # If the class or method was skipped. > skip_why = (getattr(self.__class__, '__unittest_skip_why__', '') > or getattr(testMethod, '__unittest_skip_why__', '')) > _addSkip(result, self, skip_why) > return result > > expecting_failure = ( > getattr(self, "__unittest_expecting_failure__", False) or > getattr(testMethod, "__unittest_expecting_failure__", False) > ) > outcome = _Outcome(result) > start_time = time.perf_counter() > try: > self._outcome = outcome > > with outcome.testPartExecutor(self): > self._callSetUp() > if outcome.success: > outcome.expecting_failure = expecting_failure > with outcome.testPartExecutor(self): > > self._callTestMethod(testMethod) > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:634: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > method = > > > def _callTestMethod(self, method): > > if method() is not None: > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:589: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > return func(self, *args, **kwargs) > > except tuple(exceptions) as exc: > count += 1 > if count > max_retires: > logger.debug("Exceeded retry count", exc=exc, test=self) > > raise exc > logger.debug("Retrying on error", exc=exc, test=self) > self.tearDown() > self._post_teardown() > self._pre_setup() > self.setUp() > > return wrapper(self, *args, **kwargs) > > tests/e2e/utils.py:300: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > return func(self, *args, **kwargs) > > except tuple(exceptions) as exc: > count += 1 > if count > max_retires: > logger.debug("Exceeded retry count", exc=exc, test=self) > > raise exc > logger.debug("Retrying on error", exc=exc, test=self) > self.tearDown() > self._post_teardown() > self._pre_setup() > self.setUp() > > return wrapper(self, *args, **kwargs) > > tests/e2e/utils.py:300: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > return func(self, *args, **kwargs) > > except tuple(exceptions) as exc: > count += 1 > if count > max_retires: > logger.debug("Exceeded retry count", exc=exc, test=self) > > > raise exc > > tests/e2e/utils.py:294: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > > return func(self, *args, **kwargs) > > tests/e2e/utils.py:287: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > @wraps(func) > def wrapper(*args, **kwargs): > for file in files: > content = BlueprintInstance(path=file).retrieve() > Importer.from_string(content).apply() > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:25: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > @wraps(func) > def wrapper(*args, **kwargs): > for file in files: > content = BlueprintInstance(path=file).retrieve() > Importer.from_string(content).apply() > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:25: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, file = 'system/providers-proxy.yaml' > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > @wraps(func) > def wrapper(*args, **kwargs): > for file in files: > content = BlueprintInstance(path=file).retrieve() > Importer.from_string(content).apply() > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:25: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > args = (,) > kwargs = {}, config = > > @wraps(func) > def wrapper(*args, **kwargs): > config = apps.get_app_config(app_name) > if isinstance(config, ManagedAppConfig): > config._on_startup_callback(None) > > return func(*args, **kwargs) > > .../blueprints/tests/__init__.py:43: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > @retry() > @apply_blueprint( > "default/flow-default-authentication-flow.yaml", > "default/flow-default-invalidation-flow.yaml", > ) > @apply_blueprint( > "default/flow-default-provider-authorization-implicit-consent.yaml", > "default/flow-default-provider-invalidation.yaml", > ) > @apply_blueprint( > "system/providers-oauth2.yaml", > "system/providers-proxy.yaml", > ) > @reconcile_app("authentik_crypto") > def test_proxy_simple(self): > """Test simple outpost setup with single provider""" > # set additionalHeaders to test later > self.user.attributes["additionalHeaders"] = {"X-Foo": "bar"} > self.user.save() > > proxy: ProxyProvider = ProxyProvider.objects.create( > name=generate_id(), > authorization_flow=Flow.objects.get( > slug="default-provider-authorization-implicit-consent" > ), > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > internal_host=f"http://{self.host}", > external_host="http://localhost:9000", > ) > # Ensure OAuth2 Params are set > proxy.set_oauth_defaults() > proxy.save() > # we need to create an application to actually access the proxy > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > outpost: Outpost = Outpost.objects.create( > name=generate_id(), > type=OutpostType.PROXY, > ) > outpost.providers.add(proxy) > outpost.build_user_permissions(outpost.user) > > self.start_proxy(outpost) > > # Wait until outpost healthcheck succeeds > healthcheck_retries = 0 > while healthcheck_retries < 50: # noqa: PLR2004 > if len(outpost.state) > 0: > state = outpost.state[0] > if state.last_seen: > break > healthcheck_retries += 1 > sleep(0.5) > sleep(5) > > self.driver.get("http://localhost:9000/api") > > self.login() > > tests/e2e/test_provider_proxy.py:103: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > def login(self): > """Do entire login flow and check user afterwards""" > > flow_executor = self.get_shadow_root("ak-flow-executor") > > tests/e2e/utils.py:232: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > selector = 'ak-flow-executor' > container = > > def get_shadow_root( > self, selector: str, container: WebElement | WebDriver | None = None > ) -> WebElement: > """Get shadow root element's inner shadowRoot""" > if not container: > container = self.driver > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > tests/e2e/utils.py:226: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > by = 'css selector', value = 'ak-flow-executor' > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > """Find an element given a By strategy and locator. > > :Usage: > :: > > element = driver.find_element(By.ID, 'foo') > > :rtype: WebElement > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x55ae73f66fc3 \\u003Cunknown>\\n#17 0x55ae73f77944 \\u003Cunknown>\\n#18 0x7f3149b2cac3 \\u003Cunknown>\\n"}}'} > > def check_response(self, response: Dict[str, Any]) -> None: > """Checks that a JSON response from the WebDriver does not have an > error. > > :Args: > - response - The JSON response from the WebDriver server as a dictionary > object. > > :Raises: If the response contains an error message. > """ > status = response.get("status", None) > if not status or status == ErrorCode.SUCCESS: > return > value = None > message = response.get("message", "") > screen: str = response.get("screen", "") > stacktrace = None > if isinstance(status, int): > value_json = response.get("value", None) > if value_json and isinstance(value_json, str): > import json > > try: > value = json.loads(value_json) > if len(value) == 1: > value = value["value"] > status = value.get("error", None) > if not status: > status = value.get("status", ErrorCode.UNKNOWN_ERROR) > message = value.get("value") or value.get("message") > if not isinstance(message, str): > value = message > message = message.get("message") > else: > message = value.get("message", None) > except ValueError: > pass > > exception_class: Type[WebDriverException] > e = ErrorCode() > error_codes = [item for item in dir(e) if not item.startswith("__")] > for error_code in error_codes: > error_info = getattr(ErrorCode, error_code) > if isinstance(error_info, list) and status in error_info: > exception_class = getattr(ExceptionMapping, error_code, WebDriverException) > break > else: > exception_class = WebDriverException > > if not value: > value = response["value"] > if isinstance(value, str): > raise exception_class(value) > if message == "" and "message" in value: > message = value["message"] > > screen = None # type: ignore[assignment] > if "screen" in value: > screen = value["screen"] > > stacktrace = None > st_value = value.get("stackTrace") or value.get("stacktrace") > if st_value: > if isinstance(st_value, str): > stacktrace = st_value.split("\n") > else: > stacktrace = [] > try: > for frame in st_value: > line = frame.get("lineNumber", "") > file = frame.get("fileName", "") > if line: > file = f"{file}:{line}" > meth = frame.get("methodName", "") > if "className" in frame: > meth = f"{frame['className']}.{meth}" > msg = " at %s (%s)" > msg = msg % (meth, file) > stacktrace.append(msg) > except TypeError: > pass > if exception_class == UnexpectedAlertPresentException: > alert_text = None > if "data" in value: > alert_text = value["data"].get("text") > elif "alert" in value: > alert_text = value["alert"].get("text") > raise exception_class(message, screen, stacktrace, alert_text) # type: ignore[call-arg] # mypy is not smart enough here > > raise exception_class(message, screen, stacktrace) > E selenium.common.exceptions.NoSuchElementException: Message: no such element: Unable to locate element: {"method":"css selector","selector":"ak-flow-executor"} > E (Session info: chrome=122.0.6261.69); For documentation on this error, please visit: https://www.selenium..../webdriver/troubleshooting/errors#no-such-element-exception > E Stacktrace: > E #0 0x55ae73f78793 > E #1 0x55ae73c6c1c6 > E #2 0x55ae73cb7358 > E #3 0x55ae73cb7411 > E #4 0x55ae73cfa934 > E #5 0x55ae73cd93fd > E #6 0x55ae73cf7dd9 > E #7 0x55ae73cd9173 > E #8 0x55ae73caa2d3 > E #9 0x55ae73caac9e > E #10 0x55ae73f3c8cb > E #11 0x55ae73f40745 > E #12 0x55ae73f292e1 > E #13 0x55ae73f412d2 > E #14 0x55ae73f0d17f > E #15 0x55ae73f66dc8 > E #16 0x55ae73f66fc3 > E #17 0x55ae73f77944 > E #18 0x7f3149b2cac3 > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > ```

tests.e2e.test_provider_proxy_forward.TestProviderProxyForward::test_traefik

Stack Traces | 223s run time

> > ```python > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fe298ef7880>] > args = ('group_outpost_11033f41-9620-4feb-9fec-18de17050096', {'type': 'event.update'}) > kwargs = {}, __traceback_hide__ = True > current_task = result=None> > result = None > > async def main_wrap( > self, > call_result: "Future[_R]", > exc_info: "OptExcInfo", > task_context: "Optional[List[asyncio.Task[Any]]]", > context: List[contextvars.Context], > *args: _P.args, > **kwargs: _P.kwargs, > ) -> None: > """ > Wraps the awaitable with something that puts the result into the > result/exception future. > """ > > __traceback_hide__ = True # noqa: F841 > > if context is not None: > _restore_context(context[0]) > > current_task = asyncio.current_task() > if current_task is not None and task_context is not None: > task_context.append(current_task) > > try: > # If we have an exception, run the function inside the except block > # after raising it so exc_info is correctly populated. > if exc_info[1]: > try: > > raise exc_info[1] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fe28a54f840>] > args = ('group_outpost_11033f41-9620-4feb-9fec-18de17050096', {'type': 'event.update'}) > kwargs = {}, __traceback_hide__ = True > current_task = result=None> > result = None > > async def main_wrap( > self, > call_result: "Future[_R]", > exc_info: "OptExcInfo", > task_context: "Optional[List[asyncio.Task[Any]]]", > context: List[contextvars.Context], > *args: _P.args, > **kwargs: _P.kwargs, > ) -> None: > """ > Wraps the awaitable with something that puts the result into the > result/exception future. > """ > > __traceback_hide__ = True # noqa: F841 > > if context is not None: > _restore_context(context[0]) > > current_task = asyncio.current_task() > if current_task is not None and task_context is not None: > task_context.append(current_task) > > try: > # If we have an exception, run the function inside the except block > # after raising it so exc_info is correctly populated. > if exc_info[1]: > try: > > raise exc_info[1] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > > return func(self, *args, **kwargs) > > tests/e2e/utils.py:287: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > @retry() > def test_traefik(self): > """Test traefik""" > local_config_path = ( > Path(__file__).parent / "proxy_forward_auth" / "traefik_single" / "config-static.yaml" > ) > self.run_container( > image="docker.io/library/traefik:3.1", > ports={ > "80": "80", > }, > volumes={ > local_config_path: { > "bind": "........./etc/traefik/traefik.yml", > } > }, > ) > > self.prepare() > > self.driver.get("http:.../localhost/api") > > self.login() > > tests/e2e/test_provider_proxy_forward.py:111: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > def login(self): > """Do entire login flow and check user afterwards""" > > flow_executor = self.get_shadow_root("ak-flow-executor") > > tests/e2e/utils.py:232: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > selector = 'ak-flow-executor' > container = > > def get_shadow_root( > self, selector: str, container: WebElement | WebDriver | None = None > ) -> WebElement: > """Get shadow root element's inner shadowRoot""" > if not container: > container = self.driver > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > tests/e2e/utils.py:226: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > by = 'css selector', value = 'ak-flow-executor' > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > """Find an element given a By strategy and locator. > > :Usage: > :: > > element = driver.find_element(By.ID, 'foo') > > :rtype: WebElement > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x558d4d2e6fc3 \\u003Cunknown>\\n#17 0x558d4d2f7944 \\u003Cunknown>\\n#18 0x7f5214c3cac3 \\u003Cunknown>\\n"}}'} > > def check_response(self, response: Dict[str, Any]) -> None: > """Checks that a JSON response from the WebDriver does not have an > error. > > :Args: > - response - The JSON response from the WebDriver server as a dictionary > object. > > :Raises: If the response contains an error message. > """ > status = response.get("status", None) > if not status or status == ErrorCode.SUCCESS: > return > value = None > message = response.get("message", "") > screen: str = response.get("screen", "") > stacktrace = None > if isinstance(status, int): > value_json = response.get("value", None) > if value_json and isinstance(value_json, str): > import json > > try: > value = json.loads(value_json) > if len(value) == 1: > value = value["value"] > status = value.get("error", None) > if not status: > status = value.get("status", ErrorCode.UNKNOWN_ERROR) > message = value.get("value") or value.get("message") > if not isinstance(message, str): > value = message > message = message.get("message") > else: > message = value.get("message", None) > except ValueError: > pass > > exception_class: Type[WebDriverException] > e = ErrorCode() > error_codes = [item for item in dir(e) if not item.startswith("__")] > for error_code in error_codes: > error_info = getattr(ErrorCode, error_code) > if isinstance(error_info, list) and status in error_info: > exception_class = getattr(ExceptionMapping, error_code, WebDriverException) > break > else: > exception_class = WebDriverException > > if not value: > value = response["value"] > if isinstance(value, str): > raise exception_class(value) > if message == "" and "message" in value: > message = value["message"] > > screen = None # type: ignore[assignment] > if "screen" in value: > screen = value["screen"] > > stacktrace = None > st_value = value.get("stackTrace") or value.get("stacktrace") > if st_value: > if isinstance(st_value, str): > stacktrace = st_value.split("\n") > else: > stacktrace = [] > try: > for frame in st_value: > line = frame.get("lineNumber", "") > file = frame.get("fileName", "") > if line: > file = f"{file}:{line}" > meth = frame.get("methodName", "") > if "className" in frame: > meth = f"{frame['className']}.{meth}" > msg = " at %s (%s)" > msg = msg % (meth, file) > stacktrace.append(msg) > except TypeError: > pass > if exception_class == UnexpectedAlertPresentException: > alert_text = None > if "data" in value: > alert_text = value["data"].get("text") > elif "alert" in value: > alert_text = value["alert"].get("text") > raise exception_class(message, screen, stacktrace, alert_text) # type: ignore[call-arg] # mypy is not smart enough here > > raise exception_class(message, screen, stacktrace) > E selenium.common.exceptions.NoSuchElementException: Message: no such element: Unable to locate element: {"method":"css selector","selector":"ak-flow-executor"} > E (Session info: chrome=122.0.6261.69); For documentation on this error, please visit: https://www.selenium..../webdriver/troubleshooting/errors#no-such-element-exception > E Stacktrace: > E #0 0x558d4d2f8793 > E #1 0x558d4cfec1c6 > E #2 0x558d4d037358 > E #3 0x558d4d037411 > E #4 0x558d4d07a934 > E #5 0x558d4d0593fd > E #6 0x558d4d077dd9 > E #7 0x558d4d059173 > E #8 0x558d4d02a2d3 > E #9 0x558d4d02ac9e > E #10 0x558d4d2bc8cb > E #11 0x558d4d2c0745 > E #12 0x558d4d2a92e1 > E #13 0x558d4d2c12d2 > E #14 0x558d4d28d17f > E #15 0x558d4d2e6dc8 > E #16 0x558d4d2e6fc3 > E #17 0x558d4d2f7944 > E #18 0x7f5214c3cac3 > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > During handling of the above exception, another exception occurred: > > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fe28a51de40>] > args = ('group_outpost_98cb7a3f-a3a5-4766-81a9-445084eb6d17', {'type': 'event.update'}) > kwargs = {}, __traceback_hide__ = True > current_task = result=None> > result = None > > async def main_wrap( > self, > call_result: "Future[_R]", > exc_info: "OptExcInfo", > task_context: "Optional[List[asyncio.Task[Any]]]", > context: List[contextvars.Context], > *args: _P.args, > **kwargs: _P.kwargs, > ) -> None: > """ > Wraps the awaitable with something that puts the result into the > result/exception future. > """ > > __traceback_hide__ = True # noqa: F841 > > if context is not None: > _restore_context(context[0]) > > current_task = asyncio.current_task() > if current_task is not None and task_context is not None: > task_context.append(current_task) > > try: > # If we have an exception, run the function inside the except block > # after raising it so exc_info is correctly populated. > if exc_info[1]: > try: > > raise exc_info[1] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fe298ea36c0>] > args = ('group_outpost_98cb7a3f-a3a5-4766-81a9-445084eb6d17', {'type': 'event.update'}) > kwargs = {}, __traceback_hide__ = True > current_task = result=None> > result = None > > async def main_wrap( > self, > call_result: "Future[_R]", > exc_info: "OptExcInfo", > task_context: "Optional[List[asyncio.Task[Any]]]", > context: List[contextvars.Context], > *args: _P.args, > **kwargs: _P.kwargs, > ) -> None: > """ > Wraps the awaitable with something that puts the result into the > result/exception future. > """ > > __traceback_hide__ = True # noqa: F841 > > if context is not None: > _restore_context(context[0]) > > current_task = asyncio.current_task() > if current_task is not None and task_context is not None: > task_context.append(current_task) > > try: > # If we have an exception, run the function inside the except block > # after raising it so exc_info is correctly populated. > if exc_info[1]: > try: > > raise exc_info[1] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > > return func(self, *args, **kwargs) > > tests/e2e/utils.py:287: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > @retry() > def test_traefik(self): > """Test traefik""" > local_config_path = ( > Path(__file__).parent / "proxy_forward_auth" / "traefik_single" / "config-static.yaml" > ) > self.run_container( > image="docker.io/library/traefik:3.1", > ports={ > "80": "80", > }, > volumes={ > local_config_path: { > "bind": "........./etc/traefik/traefik.yml", > } > }, > ) > > self.prepare() > > self.driver.get("http:.../localhost/api") > > self.login() > > tests/e2e/test_provider_proxy_forward.py:111: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > def login(self): > """Do entire login flow and check user afterwards""" > > flow_executor = self.get_shadow_root("ak-flow-executor") > > tests/e2e/utils.py:232: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > selector = 'ak-flow-executor' > container = > > def get_shadow_root( > self, selector: str, container: WebElement | WebDriver | None = None > ) -> WebElement: > """Get shadow root element's inner shadowRoot""" > if not container: > container = self.driver > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > tests/e2e/utils.py:226: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > by = 'css selector', value = 'ak-flow-executor' > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > """Find an element given a By strategy and locator. > > :Usage: > :: > > element = driver.find_element(By.ID, 'foo') > > :rtype: WebElement > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x5628aecaafc3 \\u003Cunknown>\\n#17 0x5628aecbb944 \\u003Cunknown>\\n#18 0x7f57095bfac3 \\u003Cunknown>\\n"}}'} > > def check_response(self, response: Dict[str, Any]) -> None: > """Checks that a JSON response from the WebDriver does not have an > error. > > :Args: > - response - The JSON response from the WebDriver server as a dictionary > object. > > :Raises: If the response contains an error message. > """ > status = response.get("status", None) > if not status or status == ErrorCode.SUCCESS: > return > value = None > message = response.get("message", "") > screen: str = response.get("screen", "") > stacktrace = None > if isinstance(status, int): > value_json = response.get("value", None) > if value_json and isinstance(value_json, str): > import json > > try: > value = json.loads(value_json) > if len(value) == 1: > value = value["value"] > status = value.get("error", None) > if not status: > status = value.get("status", ErrorCode.UNKNOWN_ERROR) > message = value.get("value") or value.get("message") > if not isinstance(message, str): > value = message > message = message.get("message") > else: > message = value.get("message", None) > except ValueError: > pass > > exception_class: Type[WebDriverException] > e = ErrorCode() > error_codes = [item for item in dir(e) if not item.startswith("__")] > for error_code in error_codes: > error_info = getattr(ErrorCode, error_code) > if isinstance(error_info, list) and status in error_info: > exception_class = getattr(ExceptionMapping, error_code, WebDriverException) > break > else: > exception_class = WebDriverException > > if not value: > value = response["value"] > if isinstance(value, str): > raise exception_class(value) > if message == "" and "message" in value: > message = value["message"] > > screen = None # type: ignore[assignment] > if "screen" in value: > screen = value["screen"] > > stacktrace = None > st_value = value.get("stackTrace") or value.get("stacktrace") > if st_value: > if isinstance(st_value, str): > stacktrace = st_value.split("\n") > else: > stacktrace = [] > try: > for frame in st_value: > line = frame.get("lineNumber", "") > file = frame.get("fileName", "") > if line: > file = f"{file}:{line}" > meth = frame.get("methodName", "") > if "className" in frame: > meth = f"{frame['className']}.{meth}" > msg = " at %s (%s)" > msg = msg % (meth, file) > stacktrace.append(msg) > except TypeError: > pass > if exception_class == UnexpectedAlertPresentException: > alert_text = None > if "data" in value: > alert_text = value["data"].get("text") > elif "alert" in value: > alert_text = value["alert"].get("text") > raise exception_class(message, screen, stacktrace, alert_text) # type: ignore[call-arg] # mypy is not smart enough here > > raise exception_class(message, screen, stacktrace) > E selenium.common.exceptions.NoSuchElementException: Message: no such element: Unable to locate element: {"method":"css selector","selector":"ak-flow-executor"} > E (Session info: chrome=122.0.6261.69); For documentation on this error, please visit: https://www.selenium..../webdriver/troubleshooting/errors#no-such-element-exception > E Stacktrace: > E #0 0x5628aecbc793 > E #1 0x5628ae9b01c6 > E #2 0x5628ae9fb358 > E #3 0x5628ae9fb411 > E #4 0x5628aea3e934 > E #5 0x5628aea1d3fd > E #6 0x5628aea3bdd9 > E #7 0x5628aea1d173 > E #8 0x5628ae9ee2d3 > E #9 0x5628ae9eec9e > E #10 0x5628aec808cb > E #11 0x5628aec84745 > E #12 0x5628aec6d2e1 > E #13 0x5628aec852d2 > E #14 0x5628aec5117f > E #15 0x5628aecaadc8 > E #16 0x5628aecaafc3 > E #17 0x5628aecbb944 > E #18 0x7f57095bfac3 > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > During handling of the above exception, another exception occurred: > > self = > test_case = > subTest = False > > @contextlib.contextmanager > def testPartExecutor(self, test_case, subTest=False): > old_success = self.success > self.success = True > try: > > yield > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:58: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > result = > > def run(self, result=None): > if result is None: > result = self.defaultTestResult() > startTestRun = getattr(result, 'startTestRun', None) > stopTestRun = getattr(result, 'stopTestRun', None) > if startTestRun is not None: > startTestRun() > else: > stopTestRun = None > > result.startTest(self) > try: > testMethod = getattr(self, self._testMethodName) > if (getattr(self.__class__, "__unittest_skip__", False) or > getattr(testMethod, "__unittest_skip__", False)): > # If the class or method was skipped. > skip_why = (getattr(self.__class__, '__unittest_skip_why__', '') > or getattr(testMethod, '__unittest_skip_why__', '')) > _addSkip(result, self, skip_why) > return result > > expecting_failure = ( > getattr(self, "__unittest_expecting_failure__", False) or > getattr(testMethod, "__unittest_expecting_failure__", False) > ) > outcome = _Outcome(result) > start_time = time.perf_counter() > try: > self._outcome = outcome > > with outcome.testPartExecutor(self): > self._callSetUp() > if outcome.success: > outcome.expecting_failure = expecting_failure > with outcome.testPartExecutor(self): > > self._callTestMethod(testMethod) > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:634: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > method = > > > def _callTestMethod(self, method): > > if method() is not None: > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:589: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > return func(self, *args, **kwargs) > > except tuple(exceptions) as exc: > count += 1 > if count > max_retires: > logger.debug("Exceeded retry count", exc=exc, test=self) > > raise exc > logger.debug("Retrying on error", exc=exc, test=self) > self.tearDown() > self._post_teardown() > self._pre_setup() > self.setUp() > > return wrapper(self, *args, **kwargs) > > tests/e2e/utils.py:300: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > return func(self, *args, **kwargs) > > except tuple(exceptions) as exc: > count += 1 > if count > max_retires: > logger.debug("Exceeded retry count", exc=exc, test=self) > > raise exc > logger.debug("Retrying on error", exc=exc, test=self) > self.tearDown() > self._post_teardown() > self._pre_setup() > self.setUp() > > return wrapper(self, *args, **kwargs) > > tests/e2e/utils.py:300: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > return func(self, *args, **kwargs) > > except tuple(exceptions) as exc: > count += 1 > if count > max_retires: > logger.debug("Exceeded retry count", exc=exc, test=self) > > > raise exc > > tests/e2e/utils.py:294: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > > return func(self, *args, **kwargs) > > tests/e2e/utils.py:287: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > @retry() > def test_traefik(self): > """Test traefik""" > local_config_path = ( > Path(__file__).parent / "proxy_forward_auth" / "traefik_single" / "config-static.yaml" > ) > self.run_container( > image="docker.io/library/traefik:3.1", > ports={ > "80": "80", > }, > volumes={ > local_config_path: { > "bind": "........./etc/traefik/traefik.yml", > } > }, > ) > > self.prepare() > > self.driver.get("http:.../localhost/api") > > self.login() > > tests/e2e/test_provider_proxy_forward.py:111: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > def login(self): > """Do entire login flow and check user afterwards""" > > flow_executor = self.get_shadow_root("ak-flow-executor") > > tests/e2e/utils.py:232: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > selector = 'ak-flow-executor' > container = > > def get_shadow_root( > self, selector: str, container: WebElement | WebDriver | None = None > ) -> WebElement: > """Get shadow root element's inner shadowRoot""" > if not container: > container = self.driver > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > tests/e2e/utils.py:226: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > by = 'css selector', value = 'ak-flow-executor' > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > """Find an element given a By strategy and locator. > > :Usage: > :: > > element = driver.find_element(By.ID, 'foo') > > :rtype: WebElement > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x55a3ca5cdfc3 \\u003Cunknown>\\n#17 0x55a3ca5de944 \\u003Cunknown>\\n#18 0x7f6991841ac3 \\u003Cunknown>\\n"}}'} > > def check_response(self, response: Dict[str, Any]) -> None: > """Checks that a JSON response from the WebDriver does not have an > error. > > :Args: > - response - The JSON response from the WebDriver server as a dictionary > object. > > :Raises: If the response contains an error message. > """ > status = response.get("status", None) > if not status or status == ErrorCode.SUCCESS: > return > value = None > message = response.get("message", "") > screen: str = response.get("screen", "") > stacktrace = None > if isinstance(status, int): > value_json = response.get("value", None) > if value_json and isinstance(value_json, str): > import json > > try: > value = json.loads(value_json) > if len(value) == 1: > value = value["value"] > status = value.get("error", None) > if not status: > status = value.get("status", ErrorCode.UNKNOWN_ERROR) > message = value.get("value") or value.get("message") > if not isinstance(message, str): > value = message > message = message.get("message") > else: > message = value.get("message", None) > except ValueError: > pass > > exception_class: Type[WebDriverException] > e = ErrorCode() > error_codes = [item for item in dir(e) if not item.startswith("__")] > for error_code in error_codes: > error_info = getattr(ErrorCode, error_code) > if isinstance(error_info, list) and status in error_info: > exception_class = getattr(ExceptionMapping, error_code, WebDriverException) > break > else: > exception_class = WebDriverException > > if not value: > value = response["value"] > if isinstance(value, str): > raise exception_class(value) > if message == "" and "message" in value: > message = value["message"] > > screen = None # type: ignore[assignment] > if "screen" in value: > screen = value["screen"] > > stacktrace = None > st_value = value.get("stackTrace") or value.get("stacktrace") > if st_value: > if isinstance(st_value, str): > stacktrace = st_value.split("\n") > else: > stacktrace = [] > try: > for frame in st_value: > line = frame.get("lineNumber", "") > file = frame.get("fileName", "") > if line: > file = f"{file}:{line}" > meth = frame.get("methodName", "") > if "className" in frame: > meth = f"{frame['className']}.{meth}" > msg = " at %s (%s)" > msg = msg % (meth, file) > stacktrace.append(msg) > except TypeError: > pass > if exception_class == UnexpectedAlertPresentException: > alert_text = None > if "data" in value: > alert_text = value["data"].get("text") > elif "alert" in value: > alert_text = value["alert"].get("text") > raise exception_class(message, screen, stacktrace, alert_text) # type: ignore[call-arg] # mypy is not smart enough here > > raise exception_class(message, screen, stacktrace) > E selenium.common.exceptions.NoSuchElementException: Message: no such element: Unable to locate element: {"method":"css selector","selector":"ak-flow-executor"} > E (Session info: chrome=122.0.6261.69); For documentation on this error, please visit: https://www.selenium..../webdriver/troubleshooting/errors#no-such-element-exception > E Stacktrace: > E #0 0x55a3ca5df793 > E #1 0x55a3ca2d31c6 > E #2 0x55a3ca31e358 > E #3 0x55a3ca31e411 > E #4 0x55a3ca361934 > E #5 0x55a3ca3403fd > E #6 0x55a3ca35edd9 > E #7 0x55a3ca340173 > E #8 0x55a3ca3112d3 > E #9 0x55a3ca311c9e > E #10 0x55a3ca5a38cb > E #11 0x55a3ca5a7745 > E #12 0x55a3ca5902e1 > E #13 0x55a3ca5a82d2 > E #14 0x55a3ca57417f > E #15 0x55a3ca5cddc8 > E #16 0x55a3ca5cdfc3 > E #17 0x55a3ca5de944 > E #18 0x7f6991841ac3 > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > ```

To view more test analytics, go to the Test Analytics Dashboard Got feedback? Let us know on Github