From 0da2609e21e9d0ba4a63d1f0e6a5d7bb7858087b Mon Sep 17 00:00:00 2001 From: Nacy Dev Date: Sat, 14 Nov 2026 10:52:25 +0700 Subject: [PATCH] fix(webdriver): handle HTTP error responses and alert text in MobileErrorHandler --- appium/webdriver/errorhandler.py | 154 +++++++++++--- test/unit/webdriver/errorhandler_test.py | 245 +++++++++++++++++++++++ 2 files changed, 371 insertions(+), 28 deletions(-) create mode 100644 test/unit/webdriver/errorhandler_test.py diff --git a/appium/webdriver/errorhandler.py b/appium/webdriver/errorhandler.py index d074b01a..2f3b6485 100644 --- a/appium/webdriver/errorhandler.py +++ b/appium/webdriver/errorhandler.py @@ -81,46 +81,144 @@ def format_stacktrace(original: None | str | Sequence) -> list[str]: return result +JSONWP_ERROR_CODE_MAP: dict[int, str] = { + 7: 'no such element', + 8: 'no such frame', + 9: 'unknown command', + 10: 'stale element reference', + 11: 'element not visible', + 12: 'invalid element state', + 13: 'unknown error', + 15: 'element not selectable', + 17: 'javascript error', + 19: 'invalid selector', + 21: 'timeout', + 23: 'no such window', + 24: 'invalid cookie domain', + 25: 'unable to set cookie', + 26: 'unexpected alert open', + 27: 'no such alert', + 28: 'script timeout', + 29: 'invalid coordinates', + 32: 'invalid selector', + 33: 'session not created', + 34: 'move target out of bounds', + 51: 'invalid selector', + 52: 'invalid selector', + 60: 'element not interactable', + 61: 'invalid argument', + 62: 'no such cookie', + 63: 'unable to capture screen', + 64: 'element click intercepted', + 65: 'detached shadow root', + 405: 'unsupported operation', +} + + +def _parse_payload_dict(payload: Any) -> dict[str, Any] | None: + """Parse an error payload into a dict if it contains a JSON object.""" + if isinstance(payload, dict): + return payload + if isinstance(payload, str): + try: + loaded = json.loads(payload) + except (json.JSONDecodeError, TypeError): + return None + return loaded if isinstance(loaded, dict) else None + return None + + +def _extract_alert_text(value: dict[str, Any]) -> str | None: + """Extract the alert prompt string from W3C `data.text` or legacy fields.""" + data = value.get('data') + if isinstance(data, dict): + text = data.get('text') + return text if isinstance(text, str) else None + if isinstance(data, str): + return data + alert = value.get('alert') + if isinstance(alert, dict): + text = alert.get('text') + return text if isinstance(text, str) else None + return None + + +def _resolve_exception_class(error: Any, message: str | None) -> type[sel_exceptions.WebDriverException]: + """Resolve a W3C error string, JSONWP numeric status, or Appium message to an exception class.""" + error_key = error + if isinstance(error, int) and not isinstance(error, bool): + error_key = JSONWP_ERROR_CODE_MAP.get(error) + exception_class: type[sel_exceptions.WebDriverException] = ( + ERROR_TO_EXC_MAPPING.get(error_key, sel_exceptions.WebDriverException) + if isinstance(error_key, str) + else sel_exceptions.WebDriverException + ) + if exception_class is sel_exceptions.WebDriverException and message: + if message == 'No such context found.': + return appium_exceptions.NoSuchContextException + if message == 'That command could not be executed in the current context.': + return appium_exceptions.InvalidSwitchToTargetException + return exception_class + + class MobileErrorHandler(errorhandler.ErrorHandler): def check_response(self, response: dict[str, Any]) -> None: """ https://www.w3.org/TR/webdriver/#errors """ + status = response.get('status') + if status == errorhandler.ErrorCode.SUCCESS and not isinstance(status, bool): + return + + has_error_status = bool(status) payload = response.get('value', '') - if isinstance(payload, dict): - payload_dict = payload - else: - try: - payload_dict = json.loads(payload) - except (json.JSONDecodeError, TypeError): - return - if not isinstance(payload_dict, dict): + payload_dict = _parse_payload_dict(payload) + fallback_msg = ( + payload if isinstance(payload, str) and payload else response.get('message') or f'HTTP error status: {status}' + ) + + error_dict: dict[str, Any] = {} + error_identifier: Any = status + if payload_dict is None: + if not has_error_status: return - value = payload_dict.get('value') - if not isinstance(value, dict): - return - error = value.get('error') - if not error: - return + message = fallback_msg + else: + nested_value = payload_dict.get('value') + if isinstance(nested_value, dict): + error_dict = nested_value + error = error_dict.get('error') + if not error and not has_error_status: + return + error_identifier = error or payload_dict.get('status') or status + message = error_dict.get('message') or payload_dict.get('message') or error or fallback_msg + else: + # Flat response dicts without an error status can be valid HTTP 200 command results. + if not has_error_status: + return + error_dict = payload_dict + error_identifier = payload_dict.get('error') or payload_dict.get('status') or status + nested_str = nested_value if isinstance(nested_value, str) and nested_value else None + message = payload_dict.get('message') or nested_str or payload_dict.get('error') or fallback_msg - message = value.get('message', error) - stacktrace = value.get('stacktrace', '') - # In theory, we should also be checking HTTP status codes. - # Java client, for example, prints a warning if the actual `error` - # value does not match to the response's HTTP status code. - exception_class: type[sel_exceptions.WebDriverException] = ERROR_TO_EXC_MAPPING.get( - error, sel_exceptions.WebDriverException + payload_map = payload_dict or {} + stacktrace = ( + error_dict.get('stacktrace') + or error_dict.get('stackTrace') + or payload_map.get('stacktrace') + or payload_map.get('stackTrace') + or response.get('stacktrace') + or response.get('stackTrace') + or '' ) - if exception_class is sel_exceptions.WebDriverException and message: - if message == 'No such context found.': - exception_class = appium_exceptions.NoSuchContextException - elif message == 'That command could not be executed in the current context.': - exception_class = appium_exceptions.InvalidSwitchToTargetException + screen = error_dict.get('screen') or payload_map.get('screen') or response.get('screen') + exception_class = _resolve_exception_class(error_identifier, message if isinstance(message, str) else None) if exception_class is sel_exceptions.UnexpectedAlertPresentException: raise sel_exceptions.UnexpectedAlertPresentException( msg=message, + screen=screen, stacktrace=format_stacktrace(stacktrace), - alert_text=value.get('data'), + alert_text=_extract_alert_text(error_dict), ) - raise exception_class(msg=message, stacktrace=format_stacktrace(stacktrace)) + raise exception_class(msg=message, screen=screen, stacktrace=format_stacktrace(stacktrace)) diff --git a/test/unit/webdriver/errorhandler_test.py b/test/unit/webdriver/errorhandler_test.py new file mode 100644 index 00000000..43b68185 --- /dev/null +++ b/test/unit/webdriver/errorhandler_test.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python + +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import httpretty +import pytest +from selenium.common.exceptions import ( + InvalidArgumentException, + InvalidSessionIdException, + NoSuchElementException, + StaleElementReferenceException, + UnexpectedAlertPresentException, + UnknownMethodException, + WebDriverException, +) +from selenium.webdriver.common.by import By + +from appium.common.exceptions import InvalidSwitchToTargetException, NoSuchContextException +from appium.webdriver.errorhandler import MobileErrorHandler, format_stacktrace +from test.unit.helper.test_helper import android_w3c_driver, appium_command + + +class TestMobileErrorHandler: + def test_w3c_error_response(self) -> None: + handler = MobileErrorHandler() + response = { + 'status': 404, + 'value': ( + '{"value": {"error": "no such element", "message": "Unable to locate element", "stacktrace": "line1\\nline2"}}' + ), + } + with pytest.raises(NoSuchElementException) as exc_info: + handler.check_response(response) + + assert exc_info.value.msg is not None + assert exc_info.value.msg.startswith('Unable to locate element') + assert exc_info.value.stacktrace == ['line1', 'line2'] + + def test_w3c_error_response_without_http_status(self) -> None: + handler = MobileErrorHandler() + response = { + 'value': '{"value": {"error": "stale element reference", "message": "Element is stale"}}', + } + with pytest.raises(StaleElementReferenceException) as exc_info: + handler.check_response(response) + + assert exc_info.value.msg is not None + assert exc_info.value.msg.startswith('Element is stale') + + def test_unexpected_alert_extracts_w3c_data_text(self) -> None: + handler = MobileErrorHandler() + response = { + 'status': 500, + 'value': ( + '{"value": {"error": "unexpected alert open", ' + '"message": "Alert is blocking command", ' + '"stacktrace": "", "data": {"text": "Allow notifications?"}}}' + ), + } + with pytest.raises(UnexpectedAlertPresentException) as exc_info: + handler.check_response(response) + + assert exc_info.value.msg is not None + assert exc_info.value.msg.startswith('Alert is blocking command') + assert exc_info.value.alert_text == 'Allow notifications?' + assert 'Alert Text: Allow notifications?' in str(exc_info.value) + + def test_unexpected_alert_extracts_legacy_alert_or_string_data(self) -> None: + handler = MobileErrorHandler() + alert_response = { + 'status': 500, + 'value': '{"value": {"error": "unexpected alert open", "message": "Blocked", "alert": {"text": "Confirm?"}}}', + } + with pytest.raises(UnexpectedAlertPresentException) as alert_exc: + handler.check_response(alert_response) + assert alert_exc.value.alert_text == 'Confirm?' + + str_response = { + 'status': 500, + 'value': '{"value": {"error": "unexpected alert open", "message": "Blocked", "data": "Legacy alert"}}', + } + with pytest.raises(UnexpectedAlertPresentException) as str_exc: + handler.check_response(str_response) + assert str_exc.value.alert_text == 'Legacy alert' + + def test_context_specific_exceptions(self) -> None: + handler = MobileErrorHandler() + + with pytest.raises(NoSuchContextException) as missing_ctx: + handler.check_response( + {'status': 400, 'value': '{"value": {"error": "unknown error", "message": "No such context found."}}'} + ) + assert missing_ctx.value.msg == 'No such context found.' + + with pytest.raises(InvalidSwitchToTargetException) as invalid_ctx: + handler.check_response( + { + 'status': 400, + 'value': ( + '{"value": {"error": "unknown error", ' + '"message": "That command could not be executed in the current context."}}' + ), + } + ) + assert invalid_ctx.value.msg == 'That command could not be executed in the current context.' + + @pytest.mark.parametrize( + ('response', 'expected_exc', 'expected_msg'), + [ + ({'status': 401, 'value': 'Authorization Required'}, WebDriverException, 'Authorization Required'), + ({'status': 502, 'value': '502 Bad Gateway'}, WebDriverException, '502 Bad Gateway'), + ({'status': 405, 'value': 'Method Not Allowed'}, UnknownMethodException, 'Method Not Allowed'), + ({'status': 500, 'value': ''}, WebDriverException, 'HTTP error status: 500'), + ({'status': 500, 'value': '["unexpected"]'}, WebDriverException, '["unexpected"]'), + ( + {'status': 404, 'value': '{"error": "invalid session id", "message": "Session expired"}'}, + InvalidSessionIdException, + 'Session expired', + ), + ( + {'status': 400, 'value': {'error': 'invalid argument', 'message': 'Invalid timeout value'}}, + InvalidArgumentException, + 'Invalid timeout value', + ), + ( + {'status': 500, 'value': '{"value": "Plain error string"}'}, + WebDriverException, + 'Plain error string', + ), + ( + {'status': 500, 'value': '{}'}, + WebDriverException, + '{}', + ), + ], + ) + def test_non_w3c_http_error_responses_raise( + self, response: dict, expected_exc: type[WebDriverException], expected_msg: str + ) -> None: + handler = MobileErrorHandler() + with pytest.raises(expected_exc) as exc_info: + handler.check_response(response) + assert exc_info.value.msg is not None + assert exc_info.value.msg.startswith(expected_msg) + + def test_jsonwp_error_with_camelcase_stacktrace(self) -> None: + handler = MobileErrorHandler() + response = { + 'status': 500, + 'value': ( + '{"status": 7, "value": {"message": "Element missing", "screen": "base64img", ' + '"stackTrace": [{"className": "AppDriver", "methodName": "find", ' + '"fileName": "AppDriver.java", "lineNumber": 42}]}}' + ), + } + with pytest.raises(NoSuchElementException) as exc_info: + handler.check_response(response) + + assert exc_info.value.msg is not None + assert exc_info.value.msg.startswith('Element missing') + assert exc_info.value.screen == 'base64img' + assert exc_info.value.stacktrace == [' at AppDriver.find (AppDriver.java:42)'] + + def test_non_error_responses_do_not_raise(self) -> None: + handler = MobileErrorHandler() + handler.check_response({'status': 0, 'value': 'iVBORw0KGgoAAAANSUhEUg=='}) + handler.check_response({'value': None}) + handler.check_response({'value': 'plain result'}) + handler.check_response({'value': {'value': {'arbitrary': 'data'}}}) + handler.check_response({'value': {'error': 'user script key', 'message': 'not a webdriver error'}}) + + def test_format_stacktrace(self) -> None: + assert format_stacktrace(None) == [] + assert format_stacktrace('line1\nline2') == ['line1', 'line2'] + assert format_stacktrace( + [ + {'className': 'Foo', 'methodName': 'bar', 'fileName': 'Foo.java', 'lineNumber': 12}, + 'invalid-frame', # type: ignore[list-item] + {'methodName': 'baz'}, + ] + ) == [' at Foo.bar (Foo.java:12)', ' at baz ()'] + + @httpretty.activate + def test_driver_raises_on_http_proxy_and_auth_errors(self) -> None: + driver = android_w3c_driver() + + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/element'), + status=401, + body='Unauthorized', + ) + with pytest.raises(WebDriverException) as auth_exc: + driver.find_element(By.ID, 'btn') + assert auth_exc.value.msg == 'Authorization Required' + + httpretty.register_uri( + httpretty.GET, + appium_command('/session/1234567890/appium/settings'), + status=502, + body='502 Bad Gateway', + ) + with pytest.raises(WebDriverException) as gateway_exc: + driver.get_settings() + assert gateway_exc.value.msg == '502 Bad Gateway' + + @httpretty.activate + def test_driver_raises_unexpected_alert_with_text(self) -> None: + driver = android_w3c_driver() + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/element'), + status=500, + body=( + '{"value": {"error": "unexpected alert open", ' + '"message": "unexpected alert open", ' + '"stacktrace": "", "data": {"text": "Allow location?"}}}' + ), + ) + with pytest.raises(UnexpectedAlertPresentException) as exc_info: + driver.find_element(By.ID, 'btn') + assert exc_info.value.alert_text == 'Allow location?' + + @httpretty.activate + def test_driver_allows_http_200_user_dict_with_error_key(self) -> None: + driver = android_w3c_driver() + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/execute/sync'), + status=200, + content_type='application/json', + body='{"value": {"error": "timeout", "message": "app payload"}}', + ) + result = driver.execute_script('return arguments[0]', {'ok': True}) + assert result == {'error': 'timeout', 'message': 'app payload'}