website/docs: prepare release notes

[BeryJu]

Details

REPLACE ME

---

Checklist

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

If an API change has been made

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

If changes to the frontend have been made

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

If applicable

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

[netlify[bot]]

✅ Deploy Preview for authentik-storybook canceled.

Name	Link
🔨 Latest commit	8d0753fd713aaf7117dc9341111ea6ac7e0a12fb
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-storybook/deploys/673f3c95a65d2f0008195464

[netlify[bot]]

✅ Deploy Preview for authentik-docs ready!

Name	Link
🔨 Latest commit	8d0753fd713aaf7117dc9341111ea6ac7e0a12fb
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-docs/deploys/673f3c95d8c51a000899d36d
😎 Deploy Preview	https://deploy-preview-12119--authentik-docs.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

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

[codecov[bot]]

:x: 6 Tests Failed:

Tests completed	Failed	Passed	Skipped
1601	6	1595	1

View the top 3 failed tests by shortest run time

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

Stack Traces | 217s run time

> > > > > ```python > > self = > > call_result = > > exc_info = (, NoSuchElementException(), ) > > task_context = None, context = [<_contextvars.Context object at 0x7f770bf1e7c0>] > > args = ('group_outpost_b7c3965d-7033-4c9a-8975-468db35d2f18', {'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 0x7f770ba0ecc0>] > > args = ('group_outpost_b7c3965d-7033-4c9a-8975-468db35d2f18', {'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 0x555f190ecfc3 \\u003Cunknown>\\n#17 0x555f190fd944 \\u003Cunknown>\\n#18 0x7f89416faac3 \\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 0x555f190fe793 > > E #1 0x555f18df21c6 > > E #2 0x555f18e3d358 > > E #3 0x555f18e3d411 > > E #4 0x555f18e80934 > > E #5 0x555f18e5f3fd > > E #6 0x555f18e7ddd9 > > E #7 0x555f18e5f173 > > E #8 0x555f18e302d3 > > E #9 0x555f18e30c9e > > E #10 0x555f190c28cb > > E #11 0x555f190c6745 > > E #12 0x555f190af2e1 > > E #13 0x555f190c72d2 > > E #14 0x555f1909317f > > E #15 0x555f190ecdc8 > > E #16 0x555f190ecfc3 > > E #17 0x555f190fd944 > > E #18 0x7f89416faac3 > > > > ../../../.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 0x7f770bd14580>] > > args = ('group_outpost_2f019cdc-8807-4b58-bef7-34c78010d0f7', {'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 0x7f770bd06780>] > > args = ('group_outpost_2f019cdc-8807-4b58-bef7-34c78010d0f7', {'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 0x563caf35afc3 \\u003Cunknown>\\n#17 0x563caf36b944 \\u003Cunknown>\\n#18 0x7fadd05c2ac3 \\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 0x563caf36c793 > > E #1 0x563caf0601c6 > > E #2 0x563caf0ab358 > > E #3 0x563caf0ab411 > > E #4 0x563caf0ee934 > > E #5 0x563caf0cd3fd > > E #6 0x563caf0ebdd9 > > E #7 0x563caf0cd173 > > E #8 0x563caf09e2d3 > > E #9 0x563caf09ec9e > > E #10 0x563caf3308cb > > E #11 0x563caf334745 > > E #12 0x563caf31d2e1 > > E #13 0x563caf3352d2 > > E #14 0x563caf30117f > > E #15 0x563caf35adc8 > > E #16 0x563caf35afc3 > > E #17 0x563caf36b944 > > E #18 0x7fadd05c2ac3 > > > > ../../../.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 0x5593e546ffc3 \\u003Cunknown>\\n#17 0x5593e5480944 \\u003Cunknown>\\n#18 0x7f7aeb6e0ac3 \\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 0x5593e5481793 > > E #1 0x5593e51751c6 > > E #2 0x5593e51c0358 > > E #3 0x5593e51c0411 > > E #4 0x5593e5203934 > > E #5 0x5593e51e23fd > > E #6 0x5593e5200dd9 > > E #7 0x5593e51e2173 > > E #8 0x5593e51b32d3 > > E #9 0x5593e51b3c9e > > E #10 0x5593e54458cb > > E #11 0x5593e5449745 > > E #12 0x5593e54322e1 > > E #13 0x5593e544a2d2 > > E #14 0x5593e541617f > > E #15 0x5593e546fdc8 > > E #16 0x5593e546ffc3 > > E #17 0x5593e5480944 > > E #18 0x7f7aeb6e0ac3 > > > > ../../../.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 | 217s run time

> > ```python > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7f770bc881c0>] > args = ('group_outpost_395277eb-0f5d-47f1-ab21-2719bb9f5ca7', {'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 0x7f770bc9f8c0>] > args = ('group_outpost_395277eb-0f5d-47f1-ab21-2719bb9f5ca7', {'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 0x55fe01f5bfc3 \\u003Cunknown>\\n#17 0x55fe01f6c944 \\u003Cunknown>\\n#18 0x7f206941dac3 \\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 0x55fe01f6d793 > E #1 0x55fe01c611c6 > E #2 0x55fe01cac358 > E #3 0x55fe01cac411 > E #4 0x55fe01cef934 > E #5 0x55fe01cce3fd > E #6 0x55fe01cecdd9 > E #7 0x55fe01cce173 > E #8 0x55fe01c9f2d3 > E #9 0x55fe01c9fc9e > E #10 0x55fe01f318cb > E #11 0x55fe01f35745 > E #12 0x55fe01f1e2e1 > E #13 0x55fe01f362d2 > E #14 0x55fe01f0217f > E #15 0x55fe01f5bdc8 > E #16 0x55fe01f5bfc3 > E #17 0x55fe01f6c944 > E #18 0x7f206941dac3 > > ../../../.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 0x7f770811a480>] > args = ('group_outpost_769f4047-4751-4272-b388-0a0acde33f49', {'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 0x7f770bb5ad80>] > args = ('group_outpost_769f4047-4751-4272-b388-0a0acde33f49', {'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 0x55c809836fc3 \\u003Cunknown>\\n#17 0x55c809847944 \\u003Cunknown>\\n#18 0x7f10c8765ac3 \\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 0x55c809848793 > E #1 0x55c80953c1c6 > E #2 0x55c809587358 > E #3 0x55c809587411 > E #4 0x55c8095ca934 > E #5 0x55c8095a93fd > E #6 0x55c8095c7dd9 > E #7 0x55c8095a9173 > E #8 0x55c80957a2d3 > E #9 0x55c80957ac9e > E #10 0x55c80980c8cb > E #11 0x55c809810745 > E #12 0x55c8097f92e1 > E #13 0x55c8098112d2 > E #14 0x55c8097dd17f > E #15 0x55c809836dc8 > E #16 0x55c809836fc3 > E #17 0x55c809847944 > E #18 0x7f10c8765ac3 > > ../../../.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 0x55d32bf2cfc3 \\u003Cunknown>\\n#17 0x55d32bf3d944 \\u003Cunknown>\\n#18 0x7fd2b6227ac3 \\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 0x55d32bf3e793 > E #1 0x55d32bc321c6 > E #2 0x55d32bc7d358 > E #3 0x55d32bc7d411 > E #4 0x55d32bcc0934 > E #5 0x55d32bc9f3fd > E #6 0x55d32bcbddd9 > E #7 0x55d32bc9f173 > E #8 0x55d32bc702d3 > E #9 0x55d32bc70c9e > E #10 0x55d32bf028cb > E #11 0x55d32bf06745 > E #12 0x55d32beef2e1 > E #13 0x55d32bf072d2 > E #14 0x55d32bed317f > E #15 0x55d32bf2cdc8 > E #16 0x55d32bf2cfc3 > E #17 0x55d32bf3d944 > E #18 0x7fd2b6227ac3 > > ../../../.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 | 221s run time

> > ```python > self = > call_result = > exc_info = (, NoSuchElementException(), ) > task_context = None, context = [<_contextvars.Context object at 0x7f770bb40480>] > args = ('group_outpost_82074463-17a6-4aeb-a7bf-9ceb7681e2cc', {'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 0x7f770bcf1340>] > args = ('group_outpost_82074463-17a6-4aeb-a7bf-9ceb7681e2cc', {'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 0x55c7fcfdbfc3 \\u003Cunknown>\\n#17 0x55c7fcfec944 \\u003Cunknown>\\n#18 0x7f92f9b90ac3 \\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 0x55c7fcfed793 > E #1 0x55c7fcce11c6 > E #2 0x55c7fcd2c358 > E #3 0x55c7fcd2c411 > E #4 0x55c7fcd6f934 > E #5 0x55c7fcd4e3fd > E #6 0x55c7fcd6cdd9 > E #7 0x55c7fcd4e173 > E #8 0x55c7fcd1f2d3 > E #9 0x55c7fcd1fc9e > E #10 0x55c7fcfb18cb > E #11 0x55c7fcfb5745 > E #12 0x55c7fcf9e2e1 > E #13 0x55c7fcfb62d2 > E #14 0x55c7fcf8217f > E #15 0x55c7fcfdbdc8 > E #16 0x55c7fcfdbfc3 > E #17 0x55c7fcfec944 > E #18 0x7f92f9b90ac3 > > ../../../.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 0x7f770bda7e80>] > args = ('group_outpost_3f9c264f-dddc-436c-8f6a-d52d661e61c4', {'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 0x7f770b928dc0>] > args = ('group_outpost_3f9c264f-dddc-436c-8f6a-d52d661e61c4', {'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 0x557006329fc3 \\u003Cunknown>\\n#17 0x55700633a944 \\u003Cunknown>\\n#18 0x7fa8e1bceac3 \\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 0x55700633b793 > E #1 0x55700602f1c6 > E #2 0x55700607a358 > E #3 0x55700607a411 > E #4 0x5570060bd934 > E #5 0x55700609c3fd > E #6 0x5570060badd9 > E #7 0x55700609c173 > E #8 0x55700606d2d3 > E #9 0x55700606dc9e > E #10 0x5570062ff8cb > E #11 0x557006303745 > E #12 0x5570062ec2e1 > E #13 0x5570063042d2 > E #14 0x5570062d017f > E #15 0x557006329dc8 > E #16 0x557006329fc3 > E #17 0x55700633a944 > E #18 0x7fa8e1bceac3 > > ../../../.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 0x560ae0393fc3 \\u003Cunknown>\\n#17 0x560ae03a4944 \\u003Cunknown>\\n#18 0x7f7ae386eac3 \\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 0x560ae03a5793 > E #1 0x560ae00991c6 > E #2 0x560ae00e4358 > E #3 0x560ae00e4411 > E #4 0x560ae0127934 > E #5 0x560ae01063fd > E #6 0x560ae0124dd9 > E #7 0x560ae0106173 > E #8 0x560ae00d72d3 > E #9 0x560ae00d7c9e > E #10 0x560ae03698cb > E #11 0x560ae036d745 > E #12 0x560ae03562e1 > E #13 0x560ae036e2d2 > E #14 0x560ae033a17f > E #15 0x560ae0393dc8 > E #16 0x560ae0393fc3 > E #17 0x560ae03a4944 > E #18 0x7f7ae386eac3 > > ../../../.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