Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -228,14 +228,38 @@ 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()
if not expected_platform
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,
Expand Down Expand Up @@ -891,6 +915,76 @@ 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.

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.
"""
log_path = get_appium_server_log_path()
if not log_path:
return subprocess.DEVNULL
try:
# buffering=1 (line-buffered) so the log is readable while Appium runs.
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 """
Expand Down Expand Up @@ -931,6 +1025,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
Expand All @@ -944,25 +1046,29 @@ 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:
appium_binary_path = os.path.normpath(appium_binary)
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(
Expand Down Expand Up @@ -1276,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
Expand Down
7 changes: 6 additions & 1 deletion Framework/Built_In_Automation/Mobile/iOS/iosOptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
23 changes: 23 additions & 0 deletions Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand All @@ -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:
Expand All @@ -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(
Expand Down
18 changes: 15 additions & 3 deletions Framework/Utilities/All_Device_Info.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")

Expand All @@ -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")
Expand Down
Loading
Loading