goauthentik / authentik

The authentication glue you need.
https://goauthentik.io
Other
13.67k stars 915 forks source link

providers/oauth2: fix redirect uri input (cherry-pick #12122) #12127

Closed gcp-cherry-pick-bot[bot] closed 17 hours ago

gcp-cherry-pick-bot[bot] commented 17 hours ago

Cherry-picked providers/oauth2: fix redirect uri input (#12122)

Signed-off-by: Jens Langhammer jens@goauthentik.io

Signed-off-by: Jens Langhammer jens@goauthentik.io

Signed-off-by: Jens Langhammer jens@goauthentik.io


Signed-off-by: Jens Langhammer jens@goauthentik.io

codecov[bot] commented 16 hours ago

:x: 6 Tests Failed:

Tests completed Failed Passed Skipped
1597 6 1591 1
View the top 2 failed tests by shortest run time > > ```python > 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 0x7f601c92f940>] > > args = ('group_outpost_221d68d9-94fd-4e98-a9a2-ca7041f3a8f1', {'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 0x7f601c917700>] > > args = ('group_outpost_221d68d9-94fd-4e98-a9a2-ca7041f3a8f1', {'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 > > """ > > 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] > > > > if by == By.ID: > > by = By.CSS_SELECTOR > > value = f'[id="{value}"]' > > elif by == By.CLASS_NAME: > > by = By.CSS_SELECTOR > > value = f".{value}" > > elif by == By.NAME: > > by = By.CSS_SELECTOR > > value = f'[name="{value}"]' > > > > > 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:748: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > 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:354: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x55b306229fc3 \\u003Cunknown>\\n#17 0x55b30623a944 \\u003Cunknown>\\n#18 0x7f6e2998eac3 \\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 0x55b30623b793 > > E #1 0x55b305f2f1c6 > > E #2 0x55b305f7a358 > > E #3 0x55b305f7a411 > > E #4 0x55b305fbd934 > > E #5 0x55b305f9c3fd > > E #6 0x55b305fbadd9 > > E #7 0x55b305f9c173 > > E #8 0x55b305f6d2d3 > > E #9 0x55b305f6dc9e > > E #10 0x55b3061ff8cb > > E #11 0x55b306203745 > > E #12 0x55b3061ec2e1 > > E #13 0x55b3062042d2 > > E #14 0x55b3061d017f > > E #15 0x55b306229dc8 > > E #16 0x55b306229fc3 > > E #17 0x55b30623a944 > > E #18 0x7f6e2998eac3 > > > > ../../../.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 0x7f601c562d00>] > > args = ('group_outpost_2d54a9e5-f733-4597-ae1c-1d89a31e9634', {'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 0x7f60261fdec0>] > > args = ('group_outpost_2d54a9e5-f733-4597-ae1c-1d89a31e9634', {'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 > > """ > > 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] > > > > if by == By.ID: > > by = By.CSS_SELECTOR > > value = f'[id="{value}"]' > > elif by == By.CLASS_NAME: > > by = By.CSS_SELECTOR > > value = f".{value}" > > elif by == By.NAME: > > by = By.CSS_SELECTOR > > value = f'[name="{value}"]' > > > > > 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:748: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > 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:354: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x5654afa10fc3 \\u003Cunknown>\\n#17 0x5654afa21944 \\u003Cunknown>\\n#18 0x7faa81e02ac3 \\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 0x5654afa22793 > > E #1 0x5654af7161c6 > > E #2 0x5654af761358 > > E #3 0x5654af761411 > > E #4 0x5654af7a4934 > > E #5 0x5654af7833fd > > E #6 0x5654af7a1dd9 > > E #7 0x5654af783173 > > E #8 0x5654af7542d3 > > E #9 0x5654af754c9e > > E #10 0x5654af9e68cb > > E #11 0x5654af9ea745 > > E #12 0x5654af9d32e1 > > E #13 0x5654af9eb2d2 > > E #14 0x5654af9b717f > > E #15 0x5654afa10dc8 > > E #16 0x5654afa10fc3 > > E #17 0x5654afa21944 > > E #18 0x7faa81e02ac3 > > > > ../../../.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 > > """ > > 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] > > > > if by == By.ID: > > by = By.CSS_SELECTOR > > value = f'[id="{value}"]' > > elif by == By.CLASS_NAME: > > by = By.CSS_SELECTOR > > value = f".{value}" > > elif by == By.NAME: > > by = By.CSS_SELECTOR > > value = f'[name="{value}"]' > > > > > 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:748: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > 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:354: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x5623c8b6ffc3 \\u003Cunknown>\\n#17 0x5623c8b80944 \\u003Cunknown>\\n#18 0x7f40c6f95ac3 \\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 0x5623c8b81793 > > E #1 0x5623c88751c6 > > E #2 0x5623c88c0358 > > E #3 0x5623c88c0411 > > E #4 0x5623c8903934 > > E #5 0x5623c88e23fd > > E #6 0x5623c8900dd9 > > E #7 0x5623c88e2173 > > E #8 0x5623c88b32d3 > > E #9 0x5623c88b3c9e > > E #10 0x5623c8b458cb > > E #11 0x5623c8b49745 > > E #12 0x5623c8b322e1 > > E #13 0x5623c8b4a2d2 > > E #14 0x5623c8b1617f > > E #15 0x5623c8b6fdc8 > > E #16 0x5623c8b6ffc3 > > E #17 0x5623c8b80944 > > E #18 0x7f40c6f95ac3 > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > ``` > >
tests.e2e.test_provider_proxy_forward.TestProviderProxyForward::test_caddy
Stack Traces | 220s run time > > ```python > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7f601c9a0140>] > args = ('group_outpost_f383d5cd-f9fc-4057-9c14-ef764b03a877', {'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 0x7f601c824500>] > args = ('group_outpost_f383d5cd-f9fc-4057-9c14-ef764b03a877', {'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_caddy(self): > """Test caddy""" > local_config_path = ( > Path(__file__).parent / "proxy_forward_auth" / "caddy_single" / "Caddyfile" > ) > self.run_container( > image="docker.io/library/caddy:2.8", > ports={ > "80": "80", > }, > volumes={ > local_config_path: { > "bind": "........./etc/caddy/Caddyfile", > } > }, > ) > > self.prepare() > > self.driver.get("http:.../localhost/api") > > self.login() > > tests/e2e/test_provider_proxy_forward.py:214: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > 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 > """ > 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] > > if by == By.ID: > by = By.CSS_SELECTOR > value = f'[id="{value}"]' > elif by == By.CLASS_NAME: > by = By.CSS_SELECTOR > value = f".{value}" > elif by == By.NAME: > by = By.CSS_SELECTOR > value = f'[name="{value}"]' > > > 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:748: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > 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:354: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x5609fc004fc3 \\u003Cunknown>\\n#17 0x5609fc015944 \\u003Cunknown>\\n#18 0x7efc8e640ac3 \\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 0x5609fc016793 > E #1 0x5609fbd0a1c6 > E #2 0x5609fbd55358 > E #3 0x5609fbd55411 > E #4 0x5609fbd98934 > E #5 0x5609fbd773fd > E #6 0x5609fbd95dd9 > E #7 0x5609fbd77173 > E #8 0x5609fbd482d3 > E #9 0x5609fbd48c9e > E #10 0x5609fbfda8cb > E #11 0x5609fbfde745 > E #12 0x5609fbfc72e1 > E #13 0x5609fbfdf2d2 > E #14 0x5609fbfab17f > E #15 0x5609fc004dc8 > E #16 0x5609fc004fc3 > E #17 0x5609fc015944 > E #18 0x7efc8e640ac3 > > ../../../.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 0x7f601cfae840>] > args = ('group_outpost_17babaf1-b441-4dd3-a269-ba622cd33079', {'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 0x7f601c638580>] > args = ('group_outpost_17babaf1-b441-4dd3-a269-ba622cd33079', {'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_caddy(self): > """Test caddy""" > local_config_path = ( > Path(__file__).parent / "proxy_forward_auth" / "caddy_single" / "Caddyfile" > ) > self.run_container( > image="docker.io/library/caddy:2.8", > ports={ > "80": "80", > }, > volumes={ > local_config_path: { > "bind": "........./etc/caddy/Caddyfile", > } > }, > ) > > self.prepare() > > self.driver.get("http:.../localhost/api") > > self.login() > > tests/e2e/test_provider_proxy_forward.py:214: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > 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 > """ > 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] > > if by == By.ID: > by = By.CSS_SELECTOR > value = f'[id="{value}"]' > elif by == By.CLASS_NAME: > by = By.CSS_SELECTOR > value = f".{value}" > elif by == By.NAME: > by = By.CSS_SELECTOR > value = f'[name="{value}"]' > > > 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:748: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > 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:354: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x55d8adf10fc3 \\u003Cunknown>\\n#17 0x55d8adf21944 \\u003Cunknown>\\n#18 0x7f340041fac3 \\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 0x55d8adf22793 > E #1 0x55d8adc161c6 > E #2 0x55d8adc61358 > E #3 0x55d8adc61411 > E #4 0x55d8adca4934 > E #5 0x55d8adc833fd > E #6 0x55d8adca1dd9 > E #7 0x55d8adc83173 > E #8 0x55d8adc542d3 > E #9 0x55d8adc54c9e > E #10 0x55d8adee68cb > E #11 0x55d8adeea745 > E #12 0x55d8aded32e1 > E #13 0x55d8adeeb2d2 > E #14 0x55d8adeb717f > E #15 0x55d8adf10dc8 > E #16 0x55d8adf10fc3 > E #17 0x55d8adf21944 > E #18 0x7f340041fac3 > > ../../../.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_caddy(self): > """Test caddy""" > local_config_path = ( > Path(__file__).parent / "proxy_forward_auth" / "caddy_single" / "Caddyfile" > ) > self.run_container( > image="docker.io/library/caddy:2.8", > ports={ > "80": "80", > }, > volumes={ > local_config_path: { > "bind": "........./etc/caddy/Caddyfile", > } > }, > ) > > self.prepare() > > self.driver.get("http:.../localhost/api") > > self.login() > > tests/e2e/test_provider_proxy_forward.py:214: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > 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 > """ > 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] > > if by == By.ID: > by = By.CSS_SELECTOR > value = f'[id="{value}"]' > elif by == By.CLASS_NAME: > by = By.CSS_SELECTOR > value = f".{value}" > elif by == By.NAME: > by = By.CSS_SELECTOR > value = f'[name="{value}"]' > > > 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:748: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > 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:354: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x564f6117ffc3 \\u003Cunknown>\\n#17 0x564f61190944 \\u003Cunknown>\\n#18 0x7f5451a27ac3 \\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 0x564f61191793 > E #1 0x564f60e851c6 > E #2 0x564f60ed0358 > E #3 0x564f60ed0411 > E #4 0x564f60f13934 > E #5 0x564f60ef23fd > E #6 0x564f60f10dd9 > E #7 0x564f60ef2173 > E #8 0x564f60ec32d3 > E #9 0x564f60ec3c9e > E #10 0x564f611558cb > E #11 0x564f61159745 > E #12 0x564f611422e1 > E #13 0x564f6115a2d2 > E #14 0x564f6112617f > E #15 0x564f6117fdc8 > E #16 0x564f6117ffc3 > E #17 0x564f61190944 > E #18 0x7f5451a27ac3 > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > ```

View the full list of 1 :snowflake: flaky tests > > ```python > tests.e2e.test_provider_proxy_forward.TestProviderProxyForward::test_nginx > ``` > > **Flake rate in main:** 23.08% (Passed 70 times, Failed 21 times) >
Stack Traces | 219s run time > > > > > ```python > > self = > > call_result = > > exc_info = (, NoSuchElementException(), ) > > task_context = None, context = [<_contextvars.Context object at 0x7f601c783840>] > > args = ('group_outpost_e3bbff65-2308-4b83-9005-41757057b7bb', {'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 0x7f6026653d40>] > > args = ('group_outpost_e3bbff65-2308-4b83-9005-41757057b7bb', {'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_nginx(self): > > """Test nginx""" > > self.prepare() > > > > # Start nginx last so all hosts are resolvable, otherwise nginx exits > > self.run_container( > > image="docker.io/library/nginx:1.27", > > ports={ > > "80": "80", > > }, > > volumes={ > > f"{Path(__file__).parent / "proxy_forward_auth" / "nginx_single" / "nginx.conf"}": { > > "bind": "........./etc/nginx/conf.d/default.conf", > > } > > }, > > ) > > > > self.driver.get("http:.../localhost/api") > > > self.login() > > > > tests/e2e/test_provider_proxy_forward.py:145: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > 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 > > """ > > 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] > > > > if by == By.ID: > > by = By.CSS_SELECTOR > > value = f'[id="{value}"]' > > elif by == By.CLASS_NAME: > > by = By.CSS_SELECTOR > > value = f".{value}" > > elif by == By.NAME: > > by = By.CSS_SELECTOR > > value = f'[name="{value}"]' > > > > > 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:748: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > 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:354: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x5616dc0a3fc3 \\u003Cunknown>\\n#17 0x5616dc0b4944 \\u003Cunknown>\\n#18 0x7fb7a7bc7ac3 \\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 0x5616dc0b5793 > > E #1 0x5616dbda91c6 > > E #2 0x5616dbdf4358 > > E #3 0x5616dbdf4411 > > E #4 0x5616dbe37934 > > E #5 0x5616dbe163fd > > E #6 0x5616dbe34dd9 > > E #7 0x5616dbe16173 > > E #8 0x5616dbde72d3 > > E #9 0x5616dbde7c9e > > E #10 0x5616dc0798cb > > E #11 0x5616dc07d745 > > E #12 0x5616dc0662e1 > > E #13 0x5616dc07e2d2 > > E #14 0x5616dc04a17f > > E #15 0x5616dc0a3dc8 > > E #16 0x5616dc0a3fc3 > > E #17 0x5616dc0b4944 > > E #18 0x7fb7a7bc7ac3 > > > > ../../../.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 0x7f601c7649c0>] > > args = ('group_outpost_c1870dc8-fbaf-400d-99fe-8d45c6b0978b', {'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 0x7f6024094040>] > > args = ('group_outpost_c1870dc8-fbaf-400d-99fe-8d45c6b0978b', {'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_nginx(self): > > """Test nginx""" > > self.prepare() > > > > # Start nginx last so all hosts are resolvable, otherwise nginx exits > > self.run_container( > > image="docker.io/library/nginx:1.27", > > ports={ > > "80": "80", > > }, > > volumes={ > > f"{Path(__file__).parent / "proxy_forward_auth" / "nginx_single" / "nginx.conf"}": { > > "bind": "........./etc/nginx/conf.d/default.conf", > > } > > }, > > ) > > > > self.driver.get("http:.../localhost/api") > > > self.login() > > > > tests/e2e/test_provider_proxy_forward.py:145: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > 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 > > """ > > 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] > > > > if by == By.ID: > > by = By.CSS_SELECTOR > > value = f'[id="{value}"]' > > elif by == By.CLASS_NAME: > > by = By.CSS_SELECTOR > > value = f".{value}" > > elif by == By.NAME: > > by = By.CSS_SELECTOR > > value = f'[name="{value}"]' > > > > > 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:748: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > 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:354: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x5573907cefc3 \\u003Cunknown>\\n#17 0x5573907df944 \\u003Cunknown>\\n#18 0x7f7358817ac3 \\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 0x5573907e0793 > > E #1 0x5573904d41c6 > > E #2 0x55739051f358 > > E #3 0x55739051f411 > > E #4 0x557390562934 > > E #5 0x5573905413fd > > E #6 0x55739055fdd9 > > E #7 0x557390541173 > > E #8 0x5573905122d3 > > E #9 0x557390512c9e > > E #10 0x5573907a48cb > > E #11 0x5573907a8745 > > E #12 0x5573907912e1 > > E #13 0x5573907a92d2 > > E #14 0x55739077517f > > E #15 0x5573907cedc8 > > E #16 0x5573907cefc3 > > E #17 0x5573907df944 > > E #18 0x7f7358817ac3 > > > > ../../../.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_nginx(self): > > """Test nginx""" > > self.prepare() > > > > # Start nginx last so all hosts are resolvable, otherwise nginx exits > > self.run_container( > > image="docker.io/library/nginx:1.27", > > ports={ > > "80": "80", > > }, > > volumes={ > > f"{Path(__file__).parent / "proxy_forward_auth" / "nginx_single" / "nginx.conf"}": { > > "bind": "........./etc/nginx/conf.d/default.conf", > > } > > }, > > ) > > > > self.driver.get("http:.../localhost/api") > > > self.login() > > > > tests/e2e/test_provider_proxy_forward.py:145: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > 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 > > """ > > 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] > > > > if by == By.ID: > > by = By.CSS_SELECTOR > > value = f'[id="{value}"]' > > elif by == By.CLASS_NAME: > > by = By.CSS_SELECTOR > > value = f".{value}" > > elif by == By.NAME: > > by = By.CSS_SELECTOR > > value = f'[name="{value}"]' > > > > > 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:748: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > 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:354: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x55b530a93fc3 \\u003Cunknown>\\n#17 0x55b530aa4944 \\u003Cunknown>\\n#18 0x7fd703027ac3 \\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 0x55b530aa5793 > > E #1 0x55b5307991c6 > > E #2 0x55b5307e4358 > > E #3 0x55b5307e4411 > > E #4 0x55b530827934 > > E #5 0x55b5308063fd > > E #6 0x55b530824dd9 > > E #7 0x55b530806173 > > E #8 0x55b5307d72d3 > > E #9 0x55b5307d7c9e > > E #10 0x55b530a698cb > > E #11 0x55b530a6d745 > > E #12 0x55b530a562e1 > > E #13 0x55b530a6e2d2 > > E #14 0x55b530a3a17f > > E #15 0x55b530a93dc8 > > E #16 0x55b530a93fc3 > > E #17 0x55b530aa4944 > > E #18 0x7fd703027ac3 > > > > ../../../.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

netlify[bot] commented 16 hours ago

Deploy Preview for authentik-docs ready!

Name Link
Latest commit 4369cf8b7a0c1a023225852b432690ef06ae35a3
Latest deploy log https://app.netlify.com/sites/authentik-docs/deploys/673f5e0b6ca4270008a95cb0
Deploy Preview https://deploy-preview-12127--authentik-docs.netlify.app
Preview on mobile
Toggle QR Code...

QR Code

Use your smartphone camera to open QR code link.

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