goauthentik / authentik

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

security: fix CVE 2024 52289 #12113

Closed BeryJu closed 19 hours ago

BeryJu commented 20 hours ago

Details

REPLACE ME


Checklist

If an API change has been made

If changes to the frontend have been made

If applicable

netlify[bot] commented 20 hours ago

Deploy Preview for authentik-storybook failed.

Name Link
Latest commit f9c9194e808f089be9d78bc9f82e54df4e82233c
Latest deploy log https://app.netlify.com/sites/authentik-storybook/deploys/673f3536246b810008af068d
netlify[bot] commented 20 hours ago

Deploy Preview for authentik-docs ready!

Name Link
Latest commit f9c9194e808f089be9d78bc9f82e54df4e82233c
Latest deploy log https://app.netlify.com/sites/authentik-docs/deploys/673f35368e38970008d9b855
Deploy Preview https://deploy-preview-12113--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.

codecov[bot] commented 20 hours ago

:x: 6 Tests Failed:

Tests completed Failed Passed Skipped
1601 6 1595 1
View the top 3 failed tests by shortest run time > > ```python > tests.e2e.test_provider_proxy.TestProviderProxy::test_proxy_basic_auth > ``` > >
Stack Traces | 218s run time > > > > > ```python > > self = > > call_result = > > exc_info = (, NoSuchElementException(), ) > > task_context = None, context = [<_contextvars.Context object at 0x7fb777d70600>] > > args = ('group_outpost_6a1e2957-ef82-46f7-a036-adfd50bba56f', {'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 0x7fb77edd7d80>] > > args = ('group_outpost_6a1e2957-ef82-46f7-a036-adfd50bba56f', {'type': 'event.update'}) > > kwargs = {}, __traceback_hide__ = True > > current_task = result=None> > > result = None > > > > async def main_wrap( > > self, > > call_result: "Future[_R]", > > exc_info: "OptExcInfo", > > task_context: "Optional[List[asyncio.Task[Any]]]", > > context: List[contextvars.Context], > > *args: _P.args, > > **kwargs: _P.kwargs, > > ) -> None: > > """ > > Wraps the awaitable with something that puts the result into the > > result/exception future. > > """ > > > > __traceback_hide__ = True # noqa: F841 > > > > if context is not None: > > _restore_context(context[0]) > > > > current_task = asyncio.current_task() > > if current_task is not None and task_context is not None: > > task_context.append(current_task) > > > > try: > > # If we have an exception, run the function inside the except block > > # after raising it so exc_info is correctly populated. > > if exc_info[1]: > > try: > > > raise exc_info[1] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > > return func(self, *args, **kwargs) > > > > tests/e2e/utils.py:287: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'system/providers-proxy.yaml' > > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, config = > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > config = apps.get_app_config(app_name) > > if isinstance(config, ManagedAppConfig): > > config._on_startup_callback(None) > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:43: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > @retry() > > @apply_blueprint( > > "default/flow-default-authentication-flow.yaml", > > "default/flow-default-invalidation-flow.yaml", > > ) > > @apply_blueprint( > > "default/flow-default-provider-authorization-implicit-consent.yaml", > > "default/flow-default-provider-invalidation.yaml", > > ) > > @apply_blueprint( > > "system/providers-oauth2.yaml", > > "system/providers-proxy.yaml", > > ) > > @reconcile_app("authentik_crypto") > > def test_proxy_basic_auth(self): > > """Test simple outpost setup with single provider""" > > cred = generate_id() > > attr = "basic-password" # nosec > > self.user.attributes["basic-username"] = cred > > self.user.attributes[attr] = cred > > self.user.save() > > > > proxy: ProxyProvider = ProxyProvider.objects.create( > > name=generate_id(), > > authorization_flow=Flow.objects.get( > > slug="default-provider-authorization-implicit-consent" > > ), > > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > > internal_host=f"http://{self.host}", > > external_host="http://localhost:9000", > > basic_auth_enabled=True, > > basic_auth_user_attribute="basic-username", > > basic_auth_password_attribute=attr, > > ) > > # Ensure OAuth2 Params are set > > proxy.set_oauth_defaults() > > proxy.save() > > # we need to create an application to actually access the proxy > > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > > outpost: Outpost = Outpost.objects.create( > > name=generate_id(), > > type=OutpostType.PROXY, > > ) > > outpost.providers.add(proxy) > > outpost.build_user_permissions(outpost.user) > > > > self.start_proxy(outpost) > > > > # Wait until outpost healthcheck succeeds > > healthcheck_retries = 0 > > while healthcheck_retries < 50: # noqa: PLR2004 > > if len(outpost.state) > 0: > > state = outpost.state[0] > > if state.last_seen: > > break > > healthcheck_retries += 1 > > sleep(0.5) > > sleep(5) > > > > self.driver.get("http://localhost:9000/api") > > > self.login() > > > > tests/e2e/test_provider_proxy.py:184: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > def login(self): > > """Do entire login flow and check user afterwards""" > > > flow_executor = self.get_shadow_root("ak-flow-executor") > > > > tests/e2e/utils.py:232: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > selector = 'ak-flow-executor' > > container = > > > > def get_shadow_root( > > self, selector: str, container: WebElement | WebDriver | None = None > > ) -> WebElement: > > """Get shadow root element's inner shadowRoot""" > > if not container: > > container = self.driver > > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > > > tests/e2e/utils.py:226: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > by = 'css selector', value = 'ak-flow-executor' > > > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > > """Find an element given a By strategy and locator. > > > > :Usage: > > :: > > > > element = driver.find_element(By.ID, 'foo') > > > > :rtype: WebElement > > """ > > by, value = self.locator_converter.convert(by, value) > > > > if isinstance(by, RelativeBy): > > elements = self.find_elements(by=by, value=value) > > if not elements: > > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > > return elements[0] > > > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > driver_command = 'findElement' > > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > > > def execute(self, driver_command: str, params: dict = None) -> dict: > > """Sends a command to be executed by a command.CommandExecutor. > > > > :Args: > > - driver_command: The name of the command to execute as a string. > > - params: A dictionary of named parameters to send with the command. > > > > :Returns: > > The command's JSON response loaded into a dictionary object. > > """ > > params = self._wrap_value(params) > > > > if self.session_id: > > if not params: > > params = {"sessionId": self.session_id} > > elif "sessionId" not in params: > > params["sessionId"] = self.session_id > > > > response = self.command_executor.execute(driver_command, params) > > if response: > > > self.error_handler.check_response(response) > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x5581a02e5fc3 \\u003Cunknown>\\n#17 0x5581a02f6944 \\u003Cunknown>\\n#18 0x7fdd2bb13ac3 \\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 0x5581a02f7793 > > E #1 0x55819ffeb1c6 > > E #2 0x5581a0036358 > > E #3 0x5581a0036411 > > E #4 0x5581a0079934 > > E #5 0x5581a00583fd > > E #6 0x5581a0076dd9 > > E #7 0x5581a0058173 > > E #8 0x5581a00292d3 > > E #9 0x5581a0029c9e > > E #10 0x5581a02bb8cb > > E #11 0x5581a02bf745 > > E #12 0x5581a02a82e1 > > E #13 0x5581a02c02d2 > > E #14 0x5581a028c17f > > E #15 0x5581a02e5dc8 > > E #16 0x5581a02e5fc3 > > E #17 0x5581a02f6944 > > E #18 0x7fdd2bb13ac3 > > > > ../../../.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 0x7fb775039c80>] > > args = ('group_outpost_8cf242a6-c366-4ede-a8b8-0cb327a4b6e1', {'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 0x7fb777a0ee00>] > > args = ('group_outpost_8cf242a6-c366-4ede-a8b8-0cb327a4b6e1', {'type': 'event.update'}) > > kwargs = {}, __traceback_hide__ = True > > current_task = result=None> > > result = None > > > > async def main_wrap( > > self, > > call_result: "Future[_R]", > > exc_info: "OptExcInfo", > > task_context: "Optional[List[asyncio.Task[Any]]]", > > context: List[contextvars.Context], > > *args: _P.args, > > **kwargs: _P.kwargs, > > ) -> None: > > """ > > Wraps the awaitable with something that puts the result into the > > result/exception future. > > """ > > > > __traceback_hide__ = True # noqa: F841 > > > > if context is not None: > > _restore_context(context[0]) > > > > current_task = asyncio.current_task() > > if current_task is not None and task_context is not None: > > task_context.append(current_task) > > > > try: > > # If we have an exception, run the function inside the except block > > # after raising it so exc_info is correctly populated. > > if exc_info[1]: > > try: > > > raise exc_info[1] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > > return func(self, *args, **kwargs) > > > > tests/e2e/utils.py:287: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'system/providers-proxy.yaml' > > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, config = > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > config = apps.get_app_config(app_name) > > if isinstance(config, ManagedAppConfig): > > config._on_startup_callback(None) > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:43: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > @retry() > > @apply_blueprint( > > "default/flow-default-authentication-flow.yaml", > > "default/flow-default-invalidation-flow.yaml", > > ) > > @apply_blueprint( > > "default/flow-default-provider-authorization-implicit-consent.yaml", > > "default/flow-default-provider-invalidation.yaml", > > ) > > @apply_blueprint( > > "system/providers-oauth2.yaml", > > "system/providers-proxy.yaml", > > ) > > @reconcile_app("authentik_crypto") > > def test_proxy_basic_auth(self): > > """Test simple outpost setup with single provider""" > > cred = generate_id() > > attr = "basic-password" # nosec > > self.user.attributes["basic-username"] = cred > > self.user.attributes[attr] = cred > > self.user.save() > > > > proxy: ProxyProvider = ProxyProvider.objects.create( > > name=generate_id(), > > authorization_flow=Flow.objects.get( > > slug="default-provider-authorization-implicit-consent" > > ), > > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > > internal_host=f"http://{self.host}", > > external_host="http://localhost:9000", > > basic_auth_enabled=True, > > basic_auth_user_attribute="basic-username", > > basic_auth_password_attribute=attr, > > ) > > # Ensure OAuth2 Params are set > > proxy.set_oauth_defaults() > > proxy.save() > > # we need to create an application to actually access the proxy > > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > > outpost: Outpost = Outpost.objects.create( > > name=generate_id(), > > type=OutpostType.PROXY, > > ) > > outpost.providers.add(proxy) > > outpost.build_user_permissions(outpost.user) > > > > self.start_proxy(outpost) > > > > # Wait until outpost healthcheck succeeds > > healthcheck_retries = 0 > > while healthcheck_retries < 50: # noqa: PLR2004 > > if len(outpost.state) > 0: > > state = outpost.state[0] > > if state.last_seen: > > break > > healthcheck_retries += 1 > > sleep(0.5) > > sleep(5) > > > > self.driver.get("http://localhost:9000/api") > > > self.login() > > > > tests/e2e/test_provider_proxy.py:184: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > def login(self): > > """Do entire login flow and check user afterwards""" > > > flow_executor = self.get_shadow_root("ak-flow-executor") > > > > tests/e2e/utils.py:232: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > selector = 'ak-flow-executor' > > container = > > > > def get_shadow_root( > > self, selector: str, container: WebElement | WebDriver | None = None > > ) -> WebElement: > > """Get shadow root element's inner shadowRoot""" > > if not container: > > container = self.driver > > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > > > tests/e2e/utils.py:226: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > by = 'css selector', value = 'ak-flow-executor' > > > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > > """Find an element given a By strategy and locator. > > > > :Usage: > > :: > > > > element = driver.find_element(By.ID, 'foo') > > > > :rtype: WebElement > > """ > > by, value = self.locator_converter.convert(by, value) > > > > if isinstance(by, RelativeBy): > > elements = self.find_elements(by=by, value=value) > > if not elements: > > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > > return elements[0] > > > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > driver_command = 'findElement' > > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > > > def execute(self, driver_command: str, params: dict = None) -> dict: > > """Sends a command to be executed by a command.CommandExecutor. > > > > :Args: > > - driver_command: The name of the command to execute as a string. > > - params: A dictionary of named parameters to send with the command. > > > > :Returns: > > The command's JSON response loaded into a dictionary object. > > """ > > params = self._wrap_value(params) > > > > if self.session_id: > > if not params: > > params = {"sessionId": self.session_id} > > elif "sessionId" not in params: > > params["sessionId"] = self.session_id > > > > response = self.command_executor.execute(driver_command, params) > > if response: > > > self.error_handler.check_response(response) > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x563a2412cfc3 \\u003Cunknown>\\n#17 0x563a2413d944 \\u003Cunknown>\\n#18 0x7fec21051ac3 \\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 0x563a2413e793 > > E #1 0x563a23e321c6 > > E #2 0x563a23e7d358 > > E #3 0x563a23e7d411 > > E #4 0x563a23ec0934 > > E #5 0x563a23e9f3fd > > E #6 0x563a23ebddd9 > > E #7 0x563a23e9f173 > > E #8 0x563a23e702d3 > > E #9 0x563a23e70c9e > > E #10 0x563a241028cb > > E #11 0x563a24106745 > > E #12 0x563a240ef2e1 > > E #13 0x563a241072d2 > > E #14 0x563a240d317f > > E #15 0x563a2412cdc8 > > E #16 0x563a2412cfc3 > > E #17 0x563a2413d944 > > E #18 0x7fec21051ac3 > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > > > During handling of the above exception, another exception occurred: > > > > self = > > test_case = > > subTest = False > > > > @contextlib.contextmanager > > def testPartExecutor(self, test_case, subTest=False): > > old_success = self.success > > self.success = True > > try: > > > yield > > > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:58: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > result = > > > > def run(self, result=None): > > if result is None: > > result = self.defaultTestResult() > > startTestRun = getattr(result, 'startTestRun', None) > > stopTestRun = getattr(result, 'stopTestRun', None) > > if startTestRun is not None: > > startTestRun() > > else: > > stopTestRun = None > > > > result.startTest(self) > > try: > > testMethod = getattr(self, self._testMethodName) > > if (getattr(self.__class__, "__unittest_skip__", False) or > > getattr(testMethod, "__unittest_skip__", False)): > > # If the class or method was skipped. > > skip_why = (getattr(self.__class__, '__unittest_skip_why__', '') > > or getattr(testMethod, '__unittest_skip_why__', '')) > > _addSkip(result, self, skip_why) > > return result > > > > expecting_failure = ( > > getattr(self, "__unittest_expecting_failure__", False) or > > getattr(testMethod, "__unittest_expecting_failure__", False) > > ) > > outcome = _Outcome(result) > > start_time = time.perf_counter() > > try: > > self._outcome = outcome > > > > with outcome.testPartExecutor(self): > > self._callSetUp() > > if outcome.success: > > outcome.expecting_failure = expecting_failure > > with outcome.testPartExecutor(self): > > > self._callTestMethod(testMethod) > > > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:634: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > method = > > > > > def _callTestMethod(self, method): > > > if method() is not None: > > > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:589: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > return func(self, *args, **kwargs) > > > > except tuple(exceptions) as exc: > > count += 1 > > if count > max_retires: > > logger.debug("Exceeded retry count", exc=exc, test=self) > > > > raise exc > > logger.debug("Retrying on error", exc=exc, test=self) > > self.tearDown() > > self._post_teardown() > > self._pre_setup() > > self.setUp() > > > return wrapper(self, *args, **kwargs) > > > > tests/e2e/utils.py:300: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > return func(self, *args, **kwargs) > > > > except tuple(exceptions) as exc: > > count += 1 > > if count > max_retires: > > logger.debug("Exceeded retry count", exc=exc, test=self) > > > > raise exc > > logger.debug("Retrying on error", exc=exc, test=self) > > self.tearDown() > > self._post_teardown() > > self._pre_setup() > > self.setUp() > > > return wrapper(self, *args, **kwargs) > > > > tests/e2e/utils.py:300: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > return func(self, *args, **kwargs) > > > > except tuple(exceptions) as exc: > > count += 1 > > if count > max_retires: > > logger.debug("Exceeded retry count", exc=exc, test=self) > > > > > raise exc > > > > tests/e2e/utils.py:294: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > > return func(self, *args, **kwargs) > > > > tests/e2e/utils.py:287: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'system/providers-proxy.yaml' > > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, config = > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > config = apps.get_app_config(app_name) > > if isinstance(config, ManagedAppConfig): > > config._on_startup_callback(None) > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:43: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > @retry() > > @apply_blueprint( > > "default/flow-default-authentication-flow.yaml", > > "default/flow-default-invalidation-flow.yaml", > > ) > > @apply_blueprint( > > "default/flow-default-provider-authorization-implicit-consent.yaml", > > "default/flow-default-provider-invalidation.yaml", > > ) > > @apply_blueprint( > > "system/providers-oauth2.yaml", > > "system/providers-proxy.yaml", > > ) > > @reconcile_app("authentik_crypto") > > def test_proxy_basic_auth(self): > > """Test simple outpost setup with single provider""" > > cred = generate_id() > > attr = "basic-password" # nosec > > self.user.attributes["basic-username"] = cred > > self.user.attributes[attr] = cred > > self.user.save() > > > > proxy: ProxyProvider = ProxyProvider.objects.create( > > name=generate_id(), > > authorization_flow=Flow.objects.get( > > slug="default-provider-authorization-implicit-consent" > > ), > > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > > internal_host=f"http://{self.host}", > > external_host="http://localhost:9000", > > basic_auth_enabled=True, > > basic_auth_user_attribute="basic-username", > > basic_auth_password_attribute=attr, > > ) > > # Ensure OAuth2 Params are set > > proxy.set_oauth_defaults() > > proxy.save() > > # we need to create an application to actually access the proxy > > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > > outpost: Outpost = Outpost.objects.create( > > name=generate_id(), > > type=OutpostType.PROXY, > > ) > > outpost.providers.add(proxy) > > outpost.build_user_permissions(outpost.user) > > > > self.start_proxy(outpost) > > > > # Wait until outpost healthcheck succeeds > > healthcheck_retries = 0 > > while healthcheck_retries < 50: # noqa: PLR2004 > > if len(outpost.state) > 0: > > state = outpost.state[0] > > if state.last_seen: > > break > > healthcheck_retries += 1 > > sleep(0.5) > > sleep(5) > > > > self.driver.get("http://localhost:9000/api") > > > self.login() > > > > tests/e2e/test_provider_proxy.py:184: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > def login(self): > > """Do entire login flow and check user afterwards""" > > > flow_executor = self.get_shadow_root("ak-flow-executor") > > > > tests/e2e/utils.py:232: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > selector = 'ak-flow-executor' > > container = > > > > def get_shadow_root( > > self, selector: str, container: WebElement | WebDriver | None = None > > ) -> WebElement: > > """Get shadow root element's inner shadowRoot""" > > if not container: > > container = self.driver > > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > > > tests/e2e/utils.py:226: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > by = 'css selector', value = 'ak-flow-executor' > > > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > > """Find an element given a By strategy and locator. > > > > :Usage: > > :: > > > > element = driver.find_element(By.ID, 'foo') > > > > :rtype: WebElement > > """ > > by, value = self.locator_converter.convert(by, value) > > > > if isinstance(by, RelativeBy): > > elements = self.find_elements(by=by, value=value) > > if not elements: > > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > > return elements[0] > > > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > driver_command = 'findElement' > > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > > > def execute(self, driver_command: str, params: dict = None) -> dict: > > """Sends a command to be executed by a command.CommandExecutor. > > > > :Args: > > - driver_command: The name of the command to execute as a string. > > - params: A dictionary of named parameters to send with the command. > > > > :Returns: > > The command's JSON response loaded into a dictionary object. > > """ > > params = self._wrap_value(params) > > > > if self.session_id: > > if not params: > > params = {"sessionId": self.session_id} > > elif "sessionId" not in params: > > params["sessionId"] = self.session_id > > > > response = self.command_executor.execute(driver_command, params) > > if response: > > > self.error_handler.check_response(response) > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x555b56b20fc3 \\u003Cunknown>\\n#17 0x555b56b31944 \\u003Cunknown>\\n#18 0x7f68bb137ac3 \\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 0x555b56b32793 > > E #1 0x555b568261c6 > > E #2 0x555b56871358 > > E #3 0x555b56871411 > > E #4 0x555b568b4934 > > E #5 0x555b568933fd > > E #6 0x555b568b1dd9 > > E #7 0x555b56893173 > > E #8 0x555b568642d3 > > E #9 0x555b56864c9e > > E #10 0x555b56af68cb > > E #11 0x555b56afa745 > > E #12 0x555b56ae32e1 > > E #13 0x555b56afb2d2 > > E #14 0x555b56ac717f > > E #15 0x555b56b20dc8 > > E #16 0x555b56b20fc3 > > E #17 0x555b56b31944 > > E #18 0x7f68bb137ac3 > > > > ../../../.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 | 219s run time > > ```python > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fb777cea840>] > args = ('group_outpost_75d18855-49df-4c63-b7b0-35697a25c627', {'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 0x7fb77767be00>] > args = ('group_outpost_75d18855-49df-4c63-b7b0-35697a25c627', {'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 > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x560fb6b44fc3 \\u003Cunknown>\\n#17 0x560fb6b55944 \\u003Cunknown>\\n#18 0x7ff78ca8fac3 \\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 0x560fb6b56793 > E #1 0x560fb684a1c6 > E #2 0x560fb6895358 > E #3 0x560fb6895411 > E #4 0x560fb68d8934 > E #5 0x560fb68b73fd > E #6 0x560fb68d5dd9 > E #7 0x560fb68b7173 > E #8 0x560fb68882d3 > E #9 0x560fb6888c9e > E #10 0x560fb6b1a8cb > E #11 0x560fb6b1e745 > E #12 0x560fb6b072e1 > E #13 0x560fb6b1f2d2 > E #14 0x560fb6aeb17f > E #15 0x560fb6b44dc8 > E #16 0x560fb6b44fc3 > E #17 0x560fb6b55944 > E #18 0x7ff78ca8fac3 > > ../../../.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 0x7fb7776b7240>] > args = ('group_outpost_d427e750-a65d-43c4-946d-fb568fbabf1d', {'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 0x7fb777628d80>] > args = ('group_outpost_d427e750-a65d-43c4-946d-fb568fbabf1d', {'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 > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x556aea8e2fc3 \\u003Cunknown>\\n#17 0x556aea8f3944 \\u003Cunknown>\\n#18 0x7fd110559ac3 \\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 0x556aea8f4793 > E #1 0x556aea5e81c6 > E #2 0x556aea633358 > E #3 0x556aea633411 > E #4 0x556aea676934 > E #5 0x556aea6553fd > E #6 0x556aea673dd9 > E #7 0x556aea655173 > E #8 0x556aea6262d3 > E #9 0x556aea626c9e > E #10 0x556aea8b88cb > E #11 0x556aea8bc745 > E #12 0x556aea8a52e1 > E #13 0x556aea8bd2d2 > E #14 0x556aea88917f > E #15 0x556aea8e2dc8 > E #16 0x556aea8e2fc3 > E #17 0x556aea8f3944 > E #18 0x7fd110559ac3 > > ../../../.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 > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x557083b83fc3 \\u003Cunknown>\\n#17 0x557083b94944 \\u003Cunknown>\\n#18 0x7f94fb543ac3 \\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 0x557083b95793 > E #1 0x5570838891c6 > E #2 0x5570838d4358 > E #3 0x5570838d4411 > E #4 0x557083917934 > E #5 0x5570838f63fd > E #6 0x557083914dd9 > E #7 0x5570838f6173 > E #8 0x5570838c72d3 > E #9 0x5570838c7c9e > E #10 0x557083b598cb > E #11 0x557083b5d745 > E #12 0x557083b462e1 > E #13 0x557083b5e2d2 > E #14 0x557083b2a17f > E #15 0x557083b83dc8 > E #16 0x557083b83fc3 > E #17 0x557083b94944 > E #18 0x7f94fb543ac3 > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > ```
tests.e2e.test_provider_proxy_forward.TestProviderProxyForward::test_traefik
Stack Traces | 219s run time > > ```python > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fb7777bd7c0>] > args = ('group_outpost_f49f9fc0-c23f-4d43-9498-05119da93aae', {'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 0x7fb7778a4e80>] > args = ('group_outpost_f49f9fc0-c23f-4d43-9498-05119da93aae', {'type': 'event.update'}) > kwargs = {}, __traceback_hide__ = True > current_task = result=None> > result = None > > async def main_wrap( > self, > call_result: "Future[_R]", > exc_info: "OptExcInfo", > task_context: "Optional[List[asyncio.Task[Any]]]", > context: List[contextvars.Context], > *args: _P.args, > **kwargs: _P.kwargs, > ) -> None: > """ > Wraps the awaitable with something that puts the result into the > result/exception future. > """ > > __traceback_hide__ = True # noqa: F841 > > if context is not None: > _restore_context(context[0]) > > current_task = asyncio.current_task() > if current_task is not None and task_context is not None: > task_context.append(current_task) > > try: > # If we have an exception, run the function inside the except block > # after raising it so exc_info is correctly populated. > if exc_info[1]: > try: > > raise exc_info[1] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > > return func(self, *args, **kwargs) > > tests/e2e/utils.py:287: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > @retry() > def test_traefik(self): > """Test traefik""" > local_config_path = ( > Path(__file__).parent / "proxy_forward_auth" / "traefik_single" / "config-static.yaml" > ) > self.run_container( > image="docker.io/library/traefik:3.1", > ports={ > "80": "80", > }, > volumes={ > local_config_path: { > "bind": "........./etc/traefik/traefik.yml", > } > }, > ) > > self.prepare() > > self.driver.get("http:.../localhost/api") > > self.login() > > tests/e2e/test_provider_proxy_forward.py:111: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > def login(self): > """Do entire login flow and check user afterwards""" > > flow_executor = self.get_shadow_root("ak-flow-executor") > > tests/e2e/utils.py:232: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > selector = 'ak-flow-executor' > container = > > def get_shadow_root( > self, selector: str, container: WebElement | WebDriver | None = None > ) -> WebElement: > """Get shadow root element's inner shadowRoot""" > if not container: > container = self.driver > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > tests/e2e/utils.py:226: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > by = 'css selector', value = 'ak-flow-executor' > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > """Find an element given a By strategy and locator. > > :Usage: > :: > > element = driver.find_element(By.ID, 'foo') > > :rtype: WebElement > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x555cc81c4fc3 \\u003Cunknown>\\n#17 0x555cc81d5944 \\u003Cunknown>\\n#18 0x7f7435070ac3 \\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 0x555cc81d6793 > E #1 0x555cc7eca1c6 > E #2 0x555cc7f15358 > E #3 0x555cc7f15411 > E #4 0x555cc7f58934 > E #5 0x555cc7f373fd > E #6 0x555cc7f55dd9 > E #7 0x555cc7f37173 > E #8 0x555cc7f082d3 > E #9 0x555cc7f08c9e > E #10 0x555cc819a8cb > E #11 0x555cc819e745 > E #12 0x555cc81872e1 > E #13 0x555cc819f2d2 > E #14 0x555cc816b17f > E #15 0x555cc81c4dc8 > E #16 0x555cc81c4fc3 > E #17 0x555cc81d5944 > E #18 0x7f7435070ac3 > > ../../../.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 0x7fb777612e40>] > args = ('group_outpost_74efc75f-902a-467d-b7bc-9afa02922017', {'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 0x7fb777cec3c0>] > args = ('group_outpost_74efc75f-902a-467d-b7bc-9afa02922017', {'type': 'event.update'}) > kwargs = {}, __traceback_hide__ = True > current_task = result=None> > result = None > > async def main_wrap( > self, > call_result: "Future[_R]", > exc_info: "OptExcInfo", > task_context: "Optional[List[asyncio.Task[Any]]]", > context: List[contextvars.Context], > *args: _P.args, > **kwargs: _P.kwargs, > ) -> None: > """ > Wraps the awaitable with something that puts the result into the > result/exception future. > """ > > __traceback_hide__ = True # noqa: F841 > > if context is not None: > _restore_context(context[0]) > > current_task = asyncio.current_task() > if current_task is not None and task_context is not None: > task_context.append(current_task) > > try: > # If we have an exception, run the function inside the except block > # after raising it so exc_info is correctly populated. > if exc_info[1]: > try: > > raise exc_info[1] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > > return func(self, *args, **kwargs) > > tests/e2e/utils.py:287: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > @retry() > def test_traefik(self): > """Test traefik""" > local_config_path = ( > Path(__file__).parent / "proxy_forward_auth" / "traefik_single" / "config-static.yaml" > ) > self.run_container( > image="docker.io/library/traefik:3.1", > ports={ > "80": "80", > }, > volumes={ > local_config_path: { > "bind": "........./etc/traefik/traefik.yml", > } > }, > ) > > self.prepare() > > self.driver.get("http:.../localhost/api") > > self.login() > > tests/e2e/test_provider_proxy_forward.py:111: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > def login(self): > """Do entire login flow and check user afterwards""" > > flow_executor = self.get_shadow_root("ak-flow-executor") > > tests/e2e/utils.py:232: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > selector = 'ak-flow-executor' > container = > > def get_shadow_root( > self, selector: str, container: WebElement | WebDriver | None = None > ) -> WebElement: > """Get shadow root element's inner shadowRoot""" > if not container: > container = self.driver > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > tests/e2e/utils.py:226: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > by = 'css selector', value = 'ak-flow-executor' > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > """Find an element given a By strategy and locator. > > :Usage: > :: > > element = driver.find_element(By.ID, 'foo') > > :rtype: WebElement > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x56430dfa4fc3 \\u003Cunknown>\\n#17 0x56430dfb5944 \\u003Cunknown>\\n#18 0x7f287b196ac3 \\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 0x56430dfb6793 > E #1 0x56430dcaa1c6 > E #2 0x56430dcf5358 > E #3 0x56430dcf5411 > E #4 0x56430dd38934 > E #5 0x56430dd173fd > E #6 0x56430dd35dd9 > E #7 0x56430dd17173 > E #8 0x56430dce82d3 > E #9 0x56430dce8c9e > E #10 0x56430df7a8cb > E #11 0x56430df7e745 > E #12 0x56430df672e1 > E #13 0x56430df7f2d2 > E #14 0x56430df4b17f > E #15 0x56430dfa4dc8 > E #16 0x56430dfa4fc3 > E #17 0x56430dfb5944 > E #18 0x7f287b196ac3 > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > During handling of the above exception, another exception occurred: > > self = > test_case = > subTest = False > > @contextlib.contextmanager > def testPartExecutor(self, test_case, subTest=False): > old_success = self.success > self.success = True > try: > > yield > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:58: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > result = > > def run(self, result=None): > if result is None: > result = self.defaultTestResult() > startTestRun = getattr(result, 'startTestRun', None) > stopTestRun = getattr(result, 'stopTestRun', None) > if startTestRun is not None: > startTestRun() > else: > stopTestRun = None > > result.startTest(self) > try: > testMethod = getattr(self, self._testMethodName) > if (getattr(self.__class__, "__unittest_skip__", False) or > getattr(testMethod, "__unittest_skip__", False)): > # If the class or method was skipped. > skip_why = (getattr(self.__class__, '__unittest_skip_why__', '') > or getattr(testMethod, '__unittest_skip_why__', '')) > _addSkip(result, self, skip_why) > return result > > expecting_failure = ( > getattr(self, "__unittest_expecting_failure__", False) or > getattr(testMethod, "__unittest_expecting_failure__", False) > ) > outcome = _Outcome(result) > start_time = time.perf_counter() > try: > self._outcome = outcome > > with outcome.testPartExecutor(self): > self._callSetUp() > if outcome.success: > outcome.expecting_failure = expecting_failure > with outcome.testPartExecutor(self): > > self._callTestMethod(testMethod) > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:634: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > method = > > > def _callTestMethod(self, method): > > if method() is not None: > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:589: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > return func(self, *args, **kwargs) > > except tuple(exceptions) as exc: > count += 1 > if count > max_retires: > logger.debug("Exceeded retry count", exc=exc, test=self) > > raise exc > logger.debug("Retrying on error", exc=exc, test=self) > self.tearDown() > self._post_teardown() > self._pre_setup() > self.setUp() > > return wrapper(self, *args, **kwargs) > > tests/e2e/utils.py:300: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > return func(self, *args, **kwargs) > > except tuple(exceptions) as exc: > count += 1 > if count > max_retires: > logger.debug("Exceeded retry count", exc=exc, test=self) > > raise exc > logger.debug("Retrying on error", exc=exc, test=self) > self.tearDown() > self._post_teardown() > self._pre_setup() > self.setUp() > > return wrapper(self, *args, **kwargs) > > tests/e2e/utils.py:300: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > return func(self, *args, **kwargs) > > except tuple(exceptions) as exc: > count += 1 > if count > max_retires: > logger.debug("Exceeded retry count", exc=exc, test=self) > > > raise exc > > tests/e2e/utils.py:294: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > args = (), kwargs = {} > > @wraps(func) > def wrapper(self: TransactionTestCase, *args, **kwargs): > """Run test again if we're below max_retries, including tearDown and > setUp. Otherwise raise the error""" > nonlocal count > try: > > return func(self, *args, **kwargs) > > tests/e2e/utils.py:287: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > @retry() > def test_traefik(self): > """Test traefik""" > local_config_path = ( > Path(__file__).parent / "proxy_forward_auth" / "traefik_single" / "config-static.yaml" > ) > self.run_container( > image="docker.io/library/traefik:3.1", > ports={ > "80": "80", > }, > volumes={ > local_config_path: { > "bind": "........./etc/traefik/traefik.yml", > } > }, > ) > > self.prepare() > > self.driver.get("http:.../localhost/api") > > self.login() > > tests/e2e/test_provider_proxy_forward.py:111: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > > def login(self): > """Do entire login flow and check user afterwards""" > > flow_executor = self.get_shadow_root("ak-flow-executor") > > tests/e2e/utils.py:232: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > selector = 'ak-flow-executor' > container = > > def get_shadow_root( > self, selector: str, container: WebElement | WebDriver | None = None > ) -> WebElement: > """Get shadow root element's inner shadowRoot""" > if not container: > container = self.driver > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > tests/e2e/utils.py:226: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > by = 'css selector', value = 'ak-flow-executor' > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > """Find an element given a By strategy and locator. > > :Usage: > :: > > element = driver.find_element(By.ID, 'foo') > > :rtype: WebElement > """ > by, value = self.locator_converter.convert(by, value) > > if isinstance(by, RelativeBy): > elements = self.find_elements(by=by, value=value) > if not elements: > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > return elements[0] > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > driver_command = 'findElement' > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > def execute(self, driver_command: str, params: dict = None) -> dict: > """Sends a command to be executed by a command.CommandExecutor. > > :Args: > - driver_command: The name of the command to execute as a string. > - params: A dictionary of named parameters to send with the command. > > :Returns: > The command's JSON response loaded into a dictionary object. > """ > params = self._wrap_value(params) > > if self.session_id: > if not params: > params = {"sessionId": self.session_id} > elif "sessionId" not in params: > params["sessionId"] = self.session_id > > response = self.command_executor.execute(driver_command, params) > if response: > > self.error_handler.check_response(response) > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > self = > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x5588b5244fc3 \\u003Cunknown>\\n#17 0x5588b5255944 \\u003Cunknown>\\n#18 0x7f3a75296ac3 \\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 0x5588b5256793 > E #1 0x5588b4f4a1c6 > E #2 0x5588b4f95358 > E #3 0x5588b4f95411 > E #4 0x5588b4fd8934 > E #5 0x5588b4fb73fd > E #6 0x5588b4fd5dd9 > E #7 0x5588b4fb7173 > E #8 0x5588b4f882d3 > E #9 0x5588b4f88c9e > E #10 0x5588b521a8cb > E #11 0x5588b521e745 > E #12 0x5588b52072e1 > E #13 0x5588b521f2d2 > E #14 0x5588b51eb17f > E #15 0x5588b5244dc8 > E #16 0x5588b5244fc3 > E #17 0x5588b5255944 > E #18 0x7f3a75296ac3 > > ../../../.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

BeryJu commented 20 hours ago

/cherry-pick version-2024.10

BeryJu commented 20 hours ago

/cherry-pick version-2024.8

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

Cherry-pick failed with Merge error 85bb638243c8d7ea42ddd3b15b3f51a90d2b8c54 into temp-cherry-pick-d31586-version-2024.10

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

Cherry-pick failed with Merge error 85bb638243c8d7ea42ddd3b15b3f51a90d2b8c54 into temp-cherry-pick-d31586-version-2024.8