From 063156a2295c7007819ac8ab57ad5898ae911ea3 Mon Sep 17 00:00:00 2001 From: mdshakib007 Date: Wed, 22 Jul 2026 11:19:47 +0600 Subject: [PATCH 1/2] Add logging redirection for Appium server and suppress command not found errors in iOS and Android device info retrieval --- .../CrossPlatform/Appium/BuiltInFunctions.py | 41 +++++- .../Mobile/iOS/iosOptions.py | 7 +- .../Web/Selenium/BuiltInFunctions.py | 23 ++++ Framework/Utilities/All_Device_Info.py | 18 ++- .../android/emulator_manager.py | 124 +++++++++++++++++- 5 files changed, 203 insertions(+), 10 deletions(-) diff --git a/Framework/Built_In_Automation/Mobile/CrossPlatform/Appium/BuiltInFunctions.py b/Framework/Built_In_Automation/Mobile/CrossPlatform/Appium/BuiltInFunctions.py index 788992ff3..c4981f550 100755 --- a/Framework/Built_In_Automation/Mobile/CrossPlatform/Appium/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Mobile/CrossPlatform/Appium/BuiltInFunctions.py @@ -891,6 +891,25 @@ def is_port_in_use(port): return s.connect_ex(("localhost", port)) == 0 +def get_appium_server_log_target(): + """Return a target for the Appium server subprocess' stdout/stderr. + + Keeps the raw Appium/ADB/HTTP output out of the ZeuZ console (matching the + Windows behaviour of running Appium in its own window) while still writing it + to a log file for debugging. Falls back to DEVNULL if the file can't be + opened so a logging problem never blocks the server from starting. + """ + try: + log_dir = os.path.join( + os.path.abspath(__file__).split("Framework")[0], "AutomationLog" + ) + os.makedirs(log_dir, exist_ok=True) + # buffering=1 (line-buffered) so the log is readable while Appium runs. + return open(os.path.join(log_dir, "appium_server.log"), "a", buffering=1) + except Exception: + return subprocess.DEVNULL + + @logger def start_appium_server(): """ Starts the external Appium server """ @@ -931,6 +950,14 @@ def start_appium_server(): ) return "zeuz_failed" + # On non-Windows platforms subprocess.Popen inherits the parent console's + # stdout/stderr, which floods the ZeuZ console with raw Appium/ADB/HTTP + # logs. Windows avoids this by launching Appium in its own minimized cmd + # window. To match that behaviour, redirect the server output to a log + # file so the console stays clean while the logs remain available for + # debugging. + appium_log_output = get_appium_server_log_target() + try: appium_server = None if sys.platform == "win32": # We need to open appium in it's own command dos box on Windows @@ -944,12 +971,16 @@ def start_appium_server(): "%s -p %s" % (appium_binary, str(appium_port)), shell=True, + stdout=appium_log_output, + stderr=subprocess.STDOUT, ) elif sys.platform == "linux" or sys.platform == "linux2": appium_server = subprocess.Popen( "%s -p %s" % (appium_binary, str(appium_port)), shell=True, + stdout=appium_log_output, + stderr=subprocess.STDOUT, ) else: try: @@ -957,12 +988,12 @@ def start_appium_server(): appium_binary_path = os.path.abspath(os.path.join(appium_binary_path, os.pardir)) env = {"PATH": str(appium_binary_path)} appium_server = subprocess.Popen( - subprocess.Popen( - "%s -p %s" - % (appium_binary, str(appium_port)), - shell=True, - ), + "%s -p %s" + % (appium_binary, str(appium_port)), + shell=True, env=env, + stdout=appium_log_output, + stderr=subprocess.STDOUT, ) except: CommonUtil.ExecLog( diff --git a/Framework/Built_In_Automation/Mobile/iOS/iosOptions.py b/Framework/Built_In_Automation/Mobile/iOS/iosOptions.py index 4ed016f6f..ecc22ec19 100755 --- a/Framework/Built_In_Automation/Mobile/iOS/iosOptions.py +++ b/Framework/Built_In_Automation/Mobile/iOS/iosOptions.py @@ -21,7 +21,12 @@ def run_program(cmd): sModuleInfo = inspect.currentframe().f_code.co_name + " : " + MODULE_NAME try: output = subprocess.check_output( - imobiledevice_path + cmd, shell=True, encoding="utf-8" + imobiledevice_path + cmd, + shell=True, + encoding="utf-8", + # Suppress shell "command not found" noise (e.g. when the + # imobiledevice tools are not installed) from leaking to the console. + stderr=subprocess.DEVNULL, ) # Execute command line program, and return STDOUT return output except: # If command produced a non-zero return code, return failed diff --git a/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py b/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py index a8be0f999..1f9f2f913 100644 --- a/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py @@ -503,6 +503,23 @@ def is_port_in_use(port): ) return "zeuz_failed" + # On non-Windows platforms subprocess.Popen inherits the parent console's + # stdout/stderr, which floods the ZeuZ console with raw Appium/ADB/HTTP + # logs. Windows avoids this by launching Appium in its own minimized cmd + # window. To match that behaviour, redirect the server output to a log + # file so the console stays clean while the logs remain available for + # debugging. + try: + _log_dir = os.path.join( + os.path.abspath(__file__).split("Framework")[0], "AutomationLog" + ) + os.makedirs(_log_dir, exist_ok=True) + appium_log_output = open( + os.path.join(_log_dir, "appium_server.log"), "a", buffering=1 + ) + except Exception: + appium_log_output = subprocess.DEVNULL + try: appium_server = None if ( @@ -520,12 +537,16 @@ def is_port_in_use(port): "%s --allow-insecure chromedriver_autodownload -p %s" % (appium_binary, str(appium_port)), shell=True, + stdout=appium_log_output, + stderr=subprocess.STDOUT, ) elif sys.platform == "linux" or sys.platform == "linux2": appium_server = subprocess.Popen( "%s --allow-insecure chromedriver_autodownload -p %s" % (appium_binary, str(appium_port)), shell=True, + stdout=appium_log_output, + stderr=subprocess.STDOUT, ) else: try: @@ -539,6 +560,8 @@ def is_port_in_use(port): % (appium_binary, str(appium_port)), shell=True, env=env, + stdout=appium_log_output, + stderr=subprocess.STDOUT, ) except: CommonUtil.ExecLog( diff --git a/Framework/Utilities/All_Device_Info.py b/Framework/Utilities/All_Device_Info.py index 2708c6844..b03d1752f 100755 --- a/Framework/Utilities/All_Device_Info.py +++ b/Framework/Utilities/All_Device_Info.py @@ -164,7 +164,13 @@ def get_all_connected_ios_info(): # Get list of UUIDs ios_list = subprocess.check_output( - "idevice_id -l", shell=True, encoding="utf-8" + "idevice_id -l", + shell=True, + encoding="utf-8", + # Suppress shell "command not found" noise (e.g. when libimobiledevice + # is not installed, common on Android-only runs) from leaking to the + # console. + stderr=subprocess.DEVNULL, ) ios_list = ios_list.split("\n") @@ -175,13 +181,19 @@ def get_all_connected_ios_info(): info = "" try: info = subprocess.check_output( - "ideviceinfo -u %s" % uuid, shell=True, encoding="utf-8" + "ideviceinfo -u %s" % uuid, + shell=True, + encoding="utf-8", + stderr=subprocess.DEVNULL, ) except: pass if "ProductType" not in info: info = subprocess.check_output( - "ideviceinfo -s -u %s" % uuid, shell=True, encoding="utf-8" + "ideviceinfo -s -u %s" % uuid, + shell=True, + encoding="utf-8", + stderr=subprocess.DEVNULL, ) # Try simple mode which gets everything we need except IMEI # info = info.encode('ascii', 'ignore') # !!!!Needed, but not working info = info.split("\n") diff --git a/Framework/install_handler/android/emulator_manager.py b/Framework/install_handler/android/emulator_manager.py index c0b05be9b..61bf7fe2f 100644 --- a/Framework/install_handler/android/emulator_manager.py +++ b/Framework/install_handler/android/emulator_manager.py @@ -73,6 +73,128 @@ def _android_process_env() -> dict[str, str]: return env +def _candidate_sdk_roots() -> list[Path]: + """SDK roots to search for an AVD's system image, most-preferred first. + + ZeuZ's managed SDK is always tried first so that setups where the image lives + inside the managed SDK (the normal Windows/Linux case) keep their current + behaviour unchanged. Only when the image is missing there do we fall back to + an SDK provided by the user's environment (e.g. an Android Studio install). + """ + roots: list[Path] = [_get_sdk_root()] + for var in ("ANDROID_SDK_ROOT", "ANDROID_HOME"): + value = os.environ.get(var, "").strip() + if value: + roots.append(Path(value)) + system = platform.system() + if system == "Darwin": + roots.append(Path.home() / "Library" / "Android" / "sdk") + elif system == "Linux": + roots.append(Path.home() / "Android" / "Sdk") + elif system == "Windows": + local_app_data = os.environ.get("LOCALAPPDATA", "").strip() + if local_app_data: + roots.append(Path(local_app_data) / "Android" / "Sdk") + + unique: list[Path] = [] + seen: set[str] = set() + for root in roots: + key = str(root) + if key not in seen: + seen.add(key) + unique.append(root) + return unique + + +def _find_avd_config_dir(avd_name: str) -> Path | None: + """Locate an AVD's ``.avd`` directory (which holds ``config.ini``). + + AVDs live under an AVD home that is independent of the SDK root. We honour the + standard override variables and fall back to ``~/.android/avd``. The + authoritative pointer is ``.ini`` (it stores the real ``path=``), with a + direct ``.avd`` lookup as a fallback. + """ + avd_homes: list[Path] = [] + for var in ("ANDROID_AVD_HOME", "ANDROID_EMULATOR_HOME"): + value = os.environ.get(var, "").strip() + if value: + avd_homes.append(Path(value)) + for var in ("ANDROID_SDK_HOME", "ANDROID_PREFS_ROOT"): + value = os.environ.get(var, "").strip() + if value: + avd_homes.append(Path(value) / ".android" / "avd") + avd_homes.append(Path.home() / ".android" / "avd") + + for avd_home in avd_homes: + pointer = avd_home / f"{avd_name}.ini" + if pointer.is_file(): + try: + for line in pointer.read_text( + encoding="utf-8", errors="ignore" + ).splitlines(): + if line.strip().startswith("path="): + avd_dir = Path(line.split("=", 1)[1].strip()) + if avd_dir.is_dir(): + return avd_dir + except Exception: + pass # Fall through to the direct-directory fallback + direct = avd_home / f"{avd_name}.avd" + if direct.is_dir(): + return direct + return None + + +def _resolve_emulator_sdk_root(avd_name: str) -> Path: + """Return the SDK root whose ``system-images`` tree holds the AVD's image. + + The emulator resolves the relative ``image.sysdir.1`` from ``config.ini`` + against ANDROID_SDK_ROOT, so pointing it at an SDK that lacks the image makes + it fail with "Broken AVD system path". We pick the first candidate SDK root + that actually contains the image; ZeuZ's managed SDK is checked first, so + setups that already work are left untouched. Falls back to the managed SDK + when nothing matches (e.g. absolute image paths, or config we can't read). + """ + default_root = _get_sdk_root() + avd_dir = _find_avd_config_dir(avd_name) + if avd_dir is None: + return default_root + + system_image_subdir = "" + try: + for line in (avd_dir / "config.ini").read_text( + encoding="utf-8", errors="ignore" + ).splitlines(): + if line.strip().startswith("image.sysdir.1="): + system_image_subdir = line.split("=", 1)[1].strip() + break + except Exception: + return default_root + + if not system_image_subdir: + return default_root + + for root in _candidate_sdk_roots(): + if (root / system_image_subdir).is_dir(): + return root + return default_root + + +def _emulator_process_env(avd_name: str) -> dict[str, str]: + """Process env for launching ``avd_name``, pointed at an SDK that has its image.""" + env = _android_process_env() + sdk_root = _resolve_emulator_sdk_root(avd_name) + if str(sdk_root) != env.get("ANDROID_SDK_ROOT"): + logger.info( + "[emulator-runtime] AVD %s system image not found under the managed " + "SDK; using SDK root %s instead.", + avd_name, + sdk_root, + ) + env["ANDROID_HOME"] = str(sdk_root) + env["ANDROID_SDK_ROOT"] = str(sdk_root) + return env + + def get_emulator_path() -> Path: executable = "emulator.exe" if platform.system() == "Windows" else "emulator" return _get_sdk_root() / "emulator" / executable @@ -258,7 +380,7 @@ def _spawn_emulator( popen_options = { "stdout": None, "stderr": subprocess.STDOUT, - "env": _android_process_env(), + "env": _emulator_process_env(avd_name), } if os.name == "nt": popen_options["creationflags"] = getattr( From d4ff213b587cb917a0ae46347b996d66c0a7c27e Mon Sep 17 00:00:00 2001 From: mdshakib007 Date: Wed, 22 Jul 2026 12:19:49 +0600 Subject: [PATCH 2/2] Enhance Appium server logging and error handling: add log path retrieval, surface recent logs on failure, and improve device type detection logic. --- .../CrossPlatform/Appium/BuiltInFunctions.py | 91 +++++++++++++++++-- 1 file changed, 85 insertions(+), 6 deletions(-) diff --git a/Framework/Built_In_Automation/Mobile/CrossPlatform/Appium/BuiltInFunctions.py b/Framework/Built_In_Automation/Mobile/CrossPlatform/Appium/BuiltInFunctions.py index c4981f550..7c3662b68 100755 --- a/Framework/Built_In_Automation/Mobile/CrossPlatform/Appium/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Mobile/CrossPlatform/Appium/BuiltInFunctions.py @@ -228,7 +228,22 @@ def find_correct_device_on_first_run(serial_or_name, device_info): if Shared_Resources.Test_Shared_Variables("dependency") else {} ) - expected_platform = str(current_dependency.get("Mobile", "")).lower().strip() + + # The devices the server deployed for THIS run are authoritative about the + # target platform (e.g. an iOS test always carries type "IOS"). Only fall + # back to the node's "Mobile" dependency when the deploy payload has no + # usable type. Using the dependency as the primary signal is wrong because + # a node can advertise "Mobile": "Android" while still running an iOS test, + deployed_types = { + str(info.get("type", "")).lower().strip() + for info in (device_info.values() if isinstance(device_info, dict) else []) + if isinstance(info, dict) and str(info.get("type", "")).strip() + } + if len(deployed_types) == 1: + expected_platform = next(iter(deployed_types)) + else: + expected_platform = str(current_dependency.get("Mobile", "")).lower().strip() + candidates = { name: info for name, info in all_device_info.items() @@ -236,6 +251,15 @@ def find_correct_device_on_first_run(serial_or_name, device_info): or str(info.get("type", "")).lower().strip() == expected_platform } + # Local detection may not surface the deployed device yet (common for iOS + # simulators that Appium boots on demand). Trust the server's device_info + if not candidates and isinstance(device_info, dict): + candidates = { + name: info + for name, info in device_info.items() + if isinstance(info, dict) and str(info.get("id", "")).strip() + } + if not candidates: CommonUtil.ExecLog( sModuleInfo, @@ -891,6 +915,18 @@ def is_port_in_use(port): return s.connect_ex(("localhost", port)) == 0 +def get_appium_server_log_path(): + """Absolute path of the file the Appium server output is redirected to.""" + try: + log_dir = os.path.join( + os.path.abspath(__file__).split("Framework")[0], "AutomationLog" + ) + os.makedirs(log_dir, exist_ok=True) + return os.path.join(log_dir, "appium_server.log") + except Exception: + return "" + + def get_appium_server_log_target(): """Return a target for the Appium server subprocess' stdout/stderr. @@ -899,17 +935,56 @@ def get_appium_server_log_target(): to a log file for debugging. Falls back to DEVNULL if the file can't be opened so a logging problem never blocks the server from starting. """ + log_path = get_appium_server_log_path() + if not log_path: + return subprocess.DEVNULL try: - log_dir = os.path.join( - os.path.abspath(__file__).split("Framework")[0], "AutomationLog" - ) - os.makedirs(log_dir, exist_ok=True) # buffering=1 (line-buffered) so the log is readable while Appium runs. - return open(os.path.join(log_dir, "appium_server.log"), "a", buffering=1) + return open(log_path, "a", buffering=1) except Exception: return subprocess.DEVNULL +# Strips ANSI colour/CSI escape sequences that Appium writes to its log. The +# character class is deliberately broad (Appium sometimes emits malformed codes +# such as "\x1b[38;5;-175m") so every sequence is cleaned, not just well-formed ones. +_ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;:?!<>=\-]*[a-zA-Z]") + + +def surface_appium_error_log(context="", max_lines=60): + """Echo the tail of the Appium server log to the ZeuZ console after a failure. + + Appium's output is normally redirected to a file so the console stays clean. + But when something actually goes wrong (e.g. an xcodebuild/WebDriverAgent build + failure), the real cause lives ONLY in that file. On failure we print the most + recent log lines so the user can see WHY it failed without digging through the + log manually. Best-effort: never raises. + """ + sModuleInfo = inspect.currentframe().f_code.co_name + " : " + MODULE_NAME + try: + from collections import deque + + log_path = get_appium_server_log_path() + if not log_path or not os.path.isfile(log_path): + return + with open(log_path, "r", encoding="utf-8", errors="ignore") as handle: + # deque keeps only the last max_lines in memory even for a big log. + recent = deque(handle, maxlen=max_lines) + cleaned = [_ANSI_ESCAPE_RE.sub("", line).rstrip() for line in recent] + cleaned = [line for line in cleaned if line.strip()] + if not cleaned: + return + CommonUtil.ExecLog( + sModuleInfo, + "Appium failed%s. Most recent Appium server log below " + "(full log: %s):" % (" - " + context if context else "", log_path), + 3, + ) + CommonUtil.ExecLog(sModuleInfo, "\n".join(cleaned), 3) + except Exception: + pass # Surfacing the error must never itself break the run + + @logger def start_appium_server(): """ Starts the external Appium server """ @@ -1307,6 +1382,10 @@ def start_appium_driver( appium_driver = None Shared_Resources.Set_Shared_Variables("appium_driver", appium_driver) CommonUtil.ExecLog(sModuleInfo, "Error during Appium setup", 3) + # The retries above are suppressed to keep the console clean; on + # final failure surface the real cause from the Appium log so the + # user sees WHY it failed (e.g. an xcodebuild/WebDriverAgent error). + surface_appium_error_log("could not create the Appium driver after %d attempts" % max_attempts) return "zeuz_failed", launch_app except Exception: return CommonUtil.Exception_Handler(sys.exc_info()), launch_app