goauthentik / authentik

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

web: bump API Client version #12118

Closed authentik-automation[bot] closed 19 hours ago

authentik-automation[bot] commented 19 hours ago

web: bump API Client version

netlify[bot] commented 19 hours ago

Deploy Preview for authentik-storybook ready!

Name Link
Latest commit 79d228cbc7eae69f30ffe08940d24985af5289db
Latest deploy log https://app.netlify.com/sites/authentik-storybook/deploys/673f3a737147340008e6cd94
Deploy Preview https://deploy-preview-12118--authentik-storybook.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.

netlify[bot] commented 19 hours ago

Deploy Preview for authentik-docs canceled.

Name Link
Latest commit 79d228cbc7eae69f30ffe08940d24985af5289db
Latest deploy log https://app.netlify.com/sites/authentik-docs/deploys/673f3a735f3ae60008ebfba7
codecov[bot] commented 19 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_simple > ``` > >
Stack Traces | 216s run time > > > > > ```python > > self = > > call_result = > > exc_info = (, NoSuchElementException(), ) > > task_context = None, context = [<_contextvars.Context object at 0x7fe03d47b2c0>] > > args = ('group_outpost_9b17fc18-8433-4ad2-a942-dd80f0bcfabc', {'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 0x7fe03d5c4e80>] > > args = ('group_outpost_9b17fc18-8433-4ad2-a942-dd80f0bcfabc', {'type': 'event.update'}) > > kwargs = {}, __traceback_hide__ = True > > current_task = result=None> > > result = None > > > > async def main_wrap( > > self, > > call_result: "Future[_R]", > > exc_info: "OptExcInfo", > > task_context: "Optional[List[asyncio.Task[Any]]]", > > context: List[contextvars.Context], > > *args: _P.args, > > **kwargs: _P.kwargs, > > ) -> None: > > """ > > Wraps the awaitable with something that puts the result into the > > result/exception future. > > """ > > > > __traceback_hide__ = True # noqa: F841 > > > > if context is not None: > > _restore_context(context[0]) > > > > current_task = asyncio.current_task() > > if current_task is not None and task_context is not None: > > task_context.append(current_task) > > > > try: > > # If we have an exception, run the function inside the except block > > # after raising it so exc_info is correctly populated. > > if exc_info[1]: > > try: > > > raise exc_info[1] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > > return func(self, *args, **kwargs) > > > > tests/e2e/utils.py:287: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'system/providers-proxy.yaml' > > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, config = > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > config = apps.get_app_config(app_name) > > if isinstance(config, ManagedAppConfig): > > config._on_startup_callback(None) > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:43: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > @retry() > > @apply_blueprint( > > "default/flow-default-authentication-flow.yaml", > > "default/flow-default-invalidation-flow.yaml", > > ) > > @apply_blueprint( > > "default/flow-default-provider-authorization-implicit-consent.yaml", > > "default/flow-default-provider-invalidation.yaml", > > ) > > @apply_blueprint( > > "system/providers-oauth2.yaml", > > "system/providers-proxy.yaml", > > ) > > @reconcile_app("authentik_crypto") > > def test_proxy_simple(self): > > """Test simple outpost setup with single provider""" > > # set additionalHeaders to test later > > self.user.attributes["additionalHeaders"] = {"X-Foo": "bar"} > > self.user.save() > > > > proxy: ProxyProvider = ProxyProvider.objects.create( > > name=generate_id(), > > authorization_flow=Flow.objects.get( > > slug="default-provider-authorization-implicit-consent" > > ), > > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > > internal_host=f"http://{self.host}", > > external_host="http://localhost:9000", > > ) > > # Ensure OAuth2 Params are set > > proxy.set_oauth_defaults() > > proxy.save() > > # we need to create an application to actually access the proxy > > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > > outpost: Outpost = Outpost.objects.create( > > name=generate_id(), > > type=OutpostType.PROXY, > > ) > > outpost.providers.add(proxy) > > outpost.build_user_permissions(outpost.user) > > > > self.start_proxy(outpost) > > > > # Wait until outpost healthcheck succeeds > > healthcheck_retries = 0 > > while healthcheck_retries < 50: # noqa: PLR2004 > > if len(outpost.state) > 0: > > state = outpost.state[0] > > if state.last_seen: > > break > > healthcheck_retries += 1 > > sleep(0.5) > > sleep(5) > > > > self.driver.get("http://localhost:9000/api") > > > self.login() > > > > tests/e2e/test_provider_proxy.py:103: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > def login(self): > > """Do entire login flow and check user afterwards""" > > > flow_executor = self.get_shadow_root("ak-flow-executor") > > > > tests/e2e/utils.py:232: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > selector = 'ak-flow-executor' > > container = > > > > def get_shadow_root( > > self, selector: str, container: WebElement | WebDriver | None = None > > ) -> WebElement: > > """Get shadow root element's inner shadowRoot""" > > if not container: > > container = self.driver > > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > > > tests/e2e/utils.py:226: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > by = 'css selector', value = 'ak-flow-executor' > > > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > > """Find an element given a By strategy and locator. > > > > :Usage: > > :: > > > > element = driver.find_element(By.ID, 'foo') > > > > :rtype: WebElement > > """ > > by, value = self.locator_converter.convert(by, value) > > > > if isinstance(by, RelativeBy): > > elements = self.find_elements(by=by, value=value) > > if not elements: > > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > > return elements[0] > > > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > driver_command = 'findElement' > > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > > > def execute(self, driver_command: str, params: dict = None) -> dict: > > """Sends a command to be executed by a command.CommandExecutor. > > > > :Args: > > - driver_command: The name of the command to execute as a string. > > - params: A dictionary of named parameters to send with the command. > > > > :Returns: > > The command's JSON response loaded into a dictionary object. > > """ > > params = self._wrap_value(params) > > > > if self.session_id: > > if not params: > > params = {"sessionId": self.session_id} > > elif "sessionId" not in params: > > params["sessionId"] = self.session_id > > > > response = self.command_executor.execute(driver_command, params) > > if response: > > > self.error_handler.check_response(response) > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x5619a8017fc3 \\u003Cunknown>\\n#17 0x5619a8028944 \\u003Cunknown>\\n#18 0x7f70af72cac3 \\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 0x5619a8029793 > > E #1 0x5619a7d1d1c6 > > E #2 0x5619a7d68358 > > E #3 0x5619a7d68411 > > E #4 0x5619a7dab934 > > E #5 0x5619a7d8a3fd > > E #6 0x5619a7da8dd9 > > E #7 0x5619a7d8a173 > > E #8 0x5619a7d5b2d3 > > E #9 0x5619a7d5bc9e > > E #10 0x5619a7fed8cb > > E #11 0x5619a7ff1745 > > E #12 0x5619a7fda2e1 > > E #13 0x5619a7ff22d2 > > E #14 0x5619a7fbe17f > > E #15 0x5619a8017dc8 > > E #16 0x5619a8017fc3 > > E #17 0x5619a8028944 > > E #18 0x7f70af72cac3 > > > > ../../../.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 0x7fe03cb6af00>] > > args = ('group_outpost_3f5dbd5a-6375-4eba-9a05-4d9af58a4b16', {'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 0x7fe03cbdc7c0>] > > args = ('group_outpost_3f5dbd5a-6375-4eba-9a05-4d9af58a4b16', {'type': 'event.update'}) > > kwargs = {}, __traceback_hide__ = True > > current_task = result=None> > > result = None > > > > async def main_wrap( > > self, > > call_result: "Future[_R]", > > exc_info: "OptExcInfo", > > task_context: "Optional[List[asyncio.Task[Any]]]", > > context: List[contextvars.Context], > > *args: _P.args, > > **kwargs: _P.kwargs, > > ) -> None: > > """ > > Wraps the awaitable with something that puts the result into the > > result/exception future. > > """ > > > > __traceback_hide__ = True # noqa: F841 > > > > if context is not None: > > _restore_context(context[0]) > > > > current_task = asyncio.current_task() > > if current_task is not None and task_context is not None: > > task_context.append(current_task) > > > > try: > > # If we have an exception, run the function inside the except block > > # after raising it so exc_info is correctly populated. > > if exc_info[1]: > > try: > > > raise exc_info[1] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12......................../site-packages/asgiref/sync.py:327: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > > return func(self, *args, **kwargs) > > > > tests/e2e/utils.py:287: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'system/providers-proxy.yaml' > > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, config = > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > config = apps.get_app_config(app_name) > > if isinstance(config, ManagedAppConfig): > > config._on_startup_callback(None) > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:43: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > @retry() > > @apply_blueprint( > > "default/flow-default-authentication-flow.yaml", > > "default/flow-default-invalidation-flow.yaml", > > ) > > @apply_blueprint( > > "default/flow-default-provider-authorization-implicit-consent.yaml", > > "default/flow-default-provider-invalidation.yaml", > > ) > > @apply_blueprint( > > "system/providers-oauth2.yaml", > > "system/providers-proxy.yaml", > > ) > > @reconcile_app("authentik_crypto") > > def test_proxy_simple(self): > > """Test simple outpost setup with single provider""" > > # set additionalHeaders to test later > > self.user.attributes["additionalHeaders"] = {"X-Foo": "bar"} > > self.user.save() > > > > proxy: ProxyProvider = ProxyProvider.objects.create( > > name=generate_id(), > > authorization_flow=Flow.objects.get( > > slug="default-provider-authorization-implicit-consent" > > ), > > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > > internal_host=f"http://{self.host}", > > external_host="http://localhost:9000", > > ) > > # Ensure OAuth2 Params are set > > proxy.set_oauth_defaults() > > proxy.save() > > # we need to create an application to actually access the proxy > > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > > outpost: Outpost = Outpost.objects.create( > > name=generate_id(), > > type=OutpostType.PROXY, > > ) > > outpost.providers.add(proxy) > > outpost.build_user_permissions(outpost.user) > > > > self.start_proxy(outpost) > > > > # Wait until outpost healthcheck succeeds > > healthcheck_retries = 0 > > while healthcheck_retries < 50: # noqa: PLR2004 > > if len(outpost.state) > 0: > > state = outpost.state[0] > > if state.last_seen: > > break > > healthcheck_retries += 1 > > sleep(0.5) > > sleep(5) > > > > self.driver.get("http://localhost:9000/api") > > > self.login() > > > > tests/e2e/test_provider_proxy.py:103: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > def login(self): > > """Do entire login flow and check user afterwards""" > > > flow_executor = self.get_shadow_root("ak-flow-executor") > > > > tests/e2e/utils.py:232: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > selector = 'ak-flow-executor' > > container = > > > > def get_shadow_root( > > self, selector: str, container: WebElement | WebDriver | None = None > > ) -> WebElement: > > """Get shadow root element's inner shadowRoot""" > > if not container: > > container = self.driver > > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > > > tests/e2e/utils.py:226: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > by = 'css selector', value = 'ak-flow-executor' > > > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > > """Find an element given a By strategy and locator. > > > > :Usage: > > :: > > > > element = driver.find_element(By.ID, 'foo') > > > > :rtype: WebElement > > """ > > by, value = self.locator_converter.convert(by, value) > > > > if isinstance(by, RelativeBy): > > elements = self.find_elements(by=by, value=value) > > if not elements: > > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > > return elements[0] > > > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > driver_command = 'findElement' > > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > > > def execute(self, driver_command: str, params: dict = None) -> dict: > > """Sends a command to be executed by a command.CommandExecutor. > > > > :Args: > > - driver_command: The name of the command to execute as a string. > > - params: A dictionary of named parameters to send with the command. > > > > :Returns: > > The command's JSON response loaded into a dictionary object. > > """ > > params = self._wrap_value(params) > > > > if self.session_id: > > if not params: > > params = {"sessionId": self.session_id} > > elif "sessionId" not in params: > > params["sessionId"] = self.session_id > > > > response = self.command_executor.execute(driver_command, params) > > if response: > > > self.error_handler.check_response(response) > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x562d3521ffc3 \\u003Cunknown>\\n#17 0x562d35230944 \\u003Cunknown>\\n#18 0x7f110cf6aac3 \\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 0x562d35231793 > > E #1 0x562d34f251c6 > > E #2 0x562d34f70358 > > E #3 0x562d34f70411 > > E #4 0x562d34fb3934 > > E #5 0x562d34f923fd > > E #6 0x562d34fb0dd9 > > E #7 0x562d34f92173 > > E #8 0x562d34f632d3 > > E #9 0x562d34f63c9e > > E #10 0x562d351f58cb > > E #11 0x562d351f9745 > > E #12 0x562d351e22e1 > > E #13 0x562d351fa2d2 > > E #14 0x562d351c617f > > E #15 0x562d3521fdc8 > > E #16 0x562d3521ffc3 > > E #17 0x562d35230944 > > E #18 0x7f110cf6aac3 > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > > > During handling of the above exception, another exception occurred: > > > > self = > > test_case = > > subTest = False > > > > @contextlib.contextmanager > > def testPartExecutor(self, test_case, subTest=False): > > old_success = self.success > > self.success = True > > try: > > > yield > > > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:58: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > result = > > > > def run(self, result=None): > > if result is None: > > result = self.defaultTestResult() > > startTestRun = getattr(result, 'startTestRun', None) > > stopTestRun = getattr(result, 'stopTestRun', None) > > if startTestRun is not None: > > startTestRun() > > else: > > stopTestRun = None > > > > result.startTest(self) > > try: > > testMethod = getattr(self, self._testMethodName) > > if (getattr(self.__class__, "__unittest_skip__", False) or > > getattr(testMethod, "__unittest_skip__", False)): > > # If the class or method was skipped. > > skip_why = (getattr(self.__class__, '__unittest_skip_why__', '') > > or getattr(testMethod, '__unittest_skip_why__', '')) > > _addSkip(result, self, skip_why) > > return result > > > > expecting_failure = ( > > getattr(self, "__unittest_expecting_failure__", False) or > > getattr(testMethod, "__unittest_expecting_failure__", False) > > ) > > outcome = _Outcome(result) > > start_time = time.perf_counter() > > try: > > self._outcome = outcome > > > > with outcome.testPartExecutor(self): > > self._callSetUp() > > if outcome.success: > > outcome.expecting_failure = expecting_failure > > with outcome.testPartExecutor(self): > > > self._callTestMethod(testMethod) > > > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:634: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > method = > > > > > def _callTestMethod(self, method): > > > if method() is not None: > > > > .../hostedtoolcache/Python/3.12.7........./x64/lib/python3.12/unittest/case.py:589: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > return func(self, *args, **kwargs) > > > > except tuple(exceptions) as exc: > > count += 1 > > if count > max_retires: > > logger.debug("Exceeded retry count", exc=exc, test=self) > > > > raise exc > > logger.debug("Retrying on error", exc=exc, test=self) > > self.tearDown() > > self._post_teardown() > > self._pre_setup() > > self.setUp() > > > return wrapper(self, *args, **kwargs) > > > > tests/e2e/utils.py:300: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > return func(self, *args, **kwargs) > > > > except tuple(exceptions) as exc: > > count += 1 > > if count > max_retires: > > logger.debug("Exceeded retry count", exc=exc, test=self) > > > > raise exc > > logger.debug("Retrying on error", exc=exc, test=self) > > self.tearDown() > > self._post_teardown() > > self._pre_setup() > > self.setUp() > > > return wrapper(self, *args, **kwargs) > > > > tests/e2e/utils.py:300: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > return func(self, *args, **kwargs) > > > > except tuple(exceptions) as exc: > > count += 1 > > if count > max_retires: > > logger.debug("Exceeded retry count", exc=exc, test=self) > > > > > raise exc > > > > tests/e2e/utils.py:294: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > args = (), kwargs = {} > > > > @wraps(func) > > def wrapper(self: TransactionTestCase, *args, **kwargs): > > """Run test again if we're below max_retries, including tearDown and > > setUp. Otherwise raise the error""" > > nonlocal count > > try: > > > return func(self, *args, **kwargs) > > > > tests/e2e/utils.py:287: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Invalidation flow\nentries:\n- attrs:\n designation: invalidation\n na...0\n stage: !KeyOf default-invalidation-logout\n target: !KeyOf flow\n model: authentik_flows.flowstagebinding\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'default/flow-default-provider-invalidation.yaml' > > content = 'version: 1\nmetadata:\n name: Default - Provider invalidation flow\nentries:\n- attrs:\n designation: invalidatio...ation: none\n identifiers:\n slug: default-provider-invalidation-flow\n model: authentik_flows.flow\n id: flow\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, file = 'system/providers-proxy.yaml' > > content = 'version: 1\nmetadata:\n labels:\n blueprints.goauthentik.io/system: "true"\n name: System - Proxy Provider - Sco...ser.group_attributes(request),\n "is_superuser": request.user.is_superuser,\n }\n }\n' > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > for file in files: > > content = BlueprintInstance(path=file).retrieve() > > Importer.from_string(content).apply() > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:25: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > args = (,) > > kwargs = {}, config = > > > > @wraps(func) > > def wrapper(*args, **kwargs): > > config = apps.get_app_config(app_name) > > if isinstance(config, ManagedAppConfig): > > config._on_startup_callback(None) > > > return func(*args, **kwargs) > > > > .../blueprints/tests/__init__.py:43: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > @retry() > > @apply_blueprint( > > "default/flow-default-authentication-flow.yaml", > > "default/flow-default-invalidation-flow.yaml", > > ) > > @apply_blueprint( > > "default/flow-default-provider-authorization-implicit-consent.yaml", > > "default/flow-default-provider-invalidation.yaml", > > ) > > @apply_blueprint( > > "system/providers-oauth2.yaml", > > "system/providers-proxy.yaml", > > ) > > @reconcile_app("authentik_crypto") > > def test_proxy_simple(self): > > """Test simple outpost setup with single provider""" > > # set additionalHeaders to test later > > self.user.attributes["additionalHeaders"] = {"X-Foo": "bar"} > > self.user.save() > > > > proxy: ProxyProvider = ProxyProvider.objects.create( > > name=generate_id(), > > authorization_flow=Flow.objects.get( > > slug="default-provider-authorization-implicit-consent" > > ), > > invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"), > > internal_host=f"http://{self.host}", > > external_host="http://localhost:9000", > > ) > > # Ensure OAuth2 Params are set > > proxy.set_oauth_defaults() > > proxy.save() > > # we need to create an application to actually access the proxy > > Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy) > > outpost: Outpost = Outpost.objects.create( > > name=generate_id(), > > type=OutpostType.PROXY, > > ) > > outpost.providers.add(proxy) > > outpost.build_user_permissions(outpost.user) > > > > self.start_proxy(outpost) > > > > # Wait until outpost healthcheck succeeds > > healthcheck_retries = 0 > > while healthcheck_retries < 50: # noqa: PLR2004 > > if len(outpost.state) > 0: > > state = outpost.state[0] > > if state.last_seen: > > break > > healthcheck_retries += 1 > > sleep(0.5) > > sleep(5) > > > > self.driver.get("http://localhost:9000/api") > > > self.login() > > > > tests/e2e/test_provider_proxy.py:103: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > > > def login(self): > > """Do entire login flow and check user afterwards""" > > > flow_executor = self.get_shadow_root("ak-flow-executor") > > > > tests/e2e/utils.py:232: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > selector = 'ak-flow-executor' > > container = > > > > def get_shadow_root( > > self, selector: str, container: WebElement | WebDriver | None = None > > ) -> WebElement: > > """Get shadow root element's inner shadowRoot""" > > if not container: > > container = self.driver > > > shadow_root = container.find_element(By.CSS_SELECTOR, selector) > > > > tests/e2e/utils.py:226: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > by = 'css selector', value = 'ak-flow-executor' > > > > def find_element(self, by=By.ID, value: Optional[str] = None) -> WebElement: > > """Find an element given a By strategy and locator. > > > > :Usage: > > :: > > > > element = driver.find_element(By.ID, 'foo') > > > > :rtype: WebElement > > """ > > by, value = self.locator_converter.convert(by, value) > > > > if isinstance(by, RelativeBy): > > elements = self.find_elements(by=by, value=value) > > if not elements: > > raise NoSuchElementException(f"Cannot locate relative element with: {by.root}") > > return elements[0] > > > > > return self.execute(Command.FIND_ELEMENT, {"using": by, "value": value})["value"] > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:766: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > driver_command = 'findElement' > > params = {'using': 'css selector', 'value': 'ak-flow-executor'} > > > > def execute(self, driver_command: str, params: dict = None) -> dict: > > """Sends a command to be executed by a command.CommandExecutor. > > > > :Args: > > - driver_command: The name of the command to execute as a string. > > - params: A dictionary of named parameters to send with the command. > > > > :Returns: > > The command's JSON response loaded into a dictionary object. > > """ > > params = self._wrap_value(params) > > > > if self.session_id: > > if not params: > > params = {"sessionId": self.session_id} > > elif "sessionId" not in params: > > params["sessionId"] = self.session_id > > > > response = self.command_executor.execute(driver_command, params) > > if response: > > > self.error_handler.check_response(response) > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/webdriver.py:380: > > _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ > > > > self = > > response = {'status': 404, 'value': '{"value":{"error":"no such element","message":"no such element: Unable to locate element: {\...\\n#16 0x55be4b061fc3 \\u003Cunknown>\\n#17 0x55be4b072944 \\u003Cunknown>\\n#18 0x7f403a659ac3 \\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 0x55be4b073793 > > E #1 0x55be4ad671c6 > > E #2 0x55be4adb2358 > > E #3 0x55be4adb2411 > > E #4 0x55be4adf5934 > > E #5 0x55be4add43fd > > E #6 0x55be4adf2dd9 > > E #7 0x55be4add4173 > > E #8 0x55be4ada52d3 > > E #9 0x55be4ada5c9e > > E #10 0x55be4b0378cb > > E #11 0x55be4b03b745 > > E #12 0x55be4b0242e1 > > E #13 0x55be4b03c2d2 > > E #14 0x55be4b00817f > > E #15 0x55be4b061dc8 > > E #16 0x55be4b061fc3 > > E #17 0x55be4b072944 > > E #18 0x7f403a659ac3 > > > > ../../../.cache............/pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/remote/errorhandler.py:229: NoSuchElementException > > ``` > >
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 0x7fe03cb8bdc0>] > args = ('group_outpost_d76d828b-f647-4038-ba6c-36130f99f3e0', {'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 0x7fe03d6cc640>] > args = ('group_outpost_d76d828b-f647-4038-ba6c-36130f99f3e0', {'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 0x55f43116afc3 \\u003Cunknown>\\n#17 0x55f43117b944 \\u003Cunknown>\\n#18 0x7f09b342fac3 \\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 0x55f43117c793 > E #1 0x55f430e701c6 > E #2 0x55f430ebb358 > E #3 0x55f430ebb411 > E #4 0x55f430efe934 > E #5 0x55f430edd3fd > E #6 0x55f430efbdd9 > E #7 0x55f430edd173 > E #8 0x55f430eae2d3 > E #9 0x55f430eaec9e > E #10 0x55f4311408cb > E #11 0x55f431144745 > E #12 0x55f43112d2e1 > E #13 0x55f4311452d2 > E #14 0x55f43111117f > E #15 0x55f43116adc8 > E #16 0x55f43116afc3 > E #17 0x55f43117b944 > E #18 0x7f09b342fac3 > > ../../../.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 0x7fe03d519940>] > args = ('group_outpost_c1f83a19-0312-45f7-aa1d-ba3f998c6b73', {'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 0x7fe03d5ef080>] > args = ('group_outpost_c1f83a19-0312-45f7-aa1d-ba3f998c6b73', {'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 0x55c019bd6fc3 \\u003Cunknown>\\n#17 0x55c019be7944 \\u003Cunknown>\\n#18 0x7efe631c8ac3 \\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 0x55c019be8793 > E #1 0x55c0198dc1c6 > E #2 0x55c019927358 > E #3 0x55c019927411 > E #4 0x55c01996a934 > E #5 0x55c0199493fd > E #6 0x55c019967dd9 > E #7 0x55c019949173 > E #8 0x55c01991a2d3 > E #9 0x55c01991ac9e > E #10 0x55c019bac8cb > E #11 0x55c019bb0745 > E #12 0x55c019b992e1 > E #13 0x55c019bb12d2 > E #14 0x55c019b7d17f > E #15 0x55c019bd6dc8 > E #16 0x55c019bd6fc3 > E #17 0x55c019be7944 > E #18 0x7efe631c8ac3 > > ../../../.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 0x55e7ad4bafc3 \\u003Cunknown>\\n#17 0x55e7ad4cb944 \\u003Cunknown>\\n#18 0x7f3d0e595ac3 \\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 0x55e7ad4cc793 > E #1 0x55e7ad1c01c6 > E #2 0x55e7ad20b358 > E #3 0x55e7ad20b411 > E #4 0x55e7ad24e934 > E #5 0x55e7ad22d3fd > E #6 0x55e7ad24bdd9 > E #7 0x55e7ad22d173 > E #8 0x55e7ad1fe2d3 > E #9 0x55e7ad1fec9e > E #10 0x55e7ad4908cb > E #11 0x55e7ad494745 > E #12 0x55e7ad47d2e1 > E #13 0x55e7ad4952d2 > E #14 0x55e7ad46117f > E #15 0x55e7ad4badc8 > E #16 0x55e7ad4bafc3 > E #17 0x55e7ad4cb944 > E #18 0x7f3d0e595ac3 > > ../../../.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 | 220s run time > > ```python > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7fe03d7143c0>] > args = ('group_outpost_dec82e59-38f5-42b5-817d-119810dafcbc', {'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 0x7fe03d66d7c0>] > args = ('group_outpost_dec82e59-38f5-42b5-817d-119810dafcbc', {'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 0x557c89b21fc3 \\u003Cunknown>\\n#17 0x557c89b32944 \\u003Cunknown>\\n#18 0x7f9b26916ac3 \\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 0x557c89b33793 > E #1 0x557c898271c6 > E #2 0x557c89872358 > E #3 0x557c89872411 > E #4 0x557c898b5934 > E #5 0x557c898943fd > E #6 0x557c898b2dd9 > E #7 0x557c89894173 > E #8 0x557c898652d3 > E #9 0x557c89865c9e > E #10 0x557c89af78cb > E #11 0x557c89afb745 > E #12 0x557c89ae42e1 > E #13 0x557c89afc2d2 > E #14 0x557c89ac817f > E #15 0x557c89b21dc8 > E #16 0x557c89b21fc3 > E #17 0x557c89b32944 > E #18 0x7f9b26916ac3 > > ../../../.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 0x7fe03cb3fec0>] > args = ('group_outpost_2c8b4337-a2bc-4df2-a68b-2fbba49bc360', {'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 0x7fe03d658f40>] > args = ('group_outpost_2c8b4337-a2bc-4df2-a68b-2fbba49bc360', {'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 0x55f6455d0fc3 \\u003Cunknown>\\n#17 0x55f6455e1944 \\u003Cunknown>\\n#18 0x7f972f34eac3 \\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 0x55f6455e2793 > E #1 0x55f6452d61c6 > E #2 0x55f645321358 > E #3 0x55f645321411 > E #4 0x55f645364934 > E #5 0x55f6453433fd > E #6 0x55f645361dd9 > E #7 0x55f645343173 > E #8 0x55f6453142d3 > E #9 0x55f645314c9e > E #10 0x55f6455a68cb > E #11 0x55f6455aa745 > E #12 0x55f6455932e1 > E #13 0x55f6455ab2d2 > E #14 0x55f64557717f > E #15 0x55f6455d0dc8 > E #16 0x55f6455d0fc3 > E #17 0x55f6455e1944 > E #18 0x7f972f34eac3 > > ../../../.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 0x55dbe0009fc3 \\u003Cunknown>\\n#17 0x55dbe001a944 \\u003Cunknown>\\n#18 0x7f810d4ccac3 \\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 0x55dbe001b793 > E #1 0x55dbdfd0f1c6 > E #2 0x55dbdfd5a358 > E #3 0x55dbdfd5a411 > E #4 0x55dbdfd9d934 > E #5 0x55dbdfd7c3fd > E #6 0x55dbdfd9add9 > E #7 0x55dbdfd7c173 > E #8 0x55dbdfd4d2d3 > E #9 0x55dbdfd4dc9e > E #10 0x55dbdffdf8cb > E #11 0x55dbdffe3745 > E #12 0x55dbdffcc2e1 > E #13 0x55dbdffe42d2 > E #14 0x55dbdffb017f > E #15 0x55dbe0009dc8 > E #16 0x55dbe0009fc3 > E #17 0x55dbe001a944 > E #18 0x7f810d4ccac3 > > ../../../.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