From 43e7fbbbe4961b9b879db2e2855f21a0d0aff35c Mon Sep 17 00:00:00 2001 From: mdshakib007 Date: Wed, 12 Aug 2026 17:34:55 +0600 Subject: [PATCH 1/4] fix the module related issue for linux & added take screenshot feature after executing a sleep --- .../Desktop/Linux/BuiltInFunctions.py | 106 ++++++++++++++++-- .../action_declarations/common.py | 2 +- Framework/Utilities/CommonUtil.py | 69 ++++++++++-- server/linux.py | 8 +- tests/test_auto_screen_capture.py | 72 ++++++++++++ 5 files changed, 231 insertions(+), 26 deletions(-) create mode 100644 tests/test_auto_screen_capture.py diff --git a/Framework/Built_In_Automation/Desktop/Linux/BuiltInFunctions.py b/Framework/Built_In_Automation/Desktop/Linux/BuiltInFunctions.py index 2dd4ef198..a4c69ac6c 100644 --- a/Framework/Built_In_Automation/Desktop/Linux/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Desktop/Linux/BuiltInFunctions.py @@ -11,37 +11,78 @@ from Framework.module_installer import install_missing_modules +# --------------------------------------------------------------------------- +# Optional system dependencies. +# +# pyatspi (AT-SPI2) is a GObject-Introspection binding shipped by the distro +# (python3-pyatspi + gir1.2-atspi-2.0) and it needs a running accessibility bus, +# so it is simply absent on headless servers, CI boxes and VMs. python-xlib is +# only used by the XComposite screenshot path, which already falls back to xwd. +# +# Importing this module therefore must never abort the process: CommonUtil +# imports it just to take a desktop screenshot, and screenshot capture +# (xdotool/xwd/XComposite based) does not need AT-SPI at all. A missing +# dependency is recorded here and reported by the AT-SPI actions themselves as +# an ordinary step failure. +# --------------------------------------------------------------------------- + try: import pyatspi from pyatspi.action import Action from pyatspi.editabletext import EditableText, Text -except ImportError: + + ATSPI_IMPORT_ERROR: Optional[str] = None +except Exception: install_missing_modules(["python3-pyatspi==1.19.0", "pygobject==3.50.1"]) try: import pyatspi from pyatspi.action import Action from pyatspi.editabletext import EditableText, Text - except ImportError: - sys.stderr.write( - "Error: system dependency is not installed. Install them by running Installer/setup_linux_inspector.sh.\n" - ) - sys.exit(1) + + ATSPI_IMPORT_ERROR = None + except Exception as atspi_error: + # pyatspi raises ImportError when the module is missing and ValueError + # ("Namespace Atspi not available") when the GI typelib is missing. + ATSPI_IMPORT_ERROR = f"{type(atspi_error).__name__}: {atspi_error}" + + # The Accessible protocol below annotates methods with these interfaces, + # and annotations in a class body are evaluated at class-creation time, + # so the names must exist even when the real interfaces do not. + Action = EditableText = Text = Any # type: ignore[misc,assignment] + + class _AtspiUnavailable: + """Stand-in for the `pyatspi` module when AT-SPI is not installed. + + Every attribute access raises, so an AT-SPI-dependent action fails as + a normal step failure with an actionable message (actions wrap their + body in try/except) instead of taking the node down. + """ + + def __getattr__(self, name: str): + raise RuntimeError(atspi_unavailable_message()) + + pyatspi = _AtspiUnavailable() # type: ignore[assignment] try: from Xlib import X, display as xlib_display from Xlib.ext import composite as xlib_composite # noqa: F401 registers ext methods from Xlib.error import XError -except ImportError: + + XLIB_IMPORT_ERROR: Optional[str] = None +except Exception: install_missing_modules(["python-xlib==0.33"]) try: from Xlib import X, display as xlib_display from Xlib.ext import composite as xlib_composite # noqa: F401 from Xlib.error import XError - except ImportError: - sys.stderr.write( - "Error: python-xlib is not installed. Install it by running Installer/setup_linux_inspector.sh.\n" - ) - sys.exit(1) + + XLIB_IMPORT_ERROR = None + except Exception as xlib_error: + XLIB_IMPORT_ERROR = f"{type(xlib_error).__name__}: {xlib_error}" + X = xlib_display = None # type: ignore[assignment] + + class XError(Exception): # type: ignore[no-redef] + """Placeholder so `except XError` stays valid without python-xlib.""" from Framework.Utilities import CommonUtil from Framework.Built_In_Automation.Shared_Resources import ( @@ -49,6 +90,33 @@ ) from Framework.Utilities.decorators import logger +_SETUP_HINT = "Install them by running Installer/setup_linux_inspector.sh." + + +def is_atspi_available() -> bool: + """True when the AT-SPI2 bindings imported successfully.""" + return ATSPI_IMPORT_ERROR is None + + +def atspi_unavailable_message() -> str: + """Actionable message for callers that need AT-SPI but cannot have it.""" + return ( + "Linux desktop automation requires AT-SPI2 (pyatspi), which is not available on " + f"this machine ({ATSPI_IMPORT_ERROR}). It needs the distro accessibility packages " + f"and a running accessibility bus, so it cannot work on a headless node. {_SETUP_HINT}" + ) + + +# Warn once, at import, rather than per action: the node keeps running, but +# whoever reads the console should know why Linux desktop actions will fail. +if ATSPI_IMPORT_ERROR: + sys.stderr.write(f"Warning: {atspi_unavailable_message()}\n") +if XLIB_IMPORT_ERROR: + sys.stderr.write( + f"Warning: python-xlib is not available ({XLIB_IMPORT_ERROR}); " + f"screenshots fall back to xwd. {_SETUP_HINT}\n" + ) + class Collection: ... @@ -328,6 +396,14 @@ def _capture_via_composite(file_path: str, winid: str) -> bool: Composite extension absent, deps missing, X errors) so the caller can fall back to a different strategy. """ + if xlib_display is None: + CommonUtil.ExecLog( + MODULE_NAME, + f"Composite capture skipped: python-xlib unavailable ({XLIB_IMPORT_ERROR})", + 4, + ) + return False + try: from PIL import Image except ImportError as e: @@ -1013,7 +1089,13 @@ def _get_frame_geometry_for_window(window_id: str) -> dict | None: Used to align captured screenshots with the AT-SPI accessibility tree, whose coordinates may differ from the X11 window's geometry by the window decoration (title bar) offset. + + Returns None when AT-SPI is unavailable, so the caller keeps the full + window pixmap instead of losing the screenshot entirely. """ + if not is_atspi_available(): + return None + geometry = _get_window_geometry(window_id) if not geometry: return None diff --git a/Framework/Built_In_Automation/Sequential_Actions/action_declarations/common.py b/Framework/Built_In_Automation/Sequential_Actions/action_declarations/common.py index a0bdae45d..7455d527d 100644 --- a/Framework/Built_In_Automation/Sequential_Actions/action_declarations/common.py +++ b/Framework/Built_In_Automation/Sequential_Actions/action_declarations/common.py @@ -1,6 +1,6 @@ declarations = ( { "name": "step result", "function": "step_result", "screenshot": "none" }, - { "name": "sleep", "function": "Sleep", "screenshot": "none" }, + { "name": "sleep", "function": "Sleep", "screenshot": "auto" }, { "name": "wait", "function": "Wait_For_Element", "screenshot": "none" }, { "name": "wait disable", "function": "Wait_For_Element", "screenshot": "none" }, { "name": "save text", "function": "Save_Text", "screenshot": "none" }, diff --git a/Framework/Utilities/CommonUtil.py b/Framework/Utilities/CommonUtil.py index 2c3c610b9..85a2a024e 100644 --- a/Framework/Utilities/CommonUtil.py +++ b/Framework/Utilities/CommonUtil.py @@ -821,6 +821,27 @@ def PhysicalAvailableMemory(): ) # Initialize global variables for TakeScreenShot() +AUTO_SCREEN_CAPTURE = "auto" + + +def _resolve_auto_screen_capture(shared_variables): + """Pick the capture type for an action that is not tied to a platform. + + Common actions (e.g. "sleep") are shared by every module, so their + declaration cannot name a platform up front. "auto" defers that choice to + here, where the drivers the test actually has open are visible. + + Only web is resolved today: a sleep is most often used to let a page settle, + and the whole point of the capture is to show the page state once the wait + is over. Everything else stays "none" so no other module changes behaviour. + Add a "mobile" branch here if the same is ever wanted for Appium runs. + """ + for driver_key in ("selenium_driver", "playwright_page"): + if shared_variables.get(driver_key) is not None: + return "web" + return "none" + + def set_screenshot_vars(shared_variables): """ Save screen capture type and selenium/appium driver objects as global variables, so TakeScreenShot() can access them """ # We can't import Shared Variables due to cyclic imports causing local runs to break, so this is the work around @@ -832,6 +853,8 @@ def set_screenshot_vars(shared_variables): try: if "screen_capture" in shared_variables: # Type of screenshot (desktop/mobile) screen_capture_type = shared_variables["screen_capture"] + if screen_capture_type == AUTO_SCREEN_CAPTURE: + screen_capture_type = _resolve_auto_screen_capture(shared_variables) if screen_capture_type == "mobile": # Appium driver object if "device_id" in shared_variables: device_id = shared_variables[ @@ -1053,6 +1076,40 @@ def _screenshot_path(image_folder, image_name, extension="png"): return os.path.join(image_folder, safe_name + "." + extension.lstrip(".")) +_linux_capture_screenshot = None # None = not resolved yet, False = resolution failed + + +def _get_linux_capture_screenshot(): + """Resolve the Linux desktop screenshot function once per process. + + The Linux module pulls in optional system dependencies (AT-SPI, python-xlib) + that are absent on headless VMs, and a module that fails to import is never + cached in sys.modules — so importing it per screenshot re-ran its dependency + install and re-raised per screenshot. Catch BaseException (it used to call + sys.exit(), and SystemExit slips straight past `except Exception`) and cache + the outcome, so a missing desktop stack costs one warning, not a traceback + on every capture. + """ + global _linux_capture_screenshot + + if _linux_capture_screenshot is None: + try: + from Framework.Built_In_Automation.Desktop.Linux.BuiltInFunctions import ( + capture_screenshot as linux_capture_screenshot, + ) + + _linux_capture_screenshot = linux_capture_screenshot + except BaseException as e: + ExecLog( + MODULE_NAME, + "Linux desktop screenshot support is unavailable: %s: %s" % (type(e).__name__, e), + 3, + ) + _linux_capture_screenshot = False + + return _linux_capture_screenshot or None + + async def Thread_ScreenShot(function_name, image_folder, Method, Driver, image_name, skip_delay=False): """Capture screen of mobile, desktop, Selenium, or Playwright.""" if performance_testing: return @@ -1084,21 +1141,15 @@ async def Thread_ScreenShot(function_name, image_folder, Method, Driver, image_n # Capture screenshot of desktop if Method == "desktop": if sys.platform in ("linux", "linux2"): - # Import Linux screenshot function for AT-SPI desktop automation - try: - if sys.platform in ("linux", "linux2"): - from Framework.Built_In_Automation.Desktop.Linux.BuiltInFunctions import capture_screenshot as linux_capture_screenshot - except Exception: - linux_capture_screenshot = None - if linux_capture_screenshot: - linux_capture_screenshot(ImageName) - else: + linux_capture_screenshot = _get_linux_capture_screenshot() + if linux_capture_screenshot is None: ExecLog( sModuleInfo, "Linux screenshot module not available", 3, ) return + linux_capture_screenshot(ImageName) elif sys.platform == "win32" or sys.platform == "darwin": bbox = _get_window_screenshot_bbox() image = ImageGrab_Mac_Win.grab(bbox) diff --git a/server/linux.py b/server/linux.py index 2cbeb99e2..8b6399ecd 100644 --- a/server/linux.py +++ b/server/linux.py @@ -49,8 +49,8 @@ def inspect(app_name: str | None = None, window_id: str | None = None): window of the app (an app may have multiple windows; see /linux/apps). """ from Framework.Built_In_Automation.Desktop.Linux import BuiltInFunctions - if BuiltInFunctions is None: - return InspectorResponse(status="error", error="Linux automation module not available") + if not BuiltInFunctions.is_atspi_available(): + return InspectorResponse(status="error", error=BuiltInFunctions.atspi_unavailable_message()) try: # Determine app name @@ -94,7 +94,7 @@ def inspect(app_name: str | None = None, window_id: str | None = None): def get_apps(): """Return available Linux applications visible to AT-SPI.""" from Framework.Built_In_Automation.Desktop.Linux import BuiltInFunctions - if BuiltInFunctions is None: + if not BuiltInFunctions.is_atspi_available(): return [] try: @@ -117,7 +117,7 @@ def get_apps(): async def upload_linux_ui_dump(): """Continuously upload Linux UI dump if changed.""" from Framework.Built_In_Automation.Desktop.Linux import BuiltInFunctions - if BuiltInFunctions is None: + if not BuiltInFunctions.is_atspi_available(): return prev_xml_hash = "" diff --git a/tests/test_auto_screen_capture.py b/tests/test_auto_screen_capture.py new file mode 100644 index 000000000..c7ad38f6a --- /dev/null +++ b/tests/test_auto_screen_capture.py @@ -0,0 +1,72 @@ +"""The common "sleep" action declares screenshot "auto", so the capture type is +chosen from whichever driver the test currently has open.""" + +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from Framework.Built_In_Automation.Sequential_Actions.action_declarations import common +from Framework.Utilities import CommonUtil + + +def _reset(monkeypatch): + monkeypatch.setattr(CommonUtil, "screen_capture_type", "none") + monkeypatch.setattr(CommonUtil, "screen_capture_driver", None) + + +def test_sleep_action_is_declared_auto(): + sleep_action = next(d for d in common.declarations if d["name"] == "sleep") + assert sleep_action["screenshot"] == "auto" + + +def test_auto_resolves_to_web_for_selenium(monkeypatch): + _reset(monkeypatch) + driver = object() + + CommonUtil.set_screenshot_vars({"screen_capture": "auto", "selenium_driver": driver}) + + assert CommonUtil.screen_capture_type == "web" + assert CommonUtil.screen_capture_driver is driver + + +def test_auto_resolves_to_web_for_playwright(monkeypatch): + _reset(monkeypatch) + page = object() + + CommonUtil.set_screenshot_vars({ + "screen_capture": "auto", + "active_web_driver_type": "playwright", + "playwright_page": page, + }) + + assert CommonUtil.screen_capture_type == "web" + assert CommonUtil.screen_capture_driver is page + + +def test_auto_resolves_to_none_without_a_web_driver(monkeypatch): + _reset(monkeypatch) + + CommonUtil.set_screenshot_vars({"screen_capture": "auto"}) + + assert CommonUtil.screen_capture_type == "none" + + +def test_auto_resolves_to_none_when_browser_was_torn_down(monkeypatch): + """A stale key left behind as None must not be mistaken for a live driver.""" + _reset(monkeypatch) + + CommonUtil.set_screenshot_vars({ + "screen_capture": "auto", + "selenium_driver": None, + "playwright_page": None, + }) + + assert CommonUtil.screen_capture_type == "none" + + +def test_explicit_types_are_untouched(monkeypatch): + for declared in ("none", "web", "mobile", "desktop"): + _reset(monkeypatch) + CommonUtil.set_screenshot_vars({"screen_capture": declared}) + assert CommonUtil.screen_capture_type == declared From d018448f3e2051c77ead8c806d73cd9af7d0c6e1 Mon Sep 17 00:00:00 2001 From: mdshakib007 Date: Wed, 12 Aug 2026 18:02:04 +0600 Subject: [PATCH 2/4] fix issue related to ui --- .../Desktop/Linux/BuiltInFunctions.py | 20 +++++++++++-------- Framework/Utilities/CommonUtil.py | 2 +- 2 files changed, 13 insertions(+), 9 deletions(-) diff --git a/Framework/Built_In_Automation/Desktop/Linux/BuiltInFunctions.py b/Framework/Built_In_Automation/Desktop/Linux/BuiltInFunctions.py index a4c69ac6c..3d9f972d8 100644 --- a/Framework/Built_In_Automation/Desktop/Linux/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Desktop/Linux/BuiltInFunctions.py @@ -14,9 +14,11 @@ # --------------------------------------------------------------------------- # Optional system dependencies. # -# pyatspi (AT-SPI2) is a GObject-Introspection binding shipped by the distro -# (python3-pyatspi + gir1.2-atspi-2.0) and it needs a running accessibility bus, -# so it is simply absent on headless servers, CI boxes and VMs. python-xlib is +# pyatspi (AT-SPI2) is a thin Python layer over the GObject-Introspection +# "Atspi" typelib, so the pip side (python3-pyatspi + pygobject) is only half of +# it: the system typelib (gir1.2-atspi-2.0 / at-spi2-core) and a running +# accessibility bus are required too. Having a display is not enough -- AT-SPI +# is just as absent on a VNC/Xvfb desktop as on a bare server. python-xlib is # only used by the XComposite screenshot path, which already falls back to xwd. # # Importing this module therefore must never abort the process: CommonUtil @@ -90,7 +92,7 @@ class XError(Exception): # type: ignore[no-redef] ) from Framework.Utilities.decorators import logger -_SETUP_HINT = "Install them by running Installer/setup_linux_inspector.sh." +_SETUP_HINT = "Installer/setup_linux_inspector.sh" def is_atspi_available() -> bool: @@ -101,9 +103,11 @@ def is_atspi_available() -> bool: def atspi_unavailable_message() -> str: """Actionable message for callers that need AT-SPI but cannot have it.""" return ( - "Linux desktop automation requires AT-SPI2 (pyatspi), which is not available on " - f"this machine ({ATSPI_IMPORT_ERROR}). It needs the distro accessibility packages " - f"and a running accessibility bus, so it cannot work on a headless node. {_SETUP_HINT}" + "Linux desktop automation requires AT-SPI2 (pyatspi), which is not importable on " + f"this machine ({ATSPI_IMPORT_ERROR}). Beyond the Python bindings it needs the system " + "AT-SPI2 typelib and a running accessibility bus -- on Debian/Ubuntu install " + "'at-spi2-core gir1.2-atspi-2.0' (or your distro's equivalent). " + f"See {_SETUP_HINT} for the X tools and accessibility settings." ) @@ -114,7 +118,7 @@ def atspi_unavailable_message() -> str: if XLIB_IMPORT_ERROR: sys.stderr.write( f"Warning: python-xlib is not available ({XLIB_IMPORT_ERROR}); " - f"screenshots fall back to xwd. {_SETUP_HINT}\n" + f"screenshots fall back to xwd. See {_SETUP_HINT}.\n" ) diff --git a/Framework/Utilities/CommonUtil.py b/Framework/Utilities/CommonUtil.py index 85a2a024e..f4c7890be 100644 --- a/Framework/Utilities/CommonUtil.py +++ b/Framework/Utilities/CommonUtil.py @@ -1083,7 +1083,7 @@ def _get_linux_capture_screenshot(): """Resolve the Linux desktop screenshot function once per process. The Linux module pulls in optional system dependencies (AT-SPI, python-xlib) - that are absent on headless VMs, and a module that fails to import is never + that are absent on most nodes, and a module that fails to import is never cached in sys.modules — so importing it per screenshot re-ran its dependency install and re-raised per screenshot. Catch BaseException (it used to call sys.exit(), and SystemExit slips straight past `except Exception`) and cache From d419fb53f2b530db4e85d5a78c8b804dd3faabbe Mon Sep 17 00:00:00 2001 From: mdshakib007 Date: Wed, 12 Aug 2026 18:56:41 +0600 Subject: [PATCH 3/4] now image capture have a max timeout, so no stuck on capture image --- Framework/Utilities/CommonUtil.py | 101 ++++++++++++++++-- tests/test_screenshot_capture_timeout.py | 129 +++++++++++++++++++++++ 2 files changed, 223 insertions(+), 7 deletions(-) create mode 100644 tests/test_screenshot_capture_timeout.py diff --git a/Framework/Utilities/CommonUtil.py b/Framework/Utilities/CommonUtil.py index f4c7890be..37efc460a 100644 --- a/Framework/Utilities/CommonUtil.py +++ b/Framework/Utilities/CommonUtil.py @@ -3,6 +3,7 @@ import selenium import sys +import asyncio import inspect import os, os.path, threading import ast @@ -1110,6 +1111,73 @@ def _get_linux_capture_screenshot(): return _linux_capture_screenshot or None +SCREENSHOT_CAPTURE_TIMEOUT_SECONDS = 60 +SCREENSHOT_CAPTURE_POLL_SECONDS = 0.05 + + +def _log_capture_timeout(sModuleInfo, function_name, Method): + ExecLog( + sModuleInfo, + "Screenshot for Action: %s Method: %s did not finish within %s seconds and was " + "abandoned -- the browser/device is not responding. Continuing the run." + % (function_name, Method, SCREENSHOT_CAPTURE_TIMEOUT_SECONDS), + 2, + ) + + +async def _capture_with_timeout(capture, sModuleInfo, function_name, Method): + """Run one screen capture bounded by a timeout, without blocking the loop. + + Selenium and Appium screenshots are synchronous HTTP calls with no read + timeout of their own -- `RemoteConnection._timeout` defaults to + `socket._GLOBAL_DEFAULT_TIMEOUT`, so urllib3 waits forever -- and + `TakeScreenShot` runs outside `_run_action_with_timeout`. A wedged browser + therefore blocked the event loop indefinitely, parking the run on the + "Capturing Screenshot" log line with no timeout able to recover it. + + The sync capture runs on a *daemon* thread and is simply abandoned on + timeout -- the same trade-off `_ActionTimeoutWorker` makes for a hung + action. It must be a daemon (and must not use the default executor, whose + threads `loop.shutdown_default_executor()` joins) so an abandoned capture + can never hold up node shutdown. Any exception is re-raised on this thread + so the existing handlers in `Thread_ScreenShot` still see it. + + `capture` is a plain callable (Selenium/Appium/desktop) or an awaitable + (Playwright, already async). Returns True when it finished in time. + """ + if inspect.isawaitable(capture): + try: + await asyncio.wait_for(capture, timeout=SCREENSHOT_CAPTURE_TIMEOUT_SECONDS) + return True + except (asyncio.TimeoutError, TimeoutError): + _log_capture_timeout(sModuleInfo, function_name, Method) + return False + + done = threading.Event() + failure = {} + + def runner(): + try: + capture() + except BaseException: + failure["exc_info"] = sys.exc_info() + finally: + done.set() + + threading.Thread(target=runner, name="zeuz_screenshot_capture", daemon=True).start() + + deadline = time.monotonic() + SCREENSHOT_CAPTURE_TIMEOUT_SECONDS + while not done.is_set(): + if time.monotonic() >= deadline: + _log_capture_timeout(sModuleInfo, function_name, Method) + return False + await asyncio.sleep(SCREENSHOT_CAPTURE_POLL_SECONDS) + + if "exc_info" in failure: + raise failure["exc_info"][1].with_traceback(failure["exc_info"][2]) + return True + + async def Thread_ScreenShot(function_name, image_folder, Method, Driver, image_name, skip_delay=False): """Capture screen of mobile, desktop, Selenium, or Playwright.""" if performance_testing: return @@ -1138,6 +1206,11 @@ async def Thread_ScreenShot(function_name, image_folder, Method, Driver, image_n if should_delay_before_capture and not skip_delay and not _wait_for_debug_screenshot_delay(sModuleInfo, function_name, Method): return + # Every capture below is bounded: a hung browser/device must not park the run. + def _desktop_grab(): + image = ImageGrab_Mac_Win.grab(_get_window_screenshot_bbox()) + image.save(ImageName, format="PNG") # Save to disk + # Capture screenshot of desktop if Method == "desktop": if sys.platform in ("linux", "linux2"): @@ -1149,23 +1222,37 @@ async def Thread_ScreenShot(function_name, image_folder, Method, Driver, image_n 3, ) return - linux_capture_screenshot(ImageName) + if not await _capture_with_timeout( + lambda: linux_capture_screenshot(ImageName), sModuleInfo, function_name, Method + ): + return elif sys.platform == "win32" or sys.platform == "darwin": - bbox = _get_window_screenshot_bbox() - image = ImageGrab_Mac_Win.grab(bbox) - image.save(ImageName, format="PNG") # Save to disk + if not await _capture_with_timeout(_desktop_grab, sModuleInfo, function_name, Method): + return # Capture screenshot of web browser elif Method == "web": # Check if it's a Playwright page or Selenium driver if is_playwright_page: - await Driver.screenshot(path=ImageName, type="jpeg", quality=PLAYWRIGHT_AUTO_SCREENSHOT_QUALITY) + captured = await _capture_with_timeout( + Driver.screenshot(path=ImageName, type="jpeg", quality=PLAYWRIGHT_AUTO_SCREENSHOT_QUALITY), + sModuleInfo, function_name, Method, + ) else: # Selenium driver - Driver.get_screenshot_as_file(ImageName) # Must be .png, otherwise an exception occurs + # Must be .png, otherwise an exception occurs + captured = await _capture_with_timeout( + lambda: Driver.get_screenshot_as_file(ImageName), sModuleInfo, function_name, Method + ) + if not captured: + return # Capture screenshot of mobile elif Method == "mobile": - Driver.save_screenshot(ImageName) # Must be .png, otherwise an exception occurs + # Must be .png, otherwise an exception occurs + if not await _capture_with_timeout( + lambda: Driver.save_screenshot(ImageName), sModuleInfo, function_name, Method + ): + return else: ExecLog( sModuleInfo, diff --git a/tests/test_screenshot_capture_timeout.py b/tests/test_screenshot_capture_timeout.py new file mode 100644 index 000000000..707792028 --- /dev/null +++ b/tests/test_screenshot_capture_timeout.py @@ -0,0 +1,129 @@ +"""Screen captures must be bounded. + +Selenium/Appium screenshots are synchronous HTTP calls with no read timeout, and +TakeScreenShot runs outside _run_action_with_timeout, so a wedged browser used to +park the whole run on the "Capturing Screenshot" line indefinitely. +""" + +import asyncio +import os +import sys +import threading +import time + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +import pytest + +from Framework.Utilities import CommonUtil + + +@pytest.fixture +def short_timeout(monkeypatch): + monkeypatch.setattr(CommonUtil, "SCREENSHOT_CAPTURE_TIMEOUT_SECONDS", 1) + + +@pytest.fixture +def logs(monkeypatch): + captured = [] + monkeypatch.setattr( + CommonUtil, + "ExecLog", + lambda module, message, level=1, *a, **k: captured.append((level, str(message))), + ) + return captured + + +def test_wedged_capture_times_out_instead_of_hanging(short_timeout, logs): + release = threading.Event() + + def wedged_browser(): + release.wait() # never released while we are timing + + async def scenario(): + start = time.perf_counter() + ok = await CommonUtil._capture_with_timeout( + wedged_browser, "mod", "Sleep", "web" + ) + return ok, time.perf_counter() - start + + try: + ok, elapsed = asyncio.run(scenario()) + finally: + release.set() + + assert ok is False + assert elapsed < 10, "capture was not bounded by the timeout" + assert any(lvl == 2 and "did not finish" in msg for lvl, msg in logs) + assert not any(lvl == 3 for lvl, _ in logs) + + +def test_event_loop_stays_responsive_while_a_capture_is_stuck(short_timeout, logs): + """The capture must not run on the event loop -- other tasks keep working.""" + release = threading.Event() + ticks = [] + + async def scenario(): + async def heartbeat(): + while True: + await asyncio.sleep(0.05) + ticks.append(1) + + hb = asyncio.create_task(heartbeat()) + await CommonUtil._capture_with_timeout(release.wait, "mod", "Sleep", "web") + hb.cancel() + + try: + asyncio.run(scenario()) + finally: + release.set() + + assert ticks, "event loop was blocked by the capture" + + +def test_successful_capture_returns_true(short_timeout, logs): + calls = [] + + async def scenario(): + return await CommonUtil._capture_with_timeout( + lambda: calls.append("captured"), "mod", "Go_To_Link", "web" + ) + + assert asyncio.run(scenario()) is True + assert calls == ["captured"] + assert not any("did not finish" in msg for _, msg in logs) + + +def test_capture_errors_still_propagate(short_timeout, logs): + """Thread_ScreenShot's WebDriverException/Exception handlers must still fire.""" + + def broken_driver(): + raise RuntimeError("browser went away") + + async def scenario(): + return await CommonUtil._capture_with_timeout( + broken_driver, "mod", "Sleep", "web" + ) + + with pytest.raises(RuntimeError, match="browser went away"): + asyncio.run(scenario()) + + +def test_awaitable_capture_is_bounded_too(short_timeout, logs): + """Playwright captures are coroutines, not callables.""" + + async def slow_playwright_screenshot(): + await asyncio.sleep(30) + + async def scenario(): + start = time.perf_counter() + ok = await CommonUtil._capture_with_timeout( + slow_playwright_screenshot(), "mod", "Sleep", "web" + ) + return ok, time.perf_counter() - start + + ok, elapsed = asyncio.run(scenario()) + + assert ok is False + assert elapsed < 10 + assert any(lvl == 2 and "did not finish" in msg for lvl, msg in logs) From 525a2f3112e9c01532736abf5fa86efd8537e3a1 Mon Sep 17 00:00:00 2001 From: mdshakib007 Date: Mon, 5 Oct 2026 11:44:23 +0600 Subject: [PATCH 4/4] Fix deploy loop crashing on network errors The deploy long-poll loop catches requests' timeout and connection errors to retry after 30 seconds, but the requests module was not imported. Any network error (server restart, network drop) raised NameError instead, so the loop crashed rather than reconnecting. Co-Authored-By: Claude Opus 5.5 --- Framework/deploy_handler/long_poll_handler.py | 1 + 1 file changed, 1 insertion(+) diff --git a/Framework/deploy_handler/long_poll_handler.py b/Framework/deploy_handler/long_poll_handler.py index 8f78267e8..f669aa3eb 100644 --- a/Framework/deploy_handler/long_poll_handler.py +++ b/Framework/deploy_handler/long_poll_handler.py @@ -6,6 +6,7 @@ import random import json import httpx +import requests from colorama import Fore from pathlib import Path from urllib.parse import urlparse